verit 0.2.0

//! Reference integration — a tiny append-only **event store** that dogfoods the
//! whole Veritate surface end to end, on one `.verit` file:
//!
//! - **`#[derive(Verit)]`** defines the event types (no hand-written schema).
//! - A **`.verit` file** (`FileWriter` / `FileView`) persists the log at rest:
//!   crash-safe appends, read back **zero-copy** from the file's own bytes.
//! - **Self-containment**: the consumer opens the file with *no registry and no
//!   schema* — the file carries everything needed to read it.
//! - **Stable record ids**: a tailing consumer checkpoints an id and resumes
//!   from it, and its checkpoint survives a compaction.
//! - **Schema evolution at rest**: an event written years ago under a narrower
//!   schema still resolves into today's `Event` type, out of the same file.
//! - **Erasure**: `purge_ids` removes *and* compacts, because plain removal only
//!   unlinks.
//! - **Crash safety**: a commit torn by a simulated crash rolls back cleanly.
//! - A **`SchemaRegistry`** + distribution bundle for the *wire* path, which is
//!   the one place a registry is still needed.
//!
//! Run it (the derive feature is needed for `#[derive(Verit)]`):
//!
//! ```text
//! cargo run -p verit --features derive --example eventstore
//! ```
//!
//! It prints what each step does and `assert!`s every round-trip, so it doubles
//! as an executable integration test.

use verit::{
    encode, Dt, FileView, FileWriter, Message, SchemaBuilder, SchemaMode, SchemaRegistry, Value,
    Verit, VeritType,
};

#[derive(Verit, Debug, Clone, Copy, PartialEq)]
#[verit(mode = "dense")]
struct Coord {
    #[verit(id = 1)]
    lat: f64,
    #[verit(id = 2)]
    lon: f64,
}

/// Today's event schema (v2): `at` and `region` were added over the years.
#[derive(Verit, Debug, Clone, PartialEq)]
struct Event {
    #[verit(id = 1)]
    seq: u64,
    #[verit(id = 2)]
    kind: String,
    #[verit(id = 3)]
    payload: String,
    #[verit(id = 4)]
    at: Option<Coord>,
    #[verit(id = 5)]
    tags: Vec<String>,
    #[verit(id = 6)]
    region: Option<String>,
}

fn event(seq: u64, kind: &str, payload: &str, tags: &[&str]) -> Event {
    Event {
        seq,
        kind: kind.into(),
        payload: payload.into(),
        at: None,
        tags: tags.iter().map(|t| t.to_string()).collect(),
        region: None,
    }
}

