alkgit_substrate/

registry.rs

//! The registry: the co-tenant tables behind the family adapter pattern
//! (ADR-019 §3–4, substrate.md §registry, POC-R's validated shape) —
//! the engine-generic half of the family.
//!
//! One trait pair, two implements per surface:
//!
//! - [`RegistryRead`] — the sync read side (`resolve`): the
//!   advertisement path's hot read. No `async`, no I/O — implementors
//!   serve in-memory state. A miss is *definitive*: implementations
//!   promise the whole record table as their cached set, so
//!   `resolve` never falls through to backing-store I/O. The
//!   unknown-repo ≡ unauthorized rule (ADR-008) is evaluated at
//!   resolve's caller — this trait returns `None` and leaks no more
//!   than that.
//! - [`RegistryWrite`] — the async write side: [`RegistryWrite::put`]
//!   / [`RegistryWrite::update`] / [`RegistryWrite::remove`] plus the
//!   reconcile machinery's [`RegistryWrite::watermark`] /
//!   [`RegistryWrite::rows_since`]. Every mutation assigns the
//!   mutation's monotone sequence number *inside* the row-write
//!   transaction (transactionally exact — the watermark's anchor,
//!   ADR-019 §5) and returns it; write-side announce (publish/notify/
//!   enqueue) is the reactor's post-commit business, never the pair's.
//!
//! The complete in-memory implementations:
//!
//! - [`InMemoryRegistry`] — the in-memory default adapter, the
//!   batteries-included degenerate case: always-populated,
//!   always-correct, single-process, stamp-free (ADR-024 §6), no
//!   engine dependency; the completeness shape in the crate's base
//!   (no feature gates — the ADR-025 placement rule).
//! - [`CachedRegistryProvider`] — the coherence contract's read side
//!   (ADR-019 §4, POC-R's named deliverable): the whole record table
//!   as one immutable snapshot swapped under a short lock; the hot
//!   path is one lock read + `Arc` clone + lookup with zero I/O;
//!   invalidation is write-through on the store's own writes plus
//!   wake-driven/tick-driven whole-table refreshes (the delivery
//!   wiring is the reactor module's); refresh failures leave the
//!   previous snapshot serving.
//! - [`RegistryBackend`] — the binding of the pair onto the core's
//!   published registry traits ([`GitRegistry`] /
//!   [`GitRegistryStore`]): resolve over the
//!   read side (a miss on the whole-table snapshot is a definitive
//!   unresolved id — [`RegistryError::NotFound`]; the unknown ≡
//!   unauthorized collapse stays at resolve's wire-layer caller),
//!   `put_repo`/`update_repo`/`remove_repo` over the write side with
//!   write-through on the optional provider. The in-memory adapter is
//!   the first complete impl (bound via
//!   [`RegistryBackend::with_cached_provider`] over
//!   [`InMemoryRegistry`] — the mem pair is test-only now, review 006
//!   R-4; the production mem composition is the entry's mem shape);
//!   the engine twins implement the same pair so the same wrapper
//!   binds over them.
//!
//! The exact Rust names pinned here are the OQ-03 substrate-side
//! inventory for this wave's members: `RegistryRead`, `RegistryWrite`,
//! [`RegistryRow`], `InMemoryRegistry`, `CachedRegistryProvider`,
//! `RegistryBackend`.

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use alkgit::validate_repo_id;
#[cfg(any(feature = "sqlite", feature = "postgres"))]
use alkgit::{AccessAction, Visibility};
use alkgit::{GitRegistry, GitRegistryStore, RegistryError, RepoRecord};
use async_trait::async_trait;
#[cfg(any(feature = "sqlite", feature = "postgres"))]
use std::collections::BTreeMap;

/// The v1 HEAD seed target — the unborn symref row both engine twins'
/// record-write transactions insert (substrate.md §refs pins the value;
/// a constructor knob is additive later, two-way — the row is
/// CAS-rewriteable).
pub const HEAD_SEED_TARGET: &str = "refs/heads/main";

/// One durable registry row — the record plus its mutation sequence
/// number.
///
/// `seq` is monotone per table (assigned inside the row's own
/// transaction); it is the reconcile sweep's watermark currency and
/// the per-seq idempotency key of every consumer downstream. First
/// mutation carries 1; sequence 0 means "everything" to
/// [`RegistryWrite::rows_since`].
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RegistryRow {
    /// The record the row carries.
    pub record: RepoRecord,
    /// The mutation's monotone sequence number.
    pub seq: i64,
}

