lib

mod.rs (42954B)

---

 use crate::backend::map_backend;
 use radroots_event::SignedEvent;
 use radroots_event_codec::Codec;
 use radroots_storage::{
     Error, EventStore,
     event::{
         AdmissionDisposition, AdmissionReceipt, AdmissionStage, BoxFuture, EventAdmission,
         EventCursor, EventId, EventPage, EventPosition, EventQuery, EventQueryBounds,
         EventSequence, SourceGeneration, StoredEventProvenance, StoredRawEvent,
         StoredVerifiedEvent, StoredVisibleEvent, VisibilityEvaluation, VisibilityInput,
         VisibilitySnapshot, evaluate_visibility,
     },
     status::{EventStoreHealth, EventStoreMode, EventStoreStatus},
 };
 use sqlx::{QueryBuilder, Row, Sqlite, SqlitePool};
 use std::{
     path::PathBuf,
     sync::{Arc, Mutex},
 };
 
 use crate::lock::WriterLock;
 use crate::status::StorageLifecycle;
 
 #[derive(Clone)]
 pub struct SqliteStorage {
     pub(crate) pool: SqlitePool,
     pub(crate) private_pool: SqlitePool,
     pub(crate) generation: SourceGeneration,
     pub(crate) mode: EventStoreMode,
     pub(crate) lifecycle: Arc<StorageLifecycle>,
     pub(crate) backup_root: Option<Arc<PathBuf>>,
     pub(crate) paths: Option<Arc<crate::Paths>>,
     pub(crate) reliability: Arc<Mutex<crate::backup::ReliabilityState>>,
 }
 
 struct StoredEventRow {
     position: EventPosition,
     raw_json: String,
     event: SignedEvent,
     stage: AdmissionStage,
 }
 
 impl SqliteStorage {
     #[allow(dead_code)] // Single-pool in-memory scaffold retained for focused backend tests.
     pub(crate) fn new(
         pool: SqlitePool,
         generation: SourceGeneration,
         mode: EventStoreMode,
     ) -> Self {
         Self {
             private_pool: pool.clone(),
             pool,
             generation,
             mode,
             lifecycle: Arc::new(StorageLifecycle::scaffold(mode)),
             backup_root: None,
             paths: None,
             reliability: Arc::new(Mutex::new(crate::backup::ReliabilityState::default())),
         }
     }
 
     #[allow(dead_code)] // Two-pool in-memory scaffold retained for focused backend tests.
     pub(crate) fn with_private_pool(
         pool: SqlitePool,
         private_pool: SqlitePool,
         generation: SourceGeneration,
         mode: EventStoreMode,
     ) -> Self {
         Self {
             pool,
             private_pool,
             generation,
             mode,
             lifecycle: Arc::new(StorageLifecycle::scaffold(mode)),
             backup_root: None,
             paths: None,
             reliability: Arc::new(Mutex::new(crate::backup::ReliabilityState::default())),
         }
     }
 
     pub(crate) fn from_opened(
         pool: SqlitePool,
         private_pool: SqlitePool,
         generation: SourceGeneration,
         options: &crate::OpenOptions,
         writer_lock: Option<WriterLock>,
     ) -> Self {
         let mode = if options.mode().is_writable() {
             EventStoreMode::ReadWrite
         } else {
             EventStoreMode::ReadOnly
         };
         Self {
             pool,
             private_pool,
             generation,
             mode,
             lifecycle: Arc::new(StorageLifecycle::new(
                 options.mode(),
                 options.busy_timeout(),
                 writer_lock,
             )),
             backup_root: options
                 .backup_root()
                 .map(|path| Arc::new(path.to_path_buf())),
             paths: Some(Arc::new(options.paths().clone())),
             reliability: Arc::new(Mutex::new(crate::backup::ReliabilityState::default())),
         }
     }
 
     pub(crate) const fn pool(&self) -> &SqlitePool {
         &self.pool
     }
 
     pub(crate) const fn private_pool(&self) -> &SqlitePool {
         &self.private_pool
     }
 
     pub(crate) const fn event_mode(&self) -> EventStoreMode {
         self.mode
     }
 
     async fn selected(
         &self,
         query: &EventQuery,
         minimum_stage: AdmissionStage,
     ) -> Result<(Vec<StoredEventRow>, Option<EventCursor>), Error> {
         self.validate_cursor(query)?;
         let after = query
             .bounds()
             .cursor()
             .map_or(0, |cursor| cursor.sequence().get());
         let fetch_limit = u64::from(query.bounds().limit()) + 1;
         let mut builder = QueryBuilder::<Sqlite>::new(
             "SELECT source_generation, source_sequence, signed_event, admission_stage, \
                     admitted_contract_id, admitted_registry_version \
              FROM radroots_runtime_events WHERE source_generation = ",
         );
         builder.push_bind(self.generation.as_bytes().as_slice());
         builder.push(" AND source_sequence > ");
         builder.push_bind(i64_from_u64(after)?);
         if minimum_stage == AdmissionStage::Verified {
             builder.push(" AND admission_stage IN ('verified', 'visible')");
         } else if minimum_stage == AdmissionStage::Visible {
             builder.push(" AND admission_stage = 'visible'");
         }
         if !query.event_ids().is_empty() {
             builder.push(" AND event_id IN (");
             let mut separated = builder.separated(", ");
             for event_id in query.event_ids() {
                 separated.push_bind(event_id.as_bytes().to_vec());
             }
             separated.push_unseparated(")");
         }
         builder.push(" ORDER BY source_sequence LIMIT ");
         builder.push_bind(i64_from_u64(fetch_limit)?);
 
         let rows = builder
             .build()
             .fetch_all(&self.pool)
             .await
             .map_err(map_backend)?;
         let mut decoded = rows
             .iter()
             .map(|row| self.decode_event_row(row))
             .collect::<Result<Vec<_>, _>>()?;
         let next = if decoded.len() > usize::from(query.bounds().limit()) {
             decoded.truncate(usize::from(query.bounds().limit()));
             decoded.last().map(|row| row.position)
         } else {
             None
         };
         Ok((decoded, next))
     }
 
     async fn current_event_rows(&self) -> Result<Vec<StoredEventRow>, Error> {
         let rows = sqlx::query(
             "SELECT source_generation, source_sequence, signed_event, admission_stage,
                     admitted_contract_id, admitted_registry_version
              FROM radroots_runtime_events
              WHERE source_generation = ?
              ORDER BY source_sequence",
         )
         .bind(self.generation.as_bytes().as_slice())
         .fetch_all(&self.pool)
         .await
         .map_err(map_backend)?;
         rows.iter().map(|row| self.decode_event_row(row)).collect()
     }
 
     fn visibility_for_rows(&self, rows: &[StoredEventRow]) -> Result<VisibilityEvaluation, Error> {
         evaluate_visibility(
             self.generation,
             rows.iter()
                 .map(|row| VisibilityInput::new(row.position, &row.event, row.stage)),
         )
     }
 
     fn validate_cursor(&self, query: &EventQuery) -> Result<(), Error> {
         if query
             .bounds()
             .cursor()
             .is_some_and(|cursor| cursor.generation() != self.generation)
         {
             return Err(Error::SourceGenerationChanged);
         }
         Ok(())
     }
 
     fn decode_event_row(&self, row: &sqlx::sqlite::SqliteRow) -> Result<StoredEventRow, Error> {
         let generation = source_generation(row.try_get("source_generation").map_err(map_corrupt)?)?;
         if generation != self.generation {
             return Err(Error::CorruptStoredEvent);
         }
         let sequence = event_sequence(row.try_get("source_sequence").map_err(map_corrupt)?)?;
         let raw_json = String::from_utf8(row.try_get("signed_event").map_err(map_corrupt)?)
             .map_err(|_| Error::CorruptStoredEvent)?;
         let event =
             Codec::decode_signed_event(raw_json.as_str()).map_err(|_| Error::CorruptStoredEvent)?;
         let stage = admission_stage(row.try_get("admission_stage").map_err(map_corrupt)?)?;
         let contract_id = row
             .try_get::<Option<String>, _>("admitted_contract_id")
             .map_err(map_corrupt)?;
         let registry_version = row
             .try_get::<Option<i64>, _>("admitted_registry_version")
             .map_err(map_corrupt)?
             .map(|value| u32::try_from(value).map_err(|_| Error::CorruptStoredEvent))
             .transpose()?;
         if contract_id.is_some() != registry_version.is_some()
             || registry_version.is_some_and(|version| {
                 version != radroots_event::contract::RegistryVersion::CURRENT.get()
             })
             || contract_id.as_deref().is_some_and(|stored| {
                 radroots_event::contract::registry_v7::validate_event_contract_for_admission(
                     event.envelope(),
                     stored,
                 )
                 .is_err()
             })
         {
             return Err(Error::CorruptStoredEvent);
         }
         Ok(StoredEventRow {
             position: EventPosition::new(generation, sequence),
             raw_json,
             event,
             stage,
         })
     }
 
     async fn store_provenance(
         transaction: &mut sqlx::Transaction<'_, Sqlite>,
         admission: &EventAdmission,
     ) -> Result<(), Error> {
         let provenance = admission.provenance();
         let cursor = provenance.cursor().map_or("", |cursor| cursor.as_str());
         sqlx::query(
             "INSERT OR IGNORE INTO radroots_runtime_event_provenance (
                event_id, transport_id, target_fingerprint, observed_at_unix_ms, cursor
              ) VALUES (?, ?, ?, ?, ?)",
         )
         .bind(admission.event_id().as_bytes().as_slice())
         .bind(provenance.transport_id().as_str())
         .bind(provenance.target().as_str())
         .bind(i64_from_u64(provenance.observed_at_unix_ms())?)
         .bind(cursor)
         .execute(&mut **transaction)
         .await
         .map_err(map_backend)?;
         Ok(())
     }
 
     pub(crate) async fn admit_transaction(
         &self,
         transaction: &mut sqlx::Transaction<'_, Sqlite>,
         admission: EventAdmission,
     ) -> Result<AdmissionReceipt, Error> {
         let existing = sqlx::query(
             "SELECT source_generation, source_sequence, signed_event, admission_stage,
                     admitted_contract_id, admitted_registry_version
              FROM radroots_runtime_events WHERE event_id = ?",
         )
         .bind(admission.event_id().as_bytes().as_slice())
         .fetch_optional(&mut **transaction)
         .await
         .map_err(map_backend)?;
 
         let (position, disposition) = if let Some(row) = existing {
             let stored = self.decode_event_row(&row)?;
             let metadata = admission_contract_metadata(&admission);
             if stored.raw_json.as_bytes() != admission.event().raw_json().as_bytes() {
                 return Err(Error::EventConflict);
             }
             if admission.stage() < stored.stage {
                 return Err(Error::AdmissionRegression);
             }
             let disposition = if admission.stage() == stored.stage {
                 AdmissionDisposition::Duplicate
             } else {
                 sqlx::query(
                     "UPDATE radroots_runtime_events
                      SET admission_stage = ?, updated_at_unix_ms = MAX(updated_at_unix_ms, ?),
                          admitted_contract_id = COALESCE(admitted_contract_id, ?),
                          admitted_registry_version = COALESCE(admitted_registry_version, ?)
                      WHERE event_id = ?",
                 )
                 .bind(stage_name(admission.stage()))
                 .bind(i64_from_u64(admission.provenance().observed_at_unix_ms())?)
                 .bind(metadata.map(|value| value.0))
                 .bind(metadata.map(|value| i64::from(value.1)))
                 .bind(admission.event_id().as_bytes().as_slice())
                 .execute(&mut **transaction)
                 .await
                 .map_err(map_backend)?;
                 AdmissionDisposition::Advanced
             };
             (stored.position, disposition)
         } else {
             let metadata = admission_contract_metadata(&admission);
             let next = sqlx::query_scalar::<_, i64>(
                 "UPDATE radroots_runtime_source_generations
                  SET sequence_head = sequence_head + 1
                  WHERE generation = ? AND state = 'active'
                  RETURNING sequence_head",
             )
             .bind(self.generation.as_bytes().as_slice())
             .fetch_optional(&mut **transaction)
             .await
             .map_err(map_backend)?
             .ok_or(Error::SourceGenerationChanged)?;
             let sequence = event_sequence(next)?;
             let observed_at = i64_from_u64(admission.provenance().observed_at_unix_ms())?;
             sqlx::query(
                 "INSERT INTO radroots_runtime_events (
                    source_generation, source_sequence, event_id, admission_stage,
                    signed_event, admitted_at_unix_ms, updated_at_unix_ms,
                    admitted_contract_id, admitted_registry_version
                  ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
             )
             .bind(self.generation.as_bytes().as_slice())
             .bind(next)
             .bind(admission.event_id().as_bytes().as_slice())
             .bind(stage_name(admission.stage()))
             .bind(admission.event().raw_json().as_bytes())
             .bind(observed_at)
             .bind(observed_at)
             .bind(metadata.map(|value| value.0))
             .bind(metadata.map(|value| i64::from(value.1)))
             .execute(&mut **transaction)
             .await
             .map_err(map_backend)?;
             (
                 EventPosition::new(self.generation, sequence),
                 AdmissionDisposition::Inserted,
             )
         };
         Self::store_provenance(transaction, &admission).await?;
         Ok(AdmissionReceipt::new(
             *admission.event_id(),
             position,
             admission.stage(),
             disposition,
         ))
     }
 }
 
 fn admission_contract_metadata(admission: &EventAdmission) -> Option<(&'static str, u32)> {
     admission.visible_event().map(|event| {
         (
             event.admitted_event().validated_event().contract_id(),
             radroots_event::contract::RegistryVersion::CURRENT.get(),
         )
     })
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 impl EventStore for SqliteStorage {
     fn status(&self) -> BoxFuture<'_, Result<EventStoreStatus, Error>> {
         Box::pin(async move {
             let row = sqlx::query(
                 "SELECT
                    COUNT(*) AS raw_events,
                    COALESCE(SUM(CASE WHEN admission_stage IN ('verified', 'visible') THEN 1 ELSE 0 END), 0)
                      AS verified_events
                  FROM radroots_runtime_events WHERE source_generation = ?",
             )
             .bind(self.generation.as_bytes().as_slice())
             .fetch_one(&self.pool)
             .await
             .map_err(map_backend)?;
             let visibility_rows = self.current_event_rows().await?;
             let visible_events = u64::try_from(
                 self.visibility_for_rows(&visibility_rows)?
                     .snapshot()
                     .visible_event_ids()
                     .len(),
             )
             .map_err(|_| Error::CorruptStoredEvent)?;
             EventStoreStatus::new(
                 self.generation,
                 self.mode,
                 EventStoreHealth::Available,
                 u64_from_i64(row.try_get("raw_events").map_err(map_corrupt)?)?,
                 u64_from_i64(row.try_get("verified_events").map_err(map_corrupt)?)?,
                 visible_events,
             )
         })
     }
 
     fn admit(&self, admission: EventAdmission) -> BoxFuture<'_, Result<AdmissionReceipt, Error>> {
         Box::pin(async move {
             if self.mode == EventStoreMode::ReadOnly {
                 return Err(Error::BackendUnavailable);
             }
             let mut transaction = self
                 .pool
                 .begin_with("BEGIN IMMEDIATE")
                 .await
                 .map_err(map_backend)?;
             let receipt = self.admit_transaction(&mut transaction, admission).await?;
             transaction.commit().await.map_err(map_backend)?;
             Ok(receipt)
         })
     }
 
     fn query_raw(
         &self,
         query: EventQuery,
     ) -> BoxFuture<'_, Result<EventPage<StoredRawEvent>, Error>> {
         Box::pin(async move {
             let (rows, next) = self.selected(&query, AdmissionStage::Raw).await?;
             let items = rows
                 .into_iter()
                 .map(|row| StoredRawEvent::new(row.position, row.event, row.stage))
                 .collect();
             EventPage::new(self.generation, items, next, query.bounds())
         })
     }
 
     fn query_verified(
         &self,
         query: EventQuery,
     ) -> BoxFuture<'_, Result<EventPage<StoredVerifiedEvent>, Error>> {
         Box::pin(async move {
             let (rows, next) = self.selected(&query, AdmissionStage::Verified).await?;
             let items = rows
                 .into_iter()
                 .map(|row| StoredVerifiedEvent::new(row.position, row.event))
                 .collect();
             EventPage::new(self.generation, items, next, query.bounds())
         })
     }
 
     fn query_visible(
         &self,
         query: EventQuery,
     ) -> BoxFuture<'_, Result<EventPage<StoredVisibleEvent>, Error>> {
         Box::pin(async move {
             self.validate_cursor(&query)?;
             let after = query
                 .bounds()
                 .cursor()
                 .map_or(0, |cursor| cursor.sequence().get());
             let rows = self.current_event_rows().await?;
             let visibility = self.visibility_for_rows(&rows)?;
             let mut selected = rows
                 .into_iter()
                 .filter(|row| {
                     row.position.sequence().get() > after
                         && query.selects(row.event.id())
                         && visibility.is_visible(row.event.id())
                 })
                 .take(usize::from(query.bounds().limit()) + 1)
                 .collect::<Vec<_>>();
             let next = if selected.len() > usize::from(query.bounds().limit()) {
                 selected.truncate(usize::from(query.bounds().limit()));
                 selected.last().map(|row| row.position)
             } else {
                 None
             };
             let items = selected
                 .into_iter()
                 .map(|row| StoredVisibleEvent::new(row.position, row.event))
                 .collect();
             EventPage::new(self.generation, items, next, query.bounds())
         })
     }
 
     fn rebuild_visibility(&self) -> BoxFuture<'_, Result<VisibilitySnapshot, Error>> {
         Box::pin(async move {
             let rows = self.current_event_rows().await?;
             Ok(self.visibility_for_rows(&rows)?.into_snapshot())
         })
     }
 
     fn query_provenance(
         &self,
         event_id: EventId,
         bounds: EventQueryBounds,
     ) -> BoxFuture<'_, Result<EventPage<StoredEventProvenance>, Error>> {
         Box::pin(async move {
             if bounds
                 .cursor()
                 .is_some_and(|cursor| cursor.generation() != self.generation)
             {
                 return Err(Error::SourceGenerationChanged);
             }
             let event_row = sqlx::query(
                 "SELECT source_generation, source_sequence
                  FROM radroots_runtime_events WHERE event_id = ?",
             )
             .bind(event_id.as_bytes().as_slice())
             .fetch_optional(&self.pool)
             .await
             .map_err(map_backend)?
             .ok_or(Error::EventNotFound)?;
             let generation = source_generation(
                 event_row
                     .try_get("source_generation")
                     .map_err(map_corrupt)?,
             )?;
             if generation != self.generation {
                 return Err(Error::CorruptStoredEvent);
             }
             let position = EventPosition::new(
                 generation,
                 event_sequence(event_row.try_get("source_sequence").map_err(map_corrupt)?)?,
             );
             let after = bounds.cursor().map_or(0, |cursor| cursor.sequence().get());
             let items = if position.sequence().get() <= after {
                 Vec::new()
             } else {
                 let rows = sqlx::query(
                     "SELECT transport_id, target_fingerprint, observed_at_unix_ms, cursor
                      FROM radroots_runtime_event_provenance
                      WHERE event_id = ?
                      ORDER BY observed_at_unix_ms, transport_id, target_fingerprint, cursor
                      LIMIT ?",
                 )
                 .bind(event_id.as_bytes().as_slice())
                 .bind(i64::from(bounds.limit()))
                 .fetch_all(&self.pool)
                 .await
                 .map_err(map_backend)?;
                 rows.iter()
                     .map(|row| {
                         let cursor = row.try_get::<String, _>("cursor").map_err(map_corrupt)?;
                         StoredEventProvenance::from_stored_parts(
                             position,
                             row.try_get::<String, _>("transport_id")
                                 .map_err(map_corrupt)?
                                 .as_str(),
                             row.try_get::<String, _>("target_fingerprint")
                                 .map_err(map_corrupt)?
                                 .as_str(),
                             u64_from_i64(row.try_get("observed_at_unix_ms").map_err(map_corrupt)?)?,
                             (!cursor.is_empty()).then_some(cursor.as_str()),
                         )
                     })
                     .collect::<Result<Vec<_>, Error>>()?
             };
             EventPage::new(self.generation, items, None, bounds)
         })
     }
 }
 
 const fn stage_name(stage: AdmissionStage) -> &'static str {
     match stage {
         AdmissionStage::Raw => "raw",
         AdmissionStage::Verified => "verified",
         AdmissionStage::Visible => "visible",
     }
 }
 
 fn admission_stage(value: String) -> Result<AdmissionStage, Error> {
     match value.as_str() {
         "raw" => Ok(AdmissionStage::Raw),
         "verified" => Ok(AdmissionStage::Verified),
         "visible" => Ok(AdmissionStage::Visible),
         _ => Err(Error::CorruptStoredEvent),
     }
 }
 
 fn source_generation(value: Vec<u8>) -> Result<SourceGeneration, Error> {
     SourceGeneration::new(value.try_into().map_err(|_| Error::CorruptStoredEvent)?)
         .map_err(|_| Error::CorruptStoredEvent)
 }
 
 fn event_sequence(value: i64) -> Result<EventSequence, Error> {
     EventSequence::new(u64_from_i64(value)?).map_err(|_| Error::CorruptStoredEvent)
 }
 
 fn i64_from_u64(value: u64) -> Result<i64, Error> {
     i64::try_from(value).map_err(|_| Error::CorruptStoredEvent)
 }
 
 fn u64_from_i64(value: i64) -> Result<u64, Error> {
     u64::try_from(value).map_err(|_| Error::CorruptStoredEvent)
 }
 
 fn map_corrupt(_: sqlx::Error) -> Error {
     Error::CorruptStoredEvent
 }
 
 #[cfg(test)]
 #[cfg_attr(coverage_nightly, coverage(off))]
 mod tests {
     use super::*;
     use crate::migration::runtime::{MIGRATIONS, migration_sql};
     use radroots_event::{
         SignedEvent,
         admission::{AdmissionPolicy, RawEvent, VisibilityPolicy, VisibleEvent},
         wire::Nip01EventWire,
     };
     use radroots_storage::event::EventQueryBounds;
     use radroots_transport::{
         Target, TransportId,
         source::{EventProvenance, FetchCursor, ObservedEvent},
     };
     use sqlx::sqlite::SqlitePoolOptions;
 
     struct Allow;
 
     impl radroots_event::admission::SignatureVerifier for Allow {
         fn verify_signature(
             &self,
             _event: &radroots_event::Event,
         ) -> Result<(), radroots_event::Error> {
             Ok(())
         }
     }
 
     impl AdmissionPolicy for Allow {
         type Error = core::convert::Infallible;
 
         fn policy_id(&self) -> &'static str {
             "test.storage-sqlite.admission.v1"
         }
 
         fn admit(
             &self,
             _event: &radroots_event::admission::ContractValidatedEvent,
         ) -> Result<(), Self::Error> {
             Ok(())
         }
     }
 
     impl VisibilityPolicy for Allow {
         type Error = core::convert::Infallible;
 
         fn policy_id(&self) -> &'static str {
             "test.storage-sqlite.visibility.v1"
         }
 
         fn make_visible(
             &self,
             _event: &radroots_event::admission::AdmittedEvent,
         ) -> Result<(), Self::Error> {
             Ok(())
         }
     }
 
     async fn store(generation: SourceGeneration) -> SqliteStorage {
         let pool = SqlitePoolOptions::new()
             .max_connections(1)
             .connect("sqlite::memory:")
             .await
             .expect("memory SQLite");
         sqlx::query("PRAGMA foreign_keys = ON")
             .execute(&pool)
             .await
             .expect("foreign keys");
         for migration in MIGRATIONS {
             sqlx::raw_sql(migration_sql(migration.version()).expect("registered SQL"))
                 .execute(&pool)
                 .await
                 .expect("runtime migration");
         }
         sqlx::query(
             "INSERT INTO radroots_runtime_source_generations (
                generation, sequence_head, state, created_at_unix_ms, retired_at_unix_ms
              ) VALUES (?, 0, 'active', 1, NULL)",
         )
         .bind(generation.as_bytes().as_slice())
         .execute(&pool)
         .await
         .expect("source generation");
         SqliteStorage::new(pool, generation, EventStoreMode::ReadWrite)
     }
 
     fn signed_event(content: &str, pretty: bool) -> SignedEvent {
         signed_event_with(content, pretty, 1_800_000_100, 0, vec![])
     }
 
     fn signed_event_with(
         content: &str,
         pretty: bool,
         created_at: u64,
         kind: u32,
         tags: Vec<Vec<String>>,
     ) -> SignedEvent {
         let mut wire = Nip01EventWire {
             id: "0".repeat(64),
             pubkey: "585591529da0bab31b3b1b1f986611cf5f435dca84f978c89ee8a40cca7103df".to_owned(),
             created_at,
             kind,
             tags,
             content: content.to_owned(),
             sig: "42".repeat(64),
             extra: Default::default(),
         };
         wire.id = wire
             .computed_event_id()
             .expect("canonical event id")
             .to_hex();
         let value = serde_json::json!({
             "id": &wire.id,
             "pubkey": &wire.pubkey,
             "created_at": wire.created_at,
             "kind": wire.kind,
             "tags": &wire.tags,
             "content": &wire.content,
             "sig": &wire.sig,
         });
         let raw_json = if pretty {
             serde_json::to_string_pretty(&value).expect("pretty event JSON")
         } else {
             value.to_string()
         };
         SignedEvent::from_wire_verified_id(wire, raw_json).expect("signed event")
     }
 
     fn observed(event: SignedEvent, at: u64, cursor: Option<&str>) -> ObservedEvent {
         let target = Target::new(TransportId::NOSTR, "wss://relay.example").expect("target");
         let mut provenance =
             EventProvenance::new(TransportId::NOSTR, target.fingerprint().clone(), at)
                 .expect("provenance");
         if let Some(cursor) = cursor {
             provenance = provenance.with_cursor(FetchCursor::parse(cursor).expect("cursor"));
         }
         ObservedEvent::new(event, provenance)
     }
 
     fn verified(event: &SignedEvent) -> radroots_event::VerifiedEvent {
         RawEvent::new(event.envelope().clone())
             .verify_id()
             .expect("event id")
             .verify_signature(&Allow)
             .expect("signature")
     }
 
     fn visible(event: &SignedEvent) -> VisibleEvent {
         let verified = verified(event);
         let validated = if event.envelope().kind_u32() == 5 {
             verified
                 .validate_contract_for_admission("radroots.social.deletion_request.v1")
                 .expect("admission-selected contract")
         } else {
             verified.validate_contract().expect("contract")
         };
         validated
             .admit_with(&Allow)
             .expect("admission")
             .make_visible_with(&Allow)
             .expect("visibility")
     }
 
     #[tokio::test]
     async fn admission_is_idempotent_monotonic_and_conflict_safe() {
         let generation = SourceGeneration::new([7; 32]).expect("generation");
         let store = store(generation).await;
         let event = signed_event(
             "{\"display_name\":\"Moss Street Farm\",\"bot\":false}",
             false,
         );
 
         let inserted = store
             .admit(EventAdmission::raw(observed(event.clone(), 10, None)))
             .await
             .expect("insert raw");
         assert_eq!(inserted.disposition(), AdmissionDisposition::Inserted);
         assert_eq!(inserted.position().sequence().get(), 1);
 
         let advanced = store
             .admit(
                 EventAdmission::visible(
                     observed(event.clone(), 11, Some("relay-page-1")),
                     visible(&event),
                 )
                 .expect("visible admission"),
             )
             .await
             .expect("advance visible");
         assert_eq!(advanced.disposition(), AdmissionDisposition::Advanced);
         assert_eq!(advanced.position(), inserted.position());
         let metadata = sqlx::query(
             "SELECT admitted_contract_id, admitted_registry_version
              FROM radroots_runtime_events WHERE event_id = ?",
         )
         .bind(event.id().as_bytes().as_slice())
         .fetch_one(&store.pool)
         .await
         .expect("event contract metadata");
         assert_eq!(
             metadata.get::<String, _>("admitted_contract_id"),
             "radroots.profile.metadata.v1"
         );
         assert_eq!(metadata.get::<i64, _>("admitted_registry_version"), 7);
 
         let duplicate = store
             .admit(
                 EventAdmission::visible(
                     observed(event.clone(), 11, Some("relay-page-1")),
                     visible(&event),
                 )
                 .expect("visible admission"),
             )
             .await
             .expect("duplicate visible");
         assert_eq!(duplicate.disposition(), AdmissionDisposition::Duplicate);
         assert_eq!(
             store
                 .admit(EventAdmission::raw(observed(event.clone(), 12, None)))
                 .await,
             Err(Error::AdmissionRegression)
         );
 
         let same_id_different_bytes = signed_event(
             "{\"display_name\":\"Moss Street Farm\",\"bot\":false}",
             true,
         );
         assert_eq!(same_id_different_bytes.id(), event.id());
         assert_eq!(
             store
                 .admit(EventAdmission::raw(observed(
                     same_id_different_bytes,
                     13,
                     None,
                 )))
                 .await,
             Err(Error::EventConflict)
         );
     }
 
     #[tokio::test]
     async fn queries_preserve_stage_bounds_cursors_and_exact_provenance() {
         let generation = SourceGeneration::new([8; 32]).expect("generation");
         let store = store(generation).await;
         let empty_status = store.status().await.expect("empty status");
         assert_eq!(empty_status.raw_events(), 0);
         assert_eq!(empty_status.verified_events(), 0);
         assert_eq!(empty_status.visible_events(), 0);
         let raw_event = signed_event("raw", false);
         let visible_event =
             signed_event("{\"display_name\":\"Visible Farm\",\"bot\":false}", false);
         store
             .admit(EventAdmission::raw(observed(raw_event.clone(), 20, None)))
             .await
             .expect("raw event");
         store
             .admit(
                 EventAdmission::visible(
                     observed(visible_event.clone(), 21, Some("relay-page-2")),
                     visible(&visible_event),
                 )
                 .expect("visible admission"),
             )
             .await
             .expect("visible event");
 
         let first = store
             .query_raw(EventQuery::all(EventQueryBounds::first(1).expect("bounds")))
             .await
             .expect("first page");
         assert_eq!(first.items().len(), 1);
         assert_eq!(first.items()[0].event(), &raw_event);
         let next = first.next_cursor().expect("continuation cursor");
         let second = store
             .query_raw(EventQuery::all(
                 EventQueryBounds::first(1).expect("bounds").after(next),
             ))
             .await
             .expect("second page");
         assert_eq!(second.items()[0].event(), &visible_event);
         assert!(second.next_cursor().is_none());
 
         let verified_page = store
             .query_verified(EventQuery::all(
                 EventQueryBounds::first(10).expect("bounds"),
             ))
             .await
             .expect("verified page");
         assert_eq!(verified_page.items().len(), 1);
         assert_eq!(verified_page.items()[0].event(), &visible_event);
         let visible_page = store
             .query_visible(
                 EventQuery::for_ids(
                     EventQueryBounds::first(10).expect("bounds"),
                     vec![*visible_event.id()],
                 )
                 .expect("id query"),
             )
             .await
             .expect("visible page");
         assert_eq!(visible_page.items().len(), 1);
 
         let provenance = store
             .query_provenance(
                 *visible_event.id(),
                 EventQueryBounds::first(10).expect("bounds"),
             )
             .await
             .expect("provenance");
         assert_eq!(provenance.items().len(), 1);
         assert_eq!(provenance.items()[0].provenance().observed_at_unix_ms(), 21);
         assert_eq!(
             provenance.items()[0]
                 .provenance()
                 .cursor()
                 .expect("cursor")
                 .as_str(),
             "relay-page-2"
         );
 
         let status = store.status().await.expect("status");
         assert_eq!(status.raw_events(), 2);
         assert_eq!(status.verified_events(), 1);
         assert_eq!(status.visible_events(), 1);
         let foreign_cursor = EventPosition::new(
             SourceGeneration::new([9; 32]).expect("foreign generation"),
             EventSequence::new(1).expect("sequence"),
         );
         assert_eq!(
             store
                 .query_raw(EventQuery::all(
                     EventQueryBounds::first(1)
                         .expect("bounds")
                         .after(foreign_cursor),
                 ))
                 .await,
             Err(Error::SourceGenerationChanged)
         );
     }
 
     #[tokio::test]
     async fn verified_replacement_and_same_count_advance_match_memory_after_reopen() {
         let generation = SourceGeneration::new([19; 32]).unwrap();
         let store = store(generation).await;
         let memory = radroots_storage::memory::MemoryStorage::new(generation);
         let old = signed_event_with(
             r#"{"display_name":"Old Farm","bot":false}"#,
             false,
             10,
             0,
             vec![],
         );
         let newer = signed_event_with("malformed profile", false, 20, 0, vec![]);
         let admissions = [
             EventAdmission::visible(observed(old.clone(), 10, None), visible(&old)).unwrap(),
             EventAdmission::raw(observed(newer.clone(), 20, None)),
             EventAdmission::verified(observed(newer.clone(), 20, None), verified(&newer)).unwrap(),
         ];
         let mut previous_digest = None;
         for (index, admission) in admissions.into_iter().enumerate() {
             assert_eq!(
                 store.admit(admission.clone()).await.unwrap(),
                 memory.admit(admission).await.unwrap()
             );
             let reopened =
                 SqliteStorage::new(store.pool.clone(), generation, EventStoreMode::ReadWrite);
             let snapshot = reopened.rebuild_visibility().await.unwrap();
             assert_eq!(snapshot, memory.rebuild_visibility().await.unwrap());
             let query = EventQuery::all(EventQueryBounds::first(10).unwrap());
             assert_eq!(
                 reopened.query_visible(query.clone()).await.unwrap(),
                 memory.query_visible(query).await.unwrap()
             );
             if index == 2 {
                 assert_ne!(previous_digest, Some(snapshot.digest()));
                 assert!(snapshot.visible_event_ids().is_empty());
                 assert_eq!(snapshot.current_heads()[0].event_id, *newer.id());
                 assert_eq!(reopened.status().await.unwrap().raw_events(), 2);
             } else {
                 assert_eq!(snapshot.visible_event_ids(), &[*old.id()]);
             }
             previous_digest = Some(snapshot.digest());
         }
     }
 
     #[tokio::test]
     async fn visibility_rebuild_survives_reopen_and_matches_current_head_deletion_queries() {
         let generation = SourceGeneration::new([18; 32]).expect("generation");
         let store = store(generation).await;
         let old = signed_event_with(
             r#"{"display_name":"Old Farm","bot":false}"#,
             false,
             1_800_000_100,
             0,
             vec![],
         );
         let current = signed_event_with(
             r#"{"display_name":"Current Farm","bot":false}"#,
             false,
             1_800_000_200,
             0,
             vec![],
         );
         let deletion = signed_event_with(
             "retired profile",
             false,
             1_800_000_300,
             5,
             vec![vec!["e".to_owned(), current.id().to_hex()]],
         );
         for (event, observed_at) in [
             (old.clone(), 100),
             (current.clone(), 200),
             (deletion.clone(), 300),
         ] {
             store
                 .admit(
                     EventAdmission::visible(
                         observed(event.clone(), observed_at, None),
                         visible(&event),
                     )
                     .expect("visible admission"),
                 )
                 .await
                 .expect("admit visible event");
         }
 
         let before = store
             .rebuild_visibility()
             .await
             .expect("visibility rebuild");
         let reopened =
             SqliteStorage::new(store.pool.clone(), generation, EventStoreMode::ReadWrite);
         let after = reopened
             .rebuild_visibility()
             .await
             .expect("reopened visibility rebuild");
         assert_eq!(before, after);
         assert_eq!(after.current_heads()[0].event_id, *current.id());
         assert_eq!(after.visible_event_ids(), &[*deletion.id()]);
         assert_eq!(after.suppressed_event_ids(), &[*current.id()]);
         assert_eq!(after.superseded_event_ids(), &[*old.id()]);
         let page = reopened
             .query_visible(EventQuery::all(
                 EventQueryBounds::first(10).expect("bounds"),
             ))
             .await
             .expect("visible page");
         assert_eq!(page.items().len(), 1);
         assert_eq!(page.items()[0].event().id(), deletion.id());
         assert_eq!(reopened.status().await.expect("status").visible_events(), 1);
     }
 
     #[tokio::test]
     async fn corrupt_rows_fail_closed_and_source_history_is_immutable() {
         let generation = SourceGeneration::new([10; 32]).expect("generation");
         let store = store(generation).await;
         let event = signed_event(
             "{\"display_name\":\"Corruption Probe\",\"bot\":false}",
             false,
         );
         store
             .admit(EventAdmission::raw(observed(event.clone(), 30, None)))
             .await
             .expect("event");
 
         let wrong_generation = SourceGeneration::new([11; 32]).expect("generation");
         let mismatched_store = SqliteStorage::new(
             store.pool.clone(),
             wrong_generation,
             EventStoreMode::ReadWrite,
         );
         assert_eq!(
             mismatched_store
                 .admit(EventAdmission::raw(observed(event, 31, None)))
                 .await,
             Err(Error::CorruptStoredEvent)
         );
 
         assert!(
             sqlx::query("DELETE FROM radroots_runtime_events")
                 .execute(&store.pool)
                 .await
                 .is_err()
         );
         assert!(
             sqlx::query("DELETE FROM radroots_runtime_source_generations")
                 .execute(&store.pool)
                 .await
                 .is_err()
         );
         sqlx::query("DROP TRIGGER radroots_runtime_events_contract_metadata_guard")
             .execute(&store.pool)
             .await
             .expect("drop metadata guard for corruption probe");
         sqlx::query(
             "UPDATE radroots_runtime_events
              SET admitted_contract_id = 'radroots.social.geochat.v1'",
         )
         .execute(&store.pool)
         .await
         .expect("forge contract metadata");
         assert_eq!(
             store
                 .query_raw(EventQuery::all(EventQueryBounds::first(1).expect("bounds"),))
                 .await,
             Err(Error::CorruptStoredEvent)
         );
         sqlx::query(
             "UPDATE radroots_runtime_events
              SET admitted_contract_id = 'radroots.social.geochat.v1',
                  admitted_registry_version = 7",
         )
         .execute(&store.pool)
         .await
         .expect("forge mismatched selected contract");
         assert_eq!(
             store
                 .query_raw(EventQuery::all(EventQueryBounds::first(1).expect("bounds"),))
                 .await,
             Err(Error::CorruptStoredEvent)
         );
         sqlx::query(
             "UPDATE radroots_runtime_events
              SET admitted_registry_version = 6",
         )
         .execute(&store.pool)
         .await
         .expect("forge registry version");
         assert_eq!(
             store
                 .query_raw(EventQuery::all(EventQueryBounds::first(1).expect("bounds"),))
                 .await,
             Err(Error::CorruptStoredEvent)
         );
         sqlx::query("PRAGMA ignore_check_constraints = ON")
             .execute(&store.pool)
             .await
             .expect("disable checks for corruption probe");
         sqlx::query("UPDATE radroots_runtime_events SET admission_stage = 'corrupt'")
             .execute(&store.pool)
             .await
             .expect("forge corrupt stage");
         assert_eq!(
             store
                 .query_raw(EventQuery::all(EventQueryBounds::first(1).expect("bounds"),))
                 .await,
             Err(Error::CorruptStoredEvent)
         );
     }
 }