tidaldb/tidal/tests/m0m10_recovery.rs

//! M0–M10 crash-recovery integration tests for the db-integration zone.
//!
//! These cover three durability/recovery contracts that had no end-to-end test
//! before the M0–M10 review:
//!
//! 1. **Vector index rebuilt on restart** — durable `EMB:` embeddings repopulate
//!    the in-memory ANN index on reopen, so vector SEARCH returns them without
//!    rewriting embeddings (the index is derived state; the entity store is the
//!    source of truth).
//! 2. **Purge tombstone replayed on crash-recovery** — a community purge that
//!    crashes BEFORE the next community checkpoint still stays purged on reopen
//!    (the durable `Tag::PurgeTombstone` is re-applied), closing the
//!    privacy/trust window where purged data silently resurrects.
//! 3. **`created_at` indexing parity across restart** — an item written without
//!    an explicit `created_at` gets the SAME `created_at` after a restart as on
//!    the live write path, because the default is materialized into the
//!    persisted metadata and the rebuild reads the identical value.

#![allow(clippy::too_many_lines, clippy::unwrap_used)]

use std::{collections::HashMap, time::Duration};

use tidaldb::{
    CommunityId, ShareIntent, SharePolicy, TempTidalHome, TidalDb, UserId,
    query::search::Search,
    schema::{DecaySpec, EntityId, EntityKind, SchemaBuilder, TextFieldType, Timestamp, Window},
    testing::{CrashInjector, CrashPoint, crash_injector::run_with_crash},
};

// ── Task 1: vector index rebuilt on restart ──────────────────────────────────

/// Schema with one Item embedding slot, used by the embedding-rebuild test.
fn embedding_schema(dims: usize) -> tidaldb::schema::Schema {
    let mut b = SchemaBuilder::new();
    let _ = b
        .signal(
            "view",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::AllTime])
        .velocity(false)
        .add();
    b.embedding_slot("content", EntityKind::Item, dims);
    b.build().unwrap()
}

#[test]
fn vector_index_rebuilt_from_durable_embeddings_on_reopen() {
    let home = TempTidalHome::new().unwrap();

    // First open: write three orthogonal item embeddings, then clean shutdown.
    {
        let db = TidalDb::builder()
            .with_data_dir(home.path())
            .with_schema(embedding_schema(4))
            .open()
            .unwrap();

        // An item needs metadata to be in the universe; write both.
        for id in 1u64..=3 {
            let mut meta = HashMap::new();
            meta.insert("title".to_string(), format!("item {id}"));
            db.write_item_with_metadata(EntityId::new(id), &meta)
                .unwrap();
        }
        db.write_item_embedding(EntityId::new(1), &[1.0, 0.0, 0.0, 0.0])
            .unwrap();
        db.write_item_embedding(EntityId::new(2), &[0.0, 1.0, 0.0, 0.0])
            .unwrap();
        db.write_item_embedding(EntityId::new(3), &[0.0, 0.0, 1.0, 0.0])
            .unwrap();

        // Sanity: the live (pre-restart) index serves the nearest neighbor.
        let live = db
            .search(
                &Search::builder()
                    .vector(vec![1.0, 0.0, 0.0, 0.0])
                    .limit(1)
                    .build()
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(live.items.len(), 1, "live index must serve a result");
        assert_eq!(live.items[0].entity_id, EntityId::new(1));

        db.shutdown().unwrap();
    }

    // Reopen: the in-memory ANN index is DERIVED state lost on shutdown. Without
    // the rebuild-from-store wiring this search returns nothing. With it, the
    // durable `EMB:content` rows repopulate the index and SEARCH works.
    let db = TidalDb::builder()
        .with_data_dir(home.path())
        .with_schema(embedding_schema(4))
        .open()
        .unwrap();

    let near_1 = db
        .search(
            &Search::builder()
                .vector(vec![1.0, 0.0, 0.0, 0.0])
                .limit(1)
                .build()
                .unwrap(),
        )
        .unwrap();
    assert_eq!(
        near_1.items.len(),
        1,
        "vector SEARCH must return a result after reopen (index rebuilt from EMB: keys)"
    );
    assert_eq!(
        near_1.items[0].entity_id,
        EntityId::new(1),
        "nearest neighbor of e0 is item 1"
    );

    // A different query direction resolves to its own nearest neighbor, proving
    // the whole index — not just one vector — was repopulated.
    let near_2 = db
        .search(
            &Search::builder()
                .vector(vec![0.0, 1.0, 0.0, 0.0])
                .limit(1)
                .build()
                .unwrap(),
        )
        .unwrap();
    assert_eq!(near_2.items.len(), 1);
    assert_eq!(near_2.items[0].entity_id, EntityId::new(2));

    db.shutdown().unwrap();
}

// ── Task 1b: item text index rebuilt from durable store on reopen ────────────

/// Schema with one indexed item text field, used by the text-rebuild test.
fn text_schema() -> tidaldb::schema::Schema {
    let mut b = SchemaBuilder::new();
    let _ = b
        .signal(
            "view",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::AllTime])
        .velocity(false)
        .add();
    b.text_field("title", TextFieldType::Text);
    b.build().unwrap()
}