/// The sync read side of the family adapter pair — the advertisement
/// path's hot read (the alkcall `OwnershipProvider` posture, POC-R's
/// validated shape).
///
/// Resolve serves in-memory state: no await, no round trip, no backing
/// I/O (the coherence contract's hot path). A `None` return is
/// definitive — implementations cache/present the whole record table,
/// so a miss is not a "not in my cache, go ask the store" signal.
pub trait RegistryRead: Send + Sync {
    /// Resolve a repo id to its record; `None` is a definitive miss.
    fn resolve(&self, repo_id: &str) -> Option<RepoRecord>;
}

/// The async write side of the family adapter pair (the alkcall
/// `OwnershipStore` posture).
///
/// Every mutation is transactional on the implementor's durable
/// surface: engine twins bump the durable `registry_seq` singleton
/// inside the row-write transaction and their row commit also carries
/// the refs-table legs the composition owes (HEAD seeding at record
/// creation, ref-row removal at repo removal — substrate.md §refs);
/// the in-memory adapter bumps a process-local counter. The returned
/// sequence number is the mutation's watermark currency; failed
/// mutations consume none.
///
/// Post-commit announce (publish/notify/enqueue) and the reactive
/// halves live in the reactor module — never inside the pair.
#[async_trait]
pub trait RegistryWrite: Send + Sync {
    /// Create a repo record. [`RegistryError::Invalid`] on a malformed
    /// id or record, [`RegistryError::AlreadyExists`] when the id is
    /// taken. Returns the mutation's sequence number.
    async fn put(&self, record: &RepoRecord) -> Result<i64, RegistryError>;

    /// Replace a repo record (the grant-mutation path, ADR-011).
    /// [`RegistryError::Invalid`] on a malformed record,
    /// [`RegistryError::NotFound`] when the id is absent. Returns the
    /// mutation's sequence number.
    async fn update(&self, record: &RepoRecord) -> Result<i64, RegistryError>;

    /// Remove a repo record. [`RegistryError::NotFound`] when the id
    /// is absent. Returns the removal's sequence number.
    async fn remove(&self, repo_id: &str) -> Result<i64, RegistryError>;

    /// The table's current watermark: the max sequence number ever
    /// assigned (0 on a fresh table). Monotone — a committed removal
    /// advances it even though it leaves no row.
    async fn watermark(&self) -> Result<i64, RegistryError>;

    /// The reconcile catch-up read: every row mutated after `seq`, in
    /// sequence order. `rows_since(0)` is the whole table — the cached
    /// provider's snapshot-refresh read (whole-table by contract).
    async fn rows_since(&self, seq: i64) -> Result<Vec<RegistryRow>, RegistryError>;
}

/// The store-side record validation shared by the family
/// ([`RegistryError::Invalid`] is closed over exactly these rejections,
/// backend.md §`RegistryError`): the repo id passes the pinned grammar
/// (the id is an opaque registry key — rejected against the grammar,
/// never decomposed; ADR-008) and grant keys are non-empty opaque
/// identity-id strings.
pub(crate) fn validate_record(record: &RepoRecord) -> Result<(), RegistryError> {
    validate_repo_id(&record.repo_id).map_err(|_| RegistryError::Invalid)?;
    if record.grants.keys().any(String::is_empty) {
        return Err(RegistryError::Invalid);
    }
    Ok(())
}

/// A committed row's record reconstruction — the serde shape's members
/// from their columns. A row the composition did not write (corrupt
/// visibility text, malformed grants JSON) is structurally invalid row
/// state: [`RegistryError::Invalid`]. Engine-neutral — both engine
/// twins decode their `git_repos` rows through it.
#[cfg(any(feature = "sqlite", feature = "postgres"))]
pub(crate) fn row_to_record(
    repo_id: &str,
    visibility_col: &str,
    grants_col: &str,
    seq: i64,
) -> Result<RegistryRow, RegistryError> {
    let visibility = match visibility_col {
        "public" => Visibility::Public,
        "private" => Visibility::Private,
        _ => return Err(RegistryError::Invalid),
    };
    let grants: BTreeMap<String, Vec<AccessAction>> =
        serde_json::from_str(grants_col).map_err(|_| RegistryError::Invalid)?;
    Ok(RegistryRow {
        record: RepoRecord {
            repo_id: repo_id.to_owned(),
            visibility,
            grants,
        },
        seq,
    })
}

/// The row's visibility column spelling (the writes' inverse of
/// [`row_to_record`]'s read) — the CHECK's domain, shared by both
/// twins.
#[cfg(any(feature = "sqlite", feature = "postgres"))]
pub(crate) fn visibility_str(visibility: Visibility) -> &'static str {
    match visibility {
        Visibility::Public => "public",
        Visibility::Private => "private",
    }
}

