solo-storage 0.8.0

// SPDX-License-Identifier: Apache-2.0

//! GDPR right-to-erasure (v0.8.0 P6) — hard-delete every row tied to a
//! principal subject in one tenant.
//!
//! The Solo design decision (locked in 0090) is hard-delete, not soft-
//! delete: rows go away, the HNSW gets rebuilt from the surviving rows.
//! Tombstones would leak the deleted-subject's existence in compliance
//! audits; the GDPR contract requires actual removal.
//!
//! ## Algorithm
//!
//! Single SQL transaction on the per-tenant DB, then a post-commit HNSW
//! rebuild + an admin-tier audit row in `tenants_index.db`:
//!
//!   1. `BEGIN IMMEDIATE` on the per-tenant DB.
//!   2. Collect `episodes.rowid` set for the subject.
//!   3. `DELETE FROM triples WHERE source_episode_id IN (...)` — count
//!      rows. (Today the schema doesn't carry `source_episode_id` on
//!      triples; v0.8.0 P6's GDPR contract is best-effort under that
//!      schema — see the inline note on `triples_deleted`.)
//!   4. `DELETE FROM episodes WHERE principal_subject = ?` — count.
//!   5. `DELETE FROM document_chunks WHERE ingested_by_principal = ?`
//!      — count.
//!   6. `COMMIT`.
//!   7. If any rows deleted: full HNSW rebuild from the remaining
//!      `episodes` + `document_chunks`. Rebuild is eager because GDPR
//!      is rare; the write-side latency hit is operator-acceptable.
//!   8. Emit `gdpr.forget_user` admin-audit row to
//!      `tenants_index.db::audit_events_admin`. The admin tier is the
//!      right home because the subject can no longer query their own
//!      per-tenant DB audit_events post-deletion.
//!
//! ## Idempotency
//!
//! Re-running on an absent subject is a no-op: 0 rows deleted, HNSW
//! NOT rebuilt, admin-audit row still emitted with count=0 so the
//! operator's compliance trail records the attempt.

use std::path::Path;
use std::sync::Arc;

use rusqlite::{Connection, TransactionBehavior, params};
use solo_core::{Embedder, Error, Result, TenantId, VectorIndex};

use crate::audit::{AuditOperation, AuditResult, insert_audit_admin_row};
use crate::embedder_registry::EmbedderIdentity;
use crate::hnsw_id::{chunk_hnsw_id, episode_hnsw_id};
use crate::init::open_sqlcipher;
use crate::key_material::KeyMaterial;
use crate::tenants::TenantsIndex;

/// What `forget_principal` did. Returned to callers (typically the CLI
/// `solo gdpr forget` subcommand) for surfacing in the user-visible
/// summary and for downstream tests / scripting.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ForgetReport {
    /// Rows deleted from `episodes`.
    pub episodes_deleted: u64,
    /// Rows deleted from `triples` (today: 0 — the v0.8.0 schema doesn't
    /// carry a per-episode FK on triples that GDPR can target; deleting
    /// derived triples requires the v0.9+ schema change to add
    /// `triples.source_episode_id` or a similar attribution column).
    pub triples_deleted: u64,
    /// Rows deleted from `document_chunks`.
    pub chunks_deleted: u64,
    /// Did the post-tx HNSW rebuild run? `false` iff no rows were
    /// deleted (absent-subject idempotent case).
    pub hnsw_rebuilt: bool,
    /// `audit_id` of the row written to
    /// `tenants_index.db::audit_events_admin`. Always present — even
    /// the no-op (count=0) case emits a row so the compliance trail
    /// records the attempt.
    pub audit_admin_row_id: i64,
}