/// BLOCKER 7: items durably persisted to the entity store but not committed to
/// Tantivy before a hard crash must still be found by BM25 SEARCH after reopen,
/// because the item text index is unconditionally rebuilt from the durable store
/// at open (the entity store is the source of truth; Tantivy is derived state).
///
/// We model the post-crash on-disk state faithfully: after a clean write the
/// store AND Tantivy both hold the items, so we delete the `text_index`
/// directory to simulate the exact failure mode the BLOCKER describes — Tantivy
/// behind the durable store because the async syncer never committed (or a
/// power-loss truncated it). Without the rebuild-at-open wiring this SEARCH
/// returns nothing; with it, the items are re-indexed from the store.
#[test]
fn item_text_index_rebuilt_from_durable_store_on_reopen() {
    let home = TempTidalHome::new().unwrap();
    let titles = [
        (1u64, "rust systems programming"),
        (2, "jazz piano improvisation"),
        (3, "italian pasta recipe"),
    ];

    // Phase 1: write items (persisted to the store) and clean-shutdown.
    {
        let db = TidalDb::builder()
            .with_data_dir(home.path())
            .with_schema(text_schema())
            .open()
            .unwrap();
        for (id, title) in titles {
            let mut meta = HashMap::new();
            meta.insert("title".to_string(), title.to_string());
            db.write_item_with_metadata(EntityId::new(id), &meta)
                .unwrap();
        }
        // Force a synchronous commit of the background text syncer so the live
        // sanity search below sees the just-written items (the default async
        // commit interval is 2s; without this flush the live read races it).
        // This is the canonical pattern used across the search tests.
        db.flush_text_index().unwrap();

        // Sanity: the live index serves a BM25 hit before any restart.
        let live = db
            .search(&Search::builder().query("rust").limit(5).build().unwrap())
            .unwrap();
        assert!(
            live.items.iter().any(|i| i.entity_id == EntityId::new(1)),
            "live text index must serve a result"
        );
        db.shutdown().unwrap();
    }

    // Simulate the crash divergence: blow away the Tantivy item index so it is
    // BEHIND the durable store, exactly as a hard crash before commit leaves it.
    let text_dir = home.path().join("text_index");
    assert!(
        text_dir.exists(),
        "text index dir must exist after a clean run"
    );
    std::fs::remove_dir_all(&text_dir).unwrap();

    // Phase 2: reopen. rebuild_text_indexes_at_open must repopulate Tantivy from
    // the durable item store so SEARCH finds the items again.
    let db = TidalDb::builder()
        .with_data_dir(home.path())
        .with_schema(text_schema())
        .open()
        .unwrap();

    for (id, query) in [(1u64, "rust"), (2, "jazz"), (3, "pasta")] {
        let results = db
            .search(&Search::builder().query(query).limit(5).build().unwrap())
            .unwrap();
        assert!(
            results
                .items
                .iter()
                .any(|i| i.entity_id == EntityId::new(id)),
            "item {id} must be found by SEARCH '{query}' after reopen \
             (text index rebuilt from the durable store)"
        );
    }

    db.shutdown().unwrap();
}

