tatara-export-worker 0.2.641

//! `tatara-export-worker` — runs one declared `ExportSpec` from an
//! ephemeral Process to completion.
//!
//! Inputs (CLI + env):
//!   * `--spec-json <path>` OR `--spec '<json>'` — the [`ExportSpec`]
//!     to run, serialized as JSON. The reconciler mounts this via a
//!     ConfigMap or stamps it inline as a Job arg.
//!   * `--process-namespace` / `--process-name` — owning Process,
//!     for run-id resolution + receipt `process_ref` stamping.
//!   * `--previous-root <hex>` — chain into the Process attestation
//!     tree. Optional; new chains start with no previous root.
//!   * `--receipt-configmap <name>` — where the worker writes its
//!     typed [`ReceiptEnvelope`]. Required.
//!   * `--receipt-key <name>` — key inside the ConfigMap (default
//!     `receipt.yaml`). The reconciler's `JobAttested` evaluator
//!     reads from `<receipt-configmap>.data.<receipt-key>`.
//!
//! The binary is intentionally thin — the pure decision logic lives
//! in the library (`lib.rs`) and is tested without infrastructure.
//! Main is argv parsing + kube/HTTP/NATS plumbing.

use anyhow::{anyhow, Context, Result};
use clap::Parser;
use kube::Client;
use std::collections::BTreeMap;
use std::path::PathBuf;
use tracing::{info, warn};

use tatara_export_worker::{
    compose_export_receipt, prepare_event_payload, resolve_run_id, resolve_subject, ExportEvent,
    ExportOutcome,
};
use tatara_process::export::{ArtifactVariant, ChannelVariant, ExportSpec};
use tatara_process::prelude::{Annotated, ErrCtxExt};
use tatara_process::string_map::BTreeMapStrExt;

#[derive(Parser, Debug)]
#[command(
    name = "tatara-export-worker",
    about = "Run one declared ExportSpec from an ephemeral Process"
)]
struct Cli {
    /// Path to a JSON file containing the ExportSpec.
    #[arg(long, conflicts_with = "spec")]
    spec_json: Option<PathBuf>,

    /// Inline JSON ExportSpec.
    #[arg(long, conflicts_with = "spec_json")]
    spec: Option<String>,

    /// Owning Process namespace (kube downward API).
    #[arg(long, env = "TATARA_PROCESS_NAMESPACE")]
    process_namespace: String,

    /// Owning Process name.
    #[arg(long, env = "TATARA_PROCESS_NAME")]
    process_name: String,

    /// Optional previous BLAKE3 root to chain this receipt into.
    #[arg(long, env = "TATARA_PREVIOUS_ROOT")]
    previous_root: Option<String>,

    /// ConfigMap to write the receipt envelope to.
    #[arg(long, env = "TATARA_RECEIPT_CONFIGMAP")]
    receipt_configmap: String,

    /// Key inside the ConfigMap for the receipt YAML payload.
    #[arg(long, env = "TATARA_RECEIPT_KEY", default_value = "receipt.yaml")]
    receipt_key: String,
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    let cli = Cli::parse();

    let spec_json = match (&cli.spec_json, &cli.spec) {
        (Some(p), None) => std::fs::read_to_string(p).context("read --spec-json")?,
        (None, Some(s)) => s.clone(),
        _ => return Err(anyhow!("exactly one of --spec-json or --spec is required")),
    };
    let spec: ExportSpec = serde_json::from_str(&spec_json).context("parse ExportSpec JSON")?;

    let run_id = resolve_run_id(&spec, &cli.process_namespace, &cli.process_name);
    let process_ref =
        tatara_process::prelude::qualified_process_ref(&cli.process_namespace, &cli.process_name);
    info!(
        run_id = %run_id,
        process_ref = %process_ref,
        "tatara-export-worker starting"
    );

    let kube = Client::try_default().await.context("kube client")?;

    // 1. Read the artifact bytes the source points to.
    let artifact_bytes = read_artifact(&spec, &kube, &cli.process_namespace, &cli.process_name)
        .await
        .context("read artifact")?;

    // 2. Build the event payload via the pure lib function.
    let signal_type = extract_signal_type(&spec);
    let event = prepare_event_payload(
        spec.source.variant().err_ctx("source")?,
        &artifact_bytes,
        &run_id,
        &signal_type,
        chrono::Utc::now(),
    );
    let event_bytes = serde_json::to_vec(&event).context("serialize event")?;

