solo-storage 0.8.1

// SPDX-License-Identifier: Apache-2.0

//! v0.8.0 P2 tests: `TenantHandle` lifecycle + `TenantRegistry`
//! lazy-load + per-tenant isolation. These exercise the registry +
//! handle abstraction end-to-end against real SQLCipher-encrypted
//! tenant DBs.

#![cfg(test)]

use crate::config::EmbedderConfig;
use crate::init::{InitParams, init};
use crate::key_material::KeyMaterial;
use crate::tenants::{TenantRegistry, TenantRegistryParams};
use crate::vector_index::HnswParams;
use solo_core::{Embedder, TenantId};
use std::sync::Arc;
use zeroize::Zeroizing;

fn fresh_init_dir(passphrase: &str) -> (tempfile::TempDir, KeyMaterial) {
    let tmp = tempfile::TempDir::new().unwrap();
    let _ = init(InitParams {
        data_dir: tmp.path().to_path_buf(),
        passphrase: Zeroizing::new(passphrase.into()),
        force: false,
        embedder: EmbedderConfig {
            name: "stub".into(),
            version: "v1".into(),
            dim: 32,
            dtype: "f32".into(),
        },
    })
    .unwrap();
    let cfg = crate::config::SoloConfig::read(&tmp.path().join("solo.config.toml")).unwrap();
    let key = KeyMaterial::derive(passphrase, &cfg.salt_bytes().unwrap()).unwrap();
    (tmp, key)
}

fn stub_embedder() -> Arc<dyn Embedder> {
    Arc::new(crate::embedder::StubEmbedder::new("stub", "v1", 32))
}

fn rt() -> tokio::runtime::Runtime {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap()
}

fn make_registry(
    data_dir: &std::path::Path,
    key: &KeyMaterial,
    runtime: &tokio::runtime::Runtime,
) -> Arc<TenantRegistry> {
    let _ = runtime;
    Arc::new(
        TenantRegistry::open(TenantRegistryParams {
            data_dir: data_dir.to_path_buf(),
            key: key.clone(),
            embedder: stub_embedder(),
            hnsw_params: HnswParams::default(),
            steward: None,
            runtime_handle: Some(runtime.handle().clone()),
        })
        .expect("open registry"),
    )
}

#[test]
fn tenant_registry_open_lists_default_tenant() {
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        let listed = registry.list_active().await.unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].tenant_id, TenantId::default_tenant());
    });
}

#[test]
fn tenant_handle_lifecycle_open_remember_recall_reopen() {
    use solo_core::{Confidence, EncodingContext, Episode, MemoryId, Tier};
    use chrono::Utc;

    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        let default_id = TenantId::default_tenant();
        let mid = {
            let h = registry.get_or_open(&default_id).await.unwrap();
            // Write one episode.
            let ep = Episode {
                memory_id: MemoryId::new(),
                ts_ms: Utc::now().timestamp_millis(),
                source_type: "user_message".into(),
                source_id: None,
                content: "lifecycle test".into(),
                encoding_context: EncodingContext::default(),
                provenance: None,
                confidence: Confidence::new(0.9).unwrap(),
                strength: 0.5,
                salience: 0.5,
                tier: Tier::Hot,
            };
            let emb = stub_embedder().embed("lifecycle test").await.unwrap();
            h.write().remember(ep, emb).await.unwrap()
        };
        // Fetch via reader pool to confirm durability.
        let mid_str = mid.to_string();
        let h = registry.get_or_open(&default_id).await.unwrap();
        let content: String = h
            .read()
            .interact(move |conn| {
                conn.query_row(
                    "SELECT content FROM episodes WHERE memory_id = ?",
                    [&mid_str],
                    |r| r.get(0),
                )
            })
            .await
            .unwrap();
        assert_eq!(content, "lifecycle test");
    });
}

#[test]
fn tenant_registry_lazy_load_returns_cached_arc() {
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        let id = TenantId::default_tenant();
        let first = registry.get_or_open(&id).await.unwrap();
        let second = registry.get_or_open(&id).await.unwrap();
        // Same Arc pointer.
        assert!(Arc::ptr_eq(&first, &second));
    });
}

#[test]
fn tenant_registry_concurrent_first_access_single_open() {
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        let id = TenantId::default_tenant();
        // 32 concurrent first-access requests. Under load, the registry
        // must serialise opens and hand back the same Arc to every
        // requester.
        let mut handles = Vec::new();
        for _ in 0..32 {
            let r = registry.clone();
            let id = id.clone();
            handles.push(tokio::spawn(async move { r.get_or_open(&id).await.unwrap() }));
        }
        let mut results: Vec<_> = Vec::new();
        for h in handles {
            results.push(h.await.unwrap());
        }
        // Every result is the same Arc.
        for r in &results[1..] {
            assert!(Arc::ptr_eq(&results[0], r));
        }
    });
}

