From d6a6a08d924299cc45156fbca0d3344423632a13 Mon Sep 17 00:00:00 2001 From: Orual Date: Tue, 23 Jun 2026 20:11:42 -0400 Subject: [PATCH] PM-32: migrate projection to the native Data value type --- src/appview/proxy.rs | 10 +- src/appview/tests.rs | 10 +- src/appview/views.rs | 23 +- src/appview/writes.rs | 4 +- src/indexing/projection.rs | 486 +++++++++++++++++++++++++------------ 5 files changed, 357 insertions(+), 176 deletions(-) diff --git a/src/appview/proxy.rs b/src/appview/proxy.rs index ae24d6f..0295d9a 100644 --- a/src/appview/proxy.rs +++ b/src/appview/proxy.rs @@ -34,6 +34,7 @@ use jacquard_axum::ExtractXrpc; use jacquard_axum::oauth::ExtractOAuthSession; use jacquard_common::error::{ClientError, ClientErrorKind, HttpError}; use jacquard_common::types::string::Did; +use jacquard_common::types::value::to_data; use jacquard_common::xrpc::{Response as XrpcResponse, XrpcClient, XrpcResp}; use polymodel_api::com_atproto::identity::resolve_handle::ResolveHandleRequest; use polymodel_api::com_atproto::repo::{ @@ -329,13 +330,20 @@ async fn eager_project( let Some((rkey, cid)) = parse_record_ref(output_buffer) else { return; }; + let record_data = match to_data(record) { + Ok(d) => d, + Err(e) => { + tracing::warn!(error = %e, %collection, "proxied record to_data failed"); + return; + } + }; let event = ProjectionEvent { seq: 0, did: did.as_ref().to_string(), collection: collection.to_string(), rkey, action: action.to_string(), - record: Some(record.clone()), + record: Some(record_data), cid: Some(cid), }; // Serialize against PM-28's decision-making writes (which read local state diff --git a/src/appview/tests.rs b/src/appview/tests.rs index d30470c..99641b3 100644 --- a/src/appview/tests.rs +++ b/src/appview/tests.rs @@ -955,9 +955,9 @@ async fn feed_recent_paginates_by_rkey() { use jacquard_common::deps::bytes::Bytes; use jacquard_common::error::{AuthError, ClientError}; +use jacquard_common::types::value::{Data, to_data}; use jacquard_common::xrpc::Response as XrpcResponse; use polymodel_api::com_atproto::repo::get_record::GetRecordResponse; -use serde_json::Value; use crate::indexing::projection::{ ProjectionEvent, ProjectionInput, project_eager_record, project_event, @@ -1124,7 +1124,7 @@ async fn override_returns_local_when_pds_reports_not_found() { assert!(body.contains("fresh"), "{body}"); } -fn poly_event(action: &str, record: Option) -> ProjectionEvent { +fn poly_event(action: &str, record: Option) -> ProjectionEvent { ProjectionEvent { seq: 1, did: DID_A.to_string(), @@ -1140,8 +1140,10 @@ fn poly_event(action: &str, record: Option) -> ProjectionEvent { async fn eager_write_marks_pending_and_firehose_clears_it() { let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); - let record = - json!({"name": "x", "license": "CC-BY-4.0", "createdAt": "2024-01-01T00:00:00.000Z"}); + let record = to_data( + &json!({"name": "x", "license": "CC-BY-4.0", "createdAt": "2024-01-01T00:00:00.000Z"}), + ) + .unwrap(); project_eager_record(&pool, &poly_event("create", Some(record.clone()))) .await .unwrap(); diff --git a/src/appview/views.rs b/src/appview/views.rs index 9980821..4a7542d 100644 --- a/src/appview/views.rs +++ b/src/appview/views.rs @@ -1,11 +1,11 @@ //! SQLite read queries and generated-view builders for the read endpoints. //! //! Records come from the `record_json` column (written by the projection); the -//! required `record` field on thing/model/part views is built from it via -//! [`to_data`]. Author handles come from `identities` (refreshed by identity -//! events) and profile display fields from `profiles`. Authored image/blob fields -//! are deserialized through generated lexicon types and resolved to Bluesky CDN -//! URLs in appview response fields; no fake CDN/media data is synthesized. +//! required `record` field on thing/model/part views is built from it. Author +//! handles come from `identities` (refreshed by identity events) and profile +//! display fields from `profiles`. Authored image/blob fields are deserialized +//! through generated lexicon types and resolved to Bluesky CDN URLs in appview +//! response fields; no fake CDN/media data is synthesized. //! //! Cursor contract: clients treat `cursor` as opaque. //! - Chronological feeds (`getAuthorThings`, `getFeed?algorithm=recent`) keyset @@ -25,10 +25,7 @@ use jacquard::identity::resolver::IdentityResolver; use jacquard::oauth::client::OAuthSession; use jacquard_common::deps::smol_str::SmolStr; use jacquard_common::types::string::{AtUri, Datetime, Did, Handle, UriValue}; -use jacquard_common::types::{ - blob::BlobRef, - value::{Data, to_data}, -}; +use jacquard_common::types::{blob::BlobRef, value::Data}; use polymodel_api::space_polymodel::library::{ Actor, FeedItem, FeedView, File, Image, ImageView, ModelView, PartView, ThingView, ThingViewBasic, ViewerState, model::Model, part::Part, thing::Thing, @@ -67,9 +64,7 @@ fn datetime_nanos(nanos: i64) -> Datetime { /// Build the required `record: Data` field from a stored `record_json` blob. fn record_data(record_json: Option<&str>) -> AppResult { let raw = record_json.ok_or_else(|| internal("content row missing record_json"))?; - let value: serde_json::Value = - serde_json::from_str(raw).map_err(|e| internal(format!("record_json parse: {e}")))?; - to_data(&value).map_err(|e| internal(format!("record encode: {e}"))) + serde_json::from_str::(raw).map_err(|e| internal(format!("record_json parse: {e}"))) } fn file_data(file_json: Option<&str>) -> AppResult { @@ -1154,12 +1149,12 @@ pub(super) async fn polymodel_profile_view( }; let stats = profile_stats(state, did.as_ref()).await?; let handle_str = resolve_handle(state, did).await?; - let record_value: serde_json::Value = serde_json::from_str(&row.record_json) + let record: Data = serde_json::from_str(&row.record_json) .map_err(|e| internal(format!("profile record_json parse: {e}")))?; Ok(Some(polymodel_api::space_polymodel::actor::ProfileView { did: did.clone(), handle: s(handle_str), - record: to_data(&record_value).map_err(|e| internal(format!("record encode: {e}")))?, + record, display_name: row.display_name.map(s), description: row.description.map(s), avatar: avatar_url(did.as_ref(), row.avatar_json.as_deref())?, diff --git a/src/appview/writes.rs b/src/appview/writes.rs index 590c35e..455a1dc 100644 --- a/src/appview/writes.rs +++ b/src/appview/writes.rs @@ -22,6 +22,7 @@ use jacquard_common::types::collection::Collection; use jacquard_common::types::ident::AtIdentifier; use jacquard_common::types::recordkey::{RecordKey, Rkey}; use jacquard_common::types::string::{AtUri, Cid, Datetime, Did}; +use jacquard_common::types::value::to_data; use polymodel_api::app_bsky::actor::profile::Profile as BskyProfile; use polymodel_api::com_atproto::repo::strong_ref::StrongRef; use polymodel_api::space_polymodel::actor::ProfileView; @@ -1255,8 +1256,7 @@ async fn eager_project_record( record: &R, action: &str, ) -> AppResult<()> { - let record = - serde_json::to_value(record).map_err(|e| internal(format!("record serialize: {e}")))?; + let record = to_data(record).map_err(|e| internal(format!("record serialize: {e}")))?; let event = ProjectionEvent { seq: 0, did: actor.as_ref().to_string(), diff --git a/src/indexing/projection.rs b/src/indexing/projection.rs index 46e26c8..0b8f12b 100644 --- a/src/indexing/projection.rs +++ b/src/indexing/projection.rs @@ -10,7 +10,12 @@ //! subject of the existing row, delete it, and only decrement if a row was //! actually removed. -use serde_json::Value; +use jacquard_common::types::value::{Data, DataDeserializerError, from_data, to_data}; +use polymodel_api::space_polymodel::{ + actor::profile::Profile, + graph::{like::Like, listitem::Listitem, save::Save, tag::Tag}, + library::{model::Model, part::Part, thing::Thing}, +}; use sqlx::{Sqlite, SqlitePool, Transaction}; /// A repo-record projection event, decoupled from Hydrant's internal types so @@ -24,8 +29,9 @@ pub struct ProjectionEvent { pub rkey: String, /// `"create"`, `"update"`, or `"delete"`. pub action: String, - /// The record body for create/update events; `None` for delete events. - pub record: Option, + /// The record body for create/update events, carried as the native AT + /// Protocol value type; `None` for delete events. + pub record: Option, pub cid: Option, } @@ -44,6 +50,7 @@ pub struct IdentityProjection { /// Account status events are not projected and are dropped by /// [`ProjectionInput::from_hydrant`]. #[derive(Debug, Clone)] +#[allow(clippy::large_enum_variant)] pub enum ProjectionInput { Record(ProjectionEvent), Identity(IdentityProjection), @@ -63,7 +70,11 @@ impl ProjectionInput { collection: rec.collection.to_string(), rkey: rec.rkey.to_string(), action: rec.action.to_string(), - record: rec.record.clone(), + record: rec.record.as_ref().and_then(|v| { + to_data(v) + .map_err(|e| tracing::warn!(error = %e, "to_data conversion failed")) + .ok() + }), cid: rec.cid.as_ref().map(|c| c.to_string()), }))); } @@ -143,7 +154,7 @@ async fn record_pending( let (Some(cid), Some(record)) = (event.cid.clone(), event.record.as_ref()) else { return Ok(()); }; - let value = record.to_string(); + let value = serde_json::to_string(record).unwrap_or_default(); let now = now_nanos(); sqlx::query!( "INSERT INTO pending_writes (uri, cid, value, written_at) VALUES (?, ?, ?, ?) \ @@ -282,48 +293,29 @@ async fn project_record( } // --------------------------------------------------------------------------- -// Field extraction helpers (records arrive as serde_json::Value). +// Typed record decode from borrowed Data. // --------------------------------------------------------------------------- -fn s_str(v: &Value, key: &str) -> Option { - v.get(key).and_then(|x| x.as_str()).map(String::from) -} - -/// URI from a single strongRef field, e.g. `derivedFrom: { uri, cid }`. -fn ref_uri(v: &Value, key: &str) -> Option { - v.get(key) - .and_then(|r| r.get("uri")) - .and_then(|u| u.as_str()) - .map(String::from) -} - -/// URIs from an array of strongRefs, e.g. `models: [{ uri, cid }, ...]`. -fn ref_uris(v: &Value, key: &str) -> Vec { - v.get(key) - .and_then(|x| x.as_array()) - .map(|arr| { - arr.iter() - .filter_map(|el| el.get("uri").and_then(|u| u.as_str()).map(String::from)) - .collect() - }) - .unwrap_or_default() -} - -fn str_array(v: &Value, key: &str) -> Vec { - v.get(key) - .and_then(|x| x.as_array()) - .map(|arr| { - arr.iter() - .filter_map(|el| el.as_str().map(String::from)) - .collect() - }) - .unwrap_or_default() -} - -fn parse_nanos(v: &Value, key: &str) -> Option { - chrono::DateTime::parse_from_rfc3339(v.get(key)?.as_str()?) - .ok() - .and_then(|dt| dt.timestamp_nanos_opt()) +/// Decode a typed record from a borrowed `&Data` body, injecting the +/// collection NSID as the `$type` tag that all polymodel record structs are +/// internally tagged on. Wire bodies and test fixtures may omit `$type`; +/// injecting it at this single decode chokepoint — through which every record +/// flows (firehose, eager writes, proxy, fixtures) — lets `from_data` succeed +/// uniformly. The tag is always inserted/replaced; this is safe because +/// dispatch is collection-keyed, so the injected value always matches the +/// target struct's `#[serde(rename)]`. +fn decode(data: &Data, collection: &str) -> Result +where + T: for<'de> serde::Deserialize<'de>, +{ + let mut tagged = data.clone(); + if let Some(obj) = tagged.as_object_mut() { + obj.0.insert( + "$type".into(), + to_data(&collection.to_string()).expect("serializing a string cannot fail"), + ); + } + from_data(&tagged) } fn now_nanos() -> i64 { @@ -344,25 +336,44 @@ async fn project_thing_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "thing record decode failed"); + return Ok(()); + } + }; let uri = at_uri(&e.did, &e.collection, &e.rkey); - // Required content fields: skip the event gracefully if absent (malformed - // record) rather than storing a blank row. let Some(cid) = e.cid.clone() else { return Ok(()); }; - let Some(name) = s_str(rec, "name") else { - return Ok(()); - }; - let summary = s_str(rec, "summary"); - let license = s_str(rec, "license"); - let instructions = s_str(rec, "instructions"); - let cover_json = rec.get("cover").map(|v| v.to_string()); - let derived_from_uri = ref_uri(rec, "derivedFrom"); - - let tags = str_array(rec, "tags"); + let name = rec.name.to_string(); + let summary = rec.summary.as_ref().map(|s| s.to_string()); + let license = rec.license.to_string(); + let instructions = rec.instructions.as_ref().map(|lines| { + lines + .iter() + .map(|s| s.to_string()) + .collect::>() + .join("\n") + }); + let cover_json = data + .as_object() + .and_then(|o| o.0.get("cover")) + .and_then(|v| serde_json::to_string(v).ok()); + let derived_from_uri = rec + .derived_from + .as_ref() + .map(|r| r.uri.as_str().to_string()); + + let tags = rec + .tags + .as_ref() + .map(|v| v.iter().map(|s| s.to_string()).collect::>()) + .unwrap_or_default(); let tags_json = if tags.is_empty() { None } else { @@ -374,12 +385,13 @@ async fn project_thing_upsert( Some(tags.join(" ")) }; - let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos); + let created_at = rec + .created_at + .as_ref() + .timestamp_nanos_opt() + .unwrap_or_else(now_nanos); let indexed_at = now_nanos(); - // The full raw record, so read endpoints can populate the required `record` - // field on thingView directly from SQLite (and the write path gets - // read-your-own-write, since Hydrant has no public single-record write API). - let record_json = rec.to_string(); + let record_json = serde_json::to_string(data).unwrap_or_default(); // UPSERT (not INSERT OR REPLACE): REPLACE performs DELETE+INSERT with a new // rowid, which skips the AFTER UPDATE FTS trigger and (with recursive_triggers @@ -421,12 +433,12 @@ async fn project_thing_upsert( sqlx::query!("DELETE FROM thing_models WHERE thing_uri = ?", uri) .execute(&mut **tx) .await?; - for (pos, model_uri) in ref_uris(rec, "models").into_iter().enumerate() { + for (pos, model_ref) in rec.models.iter().flatten().enumerate() { sqlx::query!( "INSERT OR REPLACE INTO thing_models (thing_uri, model_uri, position) \ VALUES (?, ?, ?)", uri, - model_uri, + model_ref.uri.as_str(), pos as i64, ) .execute(&mut **tx) @@ -470,20 +482,29 @@ async fn project_model_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "model record decode failed"); + return Ok(()); + } + }; let uri = at_uri(&e.did, &e.collection, &e.rkey); let Some(cid) = e.cid.clone() else { return Ok(()); }; - let Some(name) = s_str(rec, "name") else { - return Ok(()); - }; - let summary = s_str(rec, "summary"); - let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos); + let name = rec.name.to_string(); + let summary = rec.summary.as_ref().map(|s| s.to_string()); + let created_at = rec + .created_at + .as_ref() + .timestamp_nanos_opt() + .unwrap_or_else(now_nanos); let indexed_at = now_nanos(); - let record_json = rec.to_string(); + let record_json = serde_json::to_string(data).unwrap_or_default(); sqlx::query!( "INSERT INTO models \ @@ -509,12 +530,12 @@ async fn project_model_upsert( sqlx::query!("DELETE FROM model_parts WHERE model_uri = ?", uri) .execute(&mut **tx) .await?; - for (pos, part_uri) in ref_uris(rec, "parts").into_iter().enumerate() { + for (pos, part_ref) in rec.parts.iter().enumerate() { sqlx::query!( "INSERT OR REPLACE INTO model_parts (model_uri, part_uri, position) \ VALUES (?, ?, ?)", uri, - part_uri, + part_ref.uri.as_str(), pos as i64, ) .execute(&mut **tx) @@ -545,21 +566,33 @@ async fn project_part_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "part record decode failed"); + return Ok(()); + } + }; let uri = at_uri(&e.did, &e.collection, &e.rkey); let Some(cid) = e.cid.clone() else { return Ok(()); }; - let Some(name) = s_str(rec, "name") else { - return Ok(()); - }; - let format = s_str(rec, "format"); - let file_json = rec.get("file").map(|v| v.to_string()); - let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos); + let name = rec.name.to_string(); + let format = rec.format.as_ref().map(|s| s.to_string()); + let file_json = data + .as_object() + .and_then(|o| o.0.get("file")) + .and_then(|v| serde_json::to_string(v).ok()); + let created_at = rec + .created_at + .as_ref() + .timestamp_nanos_opt() + .unwrap_or_else(now_nanos); let indexed_at = now_nanos(); - let record_json = rec.to_string(); + let record_json = serde_json::to_string(data).unwrap_or_default(); sqlx::query!( "INSERT INTO parts \ @@ -613,23 +646,33 @@ async fn project_profile_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; - let display_name = s_str(rec, "displayName"); - let description = s_str(rec, "description"); - let avatar_json = rec.get("avatar").map(|v| v.to_string()); - let default_license = s_str(rec, "defaultLicense"); - let pronouns = s_str(rec, "pronouns"); - let printers_json = rec - .get("printers") - .and_then(|v| v.as_array()) - .map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string())); - let links_json = rec - .get("links") - .and_then(|v| v.as_array()) - .map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string())); - let record_json = rec.to_string(); + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "profile record decode failed"); + return Ok(()); + } + }; + let display_name = rec.display_name.as_ref().map(|s| s.to_string()); + let description = rec.description.as_ref().map(|s| s.to_string()); + let avatar_json = data + .as_object() + .and_then(|o| o.0.get("avatar")) + .and_then(|v| serde_json::to_string(v).ok()); + let default_license = rec.default_license.as_ref().map(|s| s.to_string()); + let pronouns = rec.pronouns.as_ref().map(|s| s.to_string()); + let printers_json = data + .as_object() + .and_then(|o| o.0.get("printers")) + .and_then(|v| serde_json::to_string(v).ok()); + let links_json = data + .as_object() + .and_then(|o| o.0.get("links")) + .and_then(|v| serde_json::to_string(v).ok()); + let record_json = serde_json::to_string(data).unwrap_or_default(); let indexed_at = now_nanos(); sqlx::query!( @@ -703,16 +746,25 @@ async fn project_like_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; - let Some(subject) = ref_uri(rec, "subject") else { - return Ok(()); + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "like record decode failed"); + return Ok(()); + } }; + let subject = rec.subject.uri.as_str().to_string(); let Some(cid) = e.cid.clone() else { return Ok(()); }; - let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos); + let created_at = rec + .created_at + .as_ref() + .timestamp_nanos_opt() + .unwrap_or_else(now_nanos); let res = sqlx::query!( "INSERT OR IGNORE INTO likes (did, rkey, cid, subject_uri, created_at) VALUES (?, ?, ?, ?, ?)", @@ -775,17 +827,26 @@ async fn project_save_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; - let Some(subject) = ref_uri(rec, "subject") else { - return Ok(()); + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "save record decode failed"); + return Ok(()); + } }; + let subject = rec.subject.uri.as_str().to_string(); let Some(cid) = e.cid.clone() else { return Ok(()); }; - let note = s_str(rec, "note"); - let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos); + let note = rec.note.as_ref().map(|s| s.to_string()); + let created_at = rec + .created_at + .as_ref() + .timestamp_nanos_opt() + .unwrap_or_else(now_nanos); let res = sqlx::query!( "INSERT OR IGNORE INTO saves (did, rkey, cid, subject_uri, note, created_at) \ @@ -850,19 +911,26 @@ async fn project_tag_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; - let Some(subject) = ref_uri(rec, "subject") else { - return Ok(()); - }; - let Some(tag) = s_str(rec, "tag") else { - return Ok(()); + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "tag record decode failed"); + return Ok(()); + } }; + let subject = rec.subject.uri.as_str().to_string(); + let tag = rec.tag.to_string(); let Some(cid) = e.cid.clone() else { return Ok(()); }; - let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos); + let created_at = rec + .created_at + .as_ref() + .timestamp_nanos_opt() + .unwrap_or_else(now_nanos); let res = sqlx::query!( "INSERT OR IGNORE INTO tags (did, rkey, cid, subject_uri, tag, created_at) \ @@ -923,19 +991,26 @@ async fn project_listitem_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent, ) -> anyhow::Result<()> { - let Some(rec) = &e.record else { + let Some(data) = &e.record else { return Ok(()); }; - let Some(list_uri) = ref_uri(rec, "list") else { - return Ok(()); - }; - let Some(subject_uri) = ref_uri(rec, "subject") else { - return Ok(()); + let rec = match decode::(data, &e.collection) { + Ok(r) => r, + Err(err) => { + tracing::warn!(error = %err, "listitem record decode failed"); + return Ok(()); + } }; + let list_uri = rec.list.as_str().to_string(); + let subject_uri = rec.subject.uri.as_str().to_string(); let Some(cid) = e.cid.clone() else { return Ok(()); }; - let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos); + let created_at = rec + .created_at + .as_ref() + .timestamp_nanos_opt() + .unwrap_or_else(now_nanos); // No denormalized count for listitems; list membership is queried directly. sqlx::query!( @@ -1010,13 +1085,13 @@ mod tests { collection: "space.polymodel.library.thing".to_string(), rkey: rkey.to_string(), action: "create".to_string(), - record: Some(json!({ + record: Some(to_data(&json!({ "name": name, "summary": "a thing summary", "license": "MIT", "tags": ["cube", "calibration"], "createdAt": "2024-01-15T12:00:00.000Z", - "models": models.iter().map(|m| json!({"uri": m, "cid": "c"})).collect::>(), - })), + "models": models.iter().map(|m| json!({"uri": format!("at://{did}/space.polymodel.library.model/{m}"), "cid": "c"})).collect::>(), + })).unwrap()), cid: Some("thingcid".to_string()), } } @@ -1028,10 +1103,13 @@ mod tests { collection: "space.polymodel.graph.like".to_string(), rkey: rkey.to_string(), action: "create".to_string(), - record: Some(json!({ - "subject": {"uri": subject, "cid": "c"}, - "createdAt": "2024-02-01T00:00:00.000Z", - })), + record: Some( + to_data(&json!({ + "subject": {"uri": subject, "cid": "c"}, + "createdAt": "2024-02-01T00:00:00.000Z", + })) + .unwrap(), + ), cid: Some("likecid".to_string()), } } @@ -1125,7 +1203,7 @@ mod tests { let db = test_db().await; let did = "did:plc:abc"; let uri = "at://did:plc:abc/space.polymodel.library.thing/r1"; - let e = thing_event(10, did, "r1", "Calibration Cube", &["at://m1", "at://m2"]); + let e = thing_event(10, did, "r1", "Calibration Cube", &["m1", "m2"]); project_event(&db, &ProjectionInput::Record(e)) .await .expect("project thing"); @@ -1150,14 +1228,14 @@ mod tests { // Create with two models. project_event( &db, - &ProjectionInput::Record(thing_event(1, did, "r1", "Cube", &["at://m1", "at://m2"])), + &ProjectionInput::Record(thing_event(1, did, "r1", "Cube", &["m1", "m2"])), ) .await .unwrap(); assert_eq!(junction_count(&db, uri).await, 2); // Update: same rkey, different models. Old junction rows must be replaced. - let mut upd = thing_event(2, did, "r1", "Cube v2", &["at://m3"]); + let mut upd = thing_event(2, did, "r1", "Cube v2", &["m3"]); upd.action = "update".to_string(); project_event(&db, &ProjectionInput::Record(upd)) .await @@ -1166,7 +1244,7 @@ mod tests { assert_eq!(junction_count(&db, uri).await, 1); // m1/m2 gone, m3 present. let has_m3 = sqlx::query_as::<_, CountRow>( - "SELECT COUNT(*) AS count FROM thing_models WHERE thing_uri = ? AND model_uri = 'at://m3'", + "SELECT COUNT(*) AS count FROM thing_models WHERE thing_uri = ? AND model_uri = 'at://did:plc:abc/space.polymodel.library.model/m3'", ) .bind(uri) .fetch_one(&db) @@ -1183,7 +1261,7 @@ mod tests { let uri = "at://did:plc:abc/space.polymodel.library.thing/r1"; project_event( &db, - &ProjectionInput::Record(thing_event(1, did, "r1", "Cube", &["at://m1"])), + &ProjectionInput::Record(thing_event(1, did, "r1", "Cube", &["m1"])), ) .await .unwrap(); @@ -1204,7 +1282,7 @@ mod tests { #[tokio::test] async fn like_create_increments_count() { let db = test_db().await; - let subject = "at://did:plc:author/.../thing"; + let subject = "at://did:plc:author/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(like_event(1, "did:plc:a", "k1", subject)), @@ -1225,7 +1303,7 @@ mod tests { #[tokio::test] async fn like_delete_decrements_count_two_step() { let db = test_db().await; - let subject = "at://did:plc:author/.../thing"; + let subject = "at://did:plc:author/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(like_event(1, "did:plc:a", "k1", subject)), @@ -1252,7 +1330,7 @@ mod tests { async fn like_delete_when_absent_is_idempotent() { // Deleting a like that was never there must not panic nor go negative. let db = test_db().await; - let subject = "at://did:plc:author/.../thing"; + let subject = "at://did:plc:author/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(delete_event( @@ -1271,7 +1349,7 @@ mod tests { async fn duplicate_like_does_not_double_count() { // Replay: same like event twice (e.g. eager write + firehose delivery). let db = test_db().await; - let subject = "at://did:plc:author/.../thing"; + let subject = "at://did:plc:author/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(like_event(1, "did:plc:a", "k1", subject)), @@ -1301,7 +1379,12 @@ mod tests { assert_eq!(cursor(&db).await, Some(7)); project_event( &db, - &ProjectionInput::Record(like_event(99, "did:plc:b", "k1", "at://x")), + &ProjectionInput::Record(like_event( + 99, + "did:plc:b", + "k1", + "at://did:plc:author/space.polymodel.library.thing/x", + )), ) .await .unwrap(); @@ -1311,18 +1394,21 @@ mod tests { #[tokio::test] async fn save_create_and_delete_maintain_count() { let db = test_db().await; - let subject = "at://did:plc:author/.../thing"; + let subject = "at://did:plc:author/space.polymodel.library.thing/t1"; let save = ProjectionEvent { seq: 1, did: "did:plc:a".to_string(), collection: "space.polymodel.graph.save".to_string(), rkey: "s1".to_string(), action: "create".to_string(), - record: Some(json!({ - "subject": {"uri": subject, "cid": "c"}, - "note": "useful", - "createdAt": "2024-03-01T00:00:00.000Z", - })), + record: Some( + to_data(&json!({ + "subject": {"uri": subject, "cid": "c"}, + "note": "useful", + "createdAt": "2024-03-01T00:00:00.000Z", + })) + .unwrap(), + ), cid: Some("savecid".to_string()), }; project_event(&db, &ProjectionInput::Record(save)) @@ -1361,12 +1447,15 @@ mod tests { collection: "space.polymodel.library.thing".to_string(), rkey: "r1".to_string(), action: "create".to_string(), - record: Some(json!({ - "name": "Calibration Cube", - "license": "MIT", "tags": ["cube"], - "createdAt": "2024-01-01T00:00:00.000Z", - "models": [], - })), + record: Some( + to_data(&json!({ + "name": "Calibration Cube", + "license": "MIT", "tags": ["cube"], + "createdAt": "2024-01-01T00:00:00.000Z", + "models": [], + })) + .unwrap(), + ), cid: Some("cid1".to_string()), }; project_event(&db, &ProjectionInput::Record(create.clone())) @@ -1378,12 +1467,15 @@ mod tests { update.seq = 2; update.action = "update".to_string(); update.cid = Some("cid2".to_string()); - update.record = Some(json!({ - "name": "Friendly Sphere", - "license": "MIT", "tags": ["sphere"], - "createdAt": "2024-01-02T00:00:00.000Z", - "models": [], - })); + update.record = Some( + to_data(&json!({ + "name": "Friendly Sphere", + "license": "MIT", "tags": ["sphere"], + "createdAt": "2024-01-02T00:00:00.000Z", + "models": [], + })) + .unwrap(), + ); project_event(&db, &ProjectionInput::Record(update)) .await .unwrap(); @@ -1405,4 +1497,88 @@ mod tests { .count; assert_eq!(live, 1); } + + #[tokio::test] + async fn malformed_and_incomplete_records_skip_gracefully() { + // A non-object body and a lexicon-incomplete record (Thing missing the + // required `createdAt`) must both degrade to a skip: project_event + // returns Ok, the cursor advances, and no partial row is written. + let db = test_db().await; + let did = "did:plc:abc"; + + let poison = ProjectionEvent { + seq: 1, + did: did.to_string(), + collection: "space.polymodel.library.thing".to_string(), + rkey: "bad1".to_string(), + action: "create".to_string(), + record: Some(to_data(&json!("not an object")).unwrap()), + cid: Some("cid".to_string()), + }; + project_event(&db, &ProjectionInput::Record(poison)) + .await + .expect("malformed record must not poison the stream"); + assert_eq!(cursor(&db).await, Some(1)); + + let incomplete = ProjectionEvent { + seq: 2, + did: did.to_string(), + collection: "space.polymodel.library.thing".to_string(), + rkey: "bad2".to_string(), + action: "create".to_string(), + record: Some(to_data(&json!({"name": "Incomplete", "license": "MIT"})).unwrap()), + cid: Some("cid".to_string()), + }; + project_event(&db, &ProjectionInput::Record(incomplete)) + .await + .expect("incomplete record must not poison the stream"); + assert_eq!(cursor(&db).await, Some(2)); + + let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM things") + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(rows, 0, "no partial rows should exist for skipped records"); + } + + #[tokio::test] + async fn listitem_create_projects_with_direct_list_uri() { + // Regression: the old ref_uri("list") expected a nested {uri, cid} + // object, but the lexicon type is a direct AtUri. Conformant listitems + // were silently never projected. The typed path now projects them. + let db = test_db().await; + let list_uri = "at://did:plc:abc/space.polymodel.graph.list/mylist"; + let subject_uri = "at://did:plc:xyz/space.polymodel.library.thing/t1"; + let event = ProjectionEvent { + seq: 10, + did: "did:plc:a".to_string(), + collection: "space.polymodel.graph.listitem".to_string(), + rkey: "li1".to_string(), + action: "create".to_string(), + record: Some( + to_data(&json!({ + "list": list_uri, + "subject": {"uri": subject_uri, "cid": "c"}, + "createdAt": "2024-04-01T00:00:00.000Z", + })) + .unwrap(), + ), + cid: Some("licid".to_string()), + }; + project_event(&db, &ProjectionInput::Record(event)) + .await + .unwrap(); + + let row: Option<(String, String)> = sqlx::query_as( + "SELECT list_uri, subject_uri FROM listitems WHERE did = ? AND rkey = ?", + ) + .bind("did:plc:a") + .bind("li1") + .fetch_optional(&db) + .await + .unwrap(); + let (stored_list, stored_subject) = row.expect("listitem must be projected"); + assert_eq!(stored_list, list_uri); + assert_eq!(stored_subject, subject_uri); + } } -- 2.51.2