    // 3. Ship through the chosen channel.
    let outcome = ship(&spec, &event, &event_bytes, &run_id).await;
    match &outcome {
        ExportOutcome::Shipped => info!(bytes = event_bytes.len(), "shipped"),
        ExportOutcome::Rejected(m) => warn!(reason = %m, "rejected"),
        ExportOutcome::Failed(m) => warn!(reason = %m, "failed"),
    }

    // 4. Compose typed receipt + persist to the ConfigMap.
    let receipt = compose_export_receipt(
        &spec,
        &event_bytes,
        &outcome,
        cli.previous_root.as_deref(),
        &run_id,
        Some(&process_ref),
    )
    .context("compose receipt")?;
    let receipt_yaml = serde_yaml::to_string(&receipt).context("serialize receipt")?;

    write_receipt(
        &kube,
        &cli.process_namespace,
        &cli.receipt_configmap,
        &cli.receipt_key,
        &receipt_yaml,
    )
    .await
    .context("write receipt ConfigMap")?;

    info!(
        composed_root = %receipt.composed_root,
        configmap = %cli.receipt_configmap,
        "receipt persisted"
    );

    // 5. Exit non-zero on terminal failure so the Job phase reflects
    //    it; tatara-reconciler routes Failed Jobs to Releasing →
    //    Zombie. The receipt is already persisted either way.
    if !outcome.is_shipped() {
        std::process::exit(1);
    }
    Ok(())
}

/// Pull the signal_type tag from whichever channel variant is set.
/// Used both for `ExportEvent.signal_type` and for HTTP request
/// headers.
fn extract_signal_type(spec: &ExportSpec) -> String {
    if let Some(h) = &spec.channel.http_event {
        return h.signal_type.clone();
    }
    if let Some(n) = &spec.channel.nats_subject {
        // NATS doesn't carry a separate signal_type; the subject's
        // last segment is the convention.
        return n.subject.rsplit('.').next().unwrap_or("event").to_string();
    }
    "event".into()
}

// ─── Artifact readers ──────────────────────────────────────────────