/// The in-memory default adapter — the batteries-included degenerate
/// case of the family: always-populated, always-correct,
/// single-process; the completeness shape in the crate's base (no
/// engine dependency, stamp-free per ADR-024 §6 — it builds fresh
/// every construction).
///
/// Sequence numbers are process-local but hold the same shape the
/// engine twins carry: monotone over the adapter's lifetime, assigned
/// only inside a successful mutation, never decreasing — a committed
/// removal advances the watermark while leaving no row.
pub struct InMemoryRegistry {
    rows: parking_lot::RwLock<HashMap<String, RegistryRow>>,
    seq: AtomicU64,
}

impl Default for InMemoryRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl InMemoryRegistry {
    /// A fresh, empty registry.
    pub fn new() -> Self {
        InMemoryRegistry {
            rows: parking_lot::RwLock::new(HashMap::new()),
            seq: AtomicU64::new(0),
        }
    }

    fn next_seq(&self) -> i64 {
        self.seq.fetch_add(1, Ordering::Relaxed) as i64 + 1
    }

    /// The per-push mutation-marker write's durable leg (ADR-019 §7,
    /// composition-internal — the engine twins' same-named leg, the
    /// mem adapter's shape): the process-local sequence pointer moves
    /// (the watermark's anchor); no row changes and `rows_since` never
    /// carries markers. The record must exist (the receive flow
    /// resolved it) — an absent repo's marker is refused
    /// ([`RegistryError::NotFound`], the twins' posture). Consuming
    /// composition: the mem shape's fused
    /// [`crate::pins::MemMutationLegs`].
    pub(crate) async fn record_mutation(&self, repo_id: &str) -> Result<i64, RegistryError> {
        if !self.rows.read().contains_key(repo_id) {
            return Err(RegistryError::NotFound);
        }
        Ok(self.next_seq())
    }
}

#[async_trait]
impl RegistryWrite for InMemoryRegistry {
    async fn put(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
        validate_record(record)?;
        let mut rows = self.rows.write();
        if rows.contains_key(&record.repo_id) {
            return Err(RegistryError::AlreadyExists);
        }
        let seq = self.next_seq();
        rows.insert(
            record.repo_id.clone(),
            RegistryRow {
                record: record.clone(),
                seq,
            },
        );
        Ok(seq)
    }

    async fn update(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
        validate_record(record)?;
        let mut rows = self.rows.write();
        if !rows.contains_key(&record.repo_id) {
            return Err(RegistryError::NotFound);
        }
        let seq = self.next_seq();
        rows.insert(
            record.repo_id.clone(),
            RegistryRow {
                record: record.clone(),
                seq,
            },
        );
        Ok(seq)
    }

    async fn remove(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let mut rows = self.rows.write();
        if rows.remove(repo_id).is_none() {
            return Err(RegistryError::NotFound);
        }
        Ok(self.next_seq())
    }

    async fn watermark(&self) -> Result<i64, RegistryError> {
        Ok(self.seq.load(Ordering::Relaxed) as i64)
    }

    async fn rows_since(&self, seq: i64) -> Result<Vec<RegistryRow>, RegistryError> {
        let mut rows: Vec<RegistryRow> = self
            .rows
            .read()
            .values()
            .filter(|row| row.seq > seq)
            .cloned()
            .collect();
        rows.sort_by_key(|row| row.seq);
        Ok(rows)
    }
}

impl RegistryRead for InMemoryRegistry {
    fn resolve(&self, repo_id: &str) -> Option<RepoRecord> {
        self.rows.read().get(repo_id).map(|row| row.record.clone())
    }
}

