commit e9e390200b5d32818fcdbbe408dce32ea939b72b
parent 59cbf7960c4481a8b0f462df7b2cd136c5175e35
Author: triesap <tyson@radroots.org>
Date: Fri, 7 Aug 2026 12:24:10 +0000
feat(mobile): materialize Today ingest and query
- persist deterministic Today documents and immutable snapshots
- enrich ranked cards with profiles, threads, media, and author overlays
- add SQLite schema v12 and backend-neutral projection storage
- cover rebuild, cursor, corruption, concurrency, and downstream contracts
Diffstat:
17 files changed, 3161 insertions(+), 60 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -3574,6 +3574,7 @@ dependencies = [
"radroots_identity",
"radroots_sdk",
"radroots_storage",
+ "radroots_transport",
"serde",
"serde_json",
"sha2",
diff --git a/crates/mobile_core/Cargo.toml b/crates/mobile_core/Cargo.toml
@@ -46,5 +46,6 @@ radroots_blossom = { workspace = true, default-features = false, features = [
"std",
] }
radroots_sdk = { workspace = true, features = ["memory", "sqlite"] }
+radroots_transport = { workspace = true, default-features = false, features = ["std"] }
tempfile = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt"] }
diff --git a/crates/mobile_core/src/runtime/product_surface.rs b/crates/mobile_core/src/runtime/product_surface.rs
@@ -11,6 +11,7 @@ mod identity;
mod model;
mod projection;
mod ranking;
+mod today;
pub use authoring::{
CreateAsk, CreateEvent, CreateFoodAvailability, CreatePhotoUpdate, CreateUpdate,
@@ -24,11 +25,16 @@ pub use cursor::{CursorError, CursorScope, TodayCursor, TodayCursorPosition};
pub use identity::{CARD_ID_SCHEMA_VERSION, CardId, CardIdError, CardSourceIdentity};
pub use model::{
AddCommandType, CANONICAL_ADD_COMMAND_TYPES, CANONICAL_CARD_ADD_PARITY,
- CANONICAL_TODAY_CARD_TYPES, CardAddParity, CardLifecycleState, ClassifiedCard, MediaReference,
- MediaVerificationState, ProfileSummary, SupportingProfile, ThreadReference, TodayCardType,
+ CANONICAL_TODAY_CARD_TYPES, CardAddParity, CardLifecycleState, ClassifiedCard,
+ LocalAuthorOverlay, MeSnapshot, MediaReference, MediaVerificationState, ProfileSummary,
+ SearchResult, SearchResultType, SupportingProfile, ThreadEntry, ThreadReference, TodayCard,
+ TodayCardType, TodayPage,
};
pub use projection::{ProductEventClassification, ProductEventExclusion, classify_admitted_event};
pub use ranking::{RankError, TODAY_RANK_SCHEMA_VERSION, TimeRelevance, TodayRank, TodayRankInput};
+pub use today::{
+ TodayError, TodayIngestReceipt, TodayPageRequest, TodayProjectionUpdate, TodayRefreshReceipt,
+};
use super::RadrootsRuntime;
diff --git a/crates/mobile_core/src/runtime/product_surface/cursor.rs b/crates/mobile_core/src/runtime/product_surface/cursor.rs
@@ -102,25 +102,26 @@ impl TodayCursor {
}
pub fn decode(value: &str, expected: &CursorScope) -> Result<TodayCursorPosition, CursorError> {
- let encoded = value
- .strip_prefix(CURSOR_PREFIX)
- .ok_or(CursorError::Malformed)?;
- if encoded.len() % 2 != 0
- || !encoded
- .bytes()
- .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+ let (scope, position) = decode_unbound(value)?;
+ if scope.context_id != expected.context_id
+ || scope.context_generation != expected.context_generation
{
- return Err(CursorError::Malformed);
+ return Err(CursorError::ContextMismatch);
}
- let bytes = hex::decode(encoded).map_err(|_| CursorError::Malformed)?;
- if bytes.len() < FIXED_PAYLOAD_BYTES + DIGEST_BYTES {
- return Err(CursorError::Malformed);
+ if scope.as_of != expected.as_of {
+ return Err(CursorError::SnapshotMismatch);
}
- let (payload, observed_digest) = bytes.split_at(bytes.len() - DIGEST_BYTES);
- if cursor_digest(payload).as_slice() != observed_digest {
- return Err(CursorError::Integrity);
+ if scope.store_generation != expected.store_generation
+ || scope.projection_generation != expected.projection_generation
+ {
+ return Err(CursorError::Stale);
}
- decode_payload(payload, expected)
+ Ok(position)
+ }
+
+ /// Recovers the integrity-checked frozen scope carried by an opaque cursor.
+ pub fn scope(value: &str) -> Result<CursorScope, CursorError> {
+ decode_unbound(value).map(|(scope, _)| scope)
}
pub fn as_str(&self) -> &str {
@@ -128,10 +129,29 @@ impl TodayCursor {
}
}
-fn decode_payload(
- payload: &[u8],
- expected: &CursorScope,
-) -> Result<TodayCursorPosition, CursorError> {
+fn decode_unbound(value: &str) -> Result<(CursorScope, TodayCursorPosition), CursorError> {
+ let encoded = value
+ .strip_prefix(CURSOR_PREFIX)
+ .ok_or(CursorError::Malformed)?;
+ if encoded.len() % 2 != 0
+ || !encoded
+ .bytes()
+ .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+ {
+ return Err(CursorError::Malformed);
+ }
+ let bytes = hex::decode(encoded).map_err(|_| CursorError::Malformed)?;
+ if bytes.len() < FIXED_PAYLOAD_BYTES + DIGEST_BYTES {
+ return Err(CursorError::Malformed);
+ }
+ let (payload, observed_digest) = bytes.split_at(bytes.len() - DIGEST_BYTES);
+ if cursor_digest(payload).as_slice() != observed_digest {
+ return Err(CursorError::Integrity);
+ }
+ decode_payload(payload)
+}
+
+fn decode_payload(payload: &[u8]) -> Result<(CursorScope, TodayCursorPosition), CursorError> {
let mut decoder = Decoder::new(payload);
let cursor_version = decoder.u16()?;
let rank_schema_version = decoder.u16()?;
@@ -161,27 +181,25 @@ fn decode_payload(
if !decoder.is_finished() {
return Err(CursorError::Malformed);
}
- if context_id != expected.context_id || context_generation != expected.context_generation {
- return Err(CursorError::ContextMismatch);
- }
- if as_of != expected.as_of {
- return Err(CursorError::SnapshotMismatch);
- }
- if store_generation != expected.store_generation
- || projection_generation != expected.projection_generation
- {
- return Err(CursorError::Stale);
- }
- Ok(TodayCursorPosition {
- rank: TodayRank {
- schema_version: rank_schema_version,
- algorithm_version: rank_algorithm_version,
- context_rank,
- time_relevance_rank,
- effective_at,
- card_id,
+ Ok((
+ CursorScope {
+ context_id: context_id.to_owned(),
+ context_generation,
+ as_of,
+ store_generation,
+ projection_generation,
+ },
+ TodayCursorPosition {
+ rank: TodayRank {
+ schema_version: rank_schema_version,
+ algorithm_version: rank_algorithm_version,
+ context_rank,
+ time_relevance_rank,
+ effective_at,
+ card_id,
+ },
},
- })
+ ))
}
fn validate_context_id(value: &str) -> Result<(), CursorError> {
@@ -288,6 +306,7 @@ mod tests {
TodayCursor::decode(cursor.as_str(), &scope()).expect("decode"),
position()
);
+ assert_eq!(TodayCursor::scope(cursor.as_str()).expect("scope"), scope());
}
#[test]
diff --git a/crates/mobile_core/src/runtime/product_surface/model.rs b/crates/mobile_core/src/runtime/product_surface/model.rs
@@ -124,7 +124,12 @@ pub struct ProfileSummary {
pub author_pubkey: String,
pub name: Option<String>,
pub display_name: Option<String>,
+ pub about: Option<String>,
pub picture: Option<MediaReference>,
+ pub banner: Option<MediaReference>,
+ pub nip05: Option<String>,
+ pub website: Option<String>,
+ pub lightning_address: Option<String>,
}
/// Thread enrichment identity; replies and comments never become top-level cards.
@@ -136,6 +141,26 @@ pub struct ThreadReference {
pub parent_event_id: String,
}
+/// One admitted reply or comment attached to its canonical root.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ThreadEntry {
+ pub event_id: String,
+ pub author_pubkey: String,
+ pub content: String,
+ pub authored_at: u64,
+ pub reference: ThreadReference,
+ pub author_profile: Option<ProfileSummary>,
+}
+
+/// Durable local-only authored state overlaid without changing event truth.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct LocalAuthorOverlay {
+ pub operation_id: String,
+ pub state: String,
+}
+
/// Current rendering state derived from standard event semantics.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "PascalCase")]
@@ -156,8 +181,12 @@ pub struct ClassifiedCard {
pub source_address: Option<String>,
pub author_pubkey: String,
pub contract_id: String,
+ pub title: Option<String>,
pub content: String,
pub authored_at: u64,
+ pub effective_at: u64,
+ pub event_start: Option<u64>,
+ pub event_end: Option<u64>,
pub context_rank: ContextRank,
pub inclusion_reason: String,
pub media: Vec<MediaReference>,
@@ -165,6 +194,51 @@ pub struct ClassifiedCard {
pub rank: Option<TodayRank>,
}
+/// One fully enriched Today card returned to a host.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TodayCard {
+ pub card: ClassifiedCard,
+ pub author_profile: Option<ProfileSummary>,
+ pub thread: Vec<ThreadEntry>,
+ pub local_overlay: Option<LocalAuthorOverlay>,
+}
+
+/// One frozen, cursor-addressable Today page.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TodayPage {
+ pub as_of: u64,
+ pub items: Vec<TodayCard>,
+ pub next_cursor: Option<String>,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum SearchResultType {
+ Card,
+ Profile,
+}
+
+/// One local search result governed by the same current projection as Today.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SearchResult {
+ pub result_type: SearchResultType,
+ pub stable_id: String,
+ pub card: Option<TodayCard>,
+ pub profile: Option<ProfileSummary>,
+}
+
+/// Active identity attribution and its current visible Phase 1 cards.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MeSnapshot {
+ pub public_key: String,
+ pub profile: Option<ProfileSummary>,
+ pub cards: Vec<TodayCard>,
+}
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/crates/mobile_core/src/runtime/product_surface/projection.rs b/crates/mobile_core/src/runtime/product_surface/projection.rs
@@ -145,6 +145,22 @@ fn card(
identifier,
} => Some(format!("{kind}:{author_pubkey}:{identifier}")),
};
+ let tags = event.tags_as_vec();
+ let title = tag_value(&tags, &["title", "name"]);
+ let (effective_at, event_start, event_end) = match card_type {
+ TodayCardType::Event => {
+ let start = tag_time(&tags, "start").unwrap_or_else(|| event.created_at_u64());
+ (start, Some(start), tag_time(&tags, "end"))
+ }
+ TodayCardType::FoodAvailability => (
+ tag_time(&tags, "published_at").unwrap_or_else(|| event.created_at_u64()),
+ None,
+ None,
+ ),
+ TodayCardType::Update | TodayCardType::PhotoUpdate | TodayCardType::Ask => {
+ (event.created_at_u64(), None, None)
+ }
+ };
ProductEventClassification::Card(Box::new(ClassifiedCard {
schema_version: CLASSIFIED_CARD_SCHEMA_VERSION,
card_id: CardId::derive(card_type, &source),
@@ -153,8 +169,12 @@ fn card(
source_address,
author_pubkey: event.author().to_hex(),
contract_id: admitted.contract_id().to_owned(),
+ title,
content: event.content().to_owned(),
authored_at: event.created_at_u64(),
+ effective_at,
+ event_start,
+ event_end,
context_rank: context.rank,
inclusion_reason: context.reason.to_owned(),
media,
@@ -163,6 +183,28 @@ fn card(
}))
}
+fn tag_value(tags: &[Vec<String>], names: &[&str]) -> Option<String> {
+ tags.iter().find_map(|tag| {
+ names
+ .contains(&tag.first()?.as_str())
+ .then(|| tag.get(1).cloned())
+ .flatten()
+ })
+}
+
+fn tag_time(tags: &[Vec<String>], name: &str) -> Option<u64> {
+ let value = tag_value(tags, &[name])?;
+ value.parse().ok().or_else(|| {
+ chrono::NaiveDate::parse_from_str(&value, "%Y-%m-%d")
+ .ok()?
+ .and_hms_opt(0, 0, 0)?
+ .and_utc()
+ .timestamp()
+ .try_into()
+ .ok()
+ })
+}
+
fn post_media(
projection: &radroots_event_codec::decode::post::RadrootsInboundPostProjection,
) -> Vec<MediaReference> {
diff --git a/crates/mobile_core/src/runtime/product_surface/today.rs b/crates/mobile_core/src/runtime/product_surface/today.rs
@@ -0,0 +1,2114 @@
+use std::collections::BTreeMap;
+
+use radroots_event_codec::{
+ admission::{RadrootsAdmittedEvent, admit_verified_event},
+ verify::verify_nip01_event,
+};
+use radroots_storage::{
+ EventStore, ProjectionStore,
+ event::{
+ AdmissionReceipt, EventAdmission, EventPosition, EventQuery, EventQueryBounds,
+ EventSequence,
+ },
+ projection::{
+ ProjectionCheckpoint, ProjectionDocument, ProjectionGeneration, ProjectionId,
+ ProjectionSnapshot,
+ },
+};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use thiserror::Error;
+
+use super::{
+ CardId, CardLifecycleState, ClassifiedCard, CursorError, CursorScope, LocalAuthorOverlay,
+ LocalNetwork, LocalityEvidence, MeSnapshot, MediaReference, MediaVerificationState,
+ ProductEventClassification, ProfileSummary, SearchResult, SearchResultType, SupportingProfile,
+ ThreadEntry, ThreadReference, TimeRelevance, TodayCard, TodayCardType, TodayCursor,
+ TodayCursorPosition, TodayPage, TodayRank, TodayRankInput, classify_admitted_event,
+};
+use crate::runtime::RadrootsRuntime;
+
+const TODAY_PROJECTION_ID: &str = "radroots.today.v1";
+const TODAY_PROJECTION_DOCUMENT_SCHEMA_VERSION: u16 = 1;
+const TODAY_SNAPSHOT_SCHEMA_VERSION: u16 = 1;
+const TODAY_PAGE_LIMIT_MAX: u16 = 100;
+const TODAY_SEARCH_LIMIT_MAX: u16 = 100;
+const PROJECTION_GENERATION_DOMAIN: &[u8] = b"radroots.today-projection.v1\0";
+const PROJECTION_CONTENT_DOMAIN: &[u8] = b"radroots.today-content-generation.v1\0";
+const PROJECTION_DOCUMENT_KEY_DOMAIN: &[u8] = b"radroots.today-document-key.v1\0";
+const SNAPSHOT_ID_DOMAIN: &[u8] = b"radroots.today-snapshot-id.v1\0";
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum TodayProjectionUpdate {
+ Incremental,
+ Rebuild,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TodayRefreshReceipt {
+ pub update: TodayProjectionUpdate,
+ pub source_events: u64,
+ pub visible_cards: u64,
+ pub profiles: u64,
+ pub thread_entries: u64,
+ pub content_generation: u64,
+ pub changed: bool,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TodayIngestReceipt {
+ pub event_id: String,
+ pub disposition: String,
+ pub source_sequence: u64,
+ pub projection: TodayRefreshReceipt,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct TodayPageRequest {
+ pub limit: u16,
+ pub as_of: Option<u64>,
+ pub cursor: Option<String>,
+}
+
+impl TodayPageRequest {
+ pub const fn first(limit: u16, as_of: u64) -> Self {
+ Self {
+ limit,
+ as_of: Some(as_of),
+ cursor: None,
+ }
+ }
+
+ pub fn after(limit: u16, cursor: String) -> Self {
+ Self {
+ limit,
+ as_of: None,
+ cursor: Some(cursor),
+ }
+ }
+}
+
+#[derive(Debug, Error)]
+pub enum TodayError {
+ #[error("today runtime is unavailable")]
+ RuntimeUnavailable,
+ #[error("today request is invalid")]
+ InvalidRequest,
+ #[error("today projection has not been refreshed")]
+ ProjectionMissing,
+ #[error("today frozen snapshot is unavailable")]
+ SnapshotMissing,
+ #[error("today cursor position is absent from its frozen snapshot")]
+ CursorPositionMissing,
+ #[error("today event was not admitted as visible")]
+ EventNotVisible,
+ #[error("today projection state is corrupt")]
+ CorruptProjection,
+ #[error(transparent)]
+ Cursor(#[from] CursorError),
+ #[error(transparent)]
+ Storage(#[from] radroots_storage::Error),
+ #[error("today projection serialization failed")]
+ Serialization,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ProjectedCard {
+ card: ClassifiedCard,
+ locality: Vec<LocalityTag>,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Ord, PartialOrd, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct LocalityTag {
+ kind: String,
+ value: String,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct TodayProjectionState {
+ schema_version: u16,
+ context_id: String,
+ context_generation: u64,
+ store_generation: [u8; 32],
+ source_events: u64,
+ content_generation: u64,
+ cards: Vec<ProjectedCard>,
+ profiles: BTreeMap<String, ProfileSummary>,
+ thread: Vec<ThreadEntry>,
+ overlays: BTreeMap<String, LocalAuthorOverlay>,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct FrozenTodaySnapshot {
+ schema_version: u16,
+ context_id: String,
+ context_generation: u64,
+ as_of: u64,
+ store_generation: [u8; 32],
+ projection_generation: u64,
+ items: Vec<TodayCard>,
+}
+
+impl RadrootsRuntime {
+ /// Durably admits one already verified and visibility-authorized relay observation,
+ /// then advances the selected LocalNetwork projection.
+ pub async fn phase1_ingest_visible(
+ &self,
+ admission: EventAdmission,
+ context: &LocalNetwork,
+ now_unix_seconds: u64,
+ ) -> Result<TodayIngestReceipt, TodayError> {
+ if admission.visible_event().is_none() {
+ return Err(TodayError::EventNotVisible);
+ }
+ let storage = self
+ .client
+ .storage()
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ let receipt = EventStore::admit(storage, admission).await?;
+ let projection = self
+ .phase1_refresh_today(
+ context,
+ now_unix_seconds,
+ TodayProjectionUpdate::Incremental,
+ )
+ .await?;
+ Ok(ingest_receipt(receipt, projection))
+ }
+
+ /// Materializes current visible event truth for one LocalNetwork.
+ pub async fn phase1_refresh_today(
+ &self,
+ context: &LocalNetwork,
+ now_unix_seconds: u64,
+ update: TodayProjectionUpdate,
+ ) -> Result<TodayRefreshReceipt, TodayError> {
+ if now_unix_seconds == 0 {
+ return Err(TodayError::InvalidRequest);
+ }
+ let requested_updated_at_unix_ms = now_unix_seconds
+ .checked_mul(1_000)
+ .ok_or(TodayError::InvalidRequest)?;
+ let storage = self
+ .client
+ .storage()
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ if update == TodayProjectionUpdate::Rebuild {
+ EventStore::rebuild_visibility(storage).await?;
+ }
+ let event_status = EventStore::status(storage).await?;
+ let generation = projection_generation()?;
+ let projection_id = projection_id()?;
+ let key = projection_document_key(context);
+ let prior = ProjectionStore::projection_document(
+ storage,
+ projection_id.clone(),
+ generation,
+ key.clone(),
+ )
+ .await?
+ .map(|document| decode_state(document.value()))
+ .transpose()?;
+
+ if update == TodayProjectionUpdate::Incremental
+ && prior
+ .as_ref()
+ .is_some_and(|state| state.source_events == event_status.raw_events())
+ {
+ let state = prior.expect("checked present");
+ return Ok(refresh_receipt(update, &state, false));
+ }
+
+ let visible = query_all_visible(storage).await?;
+ let local_media = prior.as_ref().map(local_media_evidence).unwrap_or_default();
+ let overlays = prior.map_or_else(BTreeMap::new, |state| state.overlays);
+ let mut state = project_state(
+ context,
+ event_status.generation().as_bytes(),
+ event_status.raw_events(),
+ visible,
+ overlays,
+ )?;
+ apply_local_media_evidence(&mut state, &local_media);
+ state.content_generation = content_generation(&state)?;
+ let encoded = encode(&state)?;
+ let changed = ProjectionStore::projection_document(
+ storage,
+ projection_id.clone(),
+ generation,
+ key.clone(),
+ )
+ .await?
+ .is_none_or(|document| document.value() != encoded);
+ ProjectionStore::put_projection_document(
+ storage,
+ projection_id.clone(),
+ generation,
+ ProjectionDocument::new(key, encoded)?,
+ )
+ .await?;
+
+ let source_position = if event_status.raw_events() == 0 {
+ None
+ } else {
+ Some(EventPosition::new(
+ event_status.generation(),
+ EventSequence::new(event_status.raw_events())?,
+ ))
+ };
+ let prior_updated_at = ProjectionStore::status(storage, projection_id.clone())
+ .await?
+ .and_then(|status| {
+ status
+ .checkpoint()
+ .map(ProjectionCheckpoint::updated_at_unix_ms)
+ })
+ .unwrap_or(0);
+ let updated_at_unix_ms = requested_updated_at_unix_ms.max(prior_updated_at);
+ ProjectionStore::checkpoint(
+ storage,
+ ProjectionCheckpoint::new(
+ projection_id,
+ generation,
+ source_position,
+ event_status.raw_events(),
+ updated_at_unix_ms,
+ )?,
+ )
+ .await?;
+ Ok(refresh_receipt(update, &state, changed))
+ }
+
+ /// Returns one page from a durable frozen Today snapshot.
+ pub async fn phase1_today_page(
+ &self,
+ context: &LocalNetwork,
+ request: TodayPageRequest,
+ ) -> Result<TodayPage, TodayError> {
+ if request.limit == 0 || request.limit > TODAY_PAGE_LIMIT_MAX {
+ return Err(TodayError::InvalidRequest);
+ }
+ let storage = self
+ .client
+ .storage()
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ let event_status = EventStore::status(storage).await?;
+ let algorithm_generation = projection_generation()?;
+ let projection_id = projection_id()?;
+
+ let (scope, snapshot, after) = if let Some(cursor) = request.cursor.as_deref() {
+ let scope = TodayCursor::scope(cursor)?;
+ if scope.context_id != context.id || scope.context_generation != context.generation {
+ return Err(CursorError::ContextMismatch.into());
+ }
+ if request.as_of.is_some_and(|as_of| as_of != scope.as_of) {
+ return Err(CursorError::SnapshotMismatch.into());
+ }
+ if scope.store_generation != *event_status.generation().as_bytes() {
+ return Err(CursorError::Stale.into());
+ }
+ let position = TodayCursor::decode(cursor, &scope)?;
+ let snapshot = load_snapshot(storage, projection_id, algorithm_generation, &scope)
+ .await?
+ .ok_or(TodayError::SnapshotMissing)?;
+ (scope, snapshot, Some(position.rank))
+ } else {
+ let as_of = request
+ .as_of
+ .filter(|value| *value != 0)
+ .ok_or(TodayError::InvalidRequest)?;
+ let state = load_state(storage, context, algorithm_generation)
+ .await?
+ .ok_or(TodayError::ProjectionMissing)?;
+ if state.store_generation != *event_status.generation().as_bytes() {
+ return Err(CursorError::Stale.into());
+ }
+ let scope = CursorScope::new(
+ context.id.clone(),
+ context.generation,
+ as_of,
+ state.store_generation,
+ state.content_generation,
+ )?;
+ let snapshot = frozen_snapshot(&state, context, as_of)?;
+ persist_snapshot(storage, algorithm_generation, &scope, &snapshot).await?;
+ (scope, snapshot, None)
+ };
+
+ page_from_snapshot(snapshot, scope, after, request.limit)
+ }
+
+ /// Searches the current local projection using Today visibility and context rules.
+ pub async fn phase1_search(
+ &self,
+ context: &LocalNetwork,
+ query: &str,
+ limit: u16,
+ as_of: u64,
+ ) -> Result<Vec<SearchResult>, TodayError> {
+ if limit == 0 || limit > TODAY_SEARCH_LIMIT_MAX || as_of == 0 {
+ return Err(TodayError::InvalidRequest);
+ }
+ let needle = query.trim().to_lowercase();
+ if needle.is_empty() || needle.len() > 256 || query.chars().any(char::is_control) {
+ return Err(TodayError::InvalidRequest);
+ }
+ let storage = self
+ .client
+ .storage()
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ let state = load_state(storage, context, projection_generation()?)
+ .await?
+ .ok_or(TodayError::ProjectionMissing)?;
+ let cards = ranked_cards(&state, context, as_of)?;
+ let mut results = Vec::new();
+ for card in cards {
+ let searchable = format!(
+ "{} {} {} {}",
+ card.card.title.as_deref().unwrap_or(""),
+ card.card.content,
+ card.card.author_pubkey,
+ card.author_profile
+ .as_ref()
+ .and_then(|profile| profile.display_name.as_deref().or(profile.name.as_deref()))
+ .unwrap_or("")
+ )
+ .to_lowercase();
+ if searchable.contains(&needle) {
+ results.push(SearchResult {
+ result_type: SearchResultType::Card,
+ stable_id: card.card.card_id.to_hex(),
+ card: Some(card),
+ profile: None,
+ });
+ if results.len() == usize::from(limit) {
+ return Ok(results);
+ }
+ }
+ }
+ for profile in state.profiles.values() {
+ let searchable = format!(
+ "{} {} {} {} {}",
+ profile.name.as_deref().unwrap_or(""),
+ profile.display_name.as_deref().unwrap_or(""),
+ profile.about.as_deref().unwrap_or(""),
+ profile.website.as_deref().unwrap_or(""),
+ profile.lightning_address.as_deref().unwrap_or("")
+ )
+ .to_lowercase();
+ if searchable.contains(&needle) {
+ results.push(SearchResult {
+ result_type: SearchResultType::Profile,
+ stable_id: profile.author_pubkey.clone(),
+ card: None,
+ profile: Some(profile.clone()),
+ });
+ if results.len() == usize::from(limit) {
+ break;
+ }
+ }
+ }
+ Ok(results)
+ }
+
+ /// Returns current active-identity attribution and visible Phase 1 content.
+ pub async fn phase1_me(
+ &self,
+ context: &LocalNetwork,
+ public_key: &str,
+ as_of: u64,
+ ) -> Result<MeSnapshot, TodayError> {
+ if !valid_public_key(public_key) || as_of == 0 {
+ return Err(TodayError::InvalidRequest);
+ }
+ if self
+ .authenticated_store_public_key_hex()
+ .is_some_and(|store_key| store_key != public_key)
+ {
+ return Err(TodayError::InvalidRequest);
+ }
+ let storage = self
+ .client
+ .storage()
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ let state = load_state(storage, context, projection_generation()?)
+ .await?
+ .ok_or(TodayError::ProjectionMissing)?;
+ let cards = ranked_cards(&state, context, as_of)?
+ .into_iter()
+ .filter(|card| card.card.author_pubkey == public_key)
+ .collect();
+ Ok(MeSnapshot {
+ public_key: public_key.to_owned(),
+ profile: state.profiles.get(public_key).cloned(),
+ cards,
+ })
+ }
+
+ /// Updates local byte-verification evidence without changing card classification.
+ pub async fn phase1_set_media_verification(
+ &self,
+ context: &LocalNetwork,
+ url: &str,
+ verification: MediaVerificationState,
+ ) -> Result<bool, TodayError> {
+ if url.is_empty() || url.len() > 8_192 || url.chars().any(char::is_control) {
+ return Err(TodayError::InvalidRequest);
+ }
+ let storage = self
+ .client
+ .storage()
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ let generation = projection_generation()?;
+ let mut state = load_state(storage, context, generation)
+ .await?
+ .ok_or(TodayError::ProjectionMissing)?;
+ let mut changed = false;
+ for projected in &mut state.cards {
+ for media in &mut projected.card.media {
+ if media.url == url && media.verification != verification {
+ media.verification = verification;
+ changed = true;
+ }
+ }
+ }
+ for profile in state.profiles.values_mut() {
+ for media in [&mut profile.picture, &mut profile.banner]
+ .into_iter()
+ .flatten()
+ {
+ if media.url == url && media.verification != verification {
+ media.verification = verification;
+ changed = true;
+ }
+ }
+ }
+ if changed {
+ refresh_thread_profiles(&mut state);
+ state.content_generation = content_generation(&state)?;
+ store_state(storage, context, generation, &state).await?;
+ }
+ Ok(changed)
+ }
+
+ /// Persists active-author delivery state as a local-only Today overlay.
+ pub async fn phase1_set_local_author_overlay(
+ &self,
+ context: &LocalNetwork,
+ card_id: CardId,
+ overlay: Option<LocalAuthorOverlay>,
+ ) -> Result<(), TodayError> {
+ let storage = self
+ .client
+ .storage()
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ let generation = projection_generation()?;
+ let mut state = load_state(storage, context, generation)
+ .await?
+ .ok_or(TodayError::ProjectionMissing)?;
+ if overlay.as_ref().is_some_and(|overlay| {
+ overlay.operation_id.is_empty()
+ || overlay.operation_id.len() > 256
+ || overlay.state.is_empty()
+ || overlay.state.len() > 96
+ || overlay.state.chars().any(char::is_control)
+ }) {
+ return Err(TodayError::InvalidRequest);
+ }
+ let key = card_id.to_hex();
+ let card = state
+ .cards
+ .iter()
+ .find(|projected| projected.card.card_id == card_id)
+ .ok_or(TodayError::InvalidRequest)?;
+ if self
+ .authenticated_store_public_key_hex()
+ .is_some_and(|store_key| store_key != card.card.author_pubkey)
+ {
+ return Err(TodayError::InvalidRequest);
+ }
+ if let Some(overlay) = overlay {
+ state.overlays.insert(key, overlay);
+ } else {
+ state.overlays.remove(&key);
+ }
+ state.content_generation = content_generation(&state)?;
+ store_state(storage, context, generation, &state).await
+ }
+}
+
+fn ingest_receipt(
+ receipt: AdmissionReceipt,
+ projection: TodayRefreshReceipt,
+) -> TodayIngestReceipt {
+ TodayIngestReceipt {
+ event_id: receipt.event_id().to_hex(),
+ disposition: format!("{:?}", receipt.disposition()).to_lowercase(),
+ source_sequence: receipt.position().sequence().get(),
+ projection,
+ }
+}
+
+fn refresh_receipt(
+ update: TodayProjectionUpdate,
+ state: &TodayProjectionState,
+ changed: bool,
+) -> TodayRefreshReceipt {
+ TodayRefreshReceipt {
+ update,
+ source_events: state.source_events,
+ visible_cards: state.cards.len().try_into().unwrap_or(u64::MAX),
+ profiles: state.profiles.len().try_into().unwrap_or(u64::MAX),
+ thread_entries: state.thread.len().try_into().unwrap_or(u64::MAX),
+ content_generation: state.content_generation,
+ changed,
+ }
+}
+
+fn local_media_evidence(state: &TodayProjectionState) -> BTreeMap<String, MediaVerificationState> {
+ let mut evidence = state
+ .cards
+ .iter()
+ .flat_map(|projected| projected.card.media.iter())
+ .filter(|media| media.verification != MediaVerificationState::Unavailable)
+ .map(|media| (media.url.clone(), media.verification))
+ .collect::<BTreeMap<_, _>>();
+ for profile in state.profiles.values() {
+ for media in [&profile.picture, &profile.banner].into_iter().flatten() {
+ if media.verification != MediaVerificationState::Unavailable {
+ evidence.insert(media.url.clone(), media.verification);
+ }
+ }
+ }
+ evidence
+}
+
+fn apply_local_media_evidence(
+ state: &mut TodayProjectionState,
+ evidence: &BTreeMap<String, MediaVerificationState>,
+) {
+ for projected in &mut state.cards {
+ for media in &mut projected.card.media {
+ if let Some(verification) = evidence.get(&media.url) {
+ media.verification = *verification;
+ }
+ }
+ }
+ for profile in state.profiles.values_mut() {
+ for media in [&mut profile.picture, &mut profile.banner]
+ .into_iter()
+ .flatten()
+ {
+ if let Some(verification) = evidence.get(&media.url) {
+ media.verification = *verification;
+ }
+ }
+ }
+ refresh_thread_profiles(state);
+}
+
+fn refresh_thread_profiles(state: &mut TodayProjectionState) {
+ for entry in &mut state.thread {
+ entry.author_profile = state.profiles.get(&entry.author_pubkey).cloned();
+ }
+}
+
+async fn query_all_visible(
+ storage: &dyn radroots_storage::Storage,
+) -> Result<Vec<radroots_storage::event::StoredVisibleEvent>, TodayError> {
+ let mut items = Vec::new();
+ let mut after = None;
+ loop {
+ let mut bounds = EventQueryBounds::first(radroots_storage::event::EVENT_QUERY_LIMIT_MAX)?;
+ if let Some(cursor) = after {
+ bounds = bounds.after(cursor);
+ }
+ let page = EventStore::query_visible(storage, EventQuery::all(bounds)).await?;
+ items.extend_from_slice(page.items());
+ let Some(next) = page.next_cursor() else {
+ break;
+ };
+ after = Some(next);
+ }
+ Ok(items)
+}
+
+fn project_state(
+ context: &LocalNetwork,
+ store_generation: &[u8; 32],
+ source_events: u64,
+ visible: Vec<radroots_storage::event::StoredVisibleEvent>,
+ overlays: BTreeMap<String, LocalAuthorOverlay>,
+) -> Result<TodayProjectionState, TodayError> {
+ let mut cards = Vec::new();
+ let mut profiles = BTreeMap::new();
+ let mut thread = Vec::new();
+ for stored in visible {
+ let verified = verify_nip01_event(stored.event().envelope().clone())
+ .map_err(|_| TodayError::CorruptProjection)?;
+ let admitted = admit_verified_event(verified).map_err(|_| TodayError::CorruptProjection)?;
+ match &admitted {
+ RadrootsAdmittedEvent::Profile(profile) => {
+ profiles.insert(
+ profile.event().author().to_hex(),
+ profile_summary(&admitted)?,
+ );
+ }
+ RadrootsAdmittedEvent::Reply(_) => {
+ thread.push(reply_entry(&admitted)?);
+ }
+ RadrootsAdmittedEvent::Comment(_) => {
+ if let Some(entry) = comment_entry(&admitted) {
+ thread.push(entry);
+ }
+ }
+ _ => {
+ let locality = locality_tags(admitted.event().tags_as_vec());
+ let evidence = locality_evidence(context.locality.as_deref(), &locality);
+ if let ProductEventClassification::Card(card) =
+ classify_admitted_event(&admitted, context.admit(evidence))
+ {
+ cards.push(ProjectedCard {
+ card: *card,
+ locality,
+ });
+ }
+ }
+ }
+ }
+ cards.sort_by_key(|projected| projected.card.card_id);
+ thread.sort_by(|left, right| left.event_id.cmp(&right.event_id));
+ for entry in &mut thread {
+ entry.author_profile = profiles.get(&entry.author_pubkey).cloned();
+ }
+ Ok(TodayProjectionState {
+ schema_version: TODAY_PROJECTION_DOCUMENT_SCHEMA_VERSION,
+ context_id: context.id.clone(),
+ context_generation: context.generation,
+ store_generation: *store_generation,
+ source_events,
+ content_generation: 0,
+ cards,
+ profiles,
+ thread,
+ overlays,
+ })
+}
+
+fn profile_summary(admitted: &RadrootsAdmittedEvent) -> Result<ProfileSummary, TodayError> {
+ let RadrootsAdmittedEvent::Profile(profile) = admitted else {
+ return Err(TodayError::CorruptProjection);
+ };
+ let metadata = profile.metadata();
+ Ok(ProfileSummary {
+ author_pubkey: profile.event().author().to_hex(),
+ name: metadata.name().map(str::to_owned),
+ display_name: metadata.display_name().map(str::to_owned),
+ about: metadata.about().map(str::to_owned),
+ picture: metadata
+ .picture()
+ .map(|value| unverified_media(value.as_str())),
+ banner: metadata
+ .banner()
+ .map(|value| unverified_media(value.as_str())),
+ nip05: metadata.nip05().map(|value| value.as_str().to_owned()),
+ website: typed_profile_extra(metadata.raw_fields(), "website"),
+ lightning_address: typed_profile_extra(metadata.raw_fields(), "lud16"),
+ })
+}
+
+fn typed_profile_extra(fields: &BTreeMap<String, serde_json::Value>, key: &str) -> Option<String> {
+ fields
+ .get(key)
+ .and_then(serde_json::Value::as_str)
+ .filter(|value| {
+ !value.is_empty() && value.len() <= 2_048 && !value.chars().any(char::is_control)
+ })
+ .map(str::to_owned)
+}
+
+fn unverified_media(url: &str) -> MediaReference {
+ MediaReference {
+ url: url.to_owned(),
+ sha256: blossom_digest(url),
+ media_type: None,
+ width: None,
+ height: None,
+ byte_size: None,
+ alt: None,
+ verification: MediaVerificationState::Unavailable,
+ }
+}
+
+fn reply_entry(admitted: &RadrootsAdmittedEvent) -> Result<ThreadEntry, TodayError> {
+ let RadrootsAdmittedEvent::Reply(reply) = admitted else {
+ return Err(TodayError::CorruptProjection);
+ };
+ Ok(ThreadEntry {
+ event_id: reply.event().id_hex(),
+ author_pubkey: reply.event().author().to_hex(),
+ content: reply.event().content().to_owned(),
+ authored_at: reply.event().created_at_u64(),
+ reference: ThreadReference {
+ profile: SupportingProfile::Reply,
+ root: reply.projection().root().event_id().to_hex(),
+ parent_event_id: reply.projection().parent().event_id().to_hex(),
+ },
+ author_profile: None,
+ })
+}
+
+fn comment_entry(admitted: &RadrootsAdmittedEvent) -> Option<ThreadEntry> {
+ let RadrootsAdmittedEvent::Comment(comment) = admitted else {
+ return None;
+ };
+ let tags = comment.event().tags_as_vec();
+ let root = tag_value(&tags, &["E", "A"])?;
+ let parent = tag_value(&tags, &["e", "a"]).unwrap_or_else(|| root.clone());
+ Some(ThreadEntry {
+ event_id: comment.event().id_hex(),
+ author_pubkey: comment.event().author().to_hex(),
+ content: comment.event().content().to_owned(),
+ authored_at: comment.event().created_at_u64(),
+ reference: ThreadReference {
+ profile: SupportingProfile::Comment,
+ root,
+ parent_event_id: parent,
+ },
+ author_profile: None,
+ })
+}
+
+fn locality_tags(tags: Vec<Vec<String>>) -> Vec<LocalityTag> {
+ let mut locality = tags
+ .into_iter()
+ .filter_map(|tag| {
+ let kind = tag.first()?.as_str();
+ if !matches!(kind, "g" | "location") {
+ return None;
+ }
+ let value = tag.get(1)?.trim().to_lowercase();
+ (!value.is_empty()).then(|| LocalityTag {
+ kind: kind.to_owned(),
+ value,
+ })
+ })
+ .collect::<Vec<_>>();
+ locality.sort();
+ locality.dedup();
+ locality
+}
+
+fn locality_evidence(selected: Option<&str>, locality: &[LocalityTag]) -> LocalityEvidence {
+ let Some(selected) = selected else {
+ return LocalityEvidence::Missing;
+ };
+ if locality.is_empty() {
+ return LocalityEvidence::Missing;
+ }
+ let selected = selected.trim().to_lowercase();
+ if locality.iter().any(|tag| {
+ tag.value == selected
+ || (tag.kind == "g"
+ && (tag.value.starts_with(&selected) || selected.starts_with(&tag.value)))
+ }) {
+ LocalityEvidence::Match
+ } else {
+ LocalityEvidence::Nonmatch
+ }
+}
+
+fn ranked_cards(
+ state: &TodayProjectionState,
+ context: &LocalNetwork,
+ as_of: u64,
+) -> Result<Vec<TodayCard>, TodayError> {
+ let mut cards = Vec::new();
+ for projected in &state.cards {
+ let evidence = locality_evidence(context.locality.as_deref(), &projected.locality);
+ let admission = match context.admit(evidence) {
+ super::LocalNetworkAdmission::Included(admission) => admission,
+ super::LocalNetworkAdmission::Excluded { .. } => continue,
+ };
+ let mut card = projected.card.clone();
+ card.context_rank = admission.rank;
+ card.inclusion_reason = admission.reason.to_owned();
+ let time = match card.card_type {
+ TodayCardType::Update | TodayCardType::PhotoUpdate | TodayCardType::Ask => {
+ TimeRelevance::Published
+ }
+ TodayCardType::FoodAvailability => TimeRelevance::FoodAvailability {
+ active: card.lifecycle == CardLifecycleState::Active,
+ },
+ TodayCardType::Event => TimeRelevance::Event {
+ start: card.event_start.unwrap_or(card.effective_at),
+ end: card.event_end,
+ },
+ };
+ let rank = TodayRank::derive(TodayRankInput {
+ card_type: card.card_type,
+ context_rank: card.context_rank,
+ as_of,
+ effective_at: card.effective_at,
+ time,
+ card_id: card.card_id,
+ })
+ .map_err(|_| TodayError::CorruptProjection)?;
+ if card.card_type == TodayCardType::Event && rank.time_relevance_rank == 0 {
+ card.lifecycle = CardLifecycleState::Past;
+ }
+ card.rank = Some(rank);
+ let roots = [
+ card.source_event_id.as_str(),
+ card.source_address.as_deref().unwrap_or(""),
+ ];
+ let card_thread = state
+ .thread
+ .iter()
+ .filter(|entry| roots.contains(&entry.reference.root.as_str()))
+ .cloned()
+ .collect();
+ cards.push(TodayCard {
+ author_profile: state.profiles.get(&card.author_pubkey).cloned(),
+ local_overlay: state.overlays.get(&card.card_id.to_hex()).cloned(),
+ card,
+ thread: card_thread,
+ });
+ }
+ cards.sort_by_key(|card| card.card.rank.expect("assigned rank"));
+ Ok(cards)
+}
+
+fn frozen_snapshot(
+ state: &TodayProjectionState,
+ context: &LocalNetwork,
+ as_of: u64,
+) -> Result<FrozenTodaySnapshot, TodayError> {
+ Ok(FrozenTodaySnapshot {
+ schema_version: TODAY_SNAPSHOT_SCHEMA_VERSION,
+ context_id: context.id.clone(),
+ context_generation: context.generation,
+ as_of,
+ store_generation: state.store_generation,
+ projection_generation: state.content_generation,
+ items: ranked_cards(state, context, as_of)?,
+ })
+}
+
+fn page_from_snapshot(
+ snapshot: FrozenTodaySnapshot,
+ scope: CursorScope,
+ after: Option<TodayRank>,
+ limit: u16,
+) -> Result<TodayPage, TodayError> {
+ validate_snapshot(&snapshot, &scope)?;
+ let start = if let Some(after) = after {
+ snapshot
+ .items
+ .iter()
+ .position(|card| card.card.rank == Some(after))
+ .map(|index| index + 1)
+ .ok_or(TodayError::CursorPositionMissing)?
+ } else {
+ 0
+ };
+ let end = start
+ .saturating_add(usize::from(limit))
+ .min(snapshot.items.len());
+ let items = snapshot.items[start..end].to_vec();
+ let next_cursor = if end < snapshot.items.len() {
+ items
+ .last()
+ .and_then(|card| card.card.rank)
+ .map(|rank| TodayCursor::encode(&scope, TodayCursorPosition { rank }))
+ .transpose()?
+ .map(|cursor| cursor.as_str().to_owned())
+ } else {
+ None
+ };
+ Ok(TodayPage {
+ as_of: snapshot.as_of,
+ items,
+ next_cursor,
+ })
+}
+
+fn validate_snapshot(
+ snapshot: &FrozenTodaySnapshot,
+ scope: &CursorScope,
+) -> Result<(), TodayError> {
+ if snapshot.schema_version != TODAY_SNAPSHOT_SCHEMA_VERSION
+ || snapshot.context_id != scope.context_id
+ || snapshot.context_generation != scope.context_generation
+ || snapshot.as_of != scope.as_of
+ || snapshot.store_generation != scope.store_generation
+ || snapshot.projection_generation != scope.projection_generation
+ {
+ return Err(TodayError::CorruptProjection);
+ }
+ Ok(())
+}
+
+async fn load_state(
+ storage: &dyn radroots_storage::Storage,
+ context: &LocalNetwork,
+ generation: ProjectionGeneration,
+) -> Result<Option<TodayProjectionState>, TodayError> {
+ ProjectionStore::projection_document(
+ storage,
+ projection_id()?,
+ generation,
+ projection_document_key(context),
+ )
+ .await?
+ .map(|document| decode_state(document.value()))
+ .transpose()
+}
+
+async fn store_state(
+ storage: &dyn radroots_storage::Storage,
+ context: &LocalNetwork,
+ generation: ProjectionGeneration,
+ state: &TodayProjectionState,
+) -> Result<(), TodayError> {
+ ProjectionStore::put_projection_document(
+ storage,
+ projection_id()?,
+ generation,
+ ProjectionDocument::new(projection_document_key(context), encode(state)?)?,
+ )
+ .await?;
+ Ok(())
+}
+
+async fn persist_snapshot(
+ storage: &dyn radroots_storage::Storage,
+ generation: ProjectionGeneration,
+ scope: &CursorScope,
+ snapshot: &FrozenTodaySnapshot,
+) -> Result<(), TodayError> {
+ ProjectionStore::put_projection_snapshot(
+ storage,
+ ProjectionSnapshot::new(
+ projection_id()?,
+ snapshot_id(scope),
+ generation,
+ scope
+ .as_of
+ .checked_mul(1_000)
+ .ok_or(TodayError::InvalidRequest)?,
+ encode(snapshot)?,
+ )?,
+ )
+ .await?;
+ Ok(())
+}
+
+async fn load_snapshot(
+ storage: &dyn radroots_storage::Storage,
+ projection_id: ProjectionId,
+ generation: ProjectionGeneration,
+ scope: &CursorScope,
+) -> Result<Option<FrozenTodaySnapshot>, TodayError> {
+ ProjectionStore::projection_snapshot(storage, projection_id, snapshot_id(scope))
+ .await?
+ .map(|snapshot| {
+ if snapshot.generation() != generation {
+ return Err(TodayError::CorruptProjection);
+ }
+ decode_snapshot(snapshot.value())
+ })
+ .transpose()
+}
+
+fn decode_state(value: &[u8]) -> Result<TodayProjectionState, TodayError> {
+ let state: TodayProjectionState = decode(value)?;
+ if state.schema_version != TODAY_PROJECTION_DOCUMENT_SCHEMA_VERSION
+ || state.content_generation == 0
+ || content_generation(&state)? != state.content_generation
+ {
+ return Err(TodayError::CorruptProjection);
+ }
+ Ok(state)
+}
+
+fn decode_snapshot(value: &[u8]) -> Result<FrozenTodaySnapshot, TodayError> {
+ let snapshot: FrozenTodaySnapshot = decode(value)?;
+ if snapshot.schema_version != TODAY_SNAPSHOT_SCHEMA_VERSION {
+ return Err(TodayError::CorruptProjection);
+ }
+ Ok(snapshot)
+}
+
+fn content_generation(state: &TodayProjectionState) -> Result<u64, TodayError> {
+ let mut canonical = state.clone();
+ canonical.content_generation = 0;
+ let digest =
+ Sha256::digest([PROJECTION_CONTENT_DOMAIN, encode(&canonical)?.as_slice()].concat());
+ let generation = u64::from_be_bytes(digest[..8].try_into().expect("digest prefix"));
+ Ok(generation.max(1))
+}
+
+fn projection_generation() -> Result<ProjectionGeneration, TodayError> {
+ ProjectionGeneration::new(Sha256::digest(PROJECTION_GENERATION_DOMAIN).into())
+ .map_err(TodayError::from)
+}
+
+fn projection_id() -> Result<ProjectionId, TodayError> {
+ ProjectionId::parse(TODAY_PROJECTION_ID).map_err(TodayError::from)
+}
+
+fn projection_document_key(context: &LocalNetwork) -> String {
+ let mut digest = Sha256::new();
+ digest.update(PROJECTION_DOCUMENT_KEY_DOMAIN);
+ digest.update(context.id.as_bytes());
+ digest.update(context.generation.to_be_bytes());
+ format!("context.{}", hex::encode(digest.finalize()))
+}
+
+fn snapshot_id(scope: &CursorScope) -> [u8; 32] {
+ let mut digest = Sha256::new();
+ digest.update(SNAPSHOT_ID_DOMAIN);
+ digest.update(scope.context_id.as_bytes());
+ digest.update(scope.context_generation.to_be_bytes());
+ digest.update(scope.as_of.to_be_bytes());
+ digest.update(scope.store_generation);
+ digest.update(scope.projection_generation.to_be_bytes());
+ digest.finalize().into()
+}
+
+fn tag_value(tags: &[Vec<String>], names: &[&str]) -> Option<String> {
+ tags.iter().find_map(|tag| {
+ names
+ .contains(&tag.first()?.as_str())
+ .then(|| tag.get(1).cloned())
+ .flatten()
+ })
+}
+
+fn blossom_digest(url: &str) -> Option<String> {
+ let path = url.split_once("://")?.1.split_once('/')?.1;
+ let candidate = path.split(['.', '/', '?', '#']).next()?;
+ (candidate.len() == 64
+ && candidate
+ .bytes()
+ .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)))
+ .then(|| candidate.to_owned())
+}
+
+fn valid_public_key(value: &str) -> bool {
+ value.len() == 64
+ && value
+ .bytes()
+ .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+}
+
+fn encode(value: &impl Serialize) -> Result<Vec<u8>, TodayError> {
+ serde_json::to_vec(value).map_err(|_| TodayError::Serialization)
+}
+
+fn decode<T: for<'de> Deserialize<'de>>(value: &[u8]) -> Result<T, TodayError> {
+ serde_json::from_slice(value).map_err(|_| TodayError::CorruptProjection)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use nostr::secp256k1::Message;
+ use nostr::{Keys, SECP256K1};
+ use radroots_event::{
+ SignedEvent,
+ admission::{AdmissionPolicy, RawEvent, VisibilityPolicy},
+ wire::{Nip01EventWire, compute_canonical_nip01_event_id},
+ };
+ use radroots_event_codec::verify::Nip01SignatureVerifier;
+ use radroots_transport::{
+ Target, TransportId,
+ source::{EventProvenance, ObservedEvent},
+ };
+
+ const SECRET: &str = "10c5304d6c9ae3a1a16f7860f1cc8f5e3a76225a2663b3a989a0d775919b7df5";
+
+ struct Allow;
+
+ impl AdmissionPolicy for Allow {
+ type Error = core::convert::Infallible;
+
+ fn policy_id(&self) -> &'static str {
+ "test.today.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.today.visibility.v1"
+ }
+
+ fn make_visible(
+ &self,
+ _event: &radroots_event::admission::AdmittedEvent,
+ ) -> Result<(), Self::Error> {
+ Ok(())
+ }
+ }
+
+ fn context(locality: Option<&str>, generation: u64) -> LocalNetwork {
+ LocalNetwork::new(
+ "victoria".into(),
+ "Victoria".into(),
+ vec!["wss://relay.example".into()],
+ locality.map(str::to_owned),
+ Vec::new(),
+ generation,
+ )
+ .expect("context")
+ }
+
+ fn keys() -> Keys {
+ Keys::parse(SECRET).expect("keys")
+ }
+
+ fn signed(kind: u32, tags: Vec<Vec<&str>>, content: &str, created_at: u64) -> SignedEvent {
+ signed_owned(
+ kind,
+ tags.into_iter()
+ .map(|tag| tag.into_iter().map(str::to_owned).collect())
+ .collect(),
+ content,
+ created_at,
+ )
+ }
+
+ fn signed_owned(
+ kind: u32,
+ tags: Vec<Vec<String>>,
+ content: &str,
+ created_at: u64,
+ ) -> SignedEvent {
+ let keys = keys();
+ let author = keys.public_key().to_string();
+ let id = compute_canonical_nip01_event_id(&author, created_at, kind, &tags, content)
+ .expect("id");
+ let message = Message::from_digest(*id.as_bytes());
+ let signature = SECP256K1.sign_schnorr_no_aux_rand(
+ &message,
+ &nostr::secp256k1::Keypair::from_secret_key(SECP256K1, keys.secret_key()),
+ );
+ let wire = Nip01EventWire {
+ id: id.to_hex(),
+ pubkey: author,
+ created_at,
+ kind,
+ tags,
+ content: content.to_owned(),
+ sig: signature.to_string(),
+ extra: Default::default(),
+ };
+ let raw = 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,
+ })
+ .to_string();
+ SignedEvent::from_wire_verified_id(wire, raw).expect("signed event")
+ }
+
+ fn visible_admission(event: SignedEvent, observed_at: u64) -> EventAdmission {
+ let selected = admit_verified_event(
+ verify_nip01_event(event.envelope().clone()).expect("codec verification"),
+ )
+ .expect("codec admission")
+ .contract_id();
+ let verified = RawEvent::new(event.envelope().clone())
+ .verify_id()
+ .expect("id")
+ .verify_signature(&Nip01SignatureVerifier)
+ .expect("signature");
+ let visible = verified
+ .validate_contract_for_admission(selected)
+ .expect("selected contract")
+ .admit_with(&Allow)
+ .expect("admission")
+ .make_visible_with(&Allow)
+ .expect("visibility");
+ let target = Target::new(TransportId::NOSTR, "wss://relay.example").expect("target");
+ let provenance = EventProvenance::new(
+ TransportId::NOSTR,
+ target.fingerprint().clone(),
+ observed_at,
+ )
+ .expect("provenance");
+ EventAdmission::visible(ObservedEvent::new(event, provenance), visible)
+ .expect("visible admission")
+ }
+
+ fn raw_admission(event: SignedEvent, observed_at: u64) -> EventAdmission {
+ let target = Target::new(TransportId::NOSTR, "wss://relay.example").expect("target");
+ let provenance = EventProvenance::new(
+ TransportId::NOSTR,
+ target.fingerprint().clone(),
+ observed_at,
+ )
+ .expect("provenance");
+ EventAdmission::raw(ObservedEvent::new(event, provenance))
+ }
+
+ fn admitted(event: &SignedEvent) -> RadrootsAdmittedEvent {
+ admit_verified_event(
+ verify_nip01_event(event.envelope().clone()).expect("codec verification"),
+ )
+ .expect("codec admission")
+ }
+
+ async fn ingest(
+ runtime: &RadrootsRuntime,
+ context: &LocalNetwork,
+ event: SignedEvent,
+ at: u64,
+ ) -> TodayIngestReceipt {
+ runtime
+ .phase1_ingest_visible(visible_admission(event, at * 1_000), context, at)
+ .await
+ .expect("ingest")
+ }
+
+ #[tokio::test]
+ async fn equal_timestamp_pages_are_complete_and_remain_frozen_across_ingest() {
+ let runtime = RadrootsRuntime::test_memory().expect("runtime");
+ let context = context(None, 1);
+ for content in ["alpha", "bravo", "charlie"] {
+ ingest(
+ &runtime,
+ &context,
+ signed(1, Vec::new(), content, 2_000_000_000),
+ 2_000_000_100,
+ )
+ .await;
+ }
+ let first = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(1, 2_000_000_200))
+ .await
+ .expect("first page");
+ assert_eq!(first.items.len(), 1);
+ let frozen_cursor = first.next_cursor.clone().expect("cursor");
+
+ ingest(
+ &runtime,
+ &context,
+ signed(1, Vec::new(), "delta", 2_000_000_001),
+ 2_000_000_101,
+ )
+ .await;
+
+ let mut ids = first
+ .items
+ .iter()
+ .map(|card| card.card.card_id.to_hex())
+ .collect::<Vec<_>>();
+ let mut cursor = Some(frozen_cursor);
+ while let Some(value) = cursor {
+ let page = runtime
+ .phase1_today_page(&context, TodayPageRequest::after(1, value))
+ .await
+ .expect("continued frozen page");
+ ids.extend(page.items.iter().map(|card| card.card.card_id.to_hex()));
+ cursor = page.next_cursor;
+ }
+ ids.sort();
+ ids.dedup();
+ assert_eq!(ids.len(), 3, "frozen snapshot has no loss or duplicates");
+
+ let current = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(100, 2_000_000_201))
+ .await
+ .expect("current page");
+ assert_eq!(current.items.len(), 4);
+ }
+
+ #[tokio::test]
+ async fn profile_thread_media_search_me_context_and_rebuild_share_one_projection() {
+ let runtime = RadrootsRuntime::test_memory().expect("runtime");
+ let context = context(Some("victoria"), 7);
+ let author = keys().public_key().to_string();
+ let profile_url = "https://blob.example/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.jpg";
+ let profile = signed(
+ 0,
+ Vec::new(),
+ &format!(
+ r#"{{"name":"moss","display_name":"Moss Farm","about":"Local roots","picture":"{profile_url}","website":"https://moss.example","lud16":"moss@example.com"}}"#
+ ),
+ 2_000_000_000,
+ );
+ ingest(&runtime, &context, profile, 2_000_000_100).await;
+
+ let digest = "b".repeat(64);
+ let photo_url = format!("https://blob.example/{digest}.jpg");
+ let photo_content = format!("Fresh field photo {photo_url}");
+ let photo = signed_owned(
+ 1,
+ vec![
+ vec!["location".into(), "Victoria".into()],
+ vec![
+ "imeta".into(),
+ format!("url {photo_url}"),
+ format!("x {digest}"),
+ "m image/jpeg".into(),
+ "dim 120x80".into(),
+ "size 345".into(),
+ "alt Fresh field photo".into(),
+ ],
+ ],
+ &photo_content,
+ 2_000_000_001,
+ );
+ let root_id = photo.id().to_hex();
+ ingest(&runtime, &context, photo, 2_000_000_101).await;
+
+ let nonmatch = signed(
+ 1,
+ vec![vec!["location", "Elsewhere"]],
+ "far away",
+ 2_000_000_002,
+ );
+ ingest(&runtime, &context, nonmatch, 2_000_000_102).await;
+
+ let reply = signed_owned(
+ 1,
+ vec![
+ vec![
+ "e".into(),
+ root_id.clone(),
+ "wss://relay.example".into(),
+ "root".into(),
+ ],
+ vec!["p".into(), author.clone()],
+ ],
+ "Looks good",
+ 2_000_000_003,
+ );
+ ingest(&runtime, &context, reply, 2_000_000_103).await;
+
+ let food = signed(
+ 30_402,
+ vec![
+ vec!["d", "today-carrots"],
+ vec!["title", "Today carrots"],
+ vec!["summary", "Fresh"],
+ vec!["published_at", "2000000004"],
+ vec!["location", "Victoria"],
+ vec!["price", "3", "CAD"],
+ vec!["radroots:price_unit", "lb"],
+ vec!["status", "active"],
+ ],
+ "Fresh carrots",
+ 2_000_000_004,
+ );
+ let food_id = food.id().to_hex();
+ ingest(&runtime, &context, food, 2_000_000_104).await;
+
+ let comment = signed_owned(
+ 1_111,
+ vec![
+ vec![
+ "E".into(),
+ food_id.clone(),
+ "wss://relay.example".into(),
+ author.clone(),
+ ],
+ vec!["K".into(), "30402".into()],
+ vec!["P".into(), author.clone(), "wss://relay.example".into()],
+ vec![
+ "e".into(),
+ food_id,
+ "wss://relay.example".into(),
+ author.clone(),
+ ],
+ vec!["k".into(), "30402".into()],
+ vec!["p".into(), author.clone(), "wss://relay.example".into()],
+ ],
+ "A NIP-22 comment",
+ 2_000_000_005,
+ );
+ ingest(&runtime, &context, comment, 2_000_000_105).await;
+
+ assert!(
+ runtime
+ .phase1_set_media_verification(
+ &context,
+ &photo_url,
+ MediaVerificationState::Verified,
+ )
+ .await
+ .expect("media evidence")
+ );
+ assert!(
+ runtime
+ .phase1_set_media_verification(
+ &context,
+ profile_url,
+ MediaVerificationState::Verified,
+ )
+ .await
+ .expect("profile media evidence")
+ );
+ let live = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(100, 2_000_000_200))
+ .await
+ .expect("page");
+ assert_eq!(live.items.len(), 2, "known locality nonmatch is excluded");
+ let card = live
+ .items
+ .iter()
+ .find(|card| card.card.card_type == TodayCardType::PhotoUpdate)
+ .unwrap_or_else(|| panic!("photo card missing from {:#?}", live.items));
+ assert_eq!(card.card.card_type, TodayCardType::PhotoUpdate);
+ assert_eq!(
+ card.card.media[0].verification,
+ MediaVerificationState::Verified
+ );
+ assert_eq!(card.thread.len(), 1);
+ assert_eq!(
+ live.items
+ .iter()
+ .find(|card| card.card.card_type == TodayCardType::FoodAvailability)
+ .expect("food card")
+ .thread
+ .len(),
+ 1
+ );
+ let profile = card.author_profile.as_ref().expect("profile enrichment");
+ assert_eq!(profile.website.as_deref(), Some("https://moss.example"));
+ assert_eq!(
+ profile.lightning_address.as_deref(),
+ Some("moss@example.com")
+ );
+ assert_eq!(
+ profile
+ .picture
+ .as_ref()
+ .expect("profile picture")
+ .verification,
+ MediaVerificationState::Verified
+ );
+
+ runtime
+ .phase1_set_local_author_overlay(
+ &context,
+ card.card.card_id,
+ Some(LocalAuthorOverlay {
+ operation_id: "publish-photo-1".into(),
+ state: "delivered".into(),
+ }),
+ )
+ .await
+ .expect("active author overlay");
+ assert!(matches!(
+ runtime
+ .phase1_set_local_author_overlay(
+ &context,
+ CardId::parse(&"f".repeat(64)).expect("unknown card id"),
+ None,
+ )
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+
+ let search = runtime
+ .phase1_search(&context, "moss farm", 10, 2_000_000_200)
+ .await
+ .expect("search");
+ assert!(search.iter().any(|result| result.profile.is_some()));
+ let me = runtime
+ .phase1_me(&context, &author, 2_000_000_200)
+ .await
+ .expect("me");
+ assert_eq!(me.cards.len(), 2);
+ assert_eq!(
+ me.profile.expect("me profile").name.as_deref(),
+ Some("moss")
+ );
+
+ let rebuilt = runtime
+ .phase1_refresh_today(&context, 2_000_000_201, TodayProjectionUpdate::Rebuild)
+ .await
+ .expect("rebuild");
+ assert!(!rebuilt.changed, "rebuild is byte-equivalent to live state");
+ let after = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(100, 2_000_000_202))
+ .await
+ .expect("rebuilt page");
+ let rebuilt_photo = after
+ .items
+ .iter()
+ .find(|card| card.card.card_type == TodayCardType::PhotoUpdate)
+ .expect("rebuilt photo");
+ assert_eq!(
+ rebuilt_photo.card.media[0].verification,
+ MediaVerificationState::Verified
+ );
+ assert_eq!(
+ rebuilt_photo
+ .author_profile
+ .as_ref()
+ .and_then(|profile| profile.picture.as_ref())
+ .expect("rebuilt profile picture")
+ .verification,
+ MediaVerificationState::Verified
+ );
+ assert_eq!(
+ rebuilt_photo
+ .local_overlay
+ .as_ref()
+ .map(|overlay| overlay.state.as_str()),
+ Some("delivered")
+ );
+ assert_eq!(rebuilt_photo.thread.len(), 1);
+ }
+
+ #[tokio::test]
+ async fn replacement_and_deletion_change_only_current_today_truth() {
+ let runtime = RadrootsRuntime::test_memory().expect("runtime");
+ let context = context(None, 1);
+ let active = signed(
+ 30_402,
+ vec![
+ vec!["d", "carrots"],
+ vec!["title", "Carrots"],
+ vec!["summary", "Fresh"],
+ vec!["published_at", "2000000000"],
+ vec!["location", "Victoria"],
+ vec!["price", "3", "CAD"],
+ vec!["radroots:price_unit", "lb"],
+ vec!["status", "active"],
+ ],
+ "available",
+ 2_000_000_000,
+ );
+ ingest(&runtime, &context, active, 2_000_000_100).await;
+ let sold = signed(
+ 30_402,
+ vec![
+ vec!["d", "carrots"],
+ vec!["title", "Carrots"],
+ vec!["summary", "Gone"],
+ vec!["published_at", "2000000001"],
+ vec!["location", "Victoria"],
+ vec!["price", "3", "CAD"],
+ vec!["radroots:price_unit", "lb"],
+ vec!["status", "sold"],
+ ],
+ "sold out",
+ 2_000_000_001,
+ );
+ let sold_id = sold.id().to_hex();
+ ingest(&runtime, &context, sold, 2_000_000_101).await;
+ let current = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(10, 2_000_000_200))
+ .await
+ .expect("current");
+ assert_eq!(current.items.len(), 1);
+ assert_eq!(current.items[0].card.source_event_id, sold_id);
+ assert_eq!(current.items[0].card.lifecycle, CardLifecycleState::Sold);
+
+ let deletion = signed_owned(5, vec![vec!["e".into(), sold_id]], "", 2_000_000_002);
+ ingest(&runtime, &context, deletion, 2_000_000_102).await;
+ let deleted = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(10, 2_000_000_201))
+ .await
+ .expect("deleted");
+ assert!(deleted.items.is_empty());
+ }
+
+ #[tokio::test]
+ async fn sqlite_reopen_preserves_materialized_state_and_frozen_cursor_pages() {
+ use crate::runtime::{
+ builder::RuntimeBuilder,
+ store::{MobileUserStoreConfig, ProtectedDataAvailability},
+ };
+
+ let root = tempfile::tempdir().expect("root");
+ let store = MobileUserStoreConfig::from_encoded(
+ root.path(),
+ &keys().public_key().to_string(),
+ "3131313131313131313131313131313131313131313131313131313131313131",
+ 2_000_000_000_000,
+ ProtectedDataAvailability::Available,
+ )
+ .expect("store");
+ std::fs::create_dir_all(store.owner_directory()).expect("owner directory");
+ let context = context(None, 1);
+ let runtime = RuntimeBuilder::new(store.clone())
+ .build()
+ .await
+ .expect("runtime");
+ ingest(
+ &runtime,
+ &context,
+ signed(1, Vec::new(), "persisted alpha", 2_000_000_000),
+ 2_000_000_100,
+ )
+ .await;
+ ingest(
+ &runtime,
+ &context,
+ signed(1, Vec::new(), "persisted bravo", 2_000_000_000),
+ 2_000_000_101,
+ )
+ .await;
+ let first = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(1, 2_000_000_200))
+ .await
+ .expect("first page");
+ let cursor = first.next_cursor.expect("frozen cursor");
+ runtime.shutdown().await.expect("shutdown");
+
+ let reopened = RuntimeBuilder::new(store).build().await.expect("reopen");
+ let second = reopened
+ .phase1_today_page(&context, TodayPageRequest::after(1, cursor))
+ .await
+ .expect("continued page after reopen");
+ assert_eq!(second.items.len(), 1);
+ assert_ne!(first.items[0].card.card_id, second.items[0].card.card_id);
+ assert!(second.next_cursor.is_none());
+ let search = reopened
+ .phase1_search(&context, "persisted", 10, 2_000_000_200)
+ .await
+ .expect("search after reopen");
+ assert_eq!(search.len(), 2);
+ reopened.shutdown().await.expect("shutdown reopened");
+ }
+
+ #[tokio::test]
+ async fn fail_closed_requests_cursors_overlays_and_projection_guards_are_executable() {
+ let mut runtime = RadrootsRuntime::test_memory().expect("runtime");
+ let context = context(None, 1);
+ let note = signed(1, Vec::new(), "guarded alpha", 2_000_000_000);
+ assert!(matches!(
+ runtime
+ .phase1_ingest_visible(raw_admission(note.clone(), 2_000_000_000_000), &context, 1)
+ .await,
+ Err(TodayError::EventNotVisible)
+ ));
+ assert!(matches!(
+ runtime
+ .phase1_refresh_today(&context, 0, TodayProjectionUpdate::Incremental)
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+ assert!(matches!(
+ runtime
+ .phase1_refresh_today(&context, u64::MAX, TodayProjectionUpdate::Incremental)
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+ let empty = runtime
+ .phase1_refresh_today(&context, 1, TodayProjectionUpdate::Incremental)
+ .await
+ .expect("empty projection");
+ assert_eq!(empty.source_events, 0);
+ assert!(empty.changed);
+ assert!(
+ !runtime
+ .phase1_refresh_today(&context, 2, TodayProjectionUpdate::Incremental)
+ .await
+ .expect("unchanged projection")
+ .changed
+ );
+
+ for request in [
+ TodayPageRequest::first(0, 1),
+ TodayPageRequest::first(TODAY_PAGE_LIMIT_MAX + 1, 1),
+ TodayPageRequest {
+ limit: 1,
+ as_of: None,
+ cursor: None,
+ },
+ TodayPageRequest::first(1, 0),
+ ] {
+ assert!(matches!(
+ runtime.phase1_today_page(&context, request).await,
+ Err(TodayError::InvalidRequest)
+ ));
+ }
+ for (query, limit, as_of) in [
+ ("valid", 0, 1),
+ ("valid", TODAY_SEARCH_LIMIT_MAX + 1, 1),
+ ("valid", 1, 0),
+ (" ", 1, 1),
+ (&"x".repeat(257), 1, 1),
+ ("bad\nquery", 1, 1),
+ ] {
+ assert!(matches!(
+ runtime.phase1_search(&context, query, limit, as_of).await,
+ Err(TodayError::InvalidRequest)
+ ));
+ }
+ assert!(matches!(
+ runtime.phase1_me(&context, "bad", 1).await,
+ Err(TodayError::InvalidRequest)
+ ));
+ assert!(matches!(
+ runtime
+ .phase1_me(&context, &keys().public_key().to_string(), 0)
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+ for url in ["", &"x".repeat(8_193), "bad\nurl"] {
+ assert!(matches!(
+ runtime
+ .phase1_set_media_verification(&context, url, MediaVerificationState::Verified,)
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+ }
+ assert!(
+ !runtime
+ .phase1_set_media_verification(
+ &context,
+ "https://blob.example/missing.jpg",
+ MediaVerificationState::Verified,
+ )
+ .await
+ .expect("unmatched media")
+ );
+
+ ingest(&runtime, &context, note.clone(), 2_000_000_100).await;
+ let page = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(1, 2_000_000_200))
+ .await
+ .expect("page");
+ let card_id = page.items[0].card.card_id;
+ let rank = page.items[0].card.rank.expect("rank");
+ let generation = projection_generation().expect("generation");
+ let storage = runtime.client.storage().expect("storage");
+ let state = load_state(storage, &context, generation)
+ .await
+ .expect("load state")
+ .expect("state");
+ let scope = CursorScope::new(
+ context.id.clone(),
+ context.generation,
+ 2_000_000_200,
+ state.store_generation,
+ state.content_generation,
+ )
+ .expect("scope");
+
+ let cursor_for = |scope: CursorScope| {
+ TodayCursor::encode(&scope, TodayCursorPosition { rank })
+ .expect("cursor")
+ .as_str()
+ .to_owned()
+ };
+ let mut wrong_context = scope.clone();
+ wrong_context.context_id = "elsewhere".into();
+ assert!(matches!(
+ runtime
+ .phase1_today_page(
+ &context,
+ TodayPageRequest::after(1, cursor_for(wrong_context))
+ )
+ .await,
+ Err(TodayError::Cursor(CursorError::ContextMismatch))
+ ));
+ let mut wrong_context_generation = scope.clone();
+ wrong_context_generation.context_generation += 1;
+ assert!(matches!(
+ runtime
+ .phase1_today_page(
+ &context,
+ TodayPageRequest::after(1, cursor_for(wrong_context_generation)),
+ )
+ .await,
+ Err(TodayError::Cursor(CursorError::ContextMismatch))
+ ));
+ assert!(matches!(
+ runtime
+ .phase1_today_page(
+ &context,
+ TodayPageRequest {
+ limit: 1,
+ as_of: Some(scope.as_of + 1),
+ cursor: Some(cursor_for(scope.clone())),
+ },
+ )
+ .await,
+ Err(TodayError::Cursor(CursorError::SnapshotMismatch))
+ ));
+ let mut stale = scope.clone();
+ stale.store_generation = [9; 32];
+ assert!(matches!(
+ runtime
+ .phase1_today_page(&context, TodayPageRequest::after(1, cursor_for(stale)))
+ .await,
+ Err(TodayError::Cursor(CursorError::Stale))
+ ));
+ let mut missing = scope.clone();
+ missing.projection_generation = missing.projection_generation.wrapping_add(1).max(1);
+ assert!(matches!(
+ runtime
+ .phase1_today_page(&context, TodayPageRequest::after(1, cursor_for(missing)))
+ .await,
+ Err(TodayError::SnapshotMissing)
+ ));
+
+ let snapshot = frozen_snapshot(&state, &context, scope.as_of).expect("snapshot");
+ assert!(validate_snapshot(&snapshot, &scope).is_ok());
+ for invalid in [
+ {
+ let mut value = snapshot.clone();
+ value.schema_version += 1;
+ value
+ },
+ {
+ let mut value = snapshot.clone();
+ value.context_id = "other".into();
+ value
+ },
+ {
+ let mut value = snapshot.clone();
+ value.context_generation += 1;
+ value
+ },
+ {
+ let mut value = snapshot.clone();
+ value.as_of += 1;
+ value
+ },
+ {
+ let mut value = snapshot.clone();
+ value.store_generation = [8; 32];
+ value
+ },
+ {
+ let mut value = snapshot.clone();
+ value.projection_generation = value.projection_generation.wrapping_add(1);
+ value
+ },
+ ] {
+ assert!(matches!(
+ validate_snapshot(&invalid, &scope),
+ Err(TodayError::CorruptProjection)
+ ));
+ }
+ let mut absent_rank = rank;
+ absent_rank.card_id = CardId::parse(&"f".repeat(64)).expect("absent card id");
+ assert!(matches!(
+ page_from_snapshot(snapshot.clone(), scope.clone(), Some(absent_rank), 1),
+ Err(TodayError::CursorPositionMissing)
+ ));
+
+ for invalid in [
+ {
+ let mut value = state.clone();
+ value.schema_version += 1;
+ value
+ },
+ {
+ let mut value = state.clone();
+ value.content_generation = 0;
+ value
+ },
+ {
+ let mut value = state.clone();
+ value.source_events += 1;
+ value
+ },
+ ] {
+ assert!(matches!(
+ decode_state(&encode(&invalid).expect("encode invalid state")),
+ Err(TodayError::CorruptProjection)
+ ));
+ }
+ let mut invalid_snapshot = snapshot.clone();
+ invalid_snapshot.schema_version += 1;
+ assert!(matches!(
+ decode_snapshot(&encode(&invalid_snapshot).expect("encode invalid snapshot")),
+ Err(TodayError::CorruptProjection)
+ ));
+ assert!(matches!(
+ decode_state(b"not-json"),
+ Err(TodayError::CorruptProjection)
+ ));
+
+ for overlay in [
+ LocalAuthorOverlay {
+ operation_id: String::new(),
+ state: "queued".into(),
+ },
+ LocalAuthorOverlay {
+ operation_id: "x".repeat(257),
+ state: "queued".into(),
+ },
+ LocalAuthorOverlay {
+ operation_id: "operation".into(),
+ state: String::new(),
+ },
+ LocalAuthorOverlay {
+ operation_id: "operation".into(),
+ state: "x".repeat(97),
+ },
+ LocalAuthorOverlay {
+ operation_id: "operation".into(),
+ state: "bad\nstate".into(),
+ },
+ ] {
+ assert!(matches!(
+ runtime
+ .phase1_set_local_author_overlay(&context, card_id, Some(overlay))
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+ }
+ let other_keys = Keys::generate();
+ runtime.store_public_key = Some(
+ radroots_identity::PublicKey::from_hex(&other_keys.public_key().to_string())
+ .expect("other public key"),
+ );
+ assert!(matches!(
+ runtime
+ .phase1_me(&context, &keys().public_key().to_string(), 1)
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+ assert!(matches!(
+ runtime
+ .phase1_set_local_author_overlay(
+ &context,
+ card_id,
+ Some(LocalAuthorOverlay {
+ operation_id: "operation".into(),
+ state: "queued".into(),
+ }),
+ )
+ .await,
+ Err(TodayError::InvalidRequest)
+ ));
+ runtime.store_public_key = None;
+ runtime
+ .phase1_set_local_author_overlay(
+ &context,
+ card_id,
+ Some(LocalAuthorOverlay {
+ operation_id: "operation".into(),
+ state: "queued".into(),
+ }),
+ )
+ .await
+ .expect("set overlay");
+ runtime
+ .phase1_set_local_author_overlay(&context, card_id, None)
+ .await
+ .expect("remove overlay");
+ assert_eq!(
+ runtime
+ .phase1_search(&context, "guarded", 1, 2_000_000_200)
+ .await
+ .expect("card-limited search")
+ .len(),
+ 1
+ );
+
+ let mut event_state = state.clone();
+ event_state.cards[0].card.card_type = TodayCardType::Event;
+ event_state.cards[0].card.event_start = Some(100);
+ event_state.cards[0].card.event_end = Some(200);
+ event_state.cards[0].card.effective_at = 100;
+ assert_eq!(
+ ranked_cards(&event_state, &context, 150).expect("live event")[0]
+ .card
+ .lifecycle,
+ CardLifecycleState::Active
+ );
+ assert_eq!(
+ ranked_cards(&event_state, &context, 200).expect("past event")[0]
+ .card
+ .lifecycle,
+ CardLifecycleState::Past
+ );
+
+ let root = admitted(¬e);
+ assert!(matches!(
+ profile_summary(&root),
+ Err(TodayError::CorruptProjection)
+ ));
+ assert!(matches!(
+ reply_entry(&root),
+ Err(TodayError::CorruptProjection)
+ ));
+ assert!(comment_entry(&root).is_none());
+ assert_eq!(tag_value(&[Vec::new()], &["x"]), None);
+ assert_eq!(tag_value(&[vec!["x".into()]], &["x"]), None);
+ assert_eq!(blossom_digest("not-a-url"), None);
+ assert_eq!(blossom_digest("https://blob.example/short"), None);
+ assert_eq!(
+ blossom_digest(&format!("https://blob.example/{}", "A".repeat(64))),
+ None
+ );
+
+ let tags = locality_tags(vec![
+ Vec::new(),
+ vec!["x".into(), "ignored".into()],
+ vec!["g".into()],
+ vec!["location".into(), " ".into()],
+ vec!["g".into(), "u10hr".into()],
+ ]);
+ assert_eq!(tags.len(), 1);
+ assert_eq!(
+ locality_evidence(Some("u10"), &tags),
+ LocalityEvidence::Match
+ );
+ assert_eq!(
+ locality_evidence(Some("u10hr7"), &tags),
+ LocalityEvidence::Match
+ );
+ assert_eq!(
+ locality_evidence(Some("other"), &tags),
+ LocalityEvidence::Nonmatch
+ );
+ assert_eq!(
+ locality_evidence(Some("selected"), &[]),
+ LocalityEvidence::Missing
+ );
+
+ let mut fields = BTreeMap::new();
+ fields.insert("value".into(), serde_json::json!(1));
+ assert_eq!(typed_profile_extra(&fields, "missing"), None);
+ assert_eq!(typed_profile_extra(&fields, "value"), None);
+ for value in [String::new(), "x".repeat(2_049), "bad\nvalue".into()] {
+ fields.insert("value".into(), serde_json::Value::String(value));
+ assert_eq!(typed_profile_extra(&fields, "value"), None);
+ }
+ }
+
+ #[test]
+ fn malformed_cursor_and_request_bounds_fail_closed() {
+ assert!(matches!(
+ TodayCursor::scope("bad"),
+ Err(CursorError::Malformed)
+ ));
+ assert!(!valid_public_key("A".repeat(64).as_str()));
+ assert_eq!(TODAY_PAGE_LIMIT_MAX, 100);
+ assert_eq!(TODAY_SEARCH_LIMIT_MAX, 100);
+ }
+}
diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs
@@ -81,6 +81,9 @@ pub enum Error {
InvalidEventIndexCheckpoint,
DuplicateEventIndexShard,
CorruptProjectionRecord,
+ InvalidProjectionDocument,
+ InvalidProjectionSnapshot,
+ CorruptProjectionDocument,
InvalidPrivateArtifactId,
InvalidPrivateArtifactKind,
InvalidPrivateArtifactSchema,
@@ -211,6 +214,9 @@ impl fmt::Display for Error {
Self::InvalidEventIndexCheckpoint => "storage event-index checkpoint is invalid",
Self::DuplicateEventIndexShard => "storage event-index shard is duplicated",
Self::CorruptProjectionRecord => "storage projection record is corrupt",
+ Self::InvalidProjectionDocument => "storage projection document is invalid",
+ Self::InvalidProjectionSnapshot => "storage projection snapshot is invalid",
+ Self::CorruptProjectionDocument => "storage projection document is corrupt",
Self::InvalidPrivateArtifactId => "storage private-artifact id is invalid",
Self::InvalidPrivateArtifactKind => "storage private-artifact kind is invalid",
Self::InvalidPrivateArtifactSchema => "storage private-artifact schema is invalid",
@@ -363,6 +369,9 @@ mod tests {
InvalidEventIndexCheckpoint,
DuplicateEventIndexShard,
CorruptProjectionRecord,
+ InvalidProjectionDocument,
+ InvalidProjectionSnapshot,
+ CorruptProjectionDocument,
InvalidPrivateArtifactId,
InvalidPrivateArtifactKind,
InvalidPrivateArtifactSchema,
diff --git a/crates/storage/src/memory.rs b/crates/storage/src/memory.rs
@@ -42,9 +42,10 @@ use crate::{
PrivateArtifactRevision, PrivateArtifactStage, PrivateArtifactStatus, PrivateArtifactStore,
},
projection::{
- EventIndexCheckpoint, EventIndexManifest, ProjectionCheckpoint, ProjectionGeneration,
- ProjectionHealth, ProjectionId, ProjectionInvalidation, ProjectionStatus, RebuildStage,
- RebuildTicket, RebuildTicketId, RebuildTransition,
+ EventIndexCheckpoint, EventIndexManifest, ProjectionCheckpoint, ProjectionDocument,
+ ProjectionGeneration, ProjectionHealth, ProjectionId, ProjectionInvalidation,
+ ProjectionSnapshot, ProjectionStatus, RebuildStage, RebuildTicket, RebuildTicketId,
+ RebuildTransition,
},
status::{
EventStoreHealth, EventStoreMode, EventStoreStatus, IntegrityHealth, IntegrityStatus,
@@ -70,6 +71,8 @@ struct State {
rebuilds: Vec<RebuildTicket>,
event_index_manifests: Vec<EventIndexManifest>,
event_index_checkpoints: Vec<EventIndexCheckpoint>,
+ projection_documents: Vec<(ProjectionId, ProjectionGeneration, ProjectionDocument)>,
+ projection_snapshots: Vec<ProjectionSnapshot>,
private_artifacts: Vec<PrivateArtifactMetadata>,
private_artifact_reseals: Vec<PrivateArtifactResealReceipt>,
backups: Vec<BackupOperation>,
@@ -101,6 +104,8 @@ impl MemoryStorage {
rebuilds: Vec::new(),
event_index_manifests: Vec::new(),
event_index_checkpoints: Vec::new(),
+ projection_documents: Vec::new(),
+ projection_snapshots: Vec::new(),
private_artifacts: Vec::new(),
private_artifact_reseals: Vec::new(),
backups: Vec::new(),
@@ -993,6 +998,90 @@ impl ProjectionStore for MemoryStorage {
Ok(())
})
}
+
+ fn put_projection_document(
+ &self,
+ projection_id: ProjectionId,
+ generation: ProjectionGeneration,
+ document: ProjectionDocument,
+ ) -> BoxFuture<'_, Result<(), Error>> {
+ Box::pin(async move {
+ let mut state = self.state()?;
+ if let Some((_, _, existing)) = state.projection_documents.iter_mut().find(
+ |(existing_id, existing_generation, existing)| {
+ existing_id == &projection_id
+ && *existing_generation == generation
+ && existing.key() == document.key()
+ },
+ ) {
+ *existing = document;
+ } else {
+ state
+ .projection_documents
+ .push((projection_id, generation, document));
+ }
+ Ok(())
+ })
+ }
+
+ fn projection_document(
+ &self,
+ projection_id: ProjectionId,
+ generation: ProjectionGeneration,
+ key: String,
+ ) -> BoxFuture<'_, Result<Option<ProjectionDocument>, Error>> {
+ Box::pin(async move {
+ Ok(self
+ .state()?
+ .projection_documents
+ .iter()
+ .find(|(existing_id, existing_generation, existing)| {
+ existing_id == &projection_id
+ && *existing_generation == generation
+ && existing.key() == key
+ })
+ .map(|(_, _, document)| document.clone()))
+ })
+ }
+
+ fn put_projection_snapshot(
+ &self,
+ snapshot: ProjectionSnapshot,
+ ) -> BoxFuture<'_, Result<(), Error>> {
+ Box::pin(async move {
+ let mut state = self.state()?;
+ if let Some(existing) = state.projection_snapshots.iter().find(|existing| {
+ existing.projection_id() == snapshot.projection_id()
+ && existing.snapshot_id() == snapshot.snapshot_id()
+ }) {
+ return if existing == &snapshot {
+ Ok(())
+ } else {
+ Err(Error::CorruptProjectionDocument)
+ };
+ }
+ state.projection_snapshots.push(snapshot);
+ Ok(())
+ })
+ }
+
+ fn projection_snapshot(
+ &self,
+ projection_id: ProjectionId,
+ snapshot_id: [u8; 32],
+ ) -> BoxFuture<'_, Result<Option<ProjectionSnapshot>, Error>> {
+ Box::pin(async move {
+ Ok(self
+ .state()?
+ .projection_snapshots
+ .iter()
+ .find(|snapshot| {
+ snapshot.projection_id() == &projection_id
+ && snapshot.snapshot_id() == &snapshot_id
+ })
+ .cloned())
+ })
+ }
}
impl PrivateArtifactStore for MemoryStorage {
diff --git a/crates/storage/src/projection.rs b/crates/storage/src/projection.rs
@@ -5,6 +5,7 @@
pub use radroots_event::EventId;
pub use radroots_transport::BoxFuture;
+use sha2::{Digest, Sha256};
use std::collections::BTreeSet;
use crate::{
@@ -842,6 +843,147 @@ pub struct ProjectionStatus {
active_rebuild: Option<RebuildTicketId>,
}
+/// Maximum UTF-8 bytes in one materialized projection document key.
+pub const PROJECTION_DOCUMENT_KEY_MAX_BYTES: usize = 512;
+/// Maximum bytes in one materialized projection document value.
+pub const PROJECTION_DOCUMENT_VALUE_MAX_BYTES: usize = 16 * 1024 * 1024;
+
+/// One backend-neutral, opaque materialized projection document.
+///
+/// Projection owners define the value encoding. Storage verifies its digest
+/// and treats the bytes as opaque so product-specific DTOs do not leak into
+/// the generic persistence boundary.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ProjectionDocument {
+ key: String,
+ value: Vec<u8>,
+ value_sha256: [u8; 32],
+}
+
+impl ProjectionDocument {
+ pub fn new(key: String, value: Vec<u8>) -> Result<Self, Error> {
+ if !valid_document_key(&key)
+ || value.is_empty()
+ || value.len() > PROJECTION_DOCUMENT_VALUE_MAX_BYTES
+ {
+ return Err(Error::InvalidProjectionDocument);
+ }
+ let value_sha256 = Sha256::digest(&value).into();
+ Ok(Self {
+ key,
+ value,
+ value_sha256,
+ })
+ }
+
+ pub fn from_stored_parts(
+ key: String,
+ value: Vec<u8>,
+ value_sha256: [u8; 32],
+ ) -> Result<Self, Error> {
+ let document = Self::new(key, value)?;
+ if document.value_sha256 != value_sha256 {
+ return Err(Error::CorruptProjectionDocument);
+ }
+ Ok(document)
+ }
+
+ pub fn key(&self) -> &str {
+ &self.key
+ }
+
+ pub fn value(&self) -> &[u8] {
+ &self.value
+ }
+
+ pub const fn value_sha256(&self) -> &[u8; 32] {
+ &self.value_sha256
+ }
+}
+
+/// One immutable, durable frozen-query snapshot.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ProjectionSnapshot {
+ projection_id: ProjectionId,
+ snapshot_id: [u8; 32],
+ generation: ProjectionGeneration,
+ created_at_unix_ms: u64,
+ value: Vec<u8>,
+ value_sha256: [u8; 32],
+}
+
+impl ProjectionSnapshot {
+ pub fn new(
+ projection_id: ProjectionId,
+ snapshot_id: [u8; 32],
+ generation: ProjectionGeneration,
+ created_at_unix_ms: u64,
+ value: Vec<u8>,
+ ) -> Result<Self, Error> {
+ if snapshot_id.iter().all(|byte| *byte == 0)
+ || created_at_unix_ms == 0
+ || value.is_empty()
+ || value.len() > PROJECTION_DOCUMENT_VALUE_MAX_BYTES
+ {
+ return Err(Error::InvalidProjectionSnapshot);
+ }
+ let value_sha256 = Sha256::digest(&value).into();
+ Ok(Self {
+ projection_id,
+ snapshot_id,
+ generation,
+ created_at_unix_ms,
+ value,
+ value_sha256,
+ })
+ }
+
+ pub fn from_stored_parts(
+ projection_id: ProjectionId,
+ snapshot_id: [u8; 32],
+ generation: ProjectionGeneration,
+ created_at_unix_ms: u64,
+ value: Vec<u8>,
+ value_sha256: [u8; 32],
+ ) -> Result<Self, Error> {
+ let snapshot = Self::new(
+ projection_id,
+ snapshot_id,
+ generation,
+ created_at_unix_ms,
+ value,
+ )?;
+ if snapshot.value_sha256 != value_sha256 {
+ return Err(Error::CorruptProjectionDocument);
+ }
+ Ok(snapshot)
+ }
+
+ pub const fn projection_id(&self) -> &ProjectionId {
+ &self.projection_id
+ }
+
+ pub const fn snapshot_id(&self) -> &[u8; 32] {
+ &self.snapshot_id
+ }
+
+ pub const fn generation(&self) -> ProjectionGeneration {
+ self.generation
+ }
+
+ pub const fn created_at_unix_ms(&self) -> u64 {
+ self.created_at_unix_ms
+ }
+
+ pub fn value(&self) -> &[u8] {
+ &self.value
+ }
+
+ pub const fn value_sha256(&self) -> &[u8; 32] {
+ &self.value_sha256
+ }
+}
+
impl ProjectionStatus {
pub fn new(
projection_id: ProjectionId,
@@ -928,6 +1070,31 @@ pub trait ProjectionStore: Send + Sync {
&self,
checkpoint: EventIndexCheckpoint,
) -> BoxFuture<'_, Result<(), Error>>;
+ /// Replaces one named materialized document for a projection generation.
+ fn put_projection_document(
+ &self,
+ projection_id: ProjectionId,
+ generation: ProjectionGeneration,
+ document: ProjectionDocument,
+ ) -> BoxFuture<'_, Result<(), Error>>;
+ /// Loads one named materialized document for an exact generation.
+ fn projection_document(
+ &self,
+ projection_id: ProjectionId,
+ generation: ProjectionGeneration,
+ key: String,
+ ) -> BoxFuture<'_, Result<Option<ProjectionDocument>, Error>>;
+ /// Persists one immutable frozen-query snapshot idempotently.
+ fn put_projection_snapshot(
+ &self,
+ snapshot: ProjectionSnapshot,
+ ) -> BoxFuture<'_, Result<(), Error>>;
+ /// Loads one immutable frozen-query snapshot by exact identity.
+ fn projection_snapshot(
+ &self,
+ projection_id: ProjectionId,
+ snapshot_id: [u8; 32],
+ ) -> BoxFuture<'_, Result<Option<ProjectionSnapshot>, Error>>;
}
fn valid_label(value: &str, max: usize) -> bool {
@@ -939,6 +1106,13 @@ fn valid_label(value: &str, max: usize) -> bool {
})
}
+fn valid_document_key(value: &str) -> bool {
+ !value.is_empty()
+ && value.len() <= PROJECTION_DOCUMENT_KEY_MAX_BYTES
+ && value == value.trim()
+ && !value.chars().any(char::is_control)
+}
+
fn valid_artifact_path(value: &str) -> bool {
!value.is_empty()
&& value.len() <= EVENT_INDEX_ARTIFACT_PATH_MAX_BYTES
@@ -971,3 +1145,92 @@ const fn bytes32_are_zero(bytes: &[u8; 32]) -> bool {
}
true
}
+
+#[cfg(test)]
+mod materialized_tests {
+ use super::*;
+
+ #[test]
+ fn materialized_document_and_snapshot_bounds_and_digests_fail_closed() {
+ assert_eq!(
+ ProjectionDocument::new(String::new(), vec![1]),
+ Err(Error::InvalidProjectionDocument)
+ );
+ assert_eq!(
+ ProjectionDocument::new("key".into(), Vec::new()),
+ Err(Error::InvalidProjectionDocument)
+ );
+ assert_eq!(
+ ProjectionDocument::new("k".repeat(PROJECTION_DOCUMENT_KEY_MAX_BYTES + 1), vec![1],),
+ Err(Error::InvalidProjectionDocument)
+ );
+ assert_eq!(
+ ProjectionDocument::new(" key".into(), vec![1]),
+ Err(Error::InvalidProjectionDocument)
+ );
+ assert_eq!(
+ ProjectionDocument::new("key\npart".into(), vec![1]),
+ Err(Error::InvalidProjectionDocument)
+ );
+ assert_eq!(
+ ProjectionDocument::new(
+ "key".into(),
+ vec![0; PROJECTION_DOCUMENT_VALUE_MAX_BYTES + 1],
+ ),
+ Err(Error::InvalidProjectionDocument)
+ );
+ let document = ProjectionDocument::new("key".into(), vec![1, 2]).unwrap();
+ assert_eq!(document.key(), "key");
+ assert_eq!(document.value(), [1, 2]);
+ assert_eq!(document.value_sha256().len(), 32);
+ assert_eq!(
+ ProjectionDocument::from_stored_parts("key".into(), vec![1, 2], [9; 32]),
+ Err(Error::CorruptProjectionDocument)
+ );
+
+ let projection_id = ProjectionId::parse("today").unwrap();
+ let generation = ProjectionGeneration::new([1; 32]).unwrap();
+ assert_eq!(
+ ProjectionSnapshot::new(projection_id.clone(), [0; 32], generation, 1, vec![1]),
+ Err(Error::InvalidProjectionSnapshot)
+ );
+ assert_eq!(
+ ProjectionSnapshot::new(projection_id.clone(), [2; 32], generation, 0, vec![1]),
+ Err(Error::InvalidProjectionSnapshot)
+ );
+ assert_eq!(
+ ProjectionSnapshot::new(projection_id.clone(), [2; 32], generation, 1, Vec::new()),
+ Err(Error::InvalidProjectionSnapshot)
+ );
+ assert_eq!(
+ ProjectionSnapshot::new(
+ projection_id.clone(),
+ [2; 32],
+ generation,
+ 1,
+ vec![0; PROJECTION_DOCUMENT_VALUE_MAX_BYTES + 1],
+ ),
+ Err(Error::InvalidProjectionSnapshot)
+ );
+ let snapshot =
+ ProjectionSnapshot::new(projection_id.clone(), [2; 32], generation, 1, vec![1])
+ .unwrap();
+ assert_eq!(snapshot.projection_id(), &projection_id);
+ assert_eq!(snapshot.snapshot_id(), &[2; 32]);
+ assert_eq!(snapshot.generation(), generation);
+ assert_eq!(snapshot.created_at_unix_ms(), 1);
+ assert_eq!(snapshot.value(), [1]);
+ assert_eq!(snapshot.value_sha256().len(), 32);
+ assert_eq!(
+ ProjectionSnapshot::from_stored_parts(
+ projection_id,
+ [2; 32],
+ generation,
+ 1,
+ vec![1],
+ [9; 32],
+ ),
+ Err(Error::CorruptProjectionDocument)
+ );
+ }
+}
diff --git a/crates/storage/tests/memory.rs b/crates/storage/tests/memory.rs
@@ -33,9 +33,10 @@ use radroots_storage::{
PrivateArtifactRevision, PrivateArtifactStore, RetentionPolicy,
},
projection::{
- InvalidationReason, ProjectionCheckpoint, ProjectionGeneration, ProjectionHealth,
- ProjectionId, ProjectionInvalidation, ProjectionRevision, RawSourceDigest, RebuildStage,
- RebuildTicket, RebuildTicketId, RebuildTransition,
+ InvalidationReason, ProjectionCheckpoint, ProjectionDocument, ProjectionGeneration,
+ ProjectionHealth, ProjectionId, ProjectionInvalidation, ProjectionRevision,
+ ProjectionSnapshot, RawSourceDigest, RebuildStage, RebuildTicket, RebuildTicketId,
+ RebuildTransition,
},
};
use radroots_transport::{
@@ -960,3 +961,84 @@ fn memory_atomic_replay_rejects_a_changed_digest_without_mutation() {
1
);
}
+
+#[test]
+fn memory_materialized_documents_replace_and_snapshots_remain_immutable() {
+ let store = MemoryStorage::default();
+ let projection_id = ProjectionId::parse("memory.today").unwrap();
+ let generation = ProjectionGeneration::new([21; 32]).unwrap();
+ block_on(store.put_projection_document(
+ projection_id.clone(),
+ generation,
+ ProjectionDocument::new("context.one".into(), vec![1]).unwrap(),
+ ))
+ .unwrap();
+ block_on(store.put_projection_document(
+ projection_id.clone(),
+ generation,
+ ProjectionDocument::new("context.one".into(), vec![2]).unwrap(),
+ ))
+ .unwrap();
+ assert_eq!(
+ block_on(store.projection_document(
+ projection_id.clone(),
+ generation,
+ "context.one".into(),
+ ))
+ .unwrap()
+ .unwrap()
+ .value(),
+ [2]
+ );
+ for (candidate_id, candidate_generation, candidate_key) in [
+ (
+ ProjectionId::parse("memory.other").unwrap(),
+ generation,
+ "context.one".to_owned(),
+ ),
+ (
+ projection_id.clone(),
+ ProjectionGeneration::new([23; 32]).unwrap(),
+ "context.one".to_owned(),
+ ),
+ (
+ projection_id.clone(),
+ generation,
+ "context.other".to_owned(),
+ ),
+ ] {
+ assert!(
+ block_on(store.projection_document(candidate_id, candidate_generation, candidate_key,))
+ .unwrap()
+ .is_none()
+ );
+ }
+ let snapshot =
+ ProjectionSnapshot::new(projection_id.clone(), [22; 32], generation, 100, vec![3]).unwrap();
+ block_on(store.put_projection_snapshot(snapshot.clone())).unwrap();
+ block_on(store.put_projection_snapshot(snapshot.clone())).unwrap();
+ assert_eq!(
+ block_on(store.projection_snapshot(projection_id.clone(), [22; 32])).unwrap(),
+ Some(snapshot)
+ );
+ assert_eq!(
+ block_on(store.put_projection_snapshot(
+ ProjectionSnapshot::new(projection_id, [22; 32], generation, 100, vec![4]).unwrap()
+ )),
+ Err(Error::CorruptProjectionDocument)
+ );
+ assert!(
+ block_on(
+ store.projection_snapshot(ProjectionId::parse("memory.other").unwrap(), [22; 32],)
+ )
+ .unwrap()
+ .is_none()
+ );
+ assert!(
+ block_on(
+ store.projection_snapshot(ProjectionId::parse("memory.today").unwrap(), [23; 32],)
+ )
+ .unwrap()
+ .is_none()
+ );
+}
diff --git a/crates/storage_sqlite/src/migration.rs b/crates/storage_sqlite/src/migration.rs
@@ -371,7 +371,7 @@ async fn metadata(
fn validate_plan(plan: &MigrationPlan) -> Result<(), Error> {
let valid = plan.minimum_version > 0
&& plan.minimum_version <= plan.current_version
- && plan.current_version <= 11
+ && plan.current_version <= 12
&& plan.steps.len() == usize::try_from(plan.current_version).unwrap_or(usize::MAX)
&& plan
.steps
@@ -487,6 +487,7 @@ const fn set_user_version_sql(version: u32) -> Option<&'static str> {
9 => Some("PRAGMA user_version = 9"),
10 => Some("PRAGMA user_version = 10"),
11 => Some("PRAGMA user_version = 11"),
+ 12 => Some("PRAGMA user_version = 12"),
_ => None,
}
}
@@ -713,7 +714,7 @@ mod tests {
.execute(&mut newer)
.await
.expect("application id");
- sqlx::raw_sql("PRAGMA user_version = 12")
+ sqlx::raw_sql("PRAGMA user_version = 13")
.execute(&mut newer)
.await
.expect("newer version");
@@ -722,10 +723,10 @@ mod tests {
Err(Error::SchemaTooNew {
database: RUNTIME_DATABASE,
supported: runtime::CURRENT_VERSION,
- actual: 12,
+ actual: 13,
})
));
- assert_eq!(pragma(&mut newer, "user_version").await, 12);
+ assert_eq!(pragma(&mut newer, "user_version").await, 13);
let mut wrong_identity = connection().await;
establish_runtime_version(&mut wrong_identity, 1).await;
diff --git a/crates/storage_sqlite/src/migration/authored_v10.rs b/crates/storage_sqlite/src/migration/authored_v10.rs
@@ -937,7 +937,7 @@ mod tests {
.fetch_one(&mut *connection)
.await
.expect("version after migration"),
- 11
+ i64::from(crate::migration::runtime::CURRENT_VERSION)
);
let raw = sqlx::query_scalar::<_, Vec<u8>>(
"SELECT signed_raw_json FROM radroots_runtime_authored_artifacts",
diff --git a/crates/storage_sqlite/src/migration/runtime/0012_materialized_projection_documents.up.sql b/crates/storage_sqlite/src/migration/runtime/0012_materialized_projection_documents.up.sql
@@ -0,0 +1,27 @@
+CREATE TABLE radroots_runtime_projection_documents (
+ projection_id TEXT NOT NULL CHECK(length(projection_id) BETWEEN 1 AND 128),
+ generation BLOB NOT NULL CHECK(length(generation) = 32),
+ document_key TEXT NOT NULL CHECK(length(document_key) BETWEEN 1 AND 512),
+ value BLOB NOT NULL CHECK(length(value) BETWEEN 1 AND 16777216),
+ value_sha256 BLOB NOT NULL CHECK(length(value_sha256) = 32),
+ PRIMARY KEY(projection_id, generation, document_key)
+) STRICT, WITHOUT ROWID;
+
+CREATE TABLE radroots_runtime_projection_snapshots (
+ projection_id TEXT NOT NULL CHECK(length(projection_id) BETWEEN 1 AND 128),
+ snapshot_id BLOB NOT NULL CHECK(length(snapshot_id) = 32),
+ generation BLOB NOT NULL CHECK(length(generation) = 32),
+ created_at_unix_ms INTEGER NOT NULL CHECK(created_at_unix_ms > 0),
+ value BLOB NOT NULL CHECK(length(value) BETWEEN 1 AND 16777216),
+ value_sha256 BLOB NOT NULL CHECK(length(value_sha256) = 32),
+ PRIMARY KEY(projection_id, snapshot_id)
+) STRICT, WITHOUT ROWID;
+
+CREATE INDEX radroots_runtime_projection_snapshots_created_idx
+ON radroots_runtime_projection_snapshots(projection_id, created_at_unix_ms, snapshot_id);
+
+CREATE TRIGGER radroots_runtime_projection_snapshots_update_guard
+BEFORE UPDATE ON radroots_runtime_projection_snapshots
+BEGIN
+ SELECT RAISE(ABORT, 'projection snapshots are immutable');
+END;
diff --git a/crates/storage_sqlite/src/migration/runtime/mod.rs b/crates/storage_sqlite/src/migration/runtime/mod.rs
@@ -6,7 +6,7 @@
/// Lowest runtime schema version this package can recognize.
pub const MINIMUM_VERSION: u32 = 1;
/// Current runtime schema version created by this package.
-pub const CURRENT_VERSION: u32 = 11;
+pub const CURRENT_VERSION: u32 = 12;
const RUNTIME_V1_SQL: &str = include_str!("0001_runtime.up.sql");
const CANONICAL_EVENT_STORAGE_V2_SQL: &str = include_str!("0002_canonical_event_storage.up.sql");
@@ -20,6 +20,8 @@ const LEGACY_IMPORT_COMMITS_V9_SQL: &str = include_str!("0009_legacy_import_comm
const PROJECTION_REBUILD_SOURCE_BINDING_V10_SQL: &str =
include_str!("0010_projection_rebuild_source_binding.up.sql");
const AUTHORED_OPERATIONS_V11_SQL: &str = include_str!("0011_authored_operations.up.sql");
+const MATERIALIZED_PROJECTION_DOCUMENTS_V12_SQL: &str =
+ include_str!("0012_materialized_projection_documents.up.sql");
/// Stable, non-SQL description of one forward runtime migration.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -402,6 +404,79 @@ const RUNTIME_V11_OBJECTS: &[&str] = &[
"radroots_runtime_source_generations_sequence_guard",
];
+const RUNTIME_V12_OBJECTS: &[&str] = &[
+ "radroots_runtime_atomic_commits",
+ "radroots_runtime_authored_artifacts",
+ "radroots_runtime_authored_artifacts_admission_ready_idx",
+ "radroots_runtime_authored_artifacts_signing_ready_idx",
+ "radroots_runtime_authored_atomic_commits",
+ "radroots_runtime_authored_atomic_commits_delete_guard",
+ "radroots_runtime_authored_atomic_commits_update_guard",
+ "radroots_runtime_authored_delivery_attempts",
+ "radroots_runtime_authored_delivery_plans",
+ "radroots_runtime_authored_delivery_ready_idx",
+ "radroots_runtime_authored_delivery_targets",
+ "radroots_runtime_authored_migration_evidence",
+ "radroots_runtime_authored_migration_evidence_delete_guard",
+ "radroots_runtime_authored_migration_evidence_update_guard",
+ "radroots_runtime_authored_operations",
+ "radroots_runtime_delivery_evidence",
+ "radroots_runtime_delivery_evidence_item_idx",
+ "radroots_runtime_event_index_checkpoints",
+ "radroots_runtime_event_index_manifests",
+ "radroots_runtime_event_index_shards",
+ "radroots_runtime_event_provenance",
+ "radroots_runtime_event_provenance_observed_idx",
+ "radroots_runtime_events",
+ "radroots_runtime_events_admission_idx",
+ "radroots_runtime_events_contract_metadata_guard",
+ "radroots_runtime_events_contract_metadata_insert_guard",
+ "radroots_runtime_events_delete_guard",
+ "radroots_runtime_events_event_id_idx",
+ "radroots_runtime_events_raw_update_guard",
+ "radroots_runtime_journal_idempotency_idx",
+ "radroots_runtime_journal_operations",
+ "radroots_runtime_journal_recovery_idx",
+ "radroots_runtime_legacy_event_staging",
+ "radroots_runtime_legacy_event_staging_delete_guard",
+ "radroots_runtime_legacy_event_staging_insert_guard",
+ "radroots_runtime_legacy_event_staging_update_guard",
+ "radroots_runtime_legacy_import_commit_delete_guard",
+ "radroots_runtime_legacy_import_commit_update_guard",
+ "radroots_runtime_legacy_import_commits",
+ "radroots_runtime_legacy_import_delete_guard",
+ "radroots_runtime_legacy_import_identity_guard",
+ "radroots_runtime_legacy_import_member_delete_guard",
+ "radroots_runtime_legacy_import_member_identity_guard",
+ "radroots_runtime_legacy_import_member_state_guard",
+ "radroots_runtime_legacy_import_members",
+ "radroots_runtime_legacy_import_state_guard",
+ "radroots_runtime_legacy_import_state_idx",
+ "radroots_runtime_legacy_imports",
+ "radroots_runtime_legacy_outbox_staging",
+ "radroots_runtime_legacy_outbox_staging_delete_guard",
+ "radroots_runtime_legacy_outbox_staging_insert_guard",
+ "radroots_runtime_legacy_outbox_staging_parent_idx",
+ "radroots_runtime_legacy_outbox_staging_update_guard",
+ "radroots_runtime_outbox_items",
+ "radroots_runtime_outbox_operation_idx",
+ "radroots_runtime_outbox_ready_idx",
+ "radroots_runtime_outbox_targets",
+ "radroots_runtime_projection_checkpoints",
+ "radroots_runtime_projection_documents",
+ "radroots_runtime_projection_invalidations",
+ "radroots_runtime_projection_rebuilds",
+ "radroots_runtime_projection_rebuilds_stage_idx",
+ "radroots_runtime_projection_snapshots",
+ "radroots_runtime_projection_snapshots_created_idx",
+ "radroots_runtime_projection_snapshots_update_guard",
+ "radroots_runtime_source_generations",
+ "radroots_runtime_source_generations_active_idx",
+ "radroots_runtime_source_generations_delete_guard",
+ "radroots_runtime_source_generations_identity_guard",
+ "radroots_runtime_source_generations_sequence_guard",
+];
+
/// Ordered, immutable runtime migration plan.
pub const MIGRATIONS: &[MigrationDescriptor] = &[
MigrationDescriptor {
@@ -470,6 +545,12 @@ pub const MIGRATIONS: &[MigrationDescriptor] = &[
up_sha256: "f07fe6695e611caa1ff2de87ee9d20356960346298f3df4158d63021fe0f9327",
owned_objects: RUNTIME_V11_OBJECTS,
},
+ MigrationDescriptor {
+ version: 12,
+ name: "materialized_projection_documents",
+ up_sha256: "c5a083ed078d2d1d9623b8c7fd6521055a59c6f4c099318779c9fa32fa17b211",
+ owned_objects: RUNTIME_V12_OBJECTS,
+ },
];
pub(crate) const fn migration_sql(version: u32) -> Option<&'static str> {
@@ -485,6 +566,7 @@ pub(crate) const fn migration_sql(version: u32) -> Option<&'static str> {
9 => Some(LEGACY_IMPORT_COMMITS_V9_SQL),
10 => Some(PROJECTION_REBUILD_SOURCE_BINDING_V10_SQL),
11 => Some(AUTHORED_OPERATIONS_V11_SQL),
+ 12 => Some(MATERIALIZED_PROJECTION_DOCUMENTS_V12_SQL),
_ => None,
}
}
@@ -528,8 +610,8 @@ mod tests {
fn migration_plan_matches_governed_snapshot() {
let snapshot = toml::from_str::<PlanSnapshot>(PLAN_SNAPSHOT).expect("valid snapshot");
assert_eq!(MINIMUM_VERSION, 1);
- assert_eq!(CURRENT_VERSION, 11);
- assert_eq!(MIGRATIONS.len(), 11);
+ assert_eq!(CURRENT_VERSION, 12);
+ assert_eq!(MIGRATIONS.len(), 12);
let migration = MIGRATIONS[8];
assert_eq!(snapshot.schema_version, 1);
assert_eq!(snapshot.database, "runtime.sqlite");
@@ -557,7 +639,7 @@ mod tests {
let sql = migration_sql(migration.version()).expect("registered SQL");
assert_eq!(format!("{:x}", Sha256::digest(sql)), migration.up_sha256());
}
- assert_eq!(migration_sql(12), None);
+ assert_eq!(migration_sql(13), None);
}
#[tokio::test]
diff --git a/crates/storage_sqlite/src/projection/mod.rs b/crates/storage_sqlite/src/projection/mod.rs
@@ -5,10 +5,10 @@ use radroots_storage::{
projection::{
ArtifactDigest, BoxFuture, EVENT_INDEX_SHARDS_MAX, EventId, EventIdRange,
EventIndexCheckpoint, EventIndexManifest, EventIndexShard, EventIndexShardCheckpoint,
- EventIndexShardId, InvalidationReason, ProjectionCheckpoint, ProjectionGeneration,
- ProjectionHealth, ProjectionId, ProjectionInvalidation, ProjectionRevision,
- ProjectionStatus, RawSourceDigest, RebuildFailure, RebuildStage, RebuildTicket,
- RebuildTicketId, RebuildTransition,
+ EventIndexShardId, InvalidationReason, ProjectionCheckpoint, ProjectionDocument,
+ ProjectionGeneration, ProjectionHealth, ProjectionId, ProjectionInvalidation,
+ ProjectionRevision, ProjectionSnapshot, ProjectionStatus, RawSourceDigest, RebuildFailure,
+ RebuildStage, RebuildTicket, RebuildTicketId, RebuildTransition,
},
};
use sqlx::{Row, Sqlite, SqliteConnection};
@@ -447,6 +447,137 @@ impl ProjectionStore for SqliteStorage {
Ok(())
})
}
+
+ fn put_projection_document(
+ &self,
+ projection_id: ProjectionId,
+ generation: ProjectionGeneration,
+ document: ProjectionDocument,
+ ) -> BoxFuture<'_, Result<(), Error>> {
+ Box::pin(async move {
+ self.require_projection_writer()?;
+ sqlx::query(
+ "INSERT INTO radroots_runtime_projection_documents (
+ projection_id, generation, document_key, value, value_sha256
+ ) VALUES (?, ?, ?, ?, ?)
+ ON CONFLICT(projection_id, generation, document_key) DO UPDATE SET
+ value = excluded.value,
+ value_sha256 = excluded.value_sha256",
+ )
+ .bind(projection_id.as_str())
+ .bind(generation.as_bytes().as_slice())
+ .bind(document.key())
+ .bind(document.value())
+ .bind(document.value_sha256().as_slice())
+ .execute(self.pool())
+ .await
+ .map_err(map_backend)?;
+ Ok(())
+ })
+ }
+
+ fn projection_document(
+ &self,
+ projection_id: ProjectionId,
+ generation: ProjectionGeneration,
+ key: String,
+ ) -> BoxFuture<'_, Result<Option<ProjectionDocument>, Error>> {
+ Box::pin(async move {
+ sqlx::query(
+ "SELECT document_key, value, value_sha256
+ FROM radroots_runtime_projection_documents
+ WHERE projection_id = ? AND generation = ? AND document_key = ?",
+ )
+ .bind(projection_id.as_str())
+ .bind(generation.as_bytes().as_slice())
+ .bind(key)
+ .fetch_optional(self.pool())
+ .await
+ .map_err(map_backend)?
+ .map(|row| {
+ ProjectionDocument::from_stored_parts(
+ row.try_get("document_key").map_err(map_corrupt)?,
+ row.try_get("value").map_err(map_corrupt)?,
+ array(row.try_get("value_sha256").map_err(map_corrupt)?)?,
+ )
+ })
+ .transpose()
+ })
+ }
+
+ fn put_projection_snapshot(
+ &self,
+ snapshot: ProjectionSnapshot,
+ ) -> BoxFuture<'_, Result<(), Error>> {
+ Box::pin(async move {
+ self.require_projection_writer()?;
+ let result = sqlx::query(
+ "INSERT INTO radroots_runtime_projection_snapshots (
+ projection_id, snapshot_id, generation, created_at_unix_ms,
+ value, value_sha256
+ ) VALUES (?, ?, ?, ?, ?, ?)
+ ON CONFLICT(projection_id, snapshot_id) DO NOTHING",
+ )
+ .bind(snapshot.projection_id().as_str())
+ .bind(snapshot.snapshot_id().as_slice())
+ .bind(snapshot.generation().as_bytes().as_slice())
+ .bind(i64_from_u64(snapshot.created_at_unix_ms())?)
+ .bind(snapshot.value())
+ .bind(snapshot.value_sha256().as_slice())
+ .execute(self.pool())
+ .await
+ .map_err(map_backend)?;
+ if result.rows_affected() == 1 {
+ return Ok(());
+ }
+ match self
+ .projection_snapshot(snapshot.projection_id().clone(), *snapshot.snapshot_id())
+ .await?
+ {
+ Some(existing) if existing == snapshot => Ok(()),
+ Some(_) => Err(Error::CorruptProjectionDocument),
+ None => Err(Error::CorruptProjectionDocument),
+ }
+ })
+ }
+
+ fn projection_snapshot(
+ &self,
+ projection_id: ProjectionId,
+ snapshot_id: [u8; 32],
+ ) -> BoxFuture<'_, Result<Option<ProjectionSnapshot>, Error>> {
+ Box::pin(async move {
+ sqlx::query(
+ "SELECT projection_id, snapshot_id, generation, created_at_unix_ms,
+ value, value_sha256
+ FROM radroots_runtime_projection_snapshots
+ WHERE projection_id = ? AND snapshot_id = ?",
+ )
+ .bind(projection_id.as_str())
+ .bind(snapshot_id.as_slice())
+ .fetch_optional(self.pool())
+ .await
+ .map_err(map_backend)?
+ .map(|row| {
+ ProjectionSnapshot::from_stored_parts(
+ ProjectionId::parse(
+ row.try_get::<String, _>("projection_id")
+ .map_err(map_corrupt)?,
+ )
+ .map_err(|_| Error::CorruptProjectionDocument)?,
+ array(row.try_get("snapshot_id").map_err(map_corrupt)?)?,
+ ProjectionGeneration::new(array(
+ row.try_get("generation").map_err(map_corrupt)?,
+ )?)
+ .map_err(|_| Error::CorruptProjectionDocument)?,
+ u64_from_i64(row.try_get("created_at_unix_ms").map_err(map_corrupt)?)?,
+ row.try_get("value").map_err(map_corrupt)?,
+ array(row.try_get("value_sha256").map_err(map_corrupt)?)?,
+ )
+ })
+ .transpose()
+ })
+ }
}
#[cfg_attr(coverage_nightly, coverage(off))]
@@ -1571,6 +1702,126 @@ mod tests {
}
#[tokio::test]
+ async fn materialized_documents_and_frozen_snapshots_round_trip_and_fail_closed() {
+ let writer = store(EventStoreMode::ReadWrite).await;
+ let id = projection_id();
+ let generation = generation(12);
+ let first = ProjectionDocument::new("context.alpha".into(), b"{\"cards\":[]}".to_vec())
+ .expect("document");
+ writer
+ .put_projection_document(id.clone(), generation, first)
+ .await
+ .expect("put document");
+ assert_eq!(
+ writer
+ .projection_document(id.clone(), generation, "context.alpha".into(),)
+ .await
+ .expect("document lookup")
+ .expect("document")
+ .value(),
+ b"{\"cards\":[]}"
+ );
+ writer
+ .put_projection_document(
+ id.clone(),
+ generation,
+ ProjectionDocument::new("context.alpha".into(), b"{\"cards\":[1]}".to_vec())
+ .expect("replacement"),
+ )
+ .await
+ .expect("replace document");
+ assert_eq!(
+ writer
+ .projection_document(id.clone(), generation, "context.alpha".into(),)
+ .await
+ .expect("document lookup")
+ .expect("document")
+ .value(),
+ b"{\"cards\":[1]}"
+ );
+
+ let snapshot = ProjectionSnapshot::new(
+ id.clone(),
+ [13; 32],
+ generation,
+ 1_000,
+ b"{\"frozen\":true}".to_vec(),
+ )
+ .expect("snapshot");
+ writer
+ .put_projection_snapshot(snapshot.clone())
+ .await
+ .expect("put snapshot");
+ writer
+ .put_projection_snapshot(snapshot.clone())
+ .await
+ .expect("idempotent snapshot replay");
+ let concurrent_snapshot = ProjectionSnapshot::new(
+ id.clone(),
+ [14; 32],
+ generation,
+ 1_001,
+ b"{\"frozen\":\"concurrent\"}".to_vec(),
+ )
+ .expect("concurrent snapshot");
+ let (left, right) = tokio::join!(
+ writer.put_projection_snapshot(concurrent_snapshot.clone()),
+ writer.put_projection_snapshot(concurrent_snapshot),
+ );
+ left.expect("concurrent left snapshot insert");
+ right.expect("concurrent right snapshot insert");
+ assert_eq!(
+ writer
+ .projection_snapshot(id.clone(), [13; 32])
+ .await
+ .expect("snapshot lookup"),
+ Some(snapshot)
+ );
+ assert_eq!(
+ writer
+ .put_projection_snapshot(
+ ProjectionSnapshot::new(
+ id.clone(),
+ [13; 32],
+ generation,
+ 1_000,
+ b"{\"frozen\":false}".to_vec(),
+ )
+ .expect("conflicting snapshot"),
+ )
+ .await,
+ Err(Error::CorruptProjectionDocument)
+ );
+
+ sqlx::query(
+ "UPDATE radroots_runtime_projection_documents
+ SET value = X'00' WHERE projection_id = ?",
+ )
+ .bind(id.as_str())
+ .execute(writer.pool())
+ .await
+ .expect("forge corrupt document");
+ assert_eq!(
+ writer
+ .projection_document(id, generation, "context.alpha".into())
+ .await,
+ Err(Error::CorruptProjectionDocument)
+ );
+
+ let read_only = store(EventStoreMode::ReadOnly).await;
+ assert_eq!(
+ read_only
+ .put_projection_document(
+ projection_id(),
+ generation,
+ ProjectionDocument::new("context.alpha".into(), vec![1]).expect("document"),
+ )
+ .await,
+ Err(Error::BackendUnavailable)
+ );
+ }
+
+ #[tokio::test]
async fn failed_rebuild_corruption_and_read_only_mode_fail_closed() {
let store = store(EventStoreMode::ReadWrite).await;
let initial = store
diff --git a/crates/sync/tests/push_enqueue.rs b/crates/sync/tests/push_enqueue.rs
@@ -384,6 +384,46 @@ impl ProjectionStore for FaultStorage {
) -> radroots_transport::BoxFuture<'_, Result<(), radroots_storage::Error>> {
ProjectionStore::put_event_index_checkpoint(self.inner.as_ref(), value)
}
+ fn put_projection_document(
+ &self,
+ projection_id: radroots_storage::projection::ProjectionId,
+ generation: radroots_storage::projection::ProjectionGeneration,
+ document: radroots_storage::projection::ProjectionDocument,
+ ) -> radroots_transport::BoxFuture<'_, Result<(), radroots_storage::Error>> {
+ ProjectionStore::put_projection_document(
+ self.inner.as_ref(),
+ projection_id,
+ generation,
+ document,
+ )
+ }
+ fn projection_document(
+ &self,
+ projection_id: radroots_storage::projection::ProjectionId,
+ generation: radroots_storage::projection::ProjectionGeneration,
+ key: String,
+ ) -> radroots_transport::BoxFuture<
+ '_,
+ Result<Option<radroots_storage::projection::ProjectionDocument>, radroots_storage::Error>,
+ > {
+ ProjectionStore::projection_document(self.inner.as_ref(), projection_id, generation, key)
+ }
+ fn put_projection_snapshot(
+ &self,
+ snapshot: radroots_storage::projection::ProjectionSnapshot,
+ ) -> radroots_transport::BoxFuture<'_, Result<(), radroots_storage::Error>> {
+ ProjectionStore::put_projection_snapshot(self.inner.as_ref(), snapshot)
+ }
+ fn projection_snapshot(
+ &self,
+ projection_id: radroots_storage::projection::ProjectionId,
+ snapshot_id: [u8; 32],
+ ) -> radroots_transport::BoxFuture<
+ '_,
+ Result<Option<radroots_storage::projection::ProjectionSnapshot>, radroots_storage::Error>,
+ > {
+ ProjectionStore::projection_snapshot(self.inner.as_ref(), projection_id, snapshot_id)
+ }
}
impl AtomicStorage for FaultStorage {
