//! Content-store garbage collection: TransitionFired subscriber, startup
//! sweep, and story-content backfill.
//!
//! Two purge paths clear the *ephemeral* `ContentKey::*` working-state keys
//! (gate output, respawn counters, merge-fixup flags, etc.) when a pipeline
//! item reaches a terminal stage (Done, Archived, Abandoned, Superseded,
//! Rejected):
//!
//! 1. **Subscriber** ([`spawn_content_gc_subscriber`]) — reacts to
//!    [`crate::pipeline_state::TransitionFired`] events and runs for new
//!    transitions in the current server session.
//!
//! 2. **Startup sweep** ([`sweep_zombie_content_on_startup`]) — cleans up
//!    zombie entries left over from sessions that predate the subscriber.
//!
//! `ContentKey::Story` — the story's markdown body — is deliberately
//! excluded from both paths (story 1222): reaching a terminal stage is not
//! deletion, and `show()` must keep returning content for Done/Archived
//! stories indefinitely. The story body is only fully purged
//! ([`purge_content_keys_for_story`]) when a story is genuinely tombstoned
//! (absent from the live CRDT index — `evict_item`/`purge_story`), which
//! the startup sweep still detects and cleans up.
//!
//! [`backfill_evicted_story_content`] is an idempotent, run-on-every-startup
//! repair pass that restores `ContentKey::Story` for any terminal-stage
//! story whose in-memory content was wiped by the pre-1222 purge bug, using
//! the durable SQLite `pipeline_items.content` shadow column (which the
//! purge never touched). Stories with no SQLite copy either are reported as
//! unrecoverable.

use crate::db::{ContentKey, all_content_ids, delete_content, get_shared_pool, write_content};
use crate::pipeline_state::{Pipeline, Stage, Status};
use crate::slog;
use crate::slog_warn;

/// Purge the eight ephemeral (non-body) `ContentKey` variants for
/// `story_id` — everything except `ContentKey::Story`.
///
/// Call this when a work item reaches a terminal stage: agent working-state
/// (gate output, respawn counters, merge-fixup flags) is no longer needed,
/// but the story's markdown body must be retained indefinitely so `show()`
/// keeps working (story 1222). Deletes for absent keys are no-ops.
pub(crate) fn purge_ephemeral_content_keys_for_story(story_id: &str) {
    delete_content(ContentKey::GateOutput(story_id));
    delete_content(ContentKey::AbortRespawnCount(story_id));
    delete_content(ContentKey::MergeMasterSpawnCount(story_id));
    delete_content(ContentKey::RunTestsOk(story_id));
    delete_content(ContentKey::CommitRecoveryPending(story_id));
    delete_content(ContentKey::MergeFixupPending(story_id));
    delete_content(ContentKey::MergeFailureKind(story_id));
    delete_content(ContentKey::MergeReport(story_id));
}

/// Purge every `ContentKey` variant for `story_id`, including the markdown
/// body (`ContentKey::Story`), from the in-memory content store.
///
/// Use this ONLY for stories that are genuinely gone — tombstoned / absent
/// from the live CRDT index via `evict_item`/`purge_story`. A terminal
/// pipeline stage (Done, Archived, …) is NOT deletion — use
/// [`purge_ephemeral_content_keys_for_story`] for that case so `show()`
/// keeps returning content (story 1222). Deletes for absent keys are no-ops.
pub(crate) fn purge_content_keys_for_story(story_id: &str) {
    delete_content(ContentKey::Story(story_id));
    purge_ephemeral_content_keys_for_story(story_id);
}

/// Spawn a background task that purges ephemeral content-store entries when a story reaches a terminal stage.
///
/// Subscribes to [`crate::pipeline_state::subscribe_transitions`]. On each
/// [`crate::pipeline_state::TransitionFired`] where `after` is `Done`,
/// `Archived`, `Abandoned`, `Superseded`, or `Rejected`, the eight ephemeral
/// `ContentKey::*` entries for that story are purged — the markdown body
/// (`ContentKey::Story`) is retained (story 1222). Lag events are logged as
/// warnings — a missed event leaves zombie entries that the next startup
/// sweep will remove.
pub(crate) fn spawn_content_gc_subscriber() {
    let mut rx = crate::pipeline_state::subscribe_transitions();
    tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(fired) => {
                    if is_terminal_stage(&fired.after) {
                        let story_id = &fired.story_id.0;
                        slog!(
                            "[content-gc] Story '{story_id}' reached terminal stage; \
                             purging ephemeral content-store entries (body retained)."
                        );
                        purge_ephemeral_content_keys_for_story(story_id);
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                    slog_warn!(
                        "[content-gc] Subscriber lagged, skipped {n} event(s). \
                         Zombie content-store entries will be cleaned by the next startup sweep."
                    );
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    });
}

