fathomdb-engine 0.8.22

use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use fathomdb_embedder_api::{Embedder, EmbedderError, EmbedderIdentity, Vector};
use fathomdb_engine::lifecycle::ProjectionStatus;
use fathomdb_engine::{Engine, EngineError, EngineOpenError, PreparedWrite};
use fathomdb_schema::SQLITE_SUFFIX;
use tempfile::TempDir;

#[derive(Clone, Debug)]
struct SleepingEmbedder {
    identity: EmbedderIdentity,
    vector: Vector,
    delay: Duration,
    fail: bool,
}

impl SleepingEmbedder {
    fn success(delay: Duration) -> Self {
        Self {
            identity: EmbedderIdentity::new("deterministic", "rev-a", 384),
            vector: unit_vector(384),
            delay,
            fail: false,
        }
    }

    fn failing(delay: Duration) -> Self {
        Self { fail: true, ..Self::success(delay) }
    }
}

impl Embedder for SleepingEmbedder {
    fn identity(&self) -> EmbedderIdentity {
        self.identity.clone()
    }

    fn embed(&self, _text: &str) -> Result<Vector, EmbedderError> {
        thread::sleep(self.delay);
        if self.fail {
            Err(EmbedderError::Failed { message: "deterministic failure".to_string() })
        } else {
            Ok(self.vector.clone())
        }
    }
}

#[derive(Clone, Debug)]
struct RoutedEmbedder {
    identity: EmbedderIdentity,
}

impl RoutedEmbedder {
    fn new(dimension: u32) -> Self {
        Self { identity: EmbedderIdentity::new("routed", "rev-a", dimension) }
    }
}

impl Embedder for RoutedEmbedder {
    fn identity(&self) -> EmbedderIdentity {
        self.identity.clone()
    }

    fn embed(&self, text: &str) -> Result<Vector, EmbedderError> {
        let mut vector = vec![0.0_f32; self.identity.dimension as usize];
        vector[match text {
            "semantic-query" | "vector-only document" => 0,
            "hybrid" | "hybrid retrieval document" => 1,
            _ => 2,
        }] = 1.0;
        Ok(vector)
    }
}

fn fixture_path(name: &str) -> (TempDir, std::path::PathBuf) {
    let dir = TempDir::new().unwrap();
    let path = dir.path().join(format!("{name}{SQLITE_SUFFIX}"));
    (dir, path)
}

fn current_test_binary() -> std::path::PathBuf {
    std::env::current_exe().expect("test binary path")
}

fn wait_with_timeout(child: &mut Child, timeout: Duration) -> bool {
    let started = Instant::now();
    loop {
        if child.try_wait().expect("poll child").is_some() {
            return true;
        }
        if started.elapsed() >= timeout {
            let _ = child.kill();
            let _ = child.wait();
            return false;
        }
        thread::sleep(Duration::from_millis(10));
    }
}

fn wait_for_status(
    engine: &Engine,
    kind: &str,
    expected: ProjectionStatus,
    timeout: Duration,
) -> ProjectionStatus {
    let started = Instant::now();
    loop {
        let observed = engine.projection_status_for_test(kind).expect("projection status");
        if observed == expected {
            return observed;
        }
        assert!(
            started.elapsed() < timeout,
            "timed out waiting for {expected:?}; saw {observed:?}"
        );
        thread::sleep(Duration::from_millis(10));
    }
}