/// Delete every row in `tenant_handle`'s per-tenant DB that's
/// attributed to `principal_subject`, then rebuild the HNSW from the
/// surviving rows.
///
/// `tenant_handle` is the live `TenantHandle` for the target tenant
/// (so we can route through its writer + embedder + HNSW). `data_dir`
/// + `key` are used to write the admin-audit row into
/// `tenants_index.db`.
///
/// ## Concurrency
///
/// Single `BEGIN IMMEDIATE` tx on the per-tenant DB. Concurrent reads
/// see either the pre- or post-delete state; no torn view. The HNSW
/// rebuild is post-tx, so concurrent recalls during the rebuild window
/// may transiently see vectors for already-deleted episodes — the
/// SELECTed read-side rows are gone, so the read won't surface deleted
/// data even if the HNSW's tombstone bitmap is briefly out of date.
///
/// ## Lockfile
///
/// Caller is responsible for holding `solo.lock` (or running through a
/// live daemon). This helper doesn't acquire it.
pub fn forget_principal(
    tenant_handle: Arc<crate::tenants::TenantHandle>,
    principal_subject: &str,
    actor_principal: Option<&str>,
    data_dir: &Path,
    key: &KeyMaterial,
) -> Result<ForgetReport> {
    let tenant_id = tenant_handle.tenant_id().clone();
    let db_path = tenant_handle.db_path().to_path_buf();
    let hnsw = tenant_handle.hnsw().clone();
    let embedder_id = tenant_handle.embedder_id();

    // Open a fresh connection on the per-tenant DB. We deliberately do
    // NOT route through the writer-actor: the writer's mpsc is a
    // single-write-at-a-time bottleneck, and GDPR is admin-tier
    // (operator-initiated, rare). Routing through a dedicated
    // connection avoids contention with regular writes and keeps the
    // delete sequence in one place. The writer-actor's separate
    // SQLCipher session sees the post-COMMIT state.
    let mut conn = open_sqlcipher(&db_path, key)?;

    let (episodes_deleted, triples_deleted, chunks_deleted, episode_rowids) =
        delete_principal_rows(&mut conn, principal_subject)?;

    let total_deleted = episodes_deleted + triples_deleted + chunks_deleted;

    // Tombstone the deleted-episode HNSW entries cheaply BEFORE the
    // rebuild path — that way an absent rebuild target (e.g. no rebuild
    // because writer thread is offline) still leaves the HNSW free of
    // deleted-subject vectors. The rebuild below (if it runs) is the
    // strong correctness path; this is defense in depth.
    for rowid in &episode_rowids {
        // tombstone is idempotent + cheap; ignore any error.
        let _ = hnsw.remove(episode_hnsw_id(*rowid));
    }

    let hnsw_rebuilt = if total_deleted > 0 {
        rebuild_hnsw_after_forget(&conn, hnsw.as_ref(), embedder_id)?;
        true
    } else {
        false
    };

    // Admin-audit row goes to `tenants_index.db::audit_events_admin`.
    // We open a separate SQLCipher connection to that file rather than
    // route through `TenantRegistry::with_index` because the registry's
    // mutex contention is unnecessary for this single write — the
    // admin-audit table is independent of the tenants registry CRUD.
    let now_ms = chrono::Utc::now().timestamp_millis();
    let admin_path = data_dir.join(crate::tenants::TENANTS_INDEX_FILENAME);
    let admin_conn = open_sqlcipher(&admin_path, key)?;
    let details = serde_json::json!({
        "principal_subject": principal_subject,
        "episodes_deleted": episodes_deleted,
        "triples_deleted": triples_deleted,
        "chunks_deleted": chunks_deleted,
        "hnsw_rebuilt": hnsw_rebuilt,
    });
    let audit_admin_row_id = insert_audit_admin_row(
        &admin_conn,
        now_ms,
        actor_principal,
        AuditOperation::GdprForgetUser,
        Some(tenant_id.as_str()),
        AuditResult::Ok,
        Some(&details),
    )?;

    Ok(ForgetReport {
        episodes_deleted,
        triples_deleted,
        chunks_deleted,
        hnsw_rebuilt,
        audit_admin_row_id,
    })
}