fn main() {
    println!("== Veritate event store — reference integration ==\n");

    let path =
        std::env::temp_dir().join(format!("veritate-eventstore-{}.verit", std::process::id()));
    let _ = std::fs::remove_file(&path);

    // ---- 1. Producer: append events to a .verit file -------------------------
    // Each event is encoded hash-only (just the 128-bit schema id on the wire).
    // The file stores the schema once, in its own schema section, so the records
    // stay compact *and* the file stays readable on its own.
    let first = Event {
        seq: 1,
        kind: "login".into(),
        payload: "user=alice".into(),
        at: Some(Coord {
            lat: 51.5,
            lon: -0.12,
        }),
        tags: vec!["auth".into()],
        region: Some("eu-west".into()),
    };
    let second = event(2, "purchase", "sku=42 qty=3", &["billing", "priority"]);

    let mut store = FileWriter::create(&path).unwrap();
    let schema = Event::verit_schema();
    // `append` returns the record id it assigned — stable from this moment, and
    // what a consumer should checkpoint against.
    let id_first = store
        .append_message(schema, &first.to_verit(SchemaMode::HashOnly).unwrap())
        .unwrap();
    let id_second = store
        .append_message(schema, &second.to_verit(SchemaMode::HashOnly).unwrap())
        .unwrap();
    store.commit().unwrap();

    let after_first_commit = std::fs::metadata(&path).unwrap().len();
    println!(
        "producer: appended 2 events → ids {id_first}, {id_second}; \
         {after_first_commit}-byte .verit file at generation {}",
        store.generation()
    );

    // ---- 2. Consumer: open the file with nothing else -------------------------
    // No registry, no schema, no sidecar. This is the whole promise of the
    // format: the bytes are enough.
    let image = std::fs::read(&path).unwrap();
    {
        let log = FileView::open(&image).unwrap();
        println!(
            "consumer: opened the file cold — {} records, generation {}, {} schema(s) inside it",
            log.len(),
            log.generation(),
            log.schemas().len()
        );
        for (i, original) in [&first, &second].iter().enumerate() {
            let back = Event::from_verit(log.get(i).unwrap()).unwrap();
            println!(
                "  [id {}] seq={} kind={} tags={:?}",
                log.record(i).unwrap().id,
                back.seq,
                back.kind,
                back.tags
            );
            assert_eq!(&back, *original, "record {i} round-trips");
        }
        // And the file describes itself with no type knowledge at all.
        println!("  dump_json(0) = {}", log.dump_json(0).unwrap());
    }
    println!();

    // ---- 3. Tailing: checkpoint an id, resume from it -------------------------
    let checkpoint = id_second;
    println!("consumer: processed through record id {checkpoint}; checkpointing there");

    let third = event(3, "logout", "user=alice", &["auth"]);
    let id_third = store
        .append_message(schema, &third.to_verit(SchemaMode::HashOnly).unwrap())
        .unwrap();

    // A legacy peer's event: an old service (v1) never knew about
    // `at`/`tags`/`region` and used a narrower `seq`. Its bytes land in the same
    // file, under their own schema — which the file then stores alongside v2.
    let legacy_schema = SchemaBuilder::new()
        .add_struct(
            "Event",
            vec![
                (1, "seq", Dt::U32), // widens to today's u64
                (2, "kind", Dt::Str),
                (3, "payload", Dt::Str),
            ],
        )
        .build("Event")
        .unwrap();
    let legacy_bytes = encode(
        &legacy_schema,
        &Value::Struct(vec![
            (1, Value::U32(99)),
            (2, Value::str("healthcheck")),
            (3, Value::str("ok")),
        ]),
        SchemaMode::HashOnly,
    )
    .unwrap();
    let id_legacy = store.append_message(&legacy_schema, &legacy_bytes).unwrap();
    store.commit().unwrap();
    println!(
        "producer: appended 2 more (ids {id_third}, {id_legacy}) — generation {}",
        store.generation()
    );

    let image = std::fs::read(&path).unwrap();
    {
        let log = FileView::open(&image).unwrap();
        // Logarithmic: the index is sorted by id, so resuming does not scan.
        let fresh: Vec<u64> = log.records_after(checkpoint).map(|r| r.id).collect();
        println!("consumer: records_after({checkpoint}) → ids {fresh:?}");
        assert_eq!(fresh, vec![id_third, id_legacy]);
        assert_eq!(
            log.schemas().len(),
            2,
            "the file now carries both schema versions"
        );

        // ---- 4. Schema evolution, at rest -------------------------------------
        // The v1 record resolves into today's reader schema using only what the
        // file kept beside it. No registry, no schema fetch — the writer schema
        // is in the file.
        let at = log.find_by_id(id_legacy).unwrap();
        let resolver = log.resolver_for(at, Event::verit_schema()).unwrap();
        let root = log.message(at).unwrap().root(&resolver).unwrap();
        println!(
            "consumer: read the v1 record through today's schema: seq={:?} kind={:?} at={:?}",
            root.get_u64(1).unwrap(),
            root.get_str(2).unwrap(),
            root.get_struct(4).unwrap().map(|_| "present"),
        );
        assert_eq!(root.get_u64(1).unwrap(), Some(99), "u32 seq widened to u64");
        assert_eq!(root.get_str(2).unwrap(), Some("healthcheck"));
        assert!(
            root.get_struct(4).unwrap().is_none(),
            "`at` did not exist in v1 data"
        );
    }
    println!();

    // ---- 5. Crash safety: a torn commit rolls back ---------------------------
    // Simulate a crash partway through the second commit by truncating the image
    // inside it. The reader falls back to the previous footer — the file reads
    // exactly as it did before the commit began.
    let torn = &image[..(after_first_commit as usize + 16).min(image.len() - 1)];
    {
        let rolled_back = FileView::open(torn).unwrap();
        println!(
            "crash sim: truncated mid-commit → opened at generation {} with {} records \
             (the torn commit never happened)",
            rolled_back.generation(),
            rolled_back.len()
        );
        assert_eq!(rolled_back.len(), 2, "rolled back to the 2-record state");
        assert_eq!(rolled_back.generation(), 2);
    }
    println!();

    // ---- 6. Erasure: removal unlinks, purge erases ---------------------------
    let pii = event(4, "profile", "email=alice@example.com", &["pii"]);
    let id_pii = store
        .append_message(schema, &pii.to_verit(SchemaMode::HashOnly).unwrap())
        .unwrap();
    store.commit().unwrap();

    store.remove_id(id_pii).unwrap();
    store.commit().unwrap();
    let image = std::fs::read(&path).unwrap();
    let still_there = image
        .windows(b"alice@example.com".len())
        .any(|w| w == b"alice@example.com");
    println!(
        "retention: removed id {id_pii} — unlinked from the index, \
         but the bytes are still in the file: {still_there}"
    );
    assert!(still_there, "removal must not erase — spec §8.2");

    // `purge_ids` is the call that actually erases: remove, then compact, in one
    // pass. (The id is already unlinked, so this just compacts.)
    store.purge_ids(&[id_pii]).unwrap();
    let image = std::fs::read(&path).unwrap();
    let gone = !image
        .windows(b"alice@example.com".len())
        .any(|w| w == b"alice@example.com");
    println!("retention: purge_ids → compacted, bytes erased: {gone}");
    assert!(gone, "purge_ids must erase");

    // ---- 7. Ids survive the compaction --------------------------------------
    // The consumer's checkpoint from step 3 is still meaningful. A *position*
    // would not have been — compaction renumbered every one of them.
    {
        let log = FileView::open(&image).unwrap();
        let fresh: Vec<u64> = log.records_after(checkpoint).map(|r| r.id).collect();
        println!(
            "consumer: after compaction, records_after({checkpoint}) → ids {fresh:?} \
             (generation reset to {})",
            log.generation()
        );
        assert_eq!(fresh, vec![id_third, id_legacy], "ids survived compaction");
        assert_eq!(log.generation(), 1, "compaction rebuilds at generation 1");
        assert!(
            log.next_record_id() > id_pii,
            "a purged record's id is retired, never reissued"
        );
    }
    println!();

    // ---- 8. The wire path still needs a registry -----------------------------
    // A *file* is self-contained. A hash-only *message* on a socket is not — it
    // carries only the 128-bit schema id. That is what the registry and its
    // distribution bundle are for, and the distinction is worth keeping straight.
    let mut producer_registry = SchemaRegistry::new();
    producer_registry.register(Event::verit_schema().clone());
    let bundle = producer_registry.to_bundle();
    let consumer_registry = SchemaRegistry::from_bundle(&bundle).unwrap();
    let on_the_wire = second.to_verit(SchemaMode::HashOnly).unwrap();
    let msg = Message::parse(&on_the_wire).unwrap();
    let resolver = consumer_registry
        .resolver_for(msg.schema_id(), Event::verit_schema())
        .unwrap();
    let root = msg.root(&resolver).unwrap();
    println!(
        "wire: a peer's hash-only message resolved via a {}-byte distribution bundle: kind={:?}",
        bundle.len(),
        root.get_str(2).unwrap()
    );
    assert_eq!(root.get_str(2).unwrap(), Some("purchase"));

    let _ = std::fs::remove_file(&path);
    println!("\n== every round-trip, the evolution read, the rollback, and the erasure passed ==");
}