#[test]
fn tenant_registry_unknown_tenant_returns_not_found() {
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        let other = TenantId::new("never-registered").unwrap();
        match registry.get_or_open(&other).await {
            Ok(_) => panic!("expected NotFound, got Ok"),
            Err(e) => assert!(
                matches!(e, solo_core::Error::NotFound(_)),
                "expected NotFound, got {e:?}"
            ),
        }
    });
}

#[test]
fn tenant_registry_forget_handle_evicts_from_cache() {
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        let id = TenantId::default_tenant();
        let first = registry.get_or_open(&id).await.unwrap();
        assert!(registry.is_open(&id).await);
        let evicted = registry.forget_handle(&id).await;
        assert!(evicted.is_some());
        assert!(!registry.is_open(&id).await);
        // The next get_or_open opens a fresh handle. The Arc identity
        // differs from the evicted one.
        let second = registry.get_or_open(&id).await.unwrap();
        assert!(!Arc::ptr_eq(&first, &second));
    });
}

#[test]
fn tenant_id_validation_rejects_uppercase() {
    let err = TenantId::new("UPPERCASE-IS-BAD").unwrap_err();
    let msg = format!("{err}");
    assert!(msg.to_lowercase().contains("invalid"), "got: {msg}");
}

#[test]
fn tenant_id_validation_rejects_too_long() {
    let too_long = "a".repeat(65);
    let err = TenantId::new(too_long).unwrap_err();
    let msg = format!("{err}");
    assert!(msg.contains("too long") || msg.contains("64"), "got: {msg}");
}

#[test]
fn tenant_per_tenant_isolation_writes_dont_cross_over() {
    use solo_core::{Confidence, EncodingContext, Episode, MemoryId, Tier};
    use chrono::Utc;
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        // Manually register a second tenant in the index + create its
        // DB. (P2 reuses the same plumbing the future `solo tenants
        // create` admin path will use; until that admin CLI ships in
        // P7, tests register tenants by going through the
        // TenantsIndex + creating the DB file manually.)
        let alpha = TenantId::new("alpha").unwrap();
        let alpha_db_name = "alpha.db";
        let alpha_db = tmp.path().join("tenants").join(alpha_db_name);
        {
            let mut conn = crate::init::open_sqlcipher(&alpha_db, &key).unwrap();
            crate::migration::run_migrations(&mut conn).unwrap();
        }
        registry
            .with_index(|idx| {
                idx.register(&alpha, alpha_db_name, Some("Alpha tenant")).unwrap();
            })
            .await;

        // Write to default tenant.
        let default_id = TenantId::default_tenant();
        let h_default = registry.get_or_open(&default_id).await.unwrap();
        let ep = Episode {
            memory_id: MemoryId::new(),
            ts_ms: Utc::now().timestamp_millis(),
            source_type: "user_message".into(),
            source_id: None,
            content: "from default tenant".into(),
            encoding_context: EncodingContext::default(),
            provenance: None,
            confidence: Confidence::new(0.9).unwrap(),
            strength: 0.5,
            salience: 0.5,
            tier: Tier::Hot,
        };
        let emb = stub_embedder().embed("from default tenant").await.unwrap();
        h_default.write().remember(ep, emb).await.unwrap();

        // Read from alpha tenant — should NOT see default's write.
        let h_alpha = registry.get_or_open(&alpha).await.unwrap();
        let count: i64 = h_alpha
            .read()
            .interact(|conn| conn.query_row("SELECT COUNT(*) FROM episodes", [], |r| r.get(0)))
            .await
            .unwrap();
        assert_eq!(count, 0, "alpha tenant must be empty");

        // Default tenant has its row.
        let count_default: i64 = h_default
            .read()
            .interact(|conn| conn.query_row("SELECT COUNT(*) FROM episodes", [], |r| r.get(0)))
            .await
            .unwrap();
        assert_eq!(count_default, 1);
    });
}

