spate 0.2.0

//! The examples index, as a snapshot test.
//!
//! The root README links a reader straight at the examples directory, so what
//! they meet there is either a map or an unannotated file listing. An example
//! missing from that map goes unfound, and nothing else in this repository
//! breaks when it happens.
//!
//! The index is rendered from four fields in each example's own header:
//!
//! ```text
//! // INDEX-TIER:  getting-started
//! // INDEX-GOAL:  build, drive and assert on a whole pipeline
//! // INDEX-TECH:  no infrastructure
//! // INDEX-NEEDS: nothing
//! ```
//!
//! Fields rather than prose, deliberately. A test rendered from an example's
//! narrative commentary fires on every wording change, and a gate that fails
//! on prose is one contributors learn to route around. A wording change around
//! these four lines cannot fail anything; a missing or orphaned entry always
//! does.
//!
//! They are plain `//` comments outside the `//!` block, so they never reach
//! docs.rs, and they read like the transclusion markers already in these
//! files. Note the marker word itself is deliberately not spelled out here:
//! `scripts/transclude.sh` scans every file under `crates/` for it, so writing
//! it in prose declares a region with an empty name and fails the gate.
//! `required-features` is **not** among them. It is in the manifest, and a
//! field duplicating the manifest is one that disagrees with it.
//!
//! To accept a change:
//!
//! ```sh
//! UPDATE_EXAMPLES_INDEX=1 cargo test -p spate --test examples_index --locked
//! ```

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::process::Command;

/// Reading tiers, in the order they appear in the file, with the blurb under
/// each heading. The tier here is where an example sits in the *guide*; it is a
/// different axis from whether a pull request can run it, which is what
/// `test = true` records.
const TIERS: &[(&str, &str, &str)] = &[
    (
        "getting-started",
        "1. Getting started",
        "No infrastructure. Read them in this order.",
    ),
    (
        "production",
        "2. Production pipelines",
        "The shapes a real deployment takes.",
    ),
    (
        "bounded-jobs",
        "3. Bounded jobs and scaling out",
        "Work that finishes, and work shared across instances.",
    ),
    (
        "operating",
        "4. Operating",
        "What the pipeline tells you, and what it does when something breaks.",
    ),
    (
        "extending",
        "5. Extending",
        "Writing your own components against the v1 contracts.",
    ),
];

const HEADER: &str = r#"<!--
Generated by crates/spate/tests/examples_index.rs from the INDEX- block in each
example's header. Do not edit by hand — accept a change with:

    UPDATE_EXAMPLES_INDEX=1 cargo test -p spate --test examples_index --locked
-->

# Examples

Every example is a whole program: build it, run it, read what it printed. Each
one's header comment says what it demonstrates and what it needs, and the
tables below group them by what you are trying to do.