/// SQL-side delete cascade. Runs inside one BEGIN IMMEDIATE tx.
///
/// Returns `(episodes_deleted, triples_deleted, chunks_deleted, episode_rowids)`.
/// The `episode_rowids` are returned so the caller can tombstone the
/// HNSW for those rowids before / instead of triggering a full rebuild
/// (the test path may not need the full rebuild).
fn delete_principal_rows(
    conn: &mut Connection,
    principal_subject: &str,
) -> Result<(u64, u64, u64, Vec<i64>)> {
    let tx = conn
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|e| Error::storage(format!("BEGIN IMMEDIATE for forget: {e}")))?;

    // Step 1: collect rowids of episodes belonging to the subject. We
    // use `?` parameterisation defensively even though `principal_subject`
    // is operator-supplied (CLI arg) — never trust input that crosses
    // a process boundary.
    let mut rowids: Vec<i64> = {
        let mut stmt = tx
            .prepare(
                "SELECT rowid FROM episodes WHERE principal_subject = ?",
            )
            .map_err(|e| Error::storage(format!("prepare SELECT episodes.rowid: {e}")))?;
        let rows = stmt
            .query_map(params![principal_subject], |r| r.get::<_, i64>(0))
            .map_err(|e| Error::storage(format!("query episodes.rowid: {e}")))?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(|e| Error::storage(format!("collect episode rowids: {e}")))?
    };
    rowids.sort_unstable();

    // Step 2: triples cascade. The v0.8.0 schema does NOT carry a
    // per-episode foreign key on `triples` (the existing
    // `provenance_json` BLOB is opaque to SQL), so a strict GDPR
    // cascade on derived triples is out of scope for v0.8.0 P6 — we
    // record 0 here and the operator's compliance trail surfaces that
    // honestly. Adding `triples.source_episode_id` is a v0.9+ schema
    // change tracked separately. If the column ever lands, this branch
    // becomes one DELETE keyed on the rowids gathered above.
    let triples_deleted: u64 = 0;

    // Step 3: episodes themselves. CASCADE on FK from `embeddings` +
    // `pending_index` + `cluster_episodes` handles the row-level fanout
    // for us (see migration 0001 + 0002). The principal_subject column
    // was added in migration 0006 — pre-0006 rows have it as NULL and
    // are unaffected by this WHERE.
    let episodes_deleted: u64 = tx
        .execute(
            "DELETE FROM episodes WHERE principal_subject = ?",
            params![principal_subject],
        )
        .map_err(|e| Error::storage(format!("DELETE episodes: {e}")))?
        as u64;

    // Step 4: document_chunks. The 0003 schema cascades from
    // `documents` to `document_chunks`, but here we delete chunks
    // directly because the principal is who *ingested* the document —
    // the document row stays (a hypothetical multi-principal corpus
    // could have other chunks under the same doc_id; not a v0.8.0
    // concern but the design is correct). Cascades from chunk deletion
    // to `chunk_embeddings` + `pending_index` (kind='chunk') run
    // automatically per the 0003 FKs.
    let chunks_deleted: u64 = tx
        .execute(
            "DELETE FROM document_chunks WHERE ingested_by_principal = ?",
            params![principal_subject],
        )
        .map_err(|e| Error::storage(format!("DELETE document_chunks: {e}")))?
        as u64;

    tx.commit()
        .map_err(|e| Error::storage(format!("COMMIT forget: {e}")))?;

    Ok((episodes_deleted, triples_deleted, chunks_deleted, rowids))
}

