memnite 0.2.1

use std::collections::{BTreeMap, HashSet};
use std::io::{BufRead, Write};
use std::path::Path;

use memnite_core::{
    detect_conflicts, new_event_id, next_lamport, normalize_project, replay, Anchor, Event,
    EventKind, Log, MemoryData, MemoryPatch, Relation, Scope, Status,
};
use memnite_ingest::{slugify, IngestSource, IngestSummary};
use memnite_stale::hash_region;
use memnite_store::{EventMetaRow, MemoryRow, Store};

use crate::error::CliError;
use crate::spec::{
    AddSpec, DoctorReport, EventCtx, ImportSummary, RelationSpec, SearchQuery, UpdatePatch,
};

/// An open Memnite instance: the event log + its SQLite projection.
pub struct App {
    pub(crate) log: Log,
    pub(crate) store: Store,
}

impl App {
    /// Open (creating if needed) a Memnite data dir: `<dir>/log/` + `<dir>/memnite.db`.
    pub fn open(data_dir: impl AsRef<Path>) -> Result<Self, CliError> {
        let data_dir = data_dir.as_ref();
        let log = Log::open(data_dir.join("log"))?;
        let store = Store::open(data_dir.join("memnite.db"))?;
        Ok(App { log, store })
    }

    /// Create a memory: stamp anchor hashes from `root`, append a `MemoryAdded`
    /// event to the log, and project it. Returns the projected `MemoryRow`.
    pub fn add(&self, spec: AddSpec, root: &Path, ctx: &EventCtx) -> Result<MemoryRow, CliError> {
        let mut anchors = Vec::with_capacity(spec.anchors.len());
        for a in &spec.anchors {
            let content_hash = hash_region(root, &a.path, a.line_start, a.line_end)?;
            anchors.push(Anchor {
                path: a.path.clone(),
                symbol: a.symbol.clone(),
                line_start: a.line_start,
                line_end: a.line_end,
                content_hash,
            });
        }

        // Canonicalize the project once, up front: the event stores the canonical
        // name and the upsert key matches by it, so name variants collapse to one
        // bucket. Pure (no I/O, no new event) — resolution from the environment is
        // the caller's job (see project_resolve).
        let project = normalize_project(&spec.project);

        // topic_key upsert: if a live memory shares (topic_key, project, scope),
        // update it in place instead of adding a new one. Look up before the spec
        // fields are moved into MemoryData.
        let target = match &spec.topic_key {
            Some(t) => self.store.find_live_by_topic(t, &project, &spec.scope)?,
            None => None,
        };

        let data = MemoryData {
            title: spec.title,
            body: spec.body,
            mem_type: spec.mem_type,
            scope: spec.scope,
            project,
            topic_key: spec.topic_key,
            anchors,
            tags: spec.tags,
        };
        let (memory_id, kind) = match target {
            Some(id) => (id, EventKind::MemoryUpdated(data)),
            None => (
                format!("mem_{}", new_event_id()),
                EventKind::MemoryAdded(data),
            ),
        };

        let event = Event {
            event_id: new_event_id(),
            lamport: next_lamport(&self.log.read_all()?),
            ts: ctx.ts.clone(),
            engine: ctx.engine.clone(),
            machine: ctx.machine.clone(),
            memory_id: memory_id.clone(),
            kind,
        };

        self.log.append(&event)?;
        self.store.apply_event(&event)?;
        self.get(&memory_id)?
            .ok_or_else(|| CliError::NotFound(memory_id))
    }

    /// Persist a session summary as a memory (`mem_type = "session_summary"`).
    /// Model-driven: the caller passes the distilled text. Plain add — summaries
    /// accumulate as history (no topic_key upsert).
    pub fn session_summary(
        &self,
        summary: String,
        project: String,
        scope: Scope,
        ctx: &EventCtx,
    ) -> Result<MemoryRow, CliError> {
        let spec = AddSpec {
            title: format!("Session summary {}", ctx.ts),
            body: summary,
            mem_type: "session_summary".to_string(),
            scope,
            project,
            topic_key: None,
            anchors: vec![],
            tags: vec![],
        };
        // No anchors → `root` is unused by `add`; a placeholder path is safe.
        self.add(spec, Path::new("."), ctx)
    }