New here, read [Getting started](#1-getting-started) in order. Nothing outside
that tier depends on reading order — pick the one matching your task.

## Running one

```sh
cargo run -p spate --example memory_pipeline
cargo run --release -p spate --features full --example kafka_avro_to_clickhouse
```

An example whose **Features** column is not `—` needs those features on the
command line. Naming one without them fails with exactly what is missing:

```text
error: target `s3_backfill` in package `spate` requires the features: `s3`, `json`
```

Examples that talk to real servers ship no `docker-compose`. Each one's header
comment names exactly what it needs and which environment variables point at
it, and every shipped config reads its endpoints through `${VAR:-default}`, so
the file you run against your own servers is the file in this directory.

Configuration comes from `SPATE_CONFIG` where an example loads YAML from disk:

```sh
SPATE_CONFIG=/etc/spate/pipeline.yaml cargo run --release -p spate \
  --features full --example kafka_avro_to_clickhouse
```

## The storefront stream

`spate-datagen` generates one dataset with nothing installed behind it: a
storefront whose payments and refunds name orders that were really placed, for
amounts matching their lines.

```text
order_placed     { order_id, customer_id, region, placed_at, lines: [{ sku, qty, unit_cents }] }
payment_captured { order_id, amount_cents }
refund_issued    { order_id, amount_cents, reason }
```

The nested `lines` array is what `flat_map` fans out, and the three event kinds
are what a split terminal separates. Sharding keys on the **order id**, which
the generator sets as each record's key: a payment and a refund carry only the
`order_id` of the order they settle, so that is the only field all three share
and the only one that can colocate them on a shard.

The types are `spate_datagen::storefront`, so an example and your own code can
share them.

"#;

/// The container link is absolute, and must stay absolute: this file ships
/// inside the published crate, where a relative path climbing out of
/// `crates/spate` resolves to nothing.
const FOOTER: &str = r#"## Containers

[`examples/docker`](https://github.com/spate-etl/spate/tree/main/examples/docker)
builds the flagship example into a distroless image, and its README covers
probes, drain timeouts and sizing.

## Related

- [User guide](https://spate.kainth.dev/docs/user-guide/) — the concepts these
  examples are worked instances of
- [`spate` on docs.rs](https://docs.rs/spate) — every type and feature named here
- [`spate-test`](https://docs.rs/spate-test) — the in-memory source and sink the
  infrastructure-free examples are built on
"#;

/// Examples the generated header names in its own prose. The tables are
/// rendered from the tree and cannot go stale; these names are hand-written, so
/// they are asserted against it instead.
const HEADER_EXAMPLES: &[&str] = &["memory_pipeline", "kafka_avro_to_clickhouse"];

struct Example {
    name: String,
    rank: u32,
    tier: String,
    goal: String,
    tech: String,
    needs: String,
    features: Vec<String>,
}

#[derive(serde::Deserialize)]
struct Metadata {
    packages: Vec<Package>,
}

#[derive(serde::Deserialize)]
struct Package {
    name: String,
    targets: Vec<Target>,
}

#[derive(serde::Deserialize)]
struct Target {
    name: String,
    kind: Vec<String>,
    /// The JSON key is hyphenated; reading it as `required_features` yields an
    /// empty list for every target and the Features column would silently
    /// render `—` throughout.
    #[serde(rename = "required-features", default)]
    required_features: Vec<String>,
    src_path: PathBuf,
}

/// Every `// INDEX-<FIELD>:` value in one source, in file order. Only a `//`
/// line counts. A `//!` line is module prose, and the module docs come first
/// in the file, so letting them match would let a sentence win over the field.
fn index_fields(src: &str, field: &str) -> Vec<String> {
    let needle = format!("INDEX-{field}:");
    src.lines()
        .map(str::trim_start)
        .filter(|l| l.starts_with("//") && !l.starts_with("//!"))
        .filter_map(|l| l.split_once(&needle))
        .map(|(_, v)| v.trim().to_string())
        .collect()
}

/// The one value an example declares for a field, or `None` when it declares
/// none.
///
/// Panics when the field is declared twice, or when its value carries a `|`.
/// Both render a plausible file that the snapshot below would then accept for
/// good. A stale copy above the live block wins silently, and a `|` splits
/// the table row into extra cells.
fn index_field(src: &str, field: &str, example: &str) -> Option<String> {
    let found = index_fields(src, field);
    assert!(
        found.len() <= 1,
        "{example} declares `// INDEX-{field}:` {} times; the first wins and the \
         rest are invisible",
        found.len()
    );
    let value = found.into_iter().next()?;
    assert!(
        !value.contains('|'),
        "{example}'s INDEX-{field} carries a `|`, which splits the table row it \
         renders into"
    );
    Some(value)
}

/// Where an example sits within its tier. An example declaring no rank takes 50
/// and falls back to name order.
fn rank(src: &str, example: &str) -> u32 {
    let Some(raw) = index_field(src, "RANK", example) else {
        return 50;
    };
    raw.parse::<u32>()
        .ok()
        .filter(|r| *r <= 999)
        .unwrap_or_else(|| panic!("{example} declares INDEX-RANK `{raw}`; a rank is 0 to 999"))
}

fn collect() -> Vec<Example> {
    let manifest = concat!(env!("CARGO_MANIFEST_DIR"), "/Cargo.toml");
    let out = Command::new(env!("CARGO"))
        .args([
            "metadata",
            "--no-deps",
            "--format-version",
            "1",
            "--locked",
            "--manifest-path",
            manifest,
        ])
        .output()
        .expect("cargo metadata");
    assert!(out.status.success(), "cargo metadata failed");
    let meta: Metadata = serde_json::from_slice(&out.stdout).expect("parse cargo metadata");
    let pkg = meta
        .packages
        .into_iter()
        .find(|p| p.name == "spate")
        .expect("the spate package");

    let mut out = Vec::new();
    for t in pkg.targets {
        if !t.kind.iter().any(|k| k == "example") {
            continue;
        }
        let src = std::fs::read_to_string(&t.src_path)
            .unwrap_or_else(|e| panic!("read {}: {e}", t.src_path.display()));
        let need = |f: &str| {
            index_field(&src, f, &t.name).unwrap_or_else(|| {
                panic!(
                    "{} has no `// INDEX-{f}:` line; the index renders four fields \
                     (TIER, GOAL, TECH, NEEDS) from each example's header",
                    t.name
                )
            })
        };
        out.push(Example {
            rank: rank(&src, &t.name),
            tier: need("TIER"),
            goal: need("GOAL"),
            tech: need("TECH"),
            needs: need("NEEDS"),
            features: t.required_features,
            name: t.name,
        });
    }
    out
}

fn render(examples: &[Example]) -> String {
    let mut by_tier: BTreeMap<&str, Vec<&Example>> = BTreeMap::new();
    for e in examples {
        by_tier.entry(e.tier.as_str()).or_default().push(e);
    }
    for (slug, _, _) in TIERS {
        if let Some(rows) = by_tier.get_mut(slug) {
            rows.sort_by(|a, b| (a.rank, &a.name).cmp(&(b.rank, &b.name)));
        }
    }

    let known: Vec<&str> = TIERS.iter().map(|(s, _, _)| *s).collect();
    for e in examples {
        assert!(
            known.contains(&e.tier.as_str()),
            "{} declares INDEX-TIER `{}`; known tiers are {}",
            e.name,
            e.tier,
            known.join(", ")
        );
    }

    for named in HEADER_EXAMPLES {
        assert!(
            HEADER.contains(named),
            "HEADER_EXAMPLES lists `{named}`, which the header does not name; the \
             assertion below it is asserting nothing"
        );
        assert!(
            examples.iter().any(|e| e.name == *named),
            "the generated header names `{named}`, which matches no example"
        );
    }

    let mut s = String::from(HEADER);
    for (slug, heading, blurb) in TIERS {
        s.push_str(&format!("## {heading}\n\n{blurb}\n\n"));
        s.push_str("| Example | What it shows | Features | Needs |\n|---|---|---|---|\n");
        for e in by_tier.get(slug).into_iter().flatten() {
            let feats = if e.features.is_empty() {
                "—".to_string()
            } else {
                format!("`{}`", e.features.join(","))
            };
            s.push_str(&format!(
                "| [`{n}`]({n}.rs) | How to {g} with **{t}** | {feats} | {needs} |\n",
                n = e.name,
                g = e.goal,
                t = e.tech,
                needs = e.needs,
            ));
        }
        s.push('\n');
    }
    s.push_str(FOOTER);
    s
}

#[test]
fn the_index_matches_the_tree() {
    let examples = collect();
    assert!(
        examples.len() >= 15,
        "only {} example(s) collected; the filter has stopped matching and this \
         test is vacuous",
        examples.len()
    );

    let path = Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/examples/README.md"));
    let want = render(&examples);

    if std::env::var_os("UPDATE_EXAMPLES_INDEX").is_some() {
        std::fs::write(path, &want).expect("write the index");
        return;
    }

    let have = std::fs::read_to_string(path).unwrap_or_default();
    assert!(
        have == want,
        "crates/spate/examples/README.md is out of date.\n\n\
         Accept the change with:\n\n    \
         UPDATE_EXAMPLES_INDEX=1 cargo test -p spate --test examples_index --locked\n"
    );
}

/// A fixture the module docs would win against if `//!` lines counted: the
/// prose comes first in the file, so the wrong reading renders a plausible row
/// and the snapshot above then holds the tree to it.
const FIXTURE: &str = "\
//! Module docs, which are prose and must not be read as fields.
//!
//! INDEX-GOAL: this sentence must not satisfy the GOAL field

// INDEX-TIER:  operating
// INDEX-GOAL:  count what an operator you wrote is doing
fn main() {}
";

#[test]
fn a_module_doc_line_cannot_satisfy_a_field() {
    assert_eq!(
        index_field(FIXTURE, "GOAL", "fixture").as_deref(),
        Some("count what an operator you wrote is doing")
    );
    assert_eq!(
        index_field(FIXTURE, "TIER", "fixture").as_deref(),
        Some("operating")
    );
    assert_eq!(index_field(FIXTURE, "ABSENT", "fixture"), None);
    assert_eq!(rank(FIXTURE, "fixture"), 50);
}

#[test]
#[should_panic(expected = "declares `// INDEX-GOAL:` 2 times")]
fn a_field_declared_twice_is_refused() {
    let src = "// INDEX-GOAL: a stale copy left above the live block\n\
               // INDEX-GOAL: the live one\n";
    let _ = index_field(src, "GOAL", "fixture");
}

#[test]
#[should_panic(expected = "splits the table row")]
fn a_pipe_in_a_value_is_refused() {
    let _ = index_field("// INDEX-GOAL: split | the row\n", "GOAL", "fixture");
}

#[test]
#[should_panic(expected = "a rank is 0 to 999")]
fn a_rank_that_does_not_parse_is_refused() {
    rank("// INDEX-RANK: 1O\n", "fixture");
}