commit 7de5e0c72729ec817b9b8561508a5eebbcf05586
parent df9848ef1d973ec91b4ba08751337897ac6df356
Author: triesap <tyson@radroots.org>
Date: Sat, 8 Aug 2026 01:34:58 +0000
feat(mobile): sync Today from bounded relay data
- propagate exact selectors through canonical sync pulls
- fetch and typed-admit Today events before projection
- expose relay sync outcomes through the native boundary
- honor public, simulator, and device context policies
Diffstat:
3 files changed, 305 insertions(+), 3 deletions(-)
diff --git a/core/crates/tera_core/src/runtime/product_surface.rs b/core/crates/tera_core/src/runtime/product_surface.rs
@@ -21,7 +21,7 @@ pub use authoring::{
};
pub use context::{
ContextAdmission, ContextRank, LocalNetwork, LocalNetworkAdmission, LocalNetworkError,
- LocalityEvidence,
+ LocalNetworkRelayPolicy, LocalityEvidence,
};
pub use cursor::{CursorError, CursorScope, TodayCursor, TodayCursorPosition};
pub use identity::{CARD_ID_SCHEMA_VERSION, CardId, CardIdError, CardSourceIdentity};
@@ -43,6 +43,8 @@ pub use ranking::{RankError, TODAY_RANK_SCHEMA_VERSION, TimeRelevance, TodayRank
pub use today::{
TodayError, TodayIngestReceipt, TodayPageRequest, TodayProjectionUpdate, TodayRefreshReceipt,
};
+#[cfg(feature = "mobile-social")]
+pub use today::{TodayRelaySyncState, TodaySyncReceipt};
use super::RadrootsRuntime;
diff --git a/core/crates/tera_core/src/runtime/product_surface/context.rs b/core/crates/tera_core/src/runtime/product_surface/context.rs
@@ -35,6 +35,15 @@ pub enum LocalNetworkError {
DuplicateAuthor,
}
+/// Host environment whose destination policy governs LocalNetwork relays.
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum LocalNetworkRelayPolicy {
+ Public,
+ Simulator,
+ Device,
+}
+
impl LocalNetwork {
pub fn new(
id: String,
@@ -44,6 +53,28 @@ impl LocalNetwork {
followed_authors: Vec<String>,
generation: u64,
) -> Result<Self, LocalNetworkError> {
+ Self::new_for_relay_policy(
+ id,
+ label,
+ relay_urls,
+ locality,
+ followed_authors,
+ generation,
+ LocalNetworkRelayPolicy::Public,
+ )
+ }
+
+ /// Constructs a context under the exact host relay destination policy.
+ #[allow(clippy::too_many_arguments)]
+ pub fn new_for_relay_policy(
+ id: String,
+ label: String,
+ relay_urls: Vec<String>,
+ locality: Option<String>,
+ followed_authors: Vec<String>,
+ generation: u64,
+ relay_policy: LocalNetworkRelayPolicy,
+ ) -> Result<Self, LocalNetworkError> {
validate_text(&id, "id")?;
validate_text(&label, "label")?;
if let Some(locality) = locality.as_deref() {
@@ -58,8 +89,15 @@ impl LocalNetwork {
if relay.is_empty() || relay.len() > RELAY_URL_MAX_BYTES {
return Err(LocalNetworkError::InvalidRelay);
}
- let relay = RelayUrl::parse(relay, RelayUrlPolicy::Public)
- .map_err(|_| LocalNetworkError::InvalidRelay)?;
+ let relay = RelayUrl::parse(
+ relay,
+ match relay_policy {
+ LocalNetworkRelayPolicy::Public => RelayUrlPolicy::Public,
+ LocalNetworkRelayPolicy::Simulator => RelayUrlPolicy::Local,
+ LocalNetworkRelayPolicy::Device => RelayUrlPolicy::PrivateNetwork,
+ },
+ )
+ .map_err(|_| LocalNetworkError::InvalidRelay)?;
if !relays.insert(relay.clone()) {
return Err(LocalNetworkError::DuplicateRelay);
}
@@ -369,5 +407,41 @@ mod tests {
),
Err(LocalNetworkError::InvalidRelay)
));
+ assert!(
+ LocalNetwork::new_for_relay_policy(
+ "id".into(),
+ "label".into(),
+ vec!["ws://127.0.0.1:7447".into()],
+ None,
+ vec![],
+ 0,
+ LocalNetworkRelayPolicy::Simulator,
+ )
+ .is_ok()
+ );
+ assert!(
+ LocalNetwork::new_for_relay_policy(
+ "id".into(),
+ "label".into(),
+ vec!["wss://192.168.1.7:7447".into()],
+ None,
+ vec![],
+ 0,
+ LocalNetworkRelayPolicy::Device,
+ )
+ .is_ok()
+ );
+ assert!(
+ LocalNetwork::new_for_relay_policy(
+ "id".into(),
+ "label".into(),
+ vec!["ws://192.168.1.7:7447".into()],
+ None,
+ vec![],
+ 0,
+ LocalNetworkRelayPolicy::Device,
+ )
+ .is_err()
+ );
}
}
diff --git a/core/crates/tera_core/src/runtime/product_surface/today.rs b/core/crates/tera_core/src/runtime/product_surface/today.rs
@@ -19,6 +19,19 @@ use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use thiserror::Error;
+#[cfg(feature = "mobile-social")]
+use radroots_event::admission::ContractValidatedEvent;
+#[cfg(feature = "mobile-social")]
+use radroots_sync::{
+ PullRequest,
+ ingest::{AdmissionDecision, AdmissionPolicy},
+ pull::PullTermination,
+};
+#[cfg(feature = "mobile-social")]
+use radroots_transport::{
+ Target, outcome::FetchTargetState, source::FetchSelector, target::TargetSet,
+};
+
use super::{
CardId, CardLifecycleState, ClassifiedCard, CursorError, CursorScope, LocalAuthorOverlay,
LocalNetwork, LocalityEvidence, MeSnapshot, MediaReference, MediaVerificationState,
@@ -33,6 +46,12 @@ 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;
+#[cfg(feature = "mobile-social")]
+const TODAY_SYNC_PAGE_LIMIT: u16 = 500;
+#[cfg(feature = "mobile-social")]
+const TODAY_SYNC_MAX_PAGES: u16 = 8;
+#[cfg(feature = "mobile-social")]
+const TODAY_SYNC_KINDS: [u32; 7] = [0, 1, 5, 1111, 30_402, 31_922, 31_923];
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";
@@ -66,6 +85,27 @@ pub struct TodayIngestReceipt {
pub projection: TodayRefreshReceipt,
}
+#[cfg(feature = "mobile-social")]
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum TodayRelaySyncState {
+ Complete,
+ Partial,
+ Offline,
+}
+
+#[cfg(feature = "mobile-social")]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TodaySyncReceipt {
+ pub relay_state: TodayRelaySyncState,
+ pub pages_fetched: u16,
+ pub events_observed: u64,
+ pub events_admitted: u64,
+ pub events_rejected: u64,
+ pub projection: TodayRefreshReceipt,
+}
+
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct TodayPageRequest {
pub limit: u16,
@@ -157,6 +197,72 @@ struct FrozenTodaySnapshot {
}
impl RadrootsRuntime {
+ /// Pulls bounded Today-relevant relay pages, canonically admits valid
+ /// observations, and materializes the selected LocalNetwork projection.
+ #[cfg(feature = "mobile-social")]
+ pub async fn phase1_sync_today(
+ &self,
+ context: &LocalNetwork,
+ now_unix_seconds: u64,
+ update: TodayProjectionUpdate,
+ ) -> Result<TodaySyncReceipt, TodayError> {
+ if now_unix_seconds == 0 {
+ return Err(TodayError::InvalidRequest);
+ }
+ let targets = context
+ .relay_urls
+ .iter()
+ .map(Target::nostr_relay)
+ .collect::<Result<Vec<_>, _>>()
+ .map_err(|_| TodayError::InvalidRequest)?;
+ let targets = TargetSet::new(targets).map_err(|_| TodayError::InvalidRequest)?;
+ let selector = FetchSelector::all()
+ .with_kinds(TODAY_SYNC_KINDS.to_vec())
+ .map_err(|_| TodayError::InvalidRequest)?;
+ let request = PullRequest::new(targets, TODAY_SYNC_PAGE_LIMIT, TODAY_SYNC_MAX_PAGES)
+ .map_err(|_| TodayError::RuntimeUnavailable)?
+ .with_selector(selector);
+ let sync = self
+ .client
+ .sync()
+ .map_err(|_| TodayError::RuntimeUnavailable)?
+ .ok_or(TodayError::RuntimeUnavailable)?;
+ let pull = sync
+ .pull(request, &TodayAdmissionPolicy)
+ .await
+ .map_err(|_| TodayError::RuntimeUnavailable)?;
+ let projection = self
+ .phase1_refresh_today(context, now_unix_seconds, update)
+ .await?;
+ let events_admitted = pull
+ .ingest_outcomes()
+ .iter()
+ .filter(|outcome| outcome.is_ok())
+ .count() as u64;
+ let events_observed = u64::try_from(pull.events_observed()).unwrap_or(u64::MAX);
+ let events_rejected = events_observed.saturating_sub(events_admitted);
+ let target_complete = !pull.target_outcomes().is_empty()
+ && pull
+ .target_outcomes()
+ .iter()
+ .all(|outcome| outcome.state() == FetchTargetState::Complete);
+ let relay_state = match pull.termination() {
+ PullTermination::Complete if target_complete => TodayRelaySyncState::Complete,
+ PullTermination::SourceFailed if pull.pages_fetched() == 0 => {
+ TodayRelaySyncState::Offline
+ }
+ _ => TodayRelaySyncState::Partial,
+ };
+ Ok(TodaySyncReceipt {
+ relay_state,
+ pages_fetched: pull.pages_fetched(),
+ events_observed,
+ events_admitted,
+ events_rejected,
+ projection,
+ })
+ }
+
/// Durably admits one already verified and visibility-authorized relay observation,
/// then advances the selected LocalNetwork projection.
pub async fn phase1_ingest_visible(
@@ -544,6 +650,27 @@ impl RadrootsRuntime {
}
}
+#[cfg(feature = "mobile-social")]
+struct TodayAdmissionPolicy;
+
+#[cfg(feature = "mobile-social")]
+impl AdmissionPolicy for TodayAdmissionPolicy {
+ fn policy_id(&self) -> &'static str {
+ "radroots.mobile.today.v1"
+ }
+
+ fn decide(&self, event: &ContractValidatedEvent) -> AdmissionDecision {
+ let admitted = verify_nip01_event(event.event().clone())
+ .ok()
+ .and_then(|event| admit_verified_event(event).ok());
+ if admitted.is_some() {
+ AdmissionDecision::Visible
+ } else {
+ AdmissionDecision::Reject
+ }
+ }
+}
+
fn ingest_receipt(
receipt: AdmissionReceipt,
projection: TodayRefreshReceipt,
@@ -1121,6 +1248,9 @@ fn decode<T: for<'de> Deserialize<'de>>(value: &[u8]) -> Result<T, TodayError> {
#[cfg(test)]
mod tests {
use super::*;
+ #[cfg(feature = "mobile-social")]
+ use std::sync::{Arc, Mutex, RwLock, atomic::AtomicBool};
+
use nostr::secp256k1::Message;
use nostr::{Keys, SECP256K1};
use radroots_event::{
@@ -1129,6 +1259,12 @@ mod tests {
wire::{Nip01EventWire, compute_canonical_nip01_event_id},
};
use radroots_event_codec::verify::Nip01SignatureVerifier;
+ #[cfg(feature = "mobile-social")]
+ use radroots_transport::{
+ Error as TransportError, EventSource, FetchPage, FetchRequest, SourceStatus,
+ outcome::{FetchTargetOutcome, FetchTargetState},
+ source::NextPage,
+ };
use radroots_transport::{
Target, TransportId,
source::{EventProvenance, ObservedEvent},
@@ -1138,6 +1274,48 @@ mod tests {
struct Allow;
+ #[cfg(feature = "mobile-social")]
+ struct TodaySource {
+ event: SignedEvent,
+ requested_kinds: Mutex<Vec<Vec<u32>>>,
+ }
+
+ #[cfg(feature = "mobile-social")]
+ impl EventSource for TodaySource {
+ fn status(
+ &self,
+ ) -> radroots_transport::BoxFuture<'_, Result<SourceStatus, TransportError>> {
+ Box::pin(async { unreachable!("Today sync does not inspect source status") })
+ }
+
+ fn fetch(
+ &self,
+ request: FetchRequest,
+ ) -> radroots_transport::BoxFuture<'_, Result<FetchPage, TransportError>> {
+ Box::pin(async move {
+ self.requested_kinds
+ .lock()
+ .expect("requested kinds")
+ .push(request.selector().kinds().to_vec());
+ let target = request.target_set().targets()[0].clone();
+ let provenance = EventProvenance::new(
+ TransportId::NOSTR,
+ target.fingerprint().clone(),
+ 2_000_000_100_000,
+ )?;
+ FetchPage::for_request(
+ &request,
+ vec![ObservedEvent::new(self.event.clone(), provenance)],
+ vec![FetchTargetOutcome::new(
+ target.fingerprint().clone(),
+ FetchTargetState::Complete,
+ )],
+ NextPage::Complete,
+ )
+ })
+ }
+ }
+
impl AdmissionPolicy for Allow {
type Error = core::convert::Infallible;
@@ -1292,6 +1470,54 @@ mod tests {
.expect("ingest")
}
+ #[cfg(feature = "mobile-social")]
+ #[tokio::test]
+ async fn relay_sync_fetches_the_exact_today_selector_and_projects_real_events() {
+ let source = Arc::new(TodaySource {
+ event: signed(1, Vec::new(), "Fresh from the field", 2_000_000_000),
+ requested_kinds: Mutex::new(Vec::new()),
+ });
+ let client = radroots_sdk::ClientBuilder::memory_default()
+ .source(source.clone())
+ .host_sync(radroots_sdk::sync::HostPolicy::standard())
+ .build()
+ .expect("client");
+ let runtime = RadrootsRuntime {
+ client,
+ started_unix_ms: 1,
+ shutting_down: AtomicBool::new(false),
+ platform_app: RwLock::new(None),
+ store_public_key: None,
+ };
+ let context = context(None, 1);
+
+ let receipt = runtime
+ .phase1_sync_today(&context, 2_000_000_200, TodayProjectionUpdate::Incremental)
+ .await
+ .expect("Today sync");
+ assert_eq!(receipt.relay_state, TodayRelaySyncState::Complete);
+ assert_eq!(receipt.pages_fetched, 1);
+ assert_eq!(receipt.events_observed, 1);
+ assert_eq!(receipt.events_admitted, 1);
+ assert_eq!(receipt.events_rejected, 0);
+ assert_eq!(receipt.projection.visible_cards, 1);
+ assert_eq!(
+ source
+ .requested_kinds
+ .lock()
+ .expect("requested kinds")
+ .as_slice(),
+ &[TODAY_SYNC_KINDS.to_vec()]
+ );
+ let page = runtime
+ .phase1_today_page(&context, TodayPageRequest::first(20, 2_000_000_200))
+ .await
+ .expect("Today page");
+ assert_eq!(page.items.len(), 1);
+ assert_eq!(page.items[0].card.card_type, TodayCardType::Update);
+ assert_eq!(page.items[0].card.content, "Fresh from the field");
+ }
+
#[tokio::test]
async fn equal_timestamp_pages_are_complete_and_remain_frozen_across_ingest() {
let runtime = RadrootsRuntime::test_memory().expect("runtime");