/// **The coherence contract as a type** (ADR-019 §4, POC-R's named
/// deliverable; substrate.md §the coherence contract) — the cached
/// read provider the advertisement path rides:
///
/// - *Cached set*: the whole record table as one immutable snapshot —
///   an `Arc` map swapped under a short lock. Whole-table is
///   deliberate: record tables are small, whole snapshots make
///   coalesced refreshes trivially correct (a refresh re-reads the
///   table; there is no delta application to get wrong), and readers
///   never block a writer.
/// - *Hot path*: one lock read + `Arc` clone + lookup. Zero I/O on the
///   steady state — the provider holds no durable-side handle at all;
///   it serves last-installed state and nothing else.
/// - *Invalidation triggers*: (1) alkstore wakes (the post-commit
///   notify hint, or sqlite's file-wide `data_version` bump from any
///   writer — the delivery wiring is the reactor's wake runner); (2)
///   the provider's own writes (write-through, via
///   [`CachedRegistryProvider::install_one`] /
///   [`CachedRegistryProvider::install_removal`] — the
///   [`RegistryBackend`] store methods call these); (3) the reconcile
///   tick (driven through
///   [`CachedRegistryProvider::refresh_from`], the same body the wake
///   handler runs).
/// - *Staleness bound*: mutation visibility ≤ one wake delivery + one
///   refresh. One `refresh_from` after a mutation is always enough to
///   surface it — the snapshot is whole-table, so there is no partial
///   state between installs.
/// - *Refresh coalescing*: correct by the snapshot swap — a burst of
///   mutations collapses into one refresh, and one refresh carries the
///   whole burst.
/// - *Failure posture*: refresh failures leave the previous snapshot
///   serving; the next wake, write, or tick retries.
pub struct CachedRegistryProvider {
    snapshot: parking_lot::RwLock<Arc<HashMap<String, RepoRecord>>>,
    refreshes: AtomicU64,
}

impl Default for CachedRegistryProvider {
    fn default() -> Self {
        Self::new()
    }
}

impl CachedRegistryProvider {
    /// A provider with an empty snapshot.
    pub fn new() -> Self {
        CachedRegistryProvider {
            snapshot: parking_lot::RwLock::new(Arc::new(HashMap::new())),
            refreshes: AtomicU64::new(0),
        }
    }

    /// Install a whole-table snapshot (the wake handler's and the
    /// reconcile tick's body — a swap, not a delta).
    pub fn install(&self, rows: Vec<RegistryRow>) {
        let map: HashMap<String, RepoRecord> = rows
            .into_iter()
            .map(|row| (row.record.repo_id.clone(), row.record))
            .collect();
        *self.snapshot.write() = Arc::new(map);
        self.refreshes.fetch_add(1, Ordering::Relaxed);
    }

    /// Write-through install of one row — the store's own write path
    /// (no table read involved).
    pub fn install_one(&self, row: RegistryRow) {
        let mut guard = self.snapshot.write();
        let mut map = (**guard).clone();
        map.insert(row.record.repo_id.clone(), row.record);
        *guard = Arc::new(map);
        self.refreshes.fetch_add(1, Ordering::Relaxed);
    }

    /// Write-through removal — the store's own removal path.
    pub fn install_removal(&self, repo_id: &str) {
        let mut guard = self.snapshot.write();
        let mut map = (**guard).clone();
        map.remove(repo_id);
        *guard = Arc::new(map);
        self.refreshes.fetch_add(1, Ordering::Relaxed);
    }

    /// Pull-and-install from the durable side — the body the wake
    /// handler and the reconcile tick both run. The whole table is
    /// re-read (`rows_since(0)`) and swapped in; on failure nothing is
    /// installed and the previous snapshot keeps serving.
    pub async fn refresh_from(&self, registry: &dyn RegistryWrite) -> Result<(), RegistryError> {
        let rows = registry.rows_since(0).await?;
        self.install(rows);
        Ok(())
    }

    /// The installed snapshot (observation without refresh —
    /// replication-style consumers and the tests' coherence asserts).
    pub fn snapshot(&self) -> Arc<HashMap<String, RepoRecord>> {
        self.snapshot.read().clone()
    }

    /// How many installs have landed (test-observation for the
    /// refresh-count pins: idle-stability across idle windows, one
    /// refresh per coalesced burst).
    pub fn refresh_count(&self) -> u64 {
        self.refreshes.load(Ordering::Relaxed)
    }
}

impl RegistryRead for CachedRegistryProvider {
    fn resolve(&self, repo_id: &str) -> Option<RepoRecord> {
        self.snapshot.read().get(repo_id).cloned()
    }
}

/// The registry backend — the pair bound onto the core's published
/// registry traits ([`GitRegistry`] /
/// [`GitRegistryStore`]); the first complete impl is the
/// in-memory binding (test-only now — review 006 R-4, the bare mem
/// pair is not a production surface; the production mem composition
/// is the entry's mem shape) and the engine twins bind through the
/// same type.
///
/// Resolve rides the read side ([`CachedRegistryProvider`] in the
/// composed assemblies; the miss maps to [`RegistryError::NotFound`]
/// — a definitive unresolved id, with the unknown ≡ unauthorized
/// collapse at the wire-layer caller, exactly as before). The store
/// methods ride the write side, announcing nothing (the reactor owns
/// post-commit announce) and, when a provider is bound with
/// [`RegistryBackend::with_cached_provider`], writing through so the
/// this-process mutation is visible to the hot path without waiting
/// for the wake-driven refresh.
pub struct RegistryBackend {
    write: Arc<dyn RegistryWrite>,
    read: Arc<dyn RegistryRead>,
    write_through: Option<Arc<CachedRegistryProvider>>,
}