async fn read_artifact(spec: &ExportSpec, kube: &Client, ns: &str, name: &str) -> Result<Vec<u8>> {
    let v = spec.source.variant().err_ctx("source")?;
    match v {
        ArtifactVariant::RunMarker(_) => Ok(Vec::new()),
        ArtifactVariant::TestReport(tr) => {
            let cm_ns = tr.namespace.as_deref().unwrap_or(ns);
            // Ns-scoped `Api<ConfigMap>` binding rides the substrate
            // primitive `tatara_process::configmap::namespaced` — pre-
            // lift this was a hand-authored 1-link `let api:
            // Api<ConfigMap> = Api::namespaced(kube.clone(), cm_ns)`
            // chain, one of FOUR workspace-wide restatements past the
            // ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold (peer sites:
            // the Receipts-arm walker below, this crate's own SSA-side
            // `write_receipt`, and `tatara-closed-loop-probe::main::
            // write_receipt`). Post-lift the ns-scoped ConfigMap
            // handle binding lives at ONE substrate owner and the
            // concrete `K = ConfigMap` type is fixed at the primitive.
            let api = tatara_process::configmap::namespaced(kube.clone(), cm_ns);
            let cm = api
                .get(&tr.configmap)
                .await
                .with_context(|| format!("get configmap {cm_ns}/{}", tr.configmap))?;
            if let Some(s) = cm.data.as_ref().and_then(|d| d.get(&tr.key)) {
                return Ok(s.as_bytes().to_vec());
            }
            if let Some(b) = cm.binary_data.as_ref().and_then(|d| d.get(&tr.key)) {
                return Ok(b.0.clone());
            }
            Err(anyhow!(
                "ConfigMap {cm_ns}/{} has no key {:?}",
                tr.configmap,
                tr.key
            ))
        }
        ArtifactVariant::ProcessSnapshot(_) => {
            // Read the owning Process as a typed CR; serialize to JSON.
            // Ns-scoped `Api<Process>` binding rides the substrate
            // primitive `tatara_process::process_api::namespaced` —
            // peer to the `tatara_process::configmap::namespaced` +
            // ProcessSnapshot arms below/above, closing the same
            // `Api::namespaced(<client>.clone(), <ns>)` shape that
            // was hand-authored at THIS site + `tatara-reconciler::
            // boundary::{evaluate_process_phase, check_depends_on}`
            // pre-lift.
            let api = tatara_process::process_api::namespaced(kube.clone(), ns);
            // Diagnostic-body head rides the substrate composer
            // `tatara_process::process_api::error_ctx` — pre-lift this
            // was a hand-authored `format!("get process {ns}/{name}")`
            // chain, one of TWO workspace-wide restatements past the
            // ★★ PRIME-DIRECTIVE ≥ 2 duplication threshold (peer at
            // `tatara-reconciler::boundary::evaluate_process_phase`'s
            // `.get_opt` fetch wrap). Post-lift the fixed `"Process"`
            // resource-kind literal + the `<ns>/<name>` qualified-ref
            // routing sit at ONE substrate owner (sibling to
            // `configmap::error_ctx` on the per-Kind × substrate-owned-
            // error-slug axis-family), closing a workspace-wide wire-
            // form drift: the pre-lift lowercase `"process"` here
            // clashed with the sibling `list::error_ctx`'s TitleCase
            // `"Processes"` plural spelling, and post-lift both
            // singular and plural axes agree on the kube-canonical
            // TitleCase kind form.
            let p = api
                .get(name)
                .await
                .with_context(|| tatara_process::process_api::error_ctx("get", ns, name))?;
            Ok(serde_json::to_vec(&p).context("serialize process")?)
        }
        ArtifactVariant::Receipts(_) => {
            // Receipts collection — list ConfigMaps in the Process's
            // namespace carrying our process annotation, parse each
            // as a YAML ReceiptEnvelope, return the JSON array.
            // Ns-scoped `Api<ConfigMap>` binding rides the substrate
            // primitive `tatara_process::configmap::namespaced` — see
            // the TestReport-arm comment above for the FOUR-site
            // lift narrative.
            let api = tatara_process::configmap::namespaced(kube.clone(), ns);
            let cms = api
                .list(&Default::default())
                .await
                .context("list configmaps")?;
            let want = tatara_process::prelude::qualified_process_ref(ns, name);
            let mut envelopes = Vec::new();
            for cm in cms.items {
                // Annotation lookup rides through the ONE substrate
                // primitive `Annotated::annotation` (blanket impl over
                // every `kube::Resource<DynamicType = ()>`) with the
                // KEY slot routed through the substrate-owner const
                // [`tatara_process::annotations::PROCESS`] — pre-lift
                // this was a hand-authored 3-line
                // `.metadata.annotations.as_ref().and_then(|m|
                // m.get("tatara.pleme.io/process")).map(|p| p == &want)
                // .unwrap_or(false)` chain on a K8s built-in
                // `ConfigMap`, the fourth workspace surface spelling
                // the annotation-lookup axis alongside the three
                // tatara-reconciler / pool-reconciler consumers already
                // routed through the peer `Process::annotation`
                // inherent (`signals::ingest`,
                // `phase_machine::released_from_annotation`,
                // `controller_pool::process_belongs_to_pool`). Post-
                // lift the four surfaces share ONE substrate owner for
                // the trait routing AND ONE substrate owner for the
                // KEY byte-shape (`annotations::PROCESS`); pre-lift
                // this READ-side callsite was the LAST workspace
                // surface still passing the bare `"tatara.pleme.io/
                // process"` literal to the trait method — every peer
                // WRITE-side (`ssapply::m.insert_str(annotations::
                // PROCESS, …)`, `render.rs::json!({ annotations::
                // PROCESS: … })`) and READ-side (`edges.rs`,
                // `render.rs::labels[annotations::PROCESS]`,
                // `phase_machine.rs`) already routed. A future rename
                // of the key (a `tatara.pleme.io/v2/process` migration,
                // a per-fleet override, a collapse into a compound
                // `tatara.pleme.io/owner` key) lands at ONE `pub const`
                // in the substrate and every downstream consumer
                // (READ + WRITE alike) picks up the shift mechanically
                // without touching this callsite.
                if cm.annotation(tatara_process::annotations::PROCESS) != Some(want.as_str()) {
                    continue;
                }
                if let Some(d) = &cm.data {
                    for (_k, v) in d {
                        if let Ok(env) = tatara_process::receipt::ReceiptEnvelope::parse_either(v) {
                            envelopes.push(env);
                        }
                    }
                }
            }
            Ok(serde_json::to_vec(&envelopes)?)
        }
    }
}

// ─── Channel shippers ──────────────────────────────────────────────

async fn ship(
    spec: &ExportSpec,
    event: &ExportEvent,
    event_bytes: &[u8],
    run_id: &str,
) -> ExportOutcome {
    let variant = match spec.channel.variant() {
        Ok(v) => v,
        Err(e) => return ExportOutcome::Failed(format!("channel: {e}")),
    };
    match variant {
        ChannelVariant::HttpEvent(h) => ship_http(h, event_bytes).await,
        ChannelVariant::NatsSubject(n) => ship_nats(n, event_bytes, run_id).await,
        ChannelVariant::Stdout(s) => ship_stdout(s, event),
    }
}