/// One-shot startup sweep: purge content-store entries for stories that have
/// already reached terminal stages or are absent from the CRDT.
///
/// Genuinely tombstoned/absent stories get a full purge (including the
/// markdown body). Stories that are still live in the CRDT but sitting in a
/// terminal stage only have their ephemeral working-state keys purged — the
/// markdown body is retained so `show()` keeps working (story 1222).
///
/// Idempotent — safe to call more than once. Intended to clean up zombie
/// entries left over from server sessions that predate the GC subscriber.
pub(crate) fn sweep_zombie_content_on_startup() {
    let raw_keys = all_content_ids();

    // Extract unique base story IDs from raw content-store keys.
    // Raw key formats: `"{id}"` (Story) or `"{id}:{suffix}"` (compound keys).
    // Story IDs never contain `:`, so splitting on `:` is unambiguous.
    let mut story_ids: Vec<String> = raw_keys
        .iter()
        .map(|k| {
            k.split_once(':')
                .map(|(base, _)| base)
                .unwrap_or(k)
                .to_string()
        })
        .collect();
    story_ids.sort();
    story_ids.dedup();

    let mut swept = 0usize;
    for story_id in &story_ids {
        match crate::crdt_state::read_item(story_id) {
            // Tombstoned or absent from the live CRDT index — genuinely
            // gone, safe to purge everything including the story body.
            None => {
                purge_content_keys_for_story(story_id);
                swept += 1;
            }
            // Still a live item, just sitting in a terminal stage — retain
            // the story body (story 1222), only purge ephemeral state.
            Some(item) if is_terminal_stage(item.stage()) => {
                purge_ephemeral_content_keys_for_story(story_id);
                swept += 1;
            }
            Some(_) => {}
        }
    }

    if swept > 0 {
        slog!(
            "[content-gc] Startup sweep purged ephemeral content-store entries for \
             {swept} zombie story(s)."
        );
    }
}

/// Restore in-memory `ContentKey::Story` content for terminal-stage stories
/// whose markdown body was wiped by the pre-1222 purge bug (which deleted
/// `ContentKey::Story` on every terminal transition), using the durable
/// SQLite `pipeline_items.content` shadow column — `delete_content` only
/// ever touched the in-memory map, so the SQLite copy survived.
///
/// Idempotent and cheap to run on every startup: stories that already have
/// in-memory content are skipped. Returns the story IDs that could not be
/// recovered (also absent from the SQLite shadow column) — the caller logs
/// this as a report (story 1222, AC4).
pub(crate) async fn backfill_evicted_story_content() -> Vec<String> {
    let Some(pool) = get_shared_pool() else {
        return Vec::new();
    };
    let Some(items) = crate::crdt_state::read_all_items() else {
        return Vec::new();
    };

    let mut restored = 0usize;
    let mut unrecoverable = Vec::new();

    for item in items {
        if !is_terminal_stage(item.stage()) {
            continue;
        }
        let story_id = item.story_id();
        if crate::db::read_content(ContentKey::Story(story_id)).is_some() {
            continue;
        }

        let row: Result<Option<(Option<String>,)>, sqlx::Error> =
            sqlx::query_as("SELECT content FROM pipeline_items WHERE id = ?1")
                .bind(story_id)
                .fetch_optional(pool)
                .await;

        match row {
            Ok(Some((Some(content),))) => {
                write_content(ContentKey::Story(story_id), &content);
                restored += 1;
            }
            _ => unrecoverable.push(story_id.to_string()),
        }
    }

    if restored > 0 {
        slog!(
            "[content-gc] Backfill restored content for {restored} terminal-stage \
             story(s) from the SQLite shadow table."
        );
    }
    if !unrecoverable.is_empty() {
        slog_warn!(
            "[content-gc] Backfill could not recover content for {} story(s) — \
             compacted away with no SQLite copy: {}",
            unrecoverable.len(),
            unrecoverable.join(", ")
        );
    }

    unrecoverable
}