impl RegistryBackend {
    /// Bind the pair as-is (no write-through leg — a resolver over an
    /// external cached provider's refresh cadence, or direct state).
    pub fn new(write: Arc<dyn RegistryWrite>, read: Arc<dyn RegistryRead>) -> Self {
        RegistryBackend {
            write,
            read,
            write_through: None,
        }
    }

    /// Bind the pair with the cached provider on both legs: resolves
    /// serve the provider, the store methods write through to it. The
    /// engine-twin binding.
    pub fn with_cached_provider(
        write: Arc<dyn RegistryWrite>,
        provider: Arc<CachedRegistryProvider>,
    ) -> Self {
        RegistryBackend {
            write: write.clone(),
            read: provider.clone(),
            write_through: Some(provider),
        }
    }

    /// The durable write side (the reactor's announce composition and
    /// reconcile sweep consume it directly).
    pub fn write(&self) -> &Arc<dyn RegistryWrite> {
        &self.write
    }

    /// The read side (the hot read).
    pub fn read(&self) -> &Arc<dyn RegistryRead> {
        &self.read
    }

    /// The write-through provider, when bound with
    /// [`RegistryBackend::with_cached_provider`].
    pub fn provider(&self) -> Option<&Arc<CachedRegistryProvider>> {
        self.write_through.as_ref()
    }
}

#[async_trait]
impl GitRegistry for RegistryBackend {
    async fn resolve(&self, repo_id: &str) -> Result<RepoRecord, RegistryError> {
        self.read.resolve(repo_id).ok_or(RegistryError::NotFound)
    }
}