    /// Fetch one memory by id.
    pub fn get(&self, memory_id: &str) -> Result<Option<MemoryRow>, CliError> {
        Ok(self.store.get(memory_id)?)
    }

    /// Rebuild the projection from the full log. Returns the event count folded.
    pub fn rebuild(&self) -> Result<usize, CliError> {
        let events = self.log.read_all()?;
        self.store.rebuild(&events)?;
        Ok(events.len())
    }

    /// Full-text search over memories, narrowed by optional filters. User text
    /// is sanitized into FTS5 syntax; a blank query returns no results. Accepts
    /// anything convertible into `SearchQuery` (e.g. a plain `&str`).
    pub fn search(&self, query: impl Into<SearchQuery>) -> Result<Vec<MemoryRow>, CliError> {
        use crate::fts::sanitize_fts;
        let query = query.into();
        let fts = sanitize_fts(&query.text, query.match_any);
        if fts.is_empty() {
            return Ok(Vec::new());
        }
        let filters = memnite_store::SearchFilters {
            mem_type: query.mem_type,
            project: query.project.map(|p| normalize_project(&p)),
            scope: query.scope,
        };
        Ok(self.store.search(&fts, &filters)?)
    }

    /// All memories currently marked stale.
    pub fn list_stale(&self) -> Result<Vec<MemoryRow>, CliError> {
        Ok(self.store.list_stale()?)
    }