/// Return `true` when `stage` is one of the terminal pipeline classifications.
///
/// Story 1086: matches via the [`Status`] projection (Done / Abandoned /
/// Superseded / Rejected) plus [`Pipeline::Archived`] for plain archived items
/// (which carry `Status::Active`).  Future Stage variants automatically
/// participate by returning the appropriate Status / Pipeline from
/// [`Stage::status`] / [`Stage::pipeline`].
fn is_terminal_stage(stage: &Stage) -> bool {
    matches!(
        stage.status(),
        Status::Done | Status::Abandoned | Status::Superseded | Status::Rejected
    ) || matches!(stage.pipeline(), Pipeline::Archived)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::{
        ContentKey, ItemMeta, ensure_content_store, read_content, write_content,
        write_item_with_content,
    };

    /// Write all eight ContentKey variants for a story.
    fn seed_all_keys(story_id: &str) {
        write_content(ContentKey::Story(story_id), "body");
        write_content(ContentKey::GateOutput(story_id), "gate");
        write_content(ContentKey::AbortRespawnCount(story_id), "1");
        write_content(ContentKey::MergeMasterSpawnCount(story_id), "2");
        write_content(ContentKey::RunTestsOk(story_id), "ok");
        write_content(ContentKey::CommitRecoveryPending(story_id), "1");
        write_content(ContentKey::MergeFixupPending(story_id), "1");
        write_content(ContentKey::MergeFailureKind(story_id), r#""GatesFailed""#);
    }

    /// Assert all eight ContentKey variants for a story are absent from the store.
    fn assert_all_keys_absent(story_id: &str) {
        assert!(
            read_content(ContentKey::Story(story_id)).is_none(),
            "Story key must be absent"
        );
        assert!(
            read_content(ContentKey::GateOutput(story_id)).is_none(),
            "GateOutput key must be absent"
        );
        assert!(
            read_content(ContentKey::AbortRespawnCount(story_id)).is_none(),
            "AbortRespawnCount key must be absent"
        );
        assert!(
            read_content(ContentKey::MergeMasterSpawnCount(story_id)).is_none(),
            "MergeMasterSpawnCount key must be absent"
        );
        assert!(
            read_content(ContentKey::RunTestsOk(story_id)).is_none(),
            "RunTestsOk key must be absent"
        );
        assert!(
            read_content(ContentKey::CommitRecoveryPending(story_id)).is_none(),
            "CommitRecoveryPending key must be absent"
        );
        assert!(
            read_content(ContentKey::MergeFixupPending(story_id)).is_none(),
            "MergeFixupPending key must be absent"
        );
        assert!(
            read_content(ContentKey::MergeFailureKind(story_id)).is_none(),
            "MergeFailureKind key must be absent"
        );
    }

    /// Assert the seven ephemeral ContentKey variants seeded by
    /// `seed_all_keys` are absent, but `ContentKey::Story` (the markdown
    /// body) is still present — story 1222.
    fn assert_ephemeral_keys_absent_but_story_present(story_id: &str) {
        assert!(
            read_content(ContentKey::Story(story_id)).is_some(),
            "Story key must be RETAINED across a terminal-stage transition (story 1222)"
        );
        assert!(
            read_content(ContentKey::GateOutput(story_id)).is_none(),
            "GateOutput key must be absent"
        );
        assert!(
            read_content(ContentKey::AbortRespawnCount(story_id)).is_none(),
            "AbortRespawnCount key must be absent"
        );
        assert!(
            read_content(ContentKey::MergeMasterSpawnCount(story_id)).is_none(),
            "MergeMasterSpawnCount key must be absent"
        );
        assert!(
            read_content(ContentKey::RunTestsOk(story_id)).is_none(),
            "RunTestsOk key must be absent"
        );
        assert!(
            read_content(ContentKey::CommitRecoveryPending(story_id)).is_none(),
            "CommitRecoveryPending key must be absent"
        );
        assert!(
            read_content(ContentKey::MergeFixupPending(story_id)).is_none(),
            "MergeFixupPending key must be absent"
        );
        assert!(
            read_content(ContentKey::MergeFailureKind(story_id)).is_none(),
            "MergeFailureKind key must be absent"
        );
    }

    /// AC1: purge_content_keys_for_story removes all eight ContentKey namespaces.
    #[test]
    fn purge_clears_all_eight_content_key_namespaces() {
        ensure_content_store();
        let id = "996_test_purge_all";
        seed_all_keys(id);

        // Verify every key is present before the purge.
        assert!(read_content(ContentKey::Story(id)).is_some());
        assert!(read_content(ContentKey::GateOutput(id)).is_some());
        assert!(read_content(ContentKey::AbortRespawnCount(id)).is_some());
        assert!(read_content(ContentKey::MergeMasterSpawnCount(id)).is_some());
        assert!(read_content(ContentKey::RunTestsOk(id)).is_some());
        assert!(read_content(ContentKey::CommitRecoveryPending(id)).is_some());
        assert!(read_content(ContentKey::MergeFixupPending(id)).is_some());
        assert!(read_content(ContentKey::MergeFailureKind(id)).is_some());

        purge_content_keys_for_story(id);

        assert_all_keys_absent(id);
    }

    /// AC1: purge_content_keys_for_story is idempotent — calling it twice on an
    /// already-empty store is safe.
    #[test]
    fn purge_is_idempotent_when_keys_already_absent() {
        ensure_content_store();
        let id = "996_test_purge_idempotent";

        // Both calls must complete without panic.
        purge_content_keys_for_story(id);
        purge_content_keys_for_story(id);
        assert_all_keys_absent(id);
    }

    /// Story 1222, AC1: the GC subscriber reacts to an Abandoned terminal
    /// transition and purges ephemeral content-store entries for the story,
    /// but RETAINS the story body so `show()` keeps working.
    #[tokio::test]
    async fn subscriber_purges_ephemeral_keys_but_retains_story_content_on_terminal_transition() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        let story_id = "996_test_sub_terminal";

        write_item_with_content(
            story_id,
            "2_current",
            "---\nname: GC Sub Test\n---\n",
            ItemMeta::named("GC Sub Test"),
        );

        seed_all_keys(story_id);

        spawn_content_gc_subscriber();

        // Transition to Abandoned (a terminal stage).
        crate::agents::lifecycle::abandon_story(story_id).expect("abandon must succeed");

        // Give the subscriber task time to run.
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        assert_ephemeral_keys_absent_but_story_present(story_id);
    }

    /// AC4: the subscriber does NOT purge content for stories that remain in
    /// active (non-terminal) stages.
    #[tokio::test]
    async fn subscriber_does_not_purge_active_story_content() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        let active_id = "996_test_sub_active";
        let terminal_id = "996_test_sub_term_2";

        for id in [active_id, terminal_id] {
            write_item_with_content(
                id,
                "2_current",
                "---\nname: GC Active Test\n---\n",
                ItemMeta::named("GC Active Test"),
            );
            seed_all_keys(id);
        }

        spawn_content_gc_subscriber();

        // Only terminate one of the two stories.
        crate::agents::lifecycle::abandon_story(terminal_id).expect("abandon must succeed");

        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        // Terminal story's ephemeral keys must be gone, but body retained
        // (story 1222).
        assert_ephemeral_keys_absent_but_story_present(terminal_id);

        // Active story's main content key must still be present.
        assert!(
            read_content(ContentKey::Story(active_id)).is_some(),
            "active story content must not be purged"
        );
    }

    /// AC2: sweep_zombie_content_on_startup purges content for a tombstoned story.
    #[test]
    fn startup_sweep_purges_tombstoned_story_content() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        let story_id = "996_test_sweep_tombstone";
        write_item_with_content(
            story_id,
            "1_backlog",
            "---\nname: Sweep Tombstone Test\n---\n",
            ItemMeta::named("Sweep Tombstone Test"),
        );
        seed_all_keys(story_id);

        // Tombstone the item (evict_item drops only ContentKey::Story).
        crate::crdt_state::evict_item(story_id).expect("evict must succeed");

        // Re-seed all keys to simulate the zombie state evict_item leaves behind.
        seed_all_keys(story_id);

        // The startup sweep must purge the remaining zombie keys.
        sweep_zombie_content_on_startup();

        assert_all_keys_absent(story_id);
    }

    /// Story 1222, AC1 + AC3: a story that reaches Done but is still LIVE in
    /// the CRDT (not tombstoned) must keep its body across the startup sweep
    /// — only ephemeral working-state keys are purged.
    #[test]
    fn startup_sweep_retains_story_content_for_live_done_item() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        let story_id = "1222_test_sweep_live_done";
        write_item_with_content(
            story_id,
            "5_done",
            "---\nname: Live Done Test\n---\n",
            ItemMeta::named("Live Done Test"),
        );
        seed_all_keys(story_id);

        sweep_zombie_content_on_startup();

        assert_ephemeral_keys_absent_but_story_present(story_id);
    }

    /// Story 1222, AC1 + AC3: same as above but for Archived, the other
    /// stage explicitly named in the AC.
    #[test]
    fn startup_sweep_retains_story_content_for_live_archived_item() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        let story_id = "1222_test_sweep_live_archived";
        write_item_with_content(
            story_id,
            "6_archived",
            "---\nname: Live Archived Test\n---\n",
            ItemMeta::named("Live Archived Test"),
        );
        seed_all_keys(story_id);

        sweep_zombie_content_on_startup();

        assert_ephemeral_keys_absent_but_story_present(story_id);
    }

    /// AC2: sweep_zombie_content_on_startup leaves active stories' content intact.
    #[test]
    fn startup_sweep_preserves_active_story_content() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        let live_id = "996_test_sweep_live";
        write_item_with_content(
            live_id,
            "2_current",
            "---\nname: Live Story\n---\n",
            ItemMeta::named("Live Story"),
        );
        write_content(ContentKey::Story(live_id), "live content");

        sweep_zombie_content_on_startup();

        assert_eq!(
            read_content(ContentKey::Story(live_id)).as_deref(),
            Some("live content"),
            "active story content must survive the startup sweep"
        );
    }

    /// AC2: sweep_zombie_content_on_startup is idempotent.
    #[test]
    fn startup_sweep_is_idempotent() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        sweep_zombie_content_on_startup();
        sweep_zombie_content_on_startup();
    }

    /// Story 1222, AC4: with no shared SQLite pool initialised, the backfill
    /// is a safe no-op (returns no unrecoverable IDs, does not panic).
    #[tokio::test]
    async fn backfill_is_noop_without_shared_pool() {
        crate::crdt_state::init_for_test();
        ensure_content_store();

        let unrecoverable = backfill_evicted_story_content().await;
        assert!(unrecoverable.is_empty());
    }

    /// Story 1222, AC4: the backfill restores in-memory Story content for a
    /// terminal-stage story whose body was wiped from the in-memory store,
    /// using the durable SQLite shadow column that the (buggy, now-fixed)
    /// terminal-transition purge never touched.
    #[tokio::test]
    async fn backfill_restores_story_content_from_sqlite_shadow_column() {
        crate::crdt_state::init_for_test();
        ensure_content_store();
        crate::db::tests::ensure_shadow_db().await;

        let story_id = "1222_test_backfill_restore";
        let body = "---\nname: Backfill Restore Test\n---\n# Body\n";

        write_item_with_content(
            story_id,
            "5_done",
            body,
            ItemMeta::named("Backfill Restore Test"),
        );

        // Let the shadow-write background task flush the insert to SQLite.
        tokio::time::sleep(std::time::Duration::from_millis(150)).await;

        // Simulate the pre-1222 bug: wipe the in-memory Story key only.
        // SQLite still has it — delete_content never touches the shadow
        // table.
        delete_content(ContentKey::Story(story_id));
        assert!(read_content(ContentKey::Story(story_id)).is_none());

        let unrecoverable = backfill_evicted_story_content().await;

        assert!(
            !unrecoverable.contains(&story_id.to_string()),
            "story with a surviving SQLite copy must not be reported unrecoverable"
        );
        assert_eq!(
            read_content(ContentKey::Story(story_id)).as_deref(),
            Some(body),
            "backfill must restore the body from the SQLite shadow column"
        );
    }

    /// Story 1222, AC4: a terminal-stage story with no SQLite copy either
    /// (never shadow-written) is reported as unrecoverable, not silently
    /// dropped.
    #[tokio::test]
    async fn backfill_reports_unrecoverable_story_with_no_sqlite_copy() {
        crate::crdt_state::init_for_test();
        ensure_content_store();
        crate::db::tests::ensure_shadow_db().await;

        let story_id = "1222_test_backfill_unrecoverable";

        // Write directly to the CRDT only — bypass write_item_with_content
        // so no shadow-write message is ever sent, and never write
        // in-memory content either, simulating a story whose SQLite copy
        // was never captured (or was compacted away).
        crate::crdt_state::write_item_str(
            story_id,
            "5_done",
            Some("Unrecoverable Test"),
            None,
            None,
            None,
        );

        let unrecoverable = backfill_evicted_story_content().await;

        assert!(
            unrecoverable.contains(&story_id.to_string()),
            "story with no in-memory AND no SQLite content must be reported unrecoverable"
        );
    }
}