#[async_trait]
impl GitRegistryStore for RegistryBackend {
    async fn put_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
        let seq = self.write.put(record).await?;
        if let Some(provider) = &self.write_through {
            provider.install_one(RegistryRow {
                record: record.clone(),
                seq,
            });
        }
        Ok(())
    }

    async fn update_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
        let seq = self.write.update(record).await?;
        if let Some(provider) = &self.write_through {
            provider.install_one(RegistryRow {
                record: record.clone(),
                seq,
            });
        }
        Ok(())
    }

    async fn remove_repo(&self, repo_id: &str) -> Result<(), RegistryError> {
        self.write.remove(repo_id).await?;
        if let Some(provider) = &self.write_through {
            provider.install_removal(repo_id);
        }
        Ok(())
    }
}

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

    use crate::testutil::{mem_registry_pair, private_record, record};

    #[tokio::test]
    async fn mem_adapter_is_complete_put_resolve_update_remove() {
        let registry = InMemoryRegistry::new();

        let put_seq = registry.put(&record("demo")).await.unwrap();
        assert_eq!(put_seq, 1);
        assert_eq!(registry.resolve("demo"), Some(record("demo")));

        let mut updated = record("demo");
        updated.visibility = alkgit::Visibility::Private;
        let update_seq = registry.update(&updated).await.unwrap();
        assert_eq!(update_seq, 2);
        assert_eq!(registry.resolve("demo"), Some(updated));

        let remove_seq = registry.remove("demo").await.unwrap();
        assert_eq!(remove_seq, 3);
        assert_eq!(registry.resolve("demo"), None);
    }

    #[tokio::test]
    async fn mem_adapter_error_map_is_pinned() {
        let registry = InMemoryRegistry::new();
        assert_eq!(
            registry.update(&record("ghost")).await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            registry.remove("ghost").await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            registry.put(&record("demo")).await.unwrap(),
            1,
            "first mutation carries seq 1"
        );
        assert_eq!(
            registry.put(&record("demo")).await.unwrap_err(),
            RegistryError::AlreadyExists
        );

        let mut bad_id = record("Owner/name");
        bad_id.repo_id = "Owner/name".to_string();
        assert_eq!(
            registry.put(&bad_id).await.unwrap_err(),
            RegistryError::Invalid
        );
        let mut empty_repo_id = record("");
        empty_repo_id.repo_id = String::new();
        assert_eq!(
            registry.put(&empty_repo_id).await.unwrap_err(),
            RegistryError::Invalid
        );
    }

    #[tokio::test]
    async fn mem_adapter_failed_mutations_consume_no_seq() {
        let registry = InMemoryRegistry::new();
        assert!(registry.put(&record("a")).await.is_ok());
        assert!(registry.put(&record("a")).await.is_err());
        assert_eq!(registry.watermark().await.unwrap(), 1);
        let seq = registry.put(&record("b")).await.unwrap();
        assert_eq!(seq, 2, "the failed duplicate put consumed nothing");
    }

    #[tokio::test]
    async fn mem_adapter_watermark_is_monotone_across_removal() {
        let registry = InMemoryRegistry::new();
        assert_eq!(registry.watermark().await.unwrap(), 0, "fresh table");
        registry.put(&record("a")).await.unwrap();
        registry.put(&record("b")).await.unwrap();
        assert_eq!(registry.watermark().await.unwrap(), 2);
        registry.remove("a").await.unwrap();
        assert_eq!(
            registry.watermark().await.unwrap(),
            3,
            "a committed removal advances the watermark though it leaves no row"
        );
        registry.remove("never-was").await.unwrap_err();
        assert_eq!(
            registry.watermark().await.unwrap(),
            3,
            "a failed removal consumes no seq"
        );
    }

    #[tokio::test]
    async fn mem_adapter_rows_since_covers_the_reconcile_recipe() {
        let registry = InMemoryRegistry::new();
        let seq_a = registry.put(&record("a")).await.unwrap();
        let seq_b = registry.put(&private_record("b")).await.unwrap();
        assert_eq!(seq_b, seq_a + 1);

        let all = registry.rows_since(0).await.unwrap();
        assert_eq!(
            all.iter().map(|row| row.seq).collect::<Vec<_>>(),
            vec![seq_a, seq_b],
            "rows_since(0) is the whole table, in sequence order"
        );

        registry.update(&record("a")).await.unwrap();
        registry.remove("b").await.unwrap();

        let catchup = registry.rows_since(seq_b).await.unwrap();
        assert_eq!(
            catchup
                .iter()
                .map(|row| (row.record.repo_id.as_str(), row.seq))
                .collect::<Vec<_>>(),
            vec![("a", seq_a + 2)],
            "the committed removal leaves no row for the catch-up read"
        );
    }

    #[tokio::test]
    async fn mem_adapter_rows_since_is_seq_ordered_and_exclusive() {
        let registry = InMemoryRegistry::new();
        let first = registry.put(&record("a")).await.unwrap();
        registry.put(&record("b")).await.unwrap();
        registry.put(&record("c")).await.unwrap();

        let rows = registry.rows_since(first).await.unwrap();
        assert_eq!(
            rows.iter()
                .map(|row| row.record.repo_id.as_str())
                .collect::<Vec<_>>(),
            vec!["b", "c"],
            "rows after `seq` only, in sequence order"
        );
        assert!(registry.rows_since(first).await.unwrap()[0].seq > first);
    }

    #[tokio::test]
    async fn resolve_hits_are_cloned_not_aliased() {
        let registry = InMemoryRegistry::new();
        registry.put(&record("demo")).await.unwrap();
        let resolved = registry.resolve("demo").unwrap();
        let via_provider = {
            let provider = CachedRegistryProvider::new();
            provider.install(registry.rows_since(0).await.unwrap());
            provider.resolve("demo").unwrap()
        };
        assert_eq!(resolved, record("demo"));
        assert_eq!(via_provider, record("demo"));
    }

    #[tokio::test]
    async fn provider_snapshot_swap_serves_the_whole_table() {
        let registry = InMemoryRegistry::new();
        let provider = CachedRegistryProvider::new();

        assert_eq!(provider.resolve("demo"), None, "fresh provider misses");
        assert_eq!(provider.refresh_count(), 0, "no installs at construction");

        registry.put(&record("demo")).await.unwrap();
        provider.install(registry.rows_since(0).await.unwrap());

        assert_eq!(provider.resolve("demo"), Some(record("demo")));
        assert_eq!(provider.snapshot().len(), 1, "whole-table install");
    }

    #[tokio::test]
    async fn provider_hot_path_never_touches_the_durable_side() {
        let registry = InMemoryRegistry::new();
        registry.put(&record("demo")).await.unwrap();
        let provider = CachedRegistryProvider::new();
        provider.install(registry.rows_since(0).await.unwrap());

        let before = registry.watermark().await.unwrap();
        for _ in 0..1_000 {
            assert_eq!(provider.resolve("demo"), Some(record("demo")));
        }
        assert_eq!(
            registry.watermark().await.unwrap(),
            before,
            "resolve is served entirely from the installed snapshot — \
             zero durable-side reads on the hot path"
        );
    }

    #[tokio::test]
    async fn provider_write_through_sees_its_own_writes_immediately() {
        let registry = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let backend = RegistryBackend::with_cached_provider(registry.clone(), provider.clone());
        let write = backend.write().clone();
        drop(backend);

        let seq = write.put(&record("demo")).await.unwrap();
        provider.install_one(RegistryRow {
            record: record("demo"),
            seq,
        });
        assert_eq!(
            provider.resolve("demo"),
            Some(record("demo")),
            "write-through install surfaces the mutation with no refresh"
        );
        assert_eq!(provider.refresh_count(), 1, "one install, no table read");

        let seq = write.put(&record("other")).await.unwrap();
        provider.install_one(RegistryRow {
            record: record("other"),
            seq,
        });
        assert_eq!(provider.resolve("demo"), Some(record("demo")));
        assert_eq!(provider.resolve("other"), Some(record("other")));

        write.remove("demo").await.unwrap();
        provider.install_removal("demo");
        assert_eq!(provider.resolve("demo"), None);
        assert_eq!(
            provider.resolve("other"),
            Some(record("other")),
            "removal is an install, not a re-read"
        );
    }

    struct FlakyCatchUp {
        registry: InMemoryRegistry,
        fail_first_rows_since: std::sync::atomic::AtomicBool,
    }

    #[async_trait]
    impl RegistryWrite for FlakyCatchUp {
        async fn put(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
            self.registry.put(record).await
        }
        async fn update(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
            self.registry.update(record).await
        }
        async fn remove(&self, repo_id: &str) -> Result<i64, RegistryError> {
            self.registry.remove(repo_id).await
        }
        async fn watermark(&self) -> Result<i64, RegistryError> {
            self.registry.watermark().await
        }
        async fn rows_since(&self, seq: i64) -> Result<Vec<RegistryRow>, RegistryError> {
            if self.fail_first_rows_since.swap(false, Ordering::Relaxed) {
                return Err(RegistryError::Io("refresh transport failed".into()));
            }
            self.registry.rows_since(seq).await
        }
    }

    #[tokio::test]
    async fn provider_refresh_failure_leaves_the_previous_snapshot_serving() {
        let flaky = FlakyCatchUp {
            registry: InMemoryRegistry::new(),
            fail_first_rows_since: std::sync::atomic::AtomicBool::new(false),
        };
        let provider = CachedRegistryProvider::new();
        let seq = flaky.put(&record("good")).await.unwrap();
        provider.install_one(RegistryRow {
            record: record("good"),
            seq,
        });

        flaky.fail_first_rows_since.store(true, Ordering::Relaxed);
        assert!(provider.refresh_from(&flaky).await.is_err());
        assert_eq!(
            provider.resolve("good"),
            Some(record("good")),
            "the failed refresh left the previous snapshot serving"
        );

        assert!(provider.refresh_from(&flaky).await.is_ok());
        assert_eq!(provider.resolve("good"), Some(record("good")));

        flaky.put(&record("late")).await.unwrap();
        assert!(provider.refresh_from(&flaky).await.is_ok());
        assert_eq!(
            provider.resolve("late"),
            Some(record("late")),
            "the next tick's refresh retries and succeeds"
        );
    }

    #[tokio::test]
    async fn provider_one_refresh_surfaces_a_burst_and_is_idempotent() {
        let registry = InMemoryRegistry::new();
        let provider = CachedRegistryProvider::new();

        for id in ["a", "b", "c"] {
            registry.put(&record(id)).await.unwrap();
        }
        provider.refresh_from(&registry).await.unwrap();
        assert_eq!(provider.refresh_count(), 1);
        for id in ["a", "b", "c"] {
            assert_eq!(
                provider.resolve(id),
                Some(record(id)),
                "one refresh carries the whole burst — coalescing is \
                 correct by the snapshot swap"
            );
        }
        assert_eq!(provider.snapshot().len(), 3);

        registry.remove("b").await.unwrap();
        provider.refresh_from(&registry).await.unwrap();
        provider.refresh_from(&registry).await.unwrap();
        assert_eq!(provider.resolve("a"), Some(record("a")));
        assert_eq!(provider.resolve("b"), None);
        assert_eq!(provider.resolve("c"), Some(record("c")));
        assert_eq!(
            provider.refresh_count(),
            3,
            "reconcile ticks re-install whole-table state — \
             idempotent by construction"
        );
    }

    #[tokio::test]
    async fn mem_backend_binds_end_to_end_put_resolve_remove() {
        let (_, store) = mem_registry_pair();

        store.put_repo(&record("demo")).await.unwrap();
        assert_eq!(
            store.resolve("demo").await.unwrap(),
            record("demo"),
            "mem-backed end-to-end: put is hot-visible through the \
             write-through provider"
        );

        let mut granted = record("demo");
        granted
            .grants
            .insert("u-7731".to_string(), vec![alkgit::AccessAction::Read]);
        store.update_repo(&granted).await.unwrap();
        assert_eq!(store.resolve("demo").await.unwrap(), granted);

        store.remove_repo("demo").await.unwrap();
        assert_eq!(
            store.resolve("demo").await.unwrap_err(),
            RegistryError::NotFound,
            "a resolve miss after removal maps to the collapse posture \
             — NotFound, the definitive unresolved id"
        );
    }

    #[tokio::test]
    async fn mem_backend_error_map_matches_the_core_traits() {
        let (_, store) = mem_registry_pair();
        assert_eq!(
            store.update_repo(&record("ghost")).await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            store.remove_repo("ghost").await.unwrap_err(),
            RegistryError::NotFound
        );
        store.put_repo(&record("demo")).await.unwrap();
        assert_eq!(
            store.put_repo(&record("demo")).await.unwrap_err(),
            RegistryError::AlreadyExists
        );

        let mut malformed = record("Bad/id");
        malformed.repo_id = "Bad/id".to_string();
        assert_eq!(
            store.put_repo(&malformed).await.unwrap_err(),
            RegistryError::Invalid
        );
        assert_eq!(
            store.resolve("demo").await.unwrap(),
            record("demo"),
            "the rejected mutation left the hot path untouched"
        );
    }

    #[tokio::test]
    async fn mem_backend_store_methods_flow_through_the_pair() {
        let registry = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let store = RegistryBackend::with_cached_provider(
            registry.clone() as Arc<dyn RegistryWrite>,
            provider.clone(),
        );

        store.put_repo(&record("demo")).await.unwrap();
        assert_eq!(registry.resolve("demo"), Some(record("demo")));
        assert_eq!(registry.watermark().await.unwrap(), 1);
        assert_eq!(
            provider.resolve("demo"),
            Some(record("demo")),
            "the provider was written through by the store, not by a \
             durable-side refresh"
        );
        assert_eq!(provider.refresh_count(), 1);

        store.remove_repo("demo").await.unwrap();
        assert_eq!(registry.resolve("demo"), None);
        assert_eq!(registry.watermark().await.unwrap(), 2);
        assert_eq!(provider.resolve("demo"), None);
    }

    #[tokio::test]
    async fn backend_binds_over_an_external_read_side_without_write_through() {
        let registry = Arc::new(InMemoryRegistry::new());
        registry.put(&record("demo")).await.unwrap();
        let store = RegistryBackend::new(registry.clone(), registry.clone());

        assert_eq!(store.resolve("demo").await.unwrap(), record("demo"));
        assert!(
            store.provider().is_none(),
            "no write-through target on the plain binding"
        );
    }

    #[tokio::test]
    async fn all_public_shapes_are_dyn_compatible() {
        let (_, store) = mem_registry_pair();
        let registry: Arc<dyn GitRegistry> = store.clone();
        let store: Arc<dyn GitRegistryStore> = store;

        let resolved = registry.resolve("nope").await;
        assert_eq!(resolved.unwrap_err(), RegistryError::NotFound);
        store.put_repo(&record("demo")).await.unwrap();
        assert_eq!(registry.resolve("demo").await.unwrap(), record("demo"));
    }

    #[tokio::test]
    async fn mem_registry_pair_exposes_the_write_side_for_the_reactor() {
        let (write, store) = mem_registry_pair();
        let seq = write.put(&record("demo")).await.unwrap();
        assert_eq!(seq, 1);
        assert_eq!(write.watermark().await.unwrap(), 1);
        assert_eq!(
            store.resolve("demo").await.unwrap_err(),
            RegistryError::NotFound,
            "a direct pair write rides the wake-driven refresh cadence — \
             never write-through"
        );
    }

    #[test]
    fn repo_record_validation_rejects_the_pinned_shapes() {
        let mut empty_key = record("demo");
        empty_key
            .grants
            .insert(String::new(), vec![alkgit::AccessAction::Read]);
        assert_eq!(validate_record(&empty_key), Err(RegistryError::Invalid));

        let mut long_id = record("x");
        long_id.repo_id = "a".repeat(101);
        assert_eq!(validate_record(&long_id), Err(RegistryError::Invalid));

        assert_eq!(validate_record(&record("owner/name")), Ok(()));
    }
}