    /// Tombstone a memory (append a `MemoryDeleted` event). Errors if the id is
    /// unknown. Re-deleting an already-deleted memory is a no-op (delete is
    /// terminal in the fold), so no duplicate event is appended.
    pub fn delete(&self, memory_id: &str, ctx: &EventCtx) -> Result<(), CliError> {
        let current = self
            .store
            .get(memory_id)?
            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))?;
        if current.status == "deleted" {
            return Ok(());
        }
        let event = Event {
            event_id: new_event_id(),
            lamport: next_lamport(&self.log.read_all()?),
            ts: ctx.ts.clone(),
            engine: ctx.engine.clone(),
            machine: ctx.machine.clone(),
            memory_id: memory_id.to_string(),
            kind: EventKind::MemoryDeleted,
        };
        self.log.append(&event)?;
        self.store.apply_event(&event)?;
        Ok(())
    }

    /// Mark a memory as reviewed: append a `MemoryReviewed` event, resetting its
    /// decay clock (`updated_ts`) without changing its status. Errors `NotFound` if
    /// the id is unknown or already deleted (a tombstone cannot be reviewed).
    pub fn mark(&self, memory_id: &str, ctx: &EventCtx) -> Result<(), CliError> {
        let current = self
            .store
            .get(memory_id)?
            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))?;
        if current.status == "deleted" {
            return Err(CliError::NotFound(memory_id.to_string()));
        }
        let event = Event {
            event_id: new_event_id(),
            lamport: next_lamport(&self.log.read_all()?),
            ts: ctx.ts.clone(),
            engine: ctx.engine.clone(),
            machine: ctx.machine.clone(),
            memory_id: memory_id.to_string(),
            kind: EventKind::MemoryReviewed,
        };
        self.log.append(&event)?;
        self.store.apply_event(&event)?;
        Ok(())
    }

    /// Live memories whose decay verdict (given `now`, an RFC3339 timestamp) is
    /// `NeedsReview`, optionally narrowed by exact `project` / `mem_type`. Read-time
    /// only — nothing is written and the log is untouched.
    pub fn review_list(
        &self,
        now: &str,
        project: Option<&str>,
        mem_type: Option<&str>,
    ) -> Result<Vec<MemoryRow>, CliError> {
        use memnite_core::{decay_state, DecayState};
        // Canonicalize the filter so a name variant matches the stored (canonical) project.
        let project = project.map(normalize_project);
        let mut out = Vec::new();
        for row in self.store.list_all()? {
            if row.status == "deleted" {
                continue;
            }
            if project.as_deref().is_some_and(|p| row.project != p) {
                continue;
            }
            if mem_type.is_some_and(|t| row.mem_type != t) {
                continue;
            }
            if decay_state(&row.mem_type, &row.updated_ts, now) == DecayState::NeedsReview {
                out.push(row);
            }
        }
        Ok(out)
    }

    /// Assert a relation `from_id → to_id`. Appends a `RelationAsserted` event and
    /// projects it. Both memories must exist, be live (not deleted), and differ.
    /// `spec.judged_by` records who classified it ("agent" for interactive/manual,
    /// an engine name for the judge).
    pub fn relate(
        &self,
        from_id: &str,
        to_id: &str,
        spec: RelationSpec,
        ctx: &EventCtx,
    ) -> Result<(), CliError> {
        if from_id == to_id {
            return Err(CliError::Invalid(
                "cannot relate a memory to itself".to_string(),
            ));
        }
        let from = self
            .store
            .get(from_id)?
            .ok_or_else(|| CliError::NotFound(from_id.to_string()))?;
        let to = self
            .store
            .get(to_id)?
            .ok_or_else(|| CliError::NotFound(to_id.to_string()))?;
        if from.status == "deleted" || to.status == "deleted" {
            return Err(CliError::Invalid(format!(
                "cannot relate a deleted memory ({from_id} or {to_id})"
            )));
        }
        if !(0.0..=1.0).contains(&spec.confidence) {
            return Err(CliError::Invalid(format!(
                "confidence must be in [0.0, 1.0], got {}",
                spec.confidence
            )));
        }

        let event = Event {
            event_id: new_event_id(),
            lamport: next_lamport(&self.log.read_all()?),
            ts: ctx.ts.clone(),
            engine: ctx.engine.clone(),
            machine: ctx.machine.clone(),
            memory_id: from_id.to_string(),
            kind: EventKind::RelationAsserted {
                to_id: to_id.to_string(),
                relation: spec.relation,
                confidence: spec.confidence,
                reason: spec.reason,
                judged_by: spec.judged_by,
            },
        };
        self.log.append(&event)?;
        self.store.apply_event(&event)?;
        Ok(())
    }

    /// Relation edges touching a memory (as source or target).
    pub fn relations_for(
        &self,
        memory_id: &str,
    ) -> Result<Vec<memnite_store::RelationRow>, CliError> {
        Ok(self.store.relations_for(memory_id)?)
    }

    /// Every relation edge.
    pub fn list_relations(&self) -> Result<Vec<memnite_store::RelationRow>, CliError> {
        Ok(self.store.list_relations()?)
    }

    /// Candidate memories worth relating to `memory_id`: FTS over its title within
    /// the same project + scope, excluding self / deleted / already-related pairs.
    /// Uses OR-match (broad recall). Returns up to `limit` rows. Callers treat a
    /// failure as "no candidates" (non-blocking) at the surface layer.
    pub fn find_candidates(&self, memory_id: &str, limit: u32) -> Result<Vec<MemoryRow>, CliError> {
        use crate::fts::sanitize_fts;
        let Some(src) = self.store.get(memory_id)? else {
            return Ok(Vec::new());
        };
        let fts = sanitize_fts(&src.title, true);
        if fts.is_empty() {
            return Ok(Vec::new());
        }
        Ok(self
            .store
            .find_candidates(&fts, &src.project, &src.scope, memory_id, limit)?)
    }

    /// Batch scan: for each live memory, find candidates and ask the injected
    /// `judge` to classify each pair. Non-`not_conflict` verdicts are asserted as
    /// `RelationAsserted` events stamped with `judged_by` (provenance is the
    /// caller's, never a hardcoded literal — keeps the audit trail honest under a
    /// swapped judge). Judge errors are swallowed per pair (never abort); the first
    /// one is sampled into `ScanSummary::first_error`.
    ///
    /// `since` limits the scan to memories updated at/after it. It must be the same
    /// RFC3339 form as the stored `updated_ts` — the comparison is lexical, so a
    /// differently-formatted timestamp will filter incorrectly.
    pub fn conflicts_scan(
        &self,
        judge: impl Fn(
            &memnite_judge::MemoryView,
            &memnite_judge::MemoryView,
        ) -> Result<memnite_judge::Verdict, memnite_judge::JudgeError>,
        judged_by: &str,
        since: Option<String>,
        limit: u32,
        ctx: &EventCtx,
    ) -> Result<crate::spec::ScanSummary, CliError> {
        let mut summary = crate::spec::ScanSummary::default();
        let mut lamport = next_lamport(&self.log.read_all()?);

        for row in self.store.list_all()? {
            if row.status == "deleted" {
                continue;
            }
            if let Some(s) = &since {
                if row.updated_ts.as_str() < s.as_str() {
                    continue;
                }
            }
            for cand in self.find_candidates(&row.memory_id, limit)? {
                summary.pairs += 1;
                let a = memnite_judge::MemoryView {
                    id: row.memory_id.clone(),
                    title: row.title.clone(),
                    body: row.body.clone(),
                };
                let b = memnite_judge::MemoryView {
                    id: cand.memory_id.clone(),
                    title: cand.title.clone(),
                    body: cand.body.clone(),
                };
                match judge(&a, &b) {
                    Ok(v) if v.relation == Relation::NotConflict => summary.not_conflict += 1,
                    Ok(v) => {
                        let event = Event {
                            event_id: new_event_id(),
                            lamport,
                            ts: ctx.ts.clone(),
                            engine: ctx.engine.clone(),
                            machine: ctx.machine.clone(),
                            memory_id: row.memory_id.clone(),
                            kind: EventKind::RelationAsserted {
                                to_id: cand.memory_id.clone(),
                                relation: v.relation,
                                confidence: v.confidence,
                                reason: v.reason,
                                judged_by: judged_by.to_string(),
                            },
                        };
                        self.log.append(&event)?;
                        self.store.apply_event(&event)?;
                        lamport += 1;
                        summary.asserted += 1;
                    }
                    Err(e) => {
                        if summary.first_error.is_none() {
                            summary.first_error = Some(e.to_string());
                        }
                        summary.errors += 1;
                    }
                }
            }
        }
        Ok(summary)
    }

    /// Run the staleness checker over every anchored, non-deleted memory. When a
    /// memory's verdict changes its status, append the verdict event and project
    /// it. Memories without anchors (nothing to verify) are skipped.
    pub fn check(
        &self,
        root: &Path,
        ctx: &EventCtx,
    ) -> Result<crate::spec::CheckSummary, CliError> {
        use crate::spec::CheckSummary;
        use memnite_stale::{check_memory, verdict_to_kind, Verdict};

        let mut summary = CheckSummary::default();
        // Read the log once; track lamport locally (incremented per emitted event)
        // instead of re-reading the full log for every changed memory.
        let mut lamport = next_lamport(&self.log.read_all()?);

        for row in self.store.list_all()? {
            if row.status == "deleted" || row.anchors.is_empty() {
                continue;
            }

            let anchors: Vec<memnite_core::Anchor> = row
                .anchors
                .iter()
                .map(|a| memnite_core::Anchor {
                    path: a.path.clone(),
                    symbol: a.symbol.clone(),
                    line_start: a.line_start,
                    line_end: a.line_end,
                    content_hash: a.content_hash.clone(),
                })
                .collect();

            let verdict = check_memory(root, &anchors)?;
            let new_status = match verdict {
                Verdict::Stable => "stable",
                Verdict::Stale(_) => "stale",
            };

            if new_status == row.status {
                summary.unchanged += 1;
                continue;
            }

            let event = Event {
                event_id: new_event_id(),
                lamport,
                ts: ctx.ts.clone(),
                engine: ctx.engine.clone(),
                machine: ctx.machine.clone(),
                memory_id: row.memory_id.clone(),
                kind: verdict_to_kind(&verdict),
            };
            self.log.append(&event)?;
            self.store.apply_event(&event)?;
            lamport += 1;

            match verdict {
                Verdict::Stable => summary.stable += 1,
                Verdict::Stale(_) => summary.stale += 1,
            }
        }

        Ok(summary)
    }

    /// Recent memories for a project (newest-updated first, deleted excluded).
    pub fn context(&self, project: &str, limit: u32) -> Result<Vec<MemoryRow>, CliError> {
        Ok(self.store.context(&normalize_project(project), limit)?)
    }

    /// Event history of one memory, oldest first.
    pub fn timeline(&self, memory_id: &str) -> Result<Vec<EventMetaRow>, CliError> {
        Ok(self.store.timeline(memory_id)?)
    }

    /// Test-only: read the raw event log. Public for integration tests that must
    /// assert on event payloads not surfaced by the projection (e.g. tags).
    #[doc(hidden)]
    pub fn read_log_for_test(&self) -> Vec<Event> {
        self.log.read_all().unwrap_or_default()
    }

    /// Test-only: wipe the projection to an empty state without touching the log.
    /// Lets integration tests simulate projection/log divergence for `doctor`.
    #[doc(hidden)]
    pub fn rebuild_from_empty_for_test(&self) {
        self.store.rebuild(&[]).unwrap();
    }

    /// Compare the log fold against the projection and report divergences. A
    /// clean projection (after any normal op, since each op projects incrementally)
    /// returns zero mismatches and `cursor_ok == true`.
    pub fn doctor(&self) -> Result<DoctorReport, CliError> {
        let events = self.log.read_all()?;
        let folded = replay(&events);
        let projected = self.store.list_all()?;

        let proj: BTreeMap<&str, &MemoryRow> = projected
            .iter()
            .map(|m| (m.memory_id.as_str(), m))
            .collect();

        let mut mismatches = Vec::new();
        for (id, m) in &folded {
            match proj.get(id.as_str()) {
                None => mismatches.push(format!("{id}: in log, missing from projection")),
                Some(p) => {
                    let want = status_str(m.status);
                    if p.status != want {
                        mismatches.push(format!("{id}: status log={want} proj={}", p.status));
                    }
                    if p.title != m.title {
                        mismatches.push(format!("{id}: title differs"));
                    }
                    if p.last_event_id != m.last_event_id {
                        mismatches.push(format!("{id}: last_event_id differs"));
                    }
                }
            }
        }
        for p in &projected {
            if !folded.contains_key(&p.memory_id) {
                mismatches.push(format!("{}: in projection, missing from log", p.memory_id));
            }
        }

        let max_lamport = events.iter().map(|e| e.lamport).max().unwrap_or(0);
        let cursor_ok = self.store.sync_cursor()?.1 == max_lamport;
        let conflicts = detect_conflicts(&events);

        Ok(DoctorReport {
            log_count: folded.len(),
            proj_count: projected.len(),
            mismatches,
            cursor_ok,
            conflicts,
        })
    }

    /// Update a memory by patch: emits a sparse `MemoryPatched` delta containing
    /// only the fields `patch` provides — untouched fields are left `None` and
    /// carry no value over the wire, enabling field-level LWW merge downstream
    /// (see `memnite_core::replay`). The LOG FOLD (not the projection) is only
    /// consulted to validate the memory exists and is not deleted; it is not used
    /// to fill in omitted fields. Anchors in the patch are re-hashed from `root`;
    /// omitted anchors are left untouched. Errors if the id is unknown. Returns
    /// the new projected row.
    pub fn update(
        &self,
        memory_id: &str,
        patch: UpdatePatch,
        root: &Path,
        ctx: &EventCtx,
    ) -> Result<MemoryRow, CliError> {
        let events = self.log.read_all()?;
        let state = replay(&events);
        let base = state
            .get(memory_id)
            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))?;

        if base.status == Status::Deleted {
            return Err(CliError::Invalid(format!(
                "{memory_id} is deleted; cannot update"
            )));
        }

        // anchors: re-hash from `root` only if the patch provides them; omitted
        // (None) means "do not touch anchors".
        let anchors = match patch.anchors {
            Some(specs) => {
                let mut out = Vec::with_capacity(specs.len());
                for a in &specs {
                    let content_hash = hash_region(root, &a.path, a.line_start, a.line_end)?;
                    out.push(Anchor {
                        path: a.path.clone(),
                        symbol: a.symbol.clone(),
                        line_start: a.line_start,
                        line_end: a.line_end,
                        content_hash,
                    });
                }
                Some(out)
            }
            None => None,
        };

        let delta = MemoryPatch {
            title: patch.title,
            body: patch.body,
            mem_type: patch.mem_type,
            scope: patch.scope,
            project: patch.project.map(|p| normalize_project(&p)),
            topic_key: patch.topic_key,
            anchors,
            tags: patch.tags,
        };

        let event = Event {
            event_id: new_event_id(),
            lamport: next_lamport(&events),
            ts: ctx.ts.clone(),
            engine: ctx.engine.clone(),
            machine: ctx.machine.clone(),
            memory_id: memory_id.to_string(),
            kind: EventKind::MemoryPatched(delta),
        };
        self.log.append(&event)?;
        self.store.apply_event(&event)?;
        self.get(memory_id)?
            .ok_or_else(|| CliError::NotFound(memory_id.to_string()))
    }

    /// Import an NDJSON event bundle: append only events whose `event_id` is not
    /// already in the local log (dedup), then rebuild the projection from the
    /// full log. Rebuild — not incremental apply — because imported events may
    /// carry a `lamport` below the local max and must be re-folded in
    /// `(lamport, event_id)` order. Corrupt lines are skipped and counted.
    pub fn import(&self, reader: impl BufRead) -> Result<ImportSummary, CliError> {
        let mut events = self.log.read_all()?;
        let mut seen: HashSet<String> = events.iter().map(|e| e.event_id.clone()).collect();
        let mut sum = ImportSummary::default();
        for line in reader.lines() {
            let line = line?;
            if line.trim().is_empty() {
                continue;
            }
            match serde_json::from_str::<Event>(&line) {
                Err(_) => sum.corrupt += 1,
                Ok(ev) => {
                    if seen.contains(&ev.event_id) {
                        sum.duplicates += 1;
                    } else {
                        self.log.append(&ev)?;
                        seen.insert(ev.event_id.clone());
                        events.push(ev);
                        sum.imported += 1;
                    }
                }
            }
        }
        if sum.imported > 0 {
            self.store.rebuild(&events)?;
        }
        Ok(sum)
    }

    /// Write the event log as an NDJSON bundle (one event per line), sorted by
    /// `(lamport, event_id)` for deterministic, diff-friendly bundles. With
    /// `since`, only events whose `lamport >= since` are written. Returns the
    /// number of events written.
    pub fn export(&self, since: Option<u64>, mut writer: impl Write) -> Result<usize, CliError> {
        let mut evs = self.log.read_all()?;
        if let Some(s) = since {
            evs.retain(|e| e.lamport >= s);
        }
        evs.sort_by(|a, b| {
            a.lamport
                .cmp(&b.lamport)
                .then_with(|| a.event_id.cmp(&b.event_id))
        });
        for e in &evs {
            writeln!(writer, "{}", serde_json::to_string(e)?)?;
        }
        Ok(evs.len())
    }

    /// Ingest memories from any `IngestSource` (source-agnostic). Each item gets a
    /// deterministic `memory_id` (`mem_ing_<source>_<key-slug>`); a new id is
    /// added, a changed one (by projected title/body/type/topic) is updated, an
    /// unchanged one is skipped. Idempotent.
    pub fn ingest(
        &self,
        source: &dyn IngestSource,
        root: &Path,
        project: &str,
        ctx: &EventCtx,
    ) -> Result<IngestSummary, CliError> {
        let mems = source.collect(root)?;
        let mut summary = IngestSummary::default();
        // Read the log once; track the lamport locally (incremented only when an
        // event is actually emitted) instead of re-reading the log per item.
        let mut lamport = next_lamport(&self.log.read_all()?);
        for m in mems {
            let id = format!("mem_ing_{}_{}", source.name(), slugify(&m.source_key));
            let existing = self.store.get(&id)?;
            if let Some(row) = &existing {
                if row.title == m.title
                    && row.body == m.body
                    && row.mem_type == m.mem_type
                    && row.topic_key == m.topic_key
                {
                    summary.unchanged += 1;
                    continue;
                }
            }
            let data = MemoryData {
                title: m.title,
                body: m.body,
                mem_type: m.mem_type,
                scope: m.scope,
                project: project.to_string(),
                topic_key: m.topic_key,
                anchors: vec![],
                tags: vec![],
            };
            let kind = if existing.is_some() {
                EventKind::MemoryUpdated(data)
            } else {
                EventKind::MemoryAdded(data)
            };
            let event = Event {
                event_id: new_event_id(),
                lamport,
                ts: ctx.ts.clone(),
                engine: ctx.engine.clone(),
                machine: ctx.machine.clone(),
                memory_id: id,
                kind,
            };
            self.log.append(&event)?;
            self.store.apply_event(&event)?;
            lamport += 1;
            if existing.is_some() {
                summary.updated += 1;
            } else {
                summary.added += 1;
            }
        }
        Ok(summary)
    }
}

/// Core `Status` → the text form stored in the projection's `status` column.
fn status_str(s: Status) -> &'static str {
    match s {
        Status::Stable => "stable",
        Status::Stale => "stale",
        Status::Deleted => "deleted",
    }
}