async fn ship_http(
    channel: &tatara_process::export::HttpEventChannel,
    event_bytes: &[u8],
) -> ExportOutcome {
    let client = match reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(15))
        .build()
    {
        Ok(c) => c,
        Err(e) => return ExportOutcome::Failed(format!("build client: {e}")),
    };
    let resp = client
        .post(channel.resolved_endpoint())
        .header("Content-Type", "application/json")
        .header("X-Tatara-Signal-Type", channel.signal_type.clone())
        .body(event_bytes.to_vec())
        .send()
        .await;
    match resp {
        Ok(r) if r.status().is_success() => ExportOutcome::Shipped,
        Ok(r) => ExportOutcome::Rejected(format!("HTTP {}", r.status())),
        Err(e) => ExportOutcome::Failed(format!("HTTP error: {e}")),
    }
}

async fn ship_nats(
    channel: &tatara_process::export::NatsSubjectChannel,
    event_bytes: &[u8],
    run_id: &str,
) -> ExportOutcome {
    let url = channel.resolved_url();
    let subject = resolve_subject(channel, run_id);
    let client = match async_nats::connect(url).await {
        Ok(c) => c,
        Err(e) => return ExportOutcome::Failed(format!("NATS connect: {e}")),
    };
    let js = async_nats::jetstream::new(client);
    let ack = js
        .publish(subject.clone(), event_bytes.to_vec().into())
        .await;
    match ack {
        Ok(fut) => match fut.await {
            Ok(_) => ExportOutcome::Shipped,
            Err(e) => ExportOutcome::Failed(format!("NATS ack: {e}")),
        },
        Err(e) => ExportOutcome::Rejected(format!("NATS publish: {e}")),
    }
}

fn ship_stdout(
    channel: &tatara_process::export::StdoutChannel,
    event: &ExportEvent,
) -> ExportOutcome {
    let result = if channel.pretty {
        serde_json::to_string_pretty(event)
    } else {
        serde_json::to_string(event)
    };
    match result {
        Ok(s) => {
            println!("{s}");
            ExportOutcome::Shipped
        }
        Err(e) => ExportOutcome::Failed(format!("serialize: {e}")),
    }
}

// ─── Receipt writer ────────────────────────────────────────────────

async fn write_receipt(
    kube: &Client,
    namespace: &str,
    configmap: &str,
    key: &str,
    payload: &str,
) -> Result<()> {
    // Ns-scoped `Api<ConfigMap>` binding rides the substrate primitive
    // `tatara_process::configmap::namespaced` — see the peer
    // `read_artifact` sites above for the FOUR-site lift narrative.
    let api = tatara_process::configmap::namespaced(kube.clone(), namespace);
    // Receipt-CM `.data` seed rides the substrate composer
    // [`tatara_process::string_map::BTreeMapStrExt::insert_str`] —
    // receiver-shape peer of
    // [`tatara_process::json_object::JsonMapStrExt::insert_str`] on
    // the K8s-canonical `BTreeMap<String, String>` carrier every
    // ConfigMap `.data` writer stamps. Pre-lift this was one of
    // THREE workspace-wide restatements of the `<map>.insert
    // (<k>.to_string(), <v>.to_string())` shape past the ★★
    // PRIME-DIRECTIVE ≥ 2 duplication threshold (peer at
    // `tatara-pool-reconciler::controller_pool::build_member_process`
    // on the pool-membership annotations map).
    let mut data = BTreeMap::new();
    data.insert_str(key, payload);
    // Wire-shape 5-link `ConfigMap { metadata: ObjectMeta { name,
    // namespace, ..Default }, data: Some(<data>), ..Default }`
    // composition rides the substrate primitive
    // `tatara_process::configmap::with_data` — pre-lift this was a
    // hand-authored struct literal, one of TWO workspace-wide
    // restatements past the ★★ PRIME-DIRECTIVE ≥ 2 duplication
    // threshold (peer at `tatara-closed-loop-probe::main::
    // write_receipt`, which stamps its own receipt-CM through the
    // same wire shape with a `Some(<labels>)` — the composer's
    // `labels: Option<...>` slot preserves both consumers' postures,
    // this site passes `None`). Post-lift the ConfigMap-body
    // composition lives at ONE substrate owner (peer of
    // `configmap::namespaced` on the same axis — the namespaced
    // binder covers the Api<ConfigMap> handle-side; this composer
    // covers the resource-body side).
    let cm = tatara_process::configmap::with_data(configmap, namespace, data, None);
    // SSA-side wire dispatch rides the substrate primitive
    // `tatara_process::patch::apply` — pre-lift this was a hand-
    // authored 2-link `let pp = apply_patch_params("tatara-export-
    // worker"); api.patch(configmap, &pp, &Patch::Apply(&cm)).await`
    // chain, one of THREE workspace-wide restatements past the ★★
    // PRIME-DIRECTIVE ≥ 2 duplication threshold (peers at
    // `tatara-reconciler::ssapply::apply_owned` +
    // `tatara-reconciler::phase_machine::transition_to_releasing`,
    // both feeding the reconciler's `FIELD_MANAGER` const). Post-
    // lift every SSA writer across two active workspace crates
    // shares ONE substrate owner for the `apply_patch_params(<mgr>)
    // + api.patch(&Patch::Apply(...))` shape.
    tatara_process::patch::apply(&api, configmap, "tatara-export-worker", &cm)
        .await
        .map(|_| ())
        .with_context(|| format!("apply configmap {namespace}/{configmap}"))?;
    Ok(())
}