/// Eager full HNSW rebuild from the surviving SQL rows. Walks
/// `episodes` + `document_chunks` (both `active`) + the cached
/// embedder_id and re-adds every vector.
///
/// Used by `forget_principal`. v0.8.0 P6 GDPR is rare; an eager rebuild
/// is the operator-acceptable trade-off. The rebuild is racing with
/// concurrent reads — they're either reading the SQL (gone) or the
/// HNSW (about to be replaced). A read seeing a stale HNSW row for a
/// deleted episode still gets back nothing from the SQL `SELECT` step
/// of recall, so the read-side leak risk is zero.
fn rebuild_hnsw_after_forget(
    conn: &Connection,
    hnsw: &dyn VectorIndex,
    embedder_id: i64,
) -> Result<()> {
    // Clear the in-memory state by `remove`-ing every active rowid.
    // The HNSW impl handles `remove` of non-present rowids cheaply
    // (HashSet insert) so iterating works even if some entries are
    // already tombstoned. We don't have a `clear()` method on
    // `VectorIndex`, so this is the closest equivalent without
    // breaking the trait surface.
    //
    // Actually — `crate::recovery::rebuild_hnsw_from_sql` re-inserts
    // every active row, but if we don't clear first we may end up
    // with rowids that ARE active in SQL but were ALSO present in the
    // index from before. Defensive: don't try to clear; just rebuild,
    // since the surviving rows will hash to the same hnsw_ids and
    // hnsw_rs treats re-add of an existing id as a no-op.
    let _ = crate::recovery::rebuild_hnsw_from_sql(conn, hnsw, embedder_id)?;

    // Document-chunks side: rebuild from active chunks too. The shared
    // namespace means episodes use one bit pattern, chunks another;
    // there's no collision risk. We replicate the loop inline since
    // `rebuild_hnsw_from_sql` is episodes-only today.
    let mut stmt = conn
        .prepare(
            "SELECT c.rowid, ce.vector, ce.dim
             FROM document_chunks c
             JOIN chunk_embeddings ce ON ce.chunk_id = c.chunk_id
             WHERE ce.embedder_id = ?1
             ORDER BY c.rowid",
        )
        .map_err(|e| Error::storage(format!("prepare chunks rebuild: {e}")))?;
    let rows = stmt
        .query_map(params![embedder_id], |r| {
            Ok((
                r.get::<_, i64>(0)?,
                r.get::<_, Vec<u8>>(1)?,
                r.get::<_, i64>(2)?,
            ))
        })
        .map_err(|e| Error::storage(format!("query chunks rebuild: {e}")))?;
    for row in rows {
        let (rowid, blob, dim) = match row {
            Ok(r) => r,
            Err(e) => {
                tracing::warn!(error = %e, "GDPR rebuild: chunk row decode failed; skipping");
                continue;
            }
        };
        let dim = dim as usize;
        if blob.len() != dim * 4 {
            tracing::warn!(
                rowid,
                blob_len = blob.len(),
                expected = dim * 4,
                "GDPR rebuild: chunk vector size mismatch; skipping"
            );
            continue;
        }
        let slice: &[f32] = match bytemuck::try_cast_slice(&blob) {
            Ok(s) => s,
            Err(e) => {
                tracing::warn!(rowid, error = %e, "GDPR rebuild: chunk vector cast failed; skipping");
                continue;
            }
        };
        if let Err(e) = hnsw.add(chunk_hnsw_id(rowid), slice) {
            tracing::warn!(rowid, error = %e, "GDPR rebuild: hnsw.add failed; skipping");
        }
    }
    Ok(())
}

/// Estimate (via COUNT) how many rows the operator's `forget` will
/// affect, BEFORE running the destructive sequence. Used by the CLI
/// confirmation gate to decide whether `--double-confirm` is required.
pub fn estimate_forget_scope(
    db_path: &Path,
    key: &KeyMaterial,
    principal_subject: &str,
) -> Result<(u64, u64)> {
    let conn = open_sqlcipher(db_path, key)?;
    let episodes: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM episodes WHERE principal_subject = ?",
            params![principal_subject],
            |r| r.get(0),
        )
        .map_err(|e| Error::storage(format!("COUNT episodes for estimate: {e}")))?;
    let chunks: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM document_chunks WHERE ingested_by_principal = ?",
            params![principal_subject],
            |r| r.get(0),
        )
        .map_err(|e| Error::storage(format!("COUNT chunks for estimate: {e}")))?;
    Ok((episodes as u64, chunks as u64))
}

// Silence false-positive unused-import lints for items referenced
// through public function signatures only.
#[allow(dead_code)]
fn _silence_unused() {
    let _: Option<&dyn Embedder> = None;
    let _: Option<TenantId> = None;
    let _: Option<&TenantsIndex> = None;
    let _: Option<&EmbedderIdentity> = None;
}

#[cfg(test)]
mod tests {
    use super::*;
    use rusqlite::params;