// ── Task 2: purge tombstone replayed after a crash before checkpoint ─────────

fn community_schema() -> tidaldb::schema::Schema {
    let mut b = SchemaBuilder::new();
    let _ = b
        .signal(
            "like",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::AllTime])
        .velocity(false)
        .add();
    b.build().unwrap()
}

fn share_policy() -> SharePolicy {
    SharePolicy::community_share(1, [ShareIntent::Like])
}

#[test]
fn purge_survives_crash_before_next_checkpoint() {
    let home = TempTidalHome::new().unwrap();
    let (community, item) = (CommunityId(1), EntityId::new(100));

    // Phase 1: seed three contributors, then CLEAN shutdown so the PRE-purge
    // community aggregate (count = 3) is persisted to the Tag::Community
    // checkpoint. The community ledger is only checkpointed on the lifecycle
    // path, so this clean shutdown is what puts the pre-purge snapshot on disk.
    {
        let db = TidalDb::builder()
            .with_data_dir(home.path())
            .with_schema(community_schema())
            .open()
            .unwrap();
        let now = Timestamp::now();
        for u in 1u64..=3 {
            db.join_community(UserId(u), community, share_policy())
                .unwrap();
            assert!(
                db.signal_for_community("like", item, 1.0, now, UserId(u), community)
                    .unwrap()
            );
        }
        assert_eq!(
            db.read_community_windowed_count(community, item, "like", Window::AllTime)
                .unwrap(),
            3
        );
        db.shutdown().unwrap();
    }

    // Phase 2: reopen, PURGE user 2 (writes a durable Tag::PurgeTombstone and
    // mutates only the in-memory aggregate -> count 2), then CRASH before the
    // next checkpoint. We arm CheckpointPreFlush so the shutdown's signal
    // checkpoint panics before any post-purge state is flushed; the community
    // checkpoint (later in shutdown) therefore never runs. The on-disk state is
    // exactly: the PRE-purge Tag::Community snapshot (count 3) + the durable
    // tombstone. The lock file is released as `db` is dropped during unwind.
    let injector = CrashInjector::new(CrashPoint::CheckpointPreFlush, 0);
    let outcome = run_with_crash(&injector, || {
        let db = TidalDb::builder()
            .with_data_dir(home.path())
            .with_schema(community_schema())
            .open()
            .unwrap();
        // Aggregate restored to the pre-purge count.
        assert_eq!(
            db.read_community_windowed_count(community, item, "like", Window::AllTime)
                .unwrap(),
            3,
            "reopen must restore the pre-purge aggregate from the checkpoint"
        );

        // Write a LOCAL signal so the signal-ledger has a checkpointable entry:
        // community-scoped contributions never touch the local aggregate
        // (local-profile-intact), so without this the shutdown checkpoint over an
        // empty ledger would not cross CheckpointPreFlush. This local state is
        // independent of the community aggregate the assertions below check.
        db.signal("like", EntityId::new(900), 1.0, Timestamp::now())
            .unwrap();

        let receipt = db
            .purge_prior_contributions(UserId(2), community, 0..=u32::MAX)
            .unwrap();
        assert_eq!(receipt.contributions_removed, 1);
        assert_eq!(
            db.read_community_windowed_count(community, item, "like", Window::AllTime)
                .unwrap(),
            2,
            "in-memory aggregate reflects the purge before the crash"
        );

        // Crash before the post-purge checkpoint flushes (CheckpointPreFlush
        // fires inside the signal-ledger checkpoint during shutdown). The
        // returned Result is irrelevant: the injector panic unwinds out of
        // `close`, so `run_with_crash` yields `Err(CrashPoint)`.
        let _ = db.close();
    });
    assert!(
        matches!(outcome, Err(CrashPoint::CheckpointPreFlush)),
        "the crash injector must have fired during the shutdown checkpoint, got {outcome:?}"
    );

    // Phase 3: reopen after the simulated crash. restore() reads the PRE-purge
    // Tag::Community checkpoint (count 3); the tombstone replay must re-apply the
    // purge so the aggregate is 2 — NOT silently resurrected to 3.
    let db = TidalDb::builder()
        .with_data_dir(home.path())
        .with_schema(community_schema())
        .open()
        .unwrap();
    assert_eq!(
        db.read_community_windowed_count(community, item, "like", Window::AllTime)
            .unwrap(),
        2,
        "purge must stay applied after a crash before checkpoint \
         (tombstone replayed; purged data must NOT resurrect)"
    );
    assert!(
        db.community_purge_watermark(community).is_some(),
        "the purge watermark must be re-established by tombstone replay"
    );
    db.shutdown().unwrap();
}