#[cfg(test)]
mod annotation_key_routing_pins {
    //! Byte-shape + routing pins for the ONE Receipts-arm READ site
    //! at `read_artifact::ArtifactVariant::Receipts` that filters
    //! ConfigMaps by the owning-Process annotation. Pre-lift the KEY
    //! slot passed to [`tatara_process::prelude::Annotated::annotation`]
    //! was a bare `"tatara.pleme.io/process"` string literal — the
    //! LAST workspace surface still spelling the annotation KEY inline
    //! after every peer WRITE-side + READ-side already routed through
    //! [`tatara_process::annotations::PROCESS`]. These pins bind the
    //! substrate owner at fail-before-pass-after granularity so a
    //! regression that re-inlined the literal at this callsite
    //! (breaking the covariance loop with every peer consumer) surfaces
    //! HERE rather than as silent operator-facing skew between the
    //! export-worker Receipts-collection filter and the ssapply /
    //! render / phase_machine consumers that stamp / read the SAME
    //! annotation on the SAME ConfigMap.
    use tatara_process::annotations;

    #[test]
    fn process_annotation_key_matches_pre_lift_wire_string() {
        // Byte-identity pin: the substrate-owner const is the SAME
        // string the pre-lift `cm.annotation("tatara.pleme.io/process")`
        // callsite spelled inline. A regression that drifted either
        // the const or the pre-lift wire form would surface HERE
        // rather than as silent skew between the export-worker
        // Receipts-collection filter and every peer consumer stamping
        // or reading the SAME K8s annotation slot.
        assert_eq!(annotations::PROCESS, "tatara.pleme.io/process");
    }

    #[test]
    fn process_annotation_key_inhabits_tatara_group_prefix() {
        // Family-membership pin: [`annotations::PROCESS`] rides in the
        // shared reverse-DNS namespace every substrate-owned annotation
        // key carries, matching the `#[kube(group = "tatara.pleme.io",
        // …)]` derive slot on every tatara CRD struct. A regression
        // that dropped or drifted the prefix at the const owner
        // surfaces HERE at this consumer-site pin (rather than only
        // at the family sweep in `tatara-process::annotations_family_
        // tests::all_share_group_prefix`), so the ONE workspace surface
        // that keys READ-side dispatch on this specific annotation
        // (the export-worker Receipts-collection filter) carries its
        // own local coherence guard against a group-segment shift
        // stranding it.
        assert!(annotations::PROCESS.starts_with(annotations::GROUP_PREFIX));
    }

    #[test]
    fn process_annotation_key_is_distinct_from_receipt_label_key() {
        // Cross-family disjointness pin: the export-worker Receipts-
        // collection filter READS through [`annotations::PROCESS`]
        // (owner-Process qualified ref); the closed-loop-probe
        // receipt-CM writer STAMPS [`annotations::RECEIPT`] (the CM's
        // receipt-envelope version marker) — two distinct annotation
        // slots on the SAME K8s ConfigMap carrier. A copy-paste that
        // collapsed either arm onto the other would silently double-
        // book the same K8s metadata slot for two orthogonal concerns
        // (owner correlation vs. envelope version enumeration); this
        // pin surfaces such a regression HERE at the ONE consumer that
        // could confuse the two.
        assert_ne!(annotations::PROCESS, annotations::RECEIPT);
    }
}