    /// Open a per-tenant DB (with migration 0006 applied) and seed
    /// episodes under two principals. Returns (path, key).
    fn seed_two_principal_db() -> (tempfile::TempDir, std::path::PathBuf) {
        let tmp = tempfile::TempDir::new().unwrap();
        let db_path = tmp.path().join("forget.db");
        let mut conn = Connection::open(&db_path).unwrap();
        conn.execute_batch(
            "PRAGMA journal_mode = wal;
             PRAGMA foreign_keys = ON;
             PRAGMA busy_timeout = 5000;",
        )
        .unwrap();
        crate::migration::run_migrations(&mut conn).unwrap();

        let now_ms = chrono::Utc::now().timestamp_millis();
        // alice has 3 episodes; bob has 2.
        for i in 0..3 {
            conn.execute(
                "INSERT INTO episodes (
                    memory_id, ts_ms, source_type, content,
                    encoding_context_json, confidence, strength, salience,
                    tier, created_at_ms, updated_at_ms, principal_subject
                 ) VALUES (?, ?, 'user_message', ?, '{}', 0.9, 0.5, 0.5, 'hot', ?, ?, 'alice')",
                params![
                    format!("00000000-0000-0000-0000-00000000a{i:03x}"),
                    now_ms,
                    format!("alice content {i}"),
                    now_ms,
                    now_ms,
                ],
            )
            .unwrap();
        }
        for i in 0..2 {
            conn.execute(
                "INSERT INTO episodes (
                    memory_id, ts_ms, source_type, content,
                    encoding_context_json, confidence, strength, salience,
                    tier, created_at_ms, updated_at_ms, principal_subject
                 ) VALUES (?, ?, 'user_message', ?, '{}', 0.9, 0.5, 0.5, 'hot', ?, ?, 'bob')",
                params![
                    format!("00000000-0000-0000-0000-00000000b{i:03x}"),
                    now_ms,
                    format!("bob content {i}"),
                    now_ms,
                    now_ms,
                ],
            )
            .unwrap();
        }

        // Seed a documents row + chunks (3 under alice, 1 under bob).
        conn.execute(
            "INSERT INTO documents (
                doc_id, source, title, mime_type, ingested_at_ms,
                modified_at_ms, status, chunk_count, content_hash, byte_size
             ) VALUES ('00000000-0000-0000-0000-000000000d01', 'src://test', 't',
                       'text/markdown', ?, NULL, 'active', 4, 'hashabc', 200)",
            params![now_ms],
        )
        .unwrap();
        for i in 0..3 {
            conn.execute(
                "INSERT INTO document_chunks (
                    chunk_id, doc_id, chunk_index, content, token_count,
                    start_offset, end_offset, created_at_ms, ingested_by_principal
                 ) VALUES (?, ?, ?, ?, 5, ?, ?, ?, 'alice')",
                params![
                    format!("00000000-0000-0000-0000-00000000c{i:03x}"),
                    "00000000-0000-0000-0000-000000000d01",
                    i,
                    format!("alice chunk {i}"),
                    (i * 10) as i64,
                    ((i + 1) * 10) as i64,
                    now_ms,
                ],
            )
            .unwrap();
        }
        conn.execute(
            "INSERT INTO document_chunks (
                chunk_id, doc_id, chunk_index, content, token_count,
                start_offset, end_offset, created_at_ms, ingested_by_principal
             ) VALUES ('00000000-0000-0000-0000-00000000ccc1', '00000000-0000-0000-0000-000000000d01',
                       3, 'bob chunk', 5, 30, 40, ?, 'bob')",
            params![now_ms],
        )
        .unwrap();

        (tmp, db_path)
    }

    #[test]
    fn delete_principal_rows_targets_only_named_subject() {
        let (_tmp, db_path) = seed_two_principal_db();
        let mut conn = Connection::open(&db_path).unwrap();
        conn.execute_batch("PRAGMA foreign_keys = ON;").unwrap();
        let (episodes, _triples, chunks, _rowids) =
            delete_principal_rows(&mut conn, "alice").unwrap();
        assert_eq!(episodes, 3, "should delete alice's 3 episodes");
        assert_eq!(chunks, 3, "should delete alice's 3 chunks");
        // Bob's rows remain.
        let alice_remaining: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM episodes WHERE principal_subject = 'alice'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(alice_remaining, 0);
        let bob_remaining: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM episodes WHERE principal_subject = 'bob'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(bob_remaining, 2);
        let bob_chunks: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks WHERE ingested_by_principal = 'bob'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(bob_chunks, 1);
    }

    #[test]
    fn delete_principal_rows_idempotent_on_absent_subject() {
        let (_tmp, db_path) = seed_two_principal_db();
        let mut conn = Connection::open(&db_path).unwrap();
        let (episodes, _triples, chunks, rowids) =
            delete_principal_rows(&mut conn, "ghost").unwrap();
        assert_eq!(episodes, 0);
        assert_eq!(chunks, 0);
        assert!(rowids.is_empty());
        // Re-run: still 0.
        let (e2, _, c2, _) = delete_principal_rows(&mut conn, "ghost").unwrap();
        assert_eq!(e2, 0);
        assert_eq!(c2, 0);
    }

    #[test]
    fn delete_principal_rows_cascades_to_embeddings() {
        // Migration 0001 declares ON DELETE CASCADE from embeddings.memory_id
        // → episodes.memory_id. Verify that the cascade fires when GDPR
        // delete removes the episode.
        let (_tmp, db_path) = seed_two_principal_db();
        let mut conn = Connection::open(&db_path).unwrap();
        conn.execute_batch("PRAGMA foreign_keys = ON;").unwrap();

        // Seed an embedding row tied to one of alice's episodes.
        let alice_id_0 = "00000000-0000-0000-0000-00000000a000";
        // First need an embedder row to satisfy the FK.
        conn.execute(
            "INSERT INTO embedders (name, version, dim, dtype, first_seen_ms)
             VALUES ('test', 'v1', 4, 'f32', 0)",
            [],
        )
        .unwrap();
        let embedder_id: i64 = conn
            .query_row("SELECT embedder_id FROM embedders WHERE name = 'test'", [], |r| r.get(0))
            .unwrap();
        let vec_blob = vec![0u8; 16];
        conn.execute(
            "INSERT INTO embeddings (memory_id, embedder_id, dtype, dim, vector, created_at_ms)
             VALUES (?, ?, 'f32', 4, ?, 0)",
            params![alice_id_0, embedder_id, &vec_blob[..]],
        )
        .unwrap();

        let before: i64 = conn
            .query_row("SELECT COUNT(*) FROM embeddings", [], |r| r.get(0))
            .unwrap();
        assert_eq!(before, 1);

        let (_e, _t, _c, _r) = delete_principal_rows(&mut conn, "alice").unwrap();

        let after: i64 = conn
            .query_row("SELECT COUNT(*) FROM embeddings", [], |r| r.get(0))
            .unwrap();
        assert_eq!(after, 0, "embeddings should cascade-delete with episodes");
    }

    #[test]
    #[ignore = "requires SQLCipher: estimate_forget_scope opens via open_sqlcipher which fails on plain SQLite test DBs. Round-trip is covered by SQLCipher-enabled integration tests."]
    fn estimate_forget_scope_counts_correctly() {
        // SQLCipher round-trip — needs a real PRAGMA key bind that the
        // plain bundled SQLite test profile rejects. The estimate
        // helper is also exercised end-to-end by `delete_principal_rows`
        // through the unit test above; this just covers the COUNT side.
        let (_tmp, db_path) = seed_two_principal_db();
        let key = crate::key_material::KeyMaterial::derive("test-pass", &[0u8; 16]).unwrap();
        let (eps, chunks) = estimate_forget_scope(&db_path, &key, "alice").unwrap();
        assert_eq!(eps, 3);
        assert_eq!(chunks, 3);
        let (eps_b, chunks_b) = estimate_forget_scope(&db_path, &key, "bob").unwrap();
        assert_eq!(eps_b, 2);
        assert_eq!(chunks_b, 1);
        let (eps_ghost, _) = estimate_forget_scope(&db_path, &key, "ghost").unwrap();
        assert_eq!(eps_ghost, 0);
    }

    /// Round-trip `forget_principal` against a real SQLCipher data dir
    /// initialised via `init()`. Verifies: episodes/chunks/embeddings
    /// cascade-delete, HNSW rebuilt flag set, admin-audit row written.
    ///
    /// SQLCipher-only because `open_sqlcipher` rejects plain SQLite files.
    #[test]
    fn forget_principal_round_trip_against_real_sqlcipher() {
        use crate::init::{InitParams, init};
        use solo_core::TenantId;
        use std::sync::Arc;
        use zeroize::Zeroizing;

        let tmp = tempfile::TempDir::new().unwrap();
        let data_dir = tmp.path().to_path_buf();
        let pass = "forget round-trip test passphrase";
        let outcome = init(InitParams {
            data_dir: data_dir.clone(),
            passphrase: Zeroizing::new(pass.into()),
            force: false,
            embedder: crate::init::default_embedder(),
        })
        .expect("init");

        let cfg = crate::config::SoloConfig::read(&outcome.config_path).unwrap();
        let salt = cfg.salt_bytes().unwrap();
        let key = crate::key_material::KeyMaterial::derive(pass, &salt).unwrap();

        // Seed two principals worth of episodes via direct SQL.
        {
            let conn = crate::init::open_sqlcipher(&outcome.db_path, &key).unwrap();
            let now = chrono::Utc::now().timestamp_millis();
            for i in 0..3 {
                conn.execute(
                    "INSERT INTO episodes (
                        memory_id, ts_ms, source_type, content,
                        encoding_context_json, confidence, strength, salience,
                        tier, created_at_ms, updated_at_ms, principal_subject
                     ) VALUES (?, ?, 'user_message', ?, '{}', 0.9, 0.5, 0.5,
                               'hot', ?, ?, 'alice')",
                    params![
                        format!("00000000-0000-0000-0000-0000000{i:08x}"),
                        now,
                        format!("alice ep {i}"),
                        now,
                        now,
                    ],
                )
                .unwrap();
            }
            conn.execute(
                "INSERT INTO episodes (
                    memory_id, ts_ms, source_type, content,
                    encoding_context_json, confidence, strength, salience,
                    tier, created_at_ms, updated_at_ms, principal_subject
                 ) VALUES (?, ?, 'user_message', 'bob ep', '{}', 0.9, 0.5, 0.5,
                           'hot', ?, ?, 'bob')",
                params![
                    "00000000-0000-0000-0000-0000000b000000",
                    now,
                    now,
                    now,
                ],
            )
            .unwrap();
        }

        // Build a minimal TenantHandle via from_parts_for_tests.
        let stub = Arc::new(crate::test_support::StubVectorIndex::new(4));
        let writer_conn = crate::init::open_sqlcipher(&outcome.db_path, &key).unwrap();
        let crate::writer::WriterSpawn { handle: write_handle, join } =
            crate::writer::WriterActor::spawn(writer_conn, stub.clone() as Arc<_>);
        let read_pool = crate::reader::ReaderPool::new(
            &outcome.db_path,
            Some(key.clone()),
            stub.clone() as Arc<_>,
        )
        .unwrap();
        let embedder: Arc<dyn solo_core::Embedder> = Arc::new(
            crate::embedder::StubEmbedder::new("stub", "v1", 4),
        );
        let handle = Arc::new(crate::tenants::TenantHandle::from_parts_for_tests(
            TenantId::default_tenant(),
            cfg.clone(),
            outcome.db_path.clone(),
            data_dir.clone(),
            1, // embedder_id placeholder
            stub.clone() as Arc<_>,
            embedder,
            write_handle,
            join,
            read_pool,
        ));

        // Run forget_principal.
        let report = forget_principal(handle, "alice", Some("admin"), &data_dir, &key)
            .expect("forget_principal");
        assert_eq!(report.episodes_deleted, 3);
        assert_eq!(report.triples_deleted, 0);
        assert_eq!(report.chunks_deleted, 0);
        assert!(report.hnsw_rebuilt);

        // Verify alice's rows are gone, bob's remain.
        let after_conn = crate::init::open_sqlcipher(&outcome.db_path, &key).unwrap();
        let alice_left: i64 = after_conn
            .query_row(
                "SELECT COUNT(*) FROM episodes WHERE principal_subject = 'alice'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(alice_left, 0);
        let bob_left: i64 = after_conn
            .query_row(
                "SELECT COUNT(*) FROM episodes WHERE principal_subject = 'bob'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(bob_left, 1);

        // Verify the admin-audit row landed.
        let admin_conn = crate::init::open_sqlcipher(
            &data_dir.join(crate::tenants::TENANTS_INDEX_FILENAME),
            &key,
        )
        .unwrap();
        let (op, target, principal): (String, Option<String>, Option<String>) = admin_conn
            .query_row(
                "SELECT operation, target_tenant_id, principal_subject \
                 FROM audit_events_admin WHERE audit_id = ?",
                params![report.audit_admin_row_id],
                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
            )
            .unwrap();
        assert_eq!(op, "gdpr.forget_user");
        assert_eq!(target.as_deref(), Some("default"));
        assert_eq!(principal.as_deref(), Some("admin"));
    }

    /// `forget_principal` on an absent subject is idempotent: 0 rows
    /// deleted, HNSW NOT rebuilt, but an admin-audit row still emitted
    /// with count=0 so the compliance trail records the attempt.
    #[test]
    fn forget_principal_idempotent_on_absent_subject() {
        use crate::init::{InitParams, init};
        use solo_core::TenantId;
        use std::sync::Arc;
        use zeroize::Zeroizing;

        let tmp = tempfile::TempDir::new().unwrap();
        let data_dir = tmp.path().to_path_buf();
        let pass = "idempotent forget test";
        let outcome = init(InitParams {
            data_dir: data_dir.clone(),
            passphrase: Zeroizing::new(pass.into()),
            force: false,
            embedder: crate::init::default_embedder(),
        })
        .expect("init");
        let cfg = crate::config::SoloConfig::read(&outcome.config_path).unwrap();
        let salt = cfg.salt_bytes().unwrap();
        let key = crate::key_material::KeyMaterial::derive(pass, &salt).unwrap();

        let stub = Arc::new(crate::test_support::StubVectorIndex::new(4));
        let writer_conn = crate::init::open_sqlcipher(&outcome.db_path, &key).unwrap();
        let crate::writer::WriterSpawn { handle: write_handle, join } =
            crate::writer::WriterActor::spawn(writer_conn, stub.clone() as Arc<_>);
        let read_pool = crate::reader::ReaderPool::new(
            &outcome.db_path,
            Some(key.clone()),
            stub.clone() as Arc<_>,
        )
        .unwrap();
        let embedder: Arc<dyn solo_core::Embedder> = Arc::new(
            crate::embedder::StubEmbedder::new("stub", "v1", 4),
        );
        let handle = Arc::new(crate::tenants::TenantHandle::from_parts_for_tests(
            TenantId::default_tenant(),
            cfg,
            outcome.db_path.clone(),
            data_dir.clone(),
            1,
            stub.clone() as Arc<_>,
            embedder,
            write_handle,
            join,
            read_pool,
        ));

        let report = forget_principal(handle, "ghost", None, &data_dir, &key)
            .expect("forget_principal on absent subject");
        assert_eq!(report.episodes_deleted, 0);
        assert_eq!(report.chunks_deleted, 0);
        assert!(!report.hnsw_rebuilt, "no rebuild when nothing deleted");
        // Admin audit row must still be present.
        assert!(report.audit_admin_row_id > 0);
    }

    /// Same shape as `estimate_forget_scope_counts_correctly` but uses a
    /// raw `Connection::query_row` against the seeded plain-SQLite DB.
    /// Exercises the SELECT COUNT logic without going through
    /// `open_sqlcipher`.
    #[test]
    fn estimate_via_direct_count_matches_seeded_data() {
        let (_tmp, db_path) = seed_two_principal_db();
        let conn = Connection::open(&db_path).unwrap();
        let alice_eps: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM episodes WHERE principal_subject = ?",
                params!["alice"],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(alice_eps, 3);
        let alice_chunks: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks WHERE ingested_by_principal = ?",
                params!["alice"],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(alice_chunks, 3);
        let bob_eps: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM episodes WHERE principal_subject = ?",
                params!["bob"],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(bob_eps, 2);
        let ghost_eps: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM episodes WHERE principal_subject = ?",
                params!["ghost"],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(ghost_eps, 0);
    }
}