#[test]
fn tenant_registry_two_tenants_have_separate_hnsw_indexes() {
    use solo_core::{Confidence, EncodingContext, Episode, MemoryId, Tier};
    use chrono::Utc;
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        // Provision a second tenant.
        let beta = TenantId::new("beta").unwrap();
        let beta_db_name = "beta.db";
        let beta_db = tmp.path().join("tenants").join(beta_db_name);
        {
            let mut conn = crate::init::open_sqlcipher(&beta_db, &key).unwrap();
            crate::migration::run_migrations(&mut conn).unwrap();
        }
        registry
            .with_index(|idx| {
                idx.register(&beta, beta_db_name, None).unwrap();
            })
            .await;

        // Write a remember to default → HNSW grows in default's index.
        let default_id = TenantId::default_tenant();
        let h_default = registry.get_or_open(&default_id).await.unwrap();
        let ep = Episode {
            memory_id: MemoryId::new(),
            ts_ms: Utc::now().timestamp_millis(),
            source_type: "test".into(),
            source_id: None,
            content: "vector test".into(),
            encoding_context: EncodingContext::default(),
            provenance: None,
            confidence: Confidence::new(0.9).unwrap(),
            strength: 0.5,
            salience: 0.5,
            tier: Tier::Hot,
        };
        let emb = stub_embedder().embed("vector test").await.unwrap();
        h_default.write().remember(ep, emb).await.unwrap();

        // Each HNSW is a separate Arc.
        let h_beta = registry.get_or_open(&beta).await.unwrap();
        assert!(!Arc::ptr_eq(h_default.hnsw(), h_beta.hnsw()));
        // Default has one vector; beta has zero.
        assert_eq!(h_default.hnsw().len(), 1);
        assert_eq!(h_beta.hnsw().len(), 0);
    });
}

#[test]
fn tenant_handle_shutdown_idempotency() {
    // Reopen the registry twice over the same data dir — second open
    // must not error and must see the persisted state.
    let (tmp, key) = fresh_init_dir("alpha-tester");
    let runtime = rt();
    runtime.block_on(async {
        {
            let registry = make_registry(tmp.path(), &key, &runtime);
            let _ = registry
                .get_or_open(&TenantId::default_tenant())
                .await
                .unwrap();
            registry.shutdown_all().await;
        }
        // Second open over the same data dir.
        let registry2 = make_registry(tmp.path(), &key, &runtime);
        let listed = registry2.list_active().await.unwrap();
        assert_eq!(listed.len(), 1);
    });
}

#[test]
fn tenant_registry_open_on_fresh_v071_install_runs_migration() {
    // Plant a v0.7.1-shape data dir (solo.db at root, no
    // tenants_index.db) and confirm the registry opens it by triggering
    // the in-place migration.
    use chrono::Utc;
    use rusqlite::params;

    let tmp = tempfile::TempDir::new().unwrap();
    let key = KeyMaterial::derive("v071-test", &[7u8; 16]).unwrap();
    let salt_hex = hex::encode([7u8; 16]);

    // Plant solo.db at root.
    let legacy = tmp.path().join("solo.db");
    {
        let mut conn = crate::init::open_sqlcipher(&legacy, &key).unwrap();
        crate::migration::run_migrations(&mut conn).unwrap();
        let now = Utc::now().timestamp_millis();
        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
             ) VALUES (?, ?, 'user_message', 'pre-upgrade',
                       '{}', 1.0, 0.5, 0.5, 'hot', ?, ?)",
            params!["00000000-0000-0000-0000-000000000001", now, now, now],
        )
        .unwrap();
    }
    // Plant a solo.config.toml so SoloConfig::read succeeds during
    // TenantHandle::open.
    let cfg_body = format!(
        "schema_version = 1\nsalt_hex = \"{salt_hex}\"\n\n[embedder]\nname = \"stub\"\nversion = \"v1\"\ndim = 32\ndtype = \"f32\"\n"
    );
    std::fs::write(tmp.path().join("solo.config.toml"), cfg_body).unwrap();

    let runtime = rt();
    runtime.block_on(async {
        let registry = make_registry(tmp.path(), &key, &runtime);
        // After open, the v0.7.1 → v0.8.0 migration should have run.
        assert!(tmp.path().join("tenants_index.db").exists());
        assert!(tmp.path().join("tenants").join("default.db").exists());
        assert!(!tmp.path().join("solo.db").exists());

        let h = registry
            .get_or_open(&TenantId::default_tenant())
            .await
            .unwrap();
        let count: i64 = h
            .read()
            .interact(|conn| conn.query_row("SELECT COUNT(*) FROM episodes", [], |r| r.get(0)))
            .await
            .unwrap();
        assert_eq!(count, 1, "pre-upgrade episode must survive");
    });
}