// ── Task 4: created_at indexing parity across restart ────────────────────────

fn recency_schema() -> tidaldb::schema::Schema {
    let mut b = SchemaBuilder::new();
    let _ = b
        .signal(
            "view",
            EntityKind::Item,
            DecaySpec::Exponential {
                half_life: Duration::from_secs(7 * 24 * 3600),
            },
        )
        .windows(&[Window::AllTime])
        .velocity(false)
        .add();
    b.build().unwrap()
}

#[test]
fn created_at_default_is_consistent_across_restart() {
    let home = TempTidalHome::new().unwrap();

    // Write two items with NO explicit created_at, in order (1 then 2). Capture
    // the materialized created_at each got on the live write path.
    let (live_ts_1, live_ts_2);
    {
        let db = TidalDb::builder()
            .with_data_dir(home.path())
            .with_schema(recency_schema())
            .open()
            .unwrap();

        let mut meta1 = HashMap::new();
        meta1.insert("title".to_string(), "first".to_string());
        db.write_item_with_metadata(EntityId::new(1), &meta1)
            .unwrap();

        let mut meta2 = HashMap::new();
        meta2.insert("title".to_string(), "second".to_string());
        db.write_item_with_metadata(EntityId::new(2), &meta2)
            .unwrap();

        // The write path materializes the default into the PERSISTED metadata.
        live_ts_1 = created_at_of(&db, EntityId::new(1));
        live_ts_2 = created_at_of(&db, EntityId::new(2));

        // Item 1 was written first, so its recency timestamp must not be later
        // than item 2's: the live recency ordering is 1 <= 2.
        assert!(
            live_ts_1 <= live_ts_2,
            "live created_at ordering must be monotonic with write order"
        );

        db.shutdown().unwrap();
    }

    // Reopen. The rebuild path reads created_at from the SAME persisted metadata
    // via the shared indexer, so the recovered created_at must be byte-identical
    // to the live value — proving the write and rebuild paths cannot diverge.
    let db = TidalDb::builder()
        .with_data_dir(home.path())
        .with_schema(recency_schema())
        .open()
        .unwrap();

    let rebuilt_ts_1 = created_at_of(&db, EntityId::new(1));
    let rebuilt_ts_2 = created_at_of(&db, EntityId::new(2));

    assert_eq!(
        rebuilt_ts_1, live_ts_1,
        "item 1 created_at must be identical after restart (no default divergence)"
    );
    assert_eq!(
        rebuilt_ts_2, live_ts_2,
        "item 2 created_at must be identical after restart"
    );
    // And the recency ordering is preserved across the restart.
    assert!(
        rebuilt_ts_1 <= rebuilt_ts_2,
        "recency ordering must survive a restart unchanged"
    );

    db.shutdown().unwrap();
}

/// Read an item's persisted `created_at` as a `u64`, asserting it is present.
fn created_at_of(db: &TidalDb, id: EntityId) -> u64 {
    let meta = db
        .get_item_metadata(id)
        .unwrap()
        .expect("item metadata must exist");
    meta.get("created_at")
        .expect("write path must materialize a created_at into persisted metadata")
        .parse::<u64>()
        .expect("created_at must be a parseable u64")
}