#[test]
fn ac_024a_second_open_rejects_quickly_with_pending_vector_work() {
    let (dir, path) = fixture_path("pending_lock");
    let embedder = Arc::new(SleepingEmbedder::success(Duration::from_millis(5)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");
    opened.engine.set_projection_scheduler_frozen_for_test(true);

    for i in 0..100 {
        opened
            .engine
            .write(&[PreparedWrite::Node {
                kind: "doc".to_string(),
                body: format!("doc {i}"),
                source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
                logical_id: None,
                state: fathomdb_engine::InitialState::Active,
                reason: None,
                valid_from: None,
                valid_until: None,
            }])
            .expect("write");
    }

    let mut child = Command::new(current_test_binary())
        .arg("--exact")
        .arg("child_second_open_locked_with_pending_vector")
        .arg("--ignored")
        .env("FATHOMDB_TEST_DB_PATH", &path)
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .expect("spawn child");

    let started = Instant::now();
    assert!(wait_with_timeout(&mut child, Duration::from_secs(1)));
    assert!(started.elapsed() <= Duration::from_secs(1));
    assert!(child.wait().expect("child status").success());
    drop(opened);
    drop(dir);
}

#[test]
#[ignore]
fn child_second_open_locked_with_pending_vector() {
    let path = std::env::var_os("FATHOMDB_TEST_DB_PATH").expect("db path");
    let started = Instant::now();
    let err = Engine::open(path).expect_err("second open must fail");
    assert!(started.elapsed() <= Duration::from_secs(1));
    assert!(matches!(err, EngineOpenError::DatabaseLocked { .. }));
}

#[test]
fn ac_025_drop_with_pending_vector_work_returns_promptly() {
    let (_dir, path) = fixture_path("drop_pending");
    let embedder = Arc::new(SleepingEmbedder::success(Duration::from_millis(5)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");
    opened.engine.set_projection_scheduler_frozen_for_test(true);

    for i in 0..1_000 {
        opened
            .engine
            .write(&[PreparedWrite::Node {
                kind: "doc".to_string(),
                body: format!("doc {i}"),
                source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
                logical_id: None,
                state: fathomdb_engine::InitialState::Active,
                reason: None,
                valid_from: None,
                valid_until: None,
            }])
            .expect("write");
    }

    let started = Instant::now();
    drop(opened);
    assert!(started.elapsed() <= Duration::from_secs(30));
}

#[test]
fn ac_029_canonical_writes_complete_under_projection_stall() {
    let (_dir, path) = fixture_path("projection_stall");
    let embedder = Arc::new(SleepingEmbedder::success(Duration::from_millis(1)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");

    // Denoise both 1,000-write wall-clock measurements with a best-of-K minimum: a single
    // noisy sample on a shared CI runner must not inflate the ratio. (0.8.9 Slice 20, F-9 —
    // this acceptance test first executed on macOS only once the pyo3 link gate was removed,
    // exposing a latent flake.) The locked P-STALL-TOL = 1.5x contract (dev/acceptance.md
    // AC-029) is UNCHANGED — only the measurement is hardened. The minimum reflects each loop's
    // true least-interfered throughput; a real back-pressure regression couples canonical
    // writes to the frozen scheduler and blows the ratio out (an effective hang of tens of
    // seconds), which the minimum still catches.
    const K: usize = 3;

    let mut baseline = Duration::MAX;
    for trial in 0..K {
        let started = Instant::now();
        for i in 0..1_000 {
            opened
                .engine
                .write(&[PreparedWrite::Node {
                    kind: "doc".to_string(),
                    body: format!("baseline {trial}-{i}"),
                    source_id: fathomdb_engine::SourceId::new("test:fixture")
                        .expect("test source id"),
                    logical_id: None,
                    state: fathomdb_engine::InitialState::Active,
                    reason: None,
                    valid_from: None,
                    valid_until: None,
                }])
                .expect("baseline write");
        }
        baseline = baseline.min(started.elapsed());
        opened.engine.drain(30_000).expect("baseline drain");
    }

    opened.engine.set_projection_scheduler_frozen_for_test(true);

    let mut stalled = Duration::MAX;
    for trial in 0..K {
        let started = Instant::now();
        for i in 0..1_000 {
            opened
                .engine
                .write(&[PreparedWrite::Node {
                    kind: "doc".to_string(),
                    body: format!("stalled {trial}-{i}"),
                    source_id: fathomdb_engine::SourceId::new("test:fixture")
                        .expect("test source id"),
                    logical_id: None,
                    state: fathomdb_engine::InitialState::Active,
                    reason: None,
                    valid_from: None,
                    valid_until: None,
                }])
                .expect("stalled write");
        }
        stalled = stalled.min(started.elapsed());
    }

    // P-STALL-TOL = 1.5x (dev/acceptance.md AC-029): a frozen projection scheduler must not
    // back-pressure canonical writes.
    assert!(
        stalled <= baseline.mul_f32(1.5),
        "P-STALL-TOL violated: baseline={baseline:?} stalled={stalled:?}"
    );
}

#[test]
fn ac_031_hybrid_search_surfaces_vector_soft_fallback_when_projection_lags() {
    let (_dir, path) = fixture_path("soft_fallback");
    let embedder = Arc::new(SleepingEmbedder::success(Duration::from_millis(1)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");
    opened.engine.set_projection_scheduler_frozen_for_test(true);

    let receipt = opened
        .engine
        .write(&[PreparedWrite::Node {
            kind: "doc".to_string(),
            body: "phase nine hybrid search".to_string(),
            source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
            logical_id: None,
            state: fathomdb_engine::InitialState::Active,
            reason: None,
            valid_from: None,
            valid_until: None,
        }])
        .expect("write");

    let result = opened.engine.search("hybrid").expect("search");
    let bodies: Vec<String> = result.results.iter().map(|h| h.body.clone()).collect();
    assert_eq!(bodies, vec!["phase nine hybrid search".to_string()]);
    assert_eq!(result.projection_cursor + 1, receipt.cursor);
    assert_eq!(
        result.soft_fallback,
        Some(fathomdb_engine::SoftFallback { branch: fathomdb_engine::SoftFallbackBranch::Vector })
    );
}

#[test]
fn hybrid_search_returns_vector_results_when_text_branch_has_no_match() {
    let (_dir, path) = fixture_path("vector_materialization");
    let embedder = Arc::new(RoutedEmbedder::new(8));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");

    opened
        .engine
        .write(&[PreparedWrite::Node {
            kind: "doc".to_string(),
            body: "vector-only document".to_string(),
            source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
            logical_id: None,
            state: fathomdb_engine::InitialState::Active,
            reason: None,
            valid_from: None,
            valid_until: None,
        }])
        .expect("write");
    opened.engine.drain(10_000).expect("drain");

    let result = opened.engine.search("semantic-query").expect("search");
    let bodies: Vec<String> = result.results.iter().map(|h| h.body.clone()).collect();
    assert_eq!(bodies, vec!["vector-only document".to_string()]);
    assert_eq!(result.soft_fallback, None);
}

#[test]
fn hybrid_search_deduplicates_rows_seen_by_text_and_vector_branches() {
    let (_dir, path) = fixture_path("hybrid_dedup");
    let embedder = Arc::new(RoutedEmbedder::new(8));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");

    opened
        .engine
        .write(&[PreparedWrite::Node {
            kind: "doc".to_string(),
            body: "hybrid retrieval document".to_string(),
            source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
            logical_id: None,
            state: fathomdb_engine::InitialState::Active,
            reason: None,
            valid_from: None,
            valid_until: None,
        }])
        .expect("write");
    opened.engine.drain(10_000).expect("drain");

    let result = opened.engine.search("hybrid").expect("search");
    let bodies: Vec<String> = result.results.iter().map(|h| h.body.clone()).collect();
    assert_eq!(bodies, vec!["hybrid retrieval document".to_string()]);
    assert_eq!(result.soft_fallback, None);
}

#[test]
fn ac_032a_drain_succeeds_when_timeout_is_sufficient() {
    let (_dir, path) = fixture_path("drain_success");
    let embedder = Arc::new(SleepingEmbedder::success(Duration::from_millis(20)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");

    let mut last_cursor = 0;
    for i in 0..10 {
        last_cursor = opened
            .engine
            .write(&[PreparedWrite::Node {
                kind: "doc".to_string(),
                body: format!("doc {i}"),
                source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
                logical_id: None,
                state: fathomdb_engine::InitialState::Active,
                reason: None,
                valid_from: None,
                valid_until: None,
            }])
            .expect("write")
            .cursor;
    }

    let started = Instant::now();
    opened.engine.drain(10_000).expect("drain");
    assert!(started.elapsed() <= Duration::from_secs(10));
    assert!(opened.engine.search("doc").unwrap().projection_cursor >= last_cursor);
}

#[test]
fn ac_032b_drain_returns_typed_timeout_when_work_does_not_finish() {
    let (_dir, path) = fixture_path("drain_timeout");
    // Each embed sleeps longer than the drain timeout T (below), so the queued work cannot
    // finish within T regardless of worker-pool size — the drain must time out. (Using a
    // per-embed sleep > T keeps the timeout deterministic where a fixed small doc count + a
    // sub-T embed would race a wider pool.)
    let embedder = Arc::new(SleepingEmbedder::success(Duration::from_secs(2)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");

    for i in 0..10 {
        opened
            .engine
            .write(&[PreparedWrite::Node {
                kind: "doc".to_string(),
                body: format!("doc {i}"),
                source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
                logical_id: None,
                state: fathomdb_engine::InitialState::Active,
                reason: None,
                valid_from: None,
                valid_until: None,
            }])
            .expect("write");
    }

    // P-DRAIN-TOL = 1.5x (dev/acceptance.md AC-032b): with timeout T, a typed timeout must be
    // returned within 1.5 x T. Use the DOCUMENTED T = 1s (acceptance.md uses drain(timeout=1s))
    // rather than an ad-hoc 100ms; the per-embed sleep (2s) exceeds T so the drain reliably
    // times out, and the 1.5s bound tolerates shared-CI scheduling jitter. The earlier 100ms T
    // (→ 150ms bound, 50ms slack) flaked on macOS once the pyo3 link gate was removed
    // (0.8.9 Slice 20, F-9). Contract UNCHANGED — only T is set to its documented value.
    const DRAIN_TIMEOUT_MS: u64 = 1_000;
    let started = Instant::now();
    let err = opened.engine.drain(DRAIN_TIMEOUT_MS).expect_err("drain must time out");
    let elapsed = started.elapsed();
    assert_eq!(err, EngineError::Scheduler);
    // 1.5 x T, computed as T * 3 / 2 to avoid float rounding.
    assert!(
        elapsed <= Duration::from_millis(DRAIN_TIMEOUT_MS * 3 / 2),
        "P-DRAIN-TOL violated: elapsed={elapsed:?}"
    );
}

#[test]
fn ac_033_provenance_growth_is_bounded_and_oldest_first() {
    let (_dir, path) = fixture_path("retention");
    let opened = Engine::open(&path).expect("open");
    opened.engine.set_provenance_row_cap_for_test(Some(10));
    opened
        .engine
        .write(&[PreparedWrite::AdminSchema {
            name: "audit".to_string(),
            kind: "append_only_log".to_string(),
            schema_json: "{}".to_string(),
            retention_json: "{}".to_string(),
        }])
        .expect("register collection");

    let mut crossed = false;
    for i in 0..30 {
        opened
            .engine
            .write(&[PreparedWrite::OpStore {
                collection: "audit".to_string(),
                record_key: format!("{i:03}"),
                schema_id: None,
                body: format!(r#"{{"i":{i}}}"#),
            }])
            .expect("append mutation");
        let row_count = opened.engine.provenance_row_count_for_test().expect("row count");
        if row_count >= 10 {
            crossed = true;
            assert!(row_count <= 11, "row_count={row_count}");
        }
    }

    assert!(crossed, "fixture must cross the configured cap");
    assert_eq!(
        opened.engine.oldest_provenance_record_key_for_test("audit").unwrap(),
        Some("020".to_string())
    );
}

#[test]
fn ac_063a_exhausted_projection_failure_is_recorded_once_and_vector_stays_absent() {
    let (_dir, path) = fixture_path("projection_failure");
    let embedder = Arc::new(SleepingEmbedder::failing(Duration::from_millis(1)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");
    opened.engine.set_projection_retry_delays_for_test(&[0, 0, 0]);

    let cursor = opened
        .engine
        .write(&[PreparedWrite::Node {
            kind: "doc".to_string(),
            body: "will fail projection".to_string(),
            source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
            logical_id: None,
            state: fathomdb_engine::InitialState::Active,
            reason: None,
            valid_from: None,
            valid_until: None,
        }])
        .expect("write")
        .cursor;

    opened.engine.drain(10_000).expect("drain");
    assert_eq!(
        wait_for_status(&opened.engine, "doc", ProjectionStatus::Failed, Duration::from_secs(2)),
        ProjectionStatus::Failed
    );
    assert_eq!(opened.engine.projection_failure_count_for_test(cursor).unwrap(), 1);
    assert!(!opened.engine.has_vector_for_cursor_for_test(cursor).unwrap());
    assert!(opened.engine.search("will").unwrap().projection_cursor >= cursor);
}

#[test]
fn ac_063b_restart_does_not_retry_terminal_projection_failures() {
    let (_dir, path) = fixture_path("projection_failure_restart");
    let failing = Arc::new(SleepingEmbedder::failing(Duration::from_millis(1)));
    let opened = Engine::open_with_embedder_for_test(&path, failing).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("vector kind");
    opened.engine.set_projection_retry_delays_for_test(&[0, 0, 0]);

    let cursor = opened
        .engine
        .write(&[PreparedWrite::Node {
            kind: "doc".to_string(),
            body: "still failed after restart".to_string(),
            source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
            logical_id: None,
            state: fathomdb_engine::InitialState::Active,
            reason: None,
            valid_from: None,
            valid_until: None,
        }])
        .expect("write")
        .cursor;
    opened.engine.drain(10_000).expect("drain");
    opened.engine.close().expect("close");

    let good = Arc::new(SleepingEmbedder::success(Duration::from_millis(1)));
    let reopened = Engine::open_with_embedder_for_test(&path, good).expect("reopen");
    thread::sleep(Duration::from_millis(200));

    assert_eq!(reopened.engine.projection_failure_count_for_test(cursor).unwrap(), 1);
    assert!(!reopened.engine.has_vector_for_cursor_for_test(cursor).unwrap());
    assert_eq!(
        reopened.engine.projection_status_for_test("doc").unwrap(),
        ProjectionStatus::Failed
    );
}

#[test]
fn projection_status_is_tracked_per_kind_not_just_global_cursor() {
    let (_dir, path) = fixture_path("projection_status_per_kind");
    let embedder = Arc::new(SleepingEmbedder::success(Duration::from_millis(1)));
    let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
    opened.engine.configure_vector_kind_for_test("doc").expect("doc vector kind");
    opened.engine.configure_vector_kind_for_test("note").expect("note vector kind");

    opened
        .engine
        .write(&[PreparedWrite::Node {
            kind: "note".to_string(),
            body: "already projected".to_string(),
            source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
            logical_id: None,
            state: fathomdb_engine::InitialState::Active,
            reason: None,
            valid_from: None,
            valid_until: None,
        }])
        .expect("note write");
    opened.engine.drain(10_000).expect("note drain");
    opened.engine.set_projection_scheduler_frozen_for_test(true);
    opened
        .engine
        .write(&[PreparedWrite::Node {
            kind: "doc".to_string(),
            body: "still pending".to_string(),
            source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
            logical_id: None,
            state: fathomdb_engine::InitialState::Active,
            reason: None,
            valid_from: None,
            valid_until: None,
        }])
        .expect("doc write");

    assert_eq!(
        opened.engine.projection_status_for_test("note").unwrap(),
        ProjectionStatus::UpToDate
    );
    assert_eq!(opened.engine.projection_status_for_test("doc").unwrap(), ProjectionStatus::Pending);
}

fn unit_vector(dimension: usize) -> Vector {
    let mut values = vec![0.0_f32; dimension];
    if dimension > 0 {
        values[0] = 1.0;
    }
    values
}