Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495//! Projection of Hydrant ATProto events into the SQLite derived tables.//!//! [`project_event`] is the single entry point. Each event is projected inside//! one SQLite transaction, and the projection cursor is persisted in that same//! transaction so crash recovery never advances past an unprojected event.//!//! All upserts are idempotent (`INSERT OR REPLACE` for content, `INSERT OR//! IGNORE` for social records) so replaying events from an earlier cursor is//! safe. Social counts are maintained with a two-step delete: look up the//! subject of the existing row, delete it, and only decrement if a row was//! actually removed.
use jacquard_common::types::string::AtUri;use jacquard_common::types::value::{Data, DataDeserializerError, from_data, to_data};use polymodel_api::space_polymodel::{ actor::profile::Profile, graph::{follow::Follow, like::Like, list::List, listitem::Listitem, save::Save, tag::Tag}, library::{File, model::Model, part::Part, thing::Thing},};use sqlx::{Sqlite, SqlitePool, Transaction};use std::collections::HashSet;
const DERIVED_LDRAW_PROVENANCE: &str = "ordinary-part";
type OrdinarySource = (Vec<String>, Vec<u8>, i64, String, String);
/// A repo-record projection event, decoupled from Hydrant's internal types so/// the projection can be unit-tested with synthetic data.#[derive(Debug, Clone)]pub struct ProjectionEvent { /// Monotonic Hydrant event id; used as the durable projection cursor. pub seq: u64, pub did: String, pub collection: String, pub rkey: String, /// `"create"`, `"update"`, or `"delete"`. pub action: String, /// The record body for create/update events, carried as the native AT /// Protocol value type; `None` for delete events. pub record: Option<Data>, pub cid: Option<String>,}
/// An identity projection event: a DID's handle may have changed. Projected into/// the `identities` table so profile/actor hydration sees handle changes. Record/// events never carry the current handle — handles live in the DID doc, not the/// record — so without this, handle changes (identity events) would be missed.#[derive(Debug, Clone)]pub struct IdentityProjection { pub seq: u64, pub did: String, pub handle: Option<String>,}
/// Input to [`project_event`]: either a record event or an identity event./// 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),}
impl ProjectionInput { /// Convert a Hydrant [`hydrant::control::Event`] into a projection input. /// /// Returns `Ok(None)` only for events that are neither record nor identity /// (e.g. account status events). Identity events are projected so handle /// changes are caught for profile hydration. pub fn from_hydrant(event: &hydrant::control::Event) -> anyhow::Result<Option<Self>> { if let Some(rec) = event.record.as_ref() { return Ok(Some(ProjectionInput::Record(ProjectionEvent { seq: event.id, did: rec.did.to_string(), collection: rec.collection.to_string(), rkey: rec.rkey.to_string(), action: rec.action.to_string(), 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()), }))); } if let Some(ident) = event.identity.as_ref() { return Ok(Some(ProjectionInput::Identity(IdentityProjection { seq: event.id, did: ident.did.to_string(), handle: ident.handle.as_ref().map(|h| h.to_string()), }))); } Ok(None) }
/// The monotonic event id (durable projection cursor) for this input. pub fn seq(&self) -> u64 { match self { ProjectionInput::Record(e) => e.seq, ProjectionInput::Identity(e) => e.seq, } }}
/// Project a single event into SQLite, advancing the cursor on success.////// Record events are dispatched to their collection rule; identity events refresh/// the DID→handle map. The cursor advances on every event either way.pub async fn project_event(db: &SqlitePool, input: &ProjectionInput) -> anyhow::Result<()> { let mut tx = db.begin().await?; let should_apply = match input { ProjectionInput::Record(event) => { event_is_current( &mut tx, event.seq, &event.did, &event.collection, &event.rkey, ) .await? } ProjectionInput::Identity(event) => { event_is_current(&mut tx, event.seq, &event.did, "#identity", "self").await? } }; if should_apply { match input { ProjectionInput::Record(event) => { project_record(&mut tx, event).await?; record_revision( &mut tx, event.seq, &event.did, &event.collection, &event.rkey, ) .await?; clear_pending(&mut tx, event).await?; } ProjectionInput::Identity(event) => { project_identity_upsert(&mut tx, event).await?; record_revision(&mut tx, event.seq, &event.did, "#identity", "self").await?; } } } else { tracing::debug!(seq = input.seq(), "discarded stale projection event"); } super::db::save_cursor(&mut tx, input.seq()).await?; tx.commit().await?; Ok(())}
async fn event_is_current( tx: &mut Transaction<'_, Sqlite>, seq: u64, did: &str, collection: &str, rkey: &str,) -> anyhow::Result<bool> { if seq == 0 { return Ok(true); } let seq = i64::try_from(seq) .map_err(|_| anyhow::anyhow!("projection sequence exceeds SQLite range"))?; let previous = sqlx::query_scalar!( "SELECT seq FROM projection_revisions WHERE did = ? AND collection = ? AND rkey = ?", did, collection, rkey, ) .fetch_optional(&mut **tx) .await?; Ok(previous.is_none_or(|previous| seq > previous))}
async fn record_revision( tx: &mut Transaction<'_, Sqlite>, seq: u64, did: &str, collection: &str, rkey: &str,) -> anyhow::Result<()> { if seq == 0 { return Ok(()); } let seq = i64::try_from(seq) .map_err(|_| anyhow::anyhow!("projection sequence exceeds SQLite range"))?; sqlx::query!( "INSERT INTO projection_revisions (did, collection, rkey, seq) VALUES (?, ?, ?, ?) \ ON CONFLICT(did, collection, rkey) DO UPDATE SET seq = excluded.seq \ WHERE excluded.seq > projection_revisions.seq", did, collection, rkey, seq, ) .execute(&mut **tx) .await?; Ok(())}
/// Eagerly project a locally confirmed write without advancing the Hydrant/// cursor.////// PM-28 write handlers call this only after the user's PDS has accepted the/// record/blob operation. Hydrant/firehose will later deliver the same record/// event; sharing [`project_record`] keeps that convergence idempotent and avoids/// a subtly different SQL path for read-your-own-writes.pub async fn project_eager_record(db: &SqlitePool, event: &ProjectionEvent) -> anyhow::Result<()> { let mut tx = db.begin().await?; project_record(&mut tx, event).await?; record_pending(&mut tx, event).await?; tx.commit().await?; Ok(())}
/// Record pending read-your-writes state for an eagerly-projected op.////// Called only from the eager path (PM-28 mediated writes + the PM-47 proxy),/// in the same transaction as [`project_record`]. A create/update marks the/// record's local cid/value pending until the firehose re-delivers it; a delete/// marks it pending-deleted. Non-`space.polymodel.*` events are skipped (only/// polymodel is indexed, so only polymodel can carry pending state).async fn record_pending( tx: &mut Transaction<'_, Sqlite>, event: &ProjectionEvent,) -> anyhow::Result<()> { if !event.collection.starts_with("space.polymodel.") { return Ok(()); } let uri = at_uri(&event.did, &event.collection, &event.rkey); match event.action.as_str() { "create" | "update" => { let (Some(cid), Some(record)) = (event.cid.clone(), event.record.as_ref()) else { return Ok(()); }; 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 (?, ?, ?, ?) \ ON CONFLICT(uri) DO UPDATE SET cid = excluded.cid, value = excluded.value, \ written_at = excluded.written_at", uri, cid, value, now, ) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM pending_deletes WHERE uri = ?", uri) .execute(&mut **tx) .await?; } "delete" => { let now = now_nanos(); sqlx::query!( "INSERT INTO pending_deletes (uri, deleted_at) VALUES (?, ?) \ ON CONFLICT(uri) DO UPDATE SET deleted_at = excluded.deleted_at", uri, now, ) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM pending_writes WHERE uri = ?", uri) .execute(&mut **tx) .await?; } _ => {} } Ok(())}
/// Clear pending read-your-writes state once the Hydrant firehose has/// re-delivered the same event (the PDS/AppView is now authoritative again).////// Called from the firehose projection path ([`project_event`]) in the same/// transaction as [`project_record`], only for `space.polymodel.*`.async fn clear_pending( tx: &mut Transaction<'_, Sqlite>, event: &ProjectionEvent,) -> anyhow::Result<()> { if !event.collection.starts_with("space.polymodel.") { return Ok(()); } let uri = at_uri(&event.did, &event.collection, &event.rkey); match event.action.as_str() { "create" | "update" => { // Clear only if the firehose event's cid matches the pending write, // so a re-ordered (older) commit can't clear a newer pending write. if let Some(cid) = &event.cid { sqlx::query!( "DELETE FROM pending_writes WHERE uri = ? AND cid = ?", uri, cid ) .execute(&mut **tx) .await?; } } "delete" => { sqlx::query!("DELETE FROM pending_deletes WHERE uri = ?", uri) .execute(&mut **tx) .await?; } _ => {} } Ok(())}
/// Dispatch a record event to its collection-specific projection rule.async fn project_record( tx: &mut Transaction<'_, Sqlite>, event: &ProjectionEvent,) -> anyhow::Result<()> { let action = event.action.as_str(); let collection = event.collection.as_str(); match (action, collection) { ("create" | "update", "space.polymodel.library.thing") => { project_thing_upsert(tx, event).await?; rebuild_ordinary_ldraw_resources(tx).await?; } ("delete", "space.polymodel.library.thing") => { project_thing_delete(tx, event).await?; rebuild_ordinary_ldraw_resources(tx).await?; } ("create" | "update", "space.polymodel.library.model") => { project_model_upsert(tx, event).await?; rebuild_ordinary_ldraw_resources(tx).await?; } ("delete", "space.polymodel.library.model") => { project_model_delete(tx, event).await?; rebuild_ordinary_ldraw_resources(tx).await?; } ("create" | "update", "space.polymodel.library.part") => { project_part_upsert(tx, event).await?; rebuild_ordinary_ldraw_resources(tx).await?; } ("delete", "space.polymodel.library.part") => { project_part_delete(tx, event).await?; rebuild_ordinary_ldraw_resources(tx).await?; } ("create" | "update", "space.polymodel.actor.profile") => { project_profile_upsert(tx, event).await?; } ("delete", "space.polymodel.actor.profile") => { project_profile_delete(tx, event).await?; } ("create", "space.polymodel.graph.follow") => { project_follow_create(tx, event).await?; } ("delete", "space.polymodel.graph.follow") => { project_follow_delete(tx, event).await?; } ("create", "space.polymodel.graph.like") => { project_like_create(tx, event).await?; } ("delete", "space.polymodel.graph.like") => { project_like_delete(tx, event).await?; } ("create", "space.polymodel.graph.save") => { project_save_create(tx, event).await?; } ("delete", "space.polymodel.graph.save") => { project_save_delete(tx, event).await?; } ("create", "space.polymodel.graph.tag") => { project_tag_create(tx, event).await?; } ("delete", "space.polymodel.graph.tag") => { project_tag_delete(tx, event).await?; } ("create" | "update", "space.polymodel.graph.list") => { project_list_upsert(tx, event).await?; } ("delete", "space.polymodel.graph.list") => { project_list_delete(tx, event).await?; } ("create", "space.polymodel.graph.listitem") => { project_listitem_create(tx, event).await?; } ("delete", "space.polymodel.graph.listitem") => { project_listitem_delete(tx, event).await?; } // account events are dropped by ProjectionInput::from_hydrant. _ => { tracing::debug!(action, collection, "no projection rule for event"); } } Ok(())}
// ---------------------------------------------------------------------------// Typed record decode from borrowed Data.// ---------------------------------------------------------------------------
/// 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<T>(data: &Data, collection: &str) -> Result<T, DataDeserializerError>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 { chrono::Utc::now() .timestamp_nanos_opt() .expect("current timestamp must fit in i64 nanoseconds")}
fn at_uri(did: &str, collection: &str, rkey: &str) -> String { format!("at://{did}/{collection}/{rkey}")}
// ---------------------------------------------------------------------------// Library content// ---------------------------------------------------------------------------
async fn project_thing_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { invalidate_thing(tx, e).await?; return Ok(()); }; let rec = match decode::<Thing>(data, &e.collection) { Ok(r) => r, Err(err) => { tracing::warn!(error = %err, "thing record decode failed; invalidating projection"); invalidate_thing(tx, e).await?; return Ok(()); } }; let uri = at_uri(&e.did, &e.collection, &e.rkey); let Some(cid) = e.cid.clone() else { return Ok(()); }; 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::<Vec<_>>() .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::<Vec<_>>()) .unwrap_or_default(); let tags_json = if tags.is_empty() { None } else { serde_json::to_string(&tags).ok() }; let tags_text = if tags.is_empty() { None } else { Some(tags.join(" ")) };
let created_at = rec .created_at .as_ref() .timestamp_nanos_opt() .unwrap_or_else(now_nanos); let indexed_at = now_nanos(); 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 // off, the sqlx default) the AFTER DELETE trigger too — corrupting things_fts // on every update/replay. ON CONFLICT DO UPDATE preserves the rowid and fires // things_au so the external-content FTS index stays in sync. sqlx::query!( "INSERT INTO things \ (did, rkey, uri, cid, name, summary, license, tags_json, tags_text, \ instructions_text, cover_json, derived_from_uri, created_at, indexed_at, record_json) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \ ON CONFLICT(did, rkey) DO UPDATE SET \ uri = excluded.uri, cid = excluded.cid, name = excluded.name, \ summary = excluded.summary, license = excluded.license, \ tags_json = excluded.tags_json, tags_text = excluded.tags_text, \ instructions_text = excluded.instructions_text, cover_json = excluded.cover_json, \ derived_from_uri = excluded.derived_from_uri, created_at = excluded.created_at, \ indexed_at = excluded.indexed_at, record_json = excluded.record_json", e.did, e.rkey, uri, cid, name, summary, license, tags_json, tags_text, instructions, cover_json, derived_from_uri, created_at, indexed_at, record_json, ) .execute(&mut **tx) .await?;
// Rebuild thing_models junction preserving array order. sqlx::query!("DELETE FROM thing_models WHERE thing_uri = ?", uri) .execute(&mut **tx) .await?; 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_ref.uri.as_str(), pos as i64, ) .execute(&mut **tx) .await?; }
// Ensure a stats row exists so feeds can inner-join. sqlx::query!( "INSERT OR IGNORE INTO content_stats (uri, like_count, save_count, tag_count) \ VALUES (?, 0, 0, 0)", uri, ) .execute(&mut **tx) .await?;
Ok(())}
async fn invalidate_thing( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let uri = at_uri(&e.did, &e.collection, &e.rkey); sqlx::query!( "DELETE FROM things WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM thing_models WHERE thing_uri = ?", uri) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM content_stats WHERE uri = ?", uri) .execute(&mut **tx) .await?; Ok(())}
async fn project_thing_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let uri = at_uri(&e.did, &e.collection, &e.rkey); sqlx::query!( "DELETE FROM things WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM thing_models WHERE thing_uri = ?", uri) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM content_stats WHERE uri = ?", uri) .execute(&mut **tx) .await?; Ok(())}
async fn project_model_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { invalidate_model(tx, e).await?; return Ok(()); }; let rec = match decode::<Model>(data, &e.collection) { Ok(r) => r, Err(err) => { tracing::warn!(error = %err, "model record decode failed; invalidating projection"); invalidate_model(tx, e).await?; return Ok(()); } }; let uri = at_uri(&e.did, &e.collection, &e.rkey); let Some(cid) = e.cid.clone() else { return Ok(()); }; 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 = serde_json::to_string(data).unwrap_or_default();
sqlx::query!( "INSERT INTO models \ (did, rkey, uri, cid, name, summary, created_at, indexed_at, record_json) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) \ ON CONFLICT(did, rkey) DO UPDATE SET \ uri = excluded.uri, cid = excluded.cid, name = excluded.name, \ summary = excluded.summary, created_at = excluded.created_at, \ indexed_at = excluded.indexed_at, record_json = excluded.record_json", e.did, e.rkey, uri, cid, name, summary, created_at, indexed_at, record_json, ) .execute(&mut **tx) .await?;
sqlx::query!("DELETE FROM model_parts WHERE model_uri = ?", uri) .execute(&mut **tx) .await?; 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_ref.uri.as_str(), pos as i64, ) .execute(&mut **tx) .await?; } Ok(())}
async fn project_model_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let uri = at_uri(&e.did, &e.collection, &e.rkey); sqlx::query!( "DELETE FROM models WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM model_parts WHERE model_uri = ?", uri) .execute(&mut **tx) .await?; Ok(())}
async fn project_part_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { invalidate_part(tx, e).await?; return Ok(()); }; let rec = match decode::<Part>(data, &e.collection) { Ok(r) => r, Err(err) => { tracing::warn!(error = %err, "part record decode failed; invalidating projection"); invalidate_part(tx, e).await?; return Ok(()); } }; let uri = at_uri(&e.did, &e.collection, &e.rkey); let Some(cid) = e.cid.clone() else { invalidate_part(tx, e).await?; return Ok(()); }; 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 = serde_json::to_string(data).unwrap_or_default();
sqlx::query!( "INSERT INTO parts \ (did, rkey, uri, cid, name, format, file_json, created_at, indexed_at, record_json) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \ ON CONFLICT(did, rkey) DO UPDATE SET \ uri = excluded.uri, cid = excluded.cid, name = excluded.name, \ format = excluded.format, file_json = excluded.file_json, \ created_at = excluded.created_at, indexed_at = excluded.indexed_at, \ record_json = excluded.record_json", e.did, e.rkey, uri, cid, name, format, file_json, created_at, indexed_at, record_json, ) .execute(&mut **tx) .await?; Ok(())}
async fn invalidate_model( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let uri = at_uri(&e.did, &e.collection, &e.rkey); sqlx::query!( "DELETE FROM models WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; sqlx::query!("DELETE FROM model_parts WHERE model_uri = ?", uri) .execute(&mut **tx) .await?; Ok(())}
async fn invalidate_part( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { sqlx::query!( "DELETE FROM parts WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; Ok(())}
async fn project_part_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { invalidate_part(tx, e).await}
/// Rebuild the resolver rows derived from ordinary Thing → Model → Part data.////// LDraw resources are not authored records in the current architecture. They/// are a projection of an ordinary Part's complete File manifest and its live/// Thing graph. Rebuilding the derived subset after each graph event makes/// firehose ordering, deletes, updates, and eager-write replay converge without/// touching PM-76 rows supplied by other producers.async fn rebuild_ordinary_ldraw_resources(tx: &mut Transaction<'_, Sqlite>) -> anyhow::Result<()> { // Memberships are derived from the complete current Thing graph. Source rows // remain the byte identity, so a shared Part is parsed once and can have any // number of project/path memberships. sqlx::query!( "DELETE FROM ldraw_resource_memberships WHERE provenance = ?", DERIVED_LDRAW_PROVENANCE, ) .execute(&mut **tx) .await?;
let rows = sqlx::query!( "SELECT p.did, p.rkey, p.uri, p.cid, p.name, p.file_json, t.uri AS project_uri, \ t.license \ FROM parts p \ JOIN model_parts mp ON mp.part_uri = p.uri \ JOIN thing_models tm ON tm.model_uri = mp.model_uri \ JOIN things t ON t.uri = tm.thing_uri \ WHERE lower(COALESCE(p.format, '')) = 'ldraw' \ ORDER BY t.uri, tm.position, mp.position, p.uri" ) .fetch_all(&mut **tx) .await?;
let mut valid_sources = HashSet::new(); let mut seen_memberships = HashSet::new(); for row in rows { let source_uri = AtUri::new_owned(&row.uri)?; let project_uri = AtUri::new_owned(&row.project_uri)?; let Ok(path) = crate::ldraw::project_part_path(&project_uri, &source_uri, &row.name) else { tracing::warn!(part = %row.uri, project = %row.project_uri, "ordinary LDraw path derivation failed"); continue; }; if !seen_memberships.insert(( row.uri.clone(), row.project_uri.clone(), path.as_str().to_owned(), )) { continue; }
let source = match parse_ordinary_source(&row.file_json, &source_uri) { Ok(source) => source, Err(error) => { tracing::warn!(part = %row.uri, error = %error, "ordinary LDraw source invalid; stale bytes remain unreadable"); continue; } }; let (blob_cids, digest, byte_length, mime_type, source_identity) = source; let ordered_blob_cids = serde_json::to_string(&blob_cids)?; let now = now_nanos(); let existing_identity: Option<String> = sqlx::query_scalar!( "SELECT source_identity FROM ldraw_verification WHERE resource_uri = ?", &row.uri ) .fetch_optional(&mut **tx) .await?; let identity_changed = existing_identity.is_some() && existing_identity.as_deref() != Some(&source_identity);
sqlx::query!( "INSERT INTO ldraw_resources \ (resource_uri, owner_did, project_uri, record_cid, rkey, root, canonical_path, \ mime_type, byte_length, sha256, ordered_blob_cids, provenance, licence, notices_json, \ created_at, updated_at) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \ ON CONFLICT(resource_uri) DO UPDATE SET \ owner_did = excluded.owner_did, record_cid = excluded.record_cid, rkey = excluded.rkey, \ mime_type = excluded.mime_type, byte_length = excluded.byte_length, sha256 = excluded.sha256, \ ordered_blob_cids = excluded.ordered_blob_cids, provenance = excluded.provenance, \ licence = excluded.licence, notices_json = excluded.notices_json, updated_at = excluded.updated_at \ WHERE ldraw_resources.provenance = excluded.provenance", row.uri, row.did, row.project_uri, row.cid, row.rkey, path.segments().next().unwrap_or("models"), path.as_str(), mime_type, byte_length, digest, ordered_blob_cids, DERIVED_LDRAW_PROVENANCE, row.license, "[]", now, now, ) .execute(&mut **tx) .await?;
sqlx::query!( "INSERT INTO ldraw_resource_memberships \ (resource_uri, project_uri, root, canonical_path, provenance) \ VALUES (?, ?, ?, ?, ?)", row.uri, row.project_uri, path.segments().next().unwrap_or("models"), path.as_str(), DERIVED_LDRAW_PROVENANCE, ) .execute(&mut **tx) .await?; valid_sources.insert(row.uri.clone());
if identity_changed { sqlx::query!( "INSERT INTO ldraw_verification \ (resource_uri, source_identity, state, diagnostic, updated_at) \ VALUES (?, ?, 'unverified', 'source-changed', ?) \ ON CONFLICT(resource_uri) DO UPDATE SET source_identity = excluded.source_identity, \ state = excluded.state, diagnostic = excluded.diagnostic, updated_at = excluded.updated_at", source_uri.as_ref(), source_identity, now, ) .execute(&mut **tx) .await?; } else if existing_identity.is_none() { sqlx::query!( "INSERT INTO ldraw_verification \ (resource_uri, source_identity, state, diagnostic, updated_at) \ VALUES (?, ?, 'verified', NULL, ?)", source_uri.as_ref(), source_identity, now, ) .execute(&mut **tx) .await?; } }
// Remove only ordinary sources no longer represented by a valid current // graph. PM-76-owned rows and their verification state are untouched. let stale = sqlx::query!( "SELECT resource_uri AS \"resource_uri!\" FROM ldraw_resources WHERE provenance = ?", DERIVED_LDRAW_PROVENANCE, ) .fetch_all(&mut **tx) .await?; for row in stale { if valid_sources.contains(&row.resource_uri) { continue; } sqlx::query!( "DELETE FROM ldraw_manifest_files WHERE resource_uri = ? OR target_resource_uri = ?", row.resource_uri, row.resource_uri, ) .execute(&mut **tx) .await?; sqlx::query!( "DELETE FROM ldraw_verification WHERE resource_uri = ?", row.resource_uri ) .execute(&mut **tx) .await?; sqlx::query!( "DELETE FROM ldraw_resources WHERE resource_uri = ?", row.resource_uri ) .execute(&mut **tx) .await?; } Ok(())}
fn parse_ordinary_source(file_json: &str, resource_uri: &AtUri) -> anyhow::Result<OrdinarySource> { let file: File = serde_json::from_str(file_json)?; let digest = file .digest .as_ref() .filter(|digest| digest.len() == 32) .ok_or_else(|| anyhow::anyhow!("digest must contain 32 bytes"))?; let mut chunks = file.chunks.iter().collect::<Vec<_>>(); chunks.sort_by_key(|chunk| chunk.offset); let mut next_offset = 0_i64; let mut blob_cids = Vec::with_capacity(chunks.len()); for chunk in chunks { if chunk.offset < 0 || chunk.size < 0 || chunk.offset != next_offset || chunk.blob.blob().size as i64 != chunk.size { return Err(anyhow::anyhow!("file chunks are not contiguous")); } next_offset = next_offset .checked_add(chunk.size) .ok_or_else(|| anyhow::anyhow!("file size overflows"))?; blob_cids.push(chunk.blob.blob().cid().as_str().to_owned()); } if blob_cids.is_empty() || next_offset != file.size || file.size < 0 { return Err(anyhow::anyhow!("file size does not match chunks")); } let digest = digest.to_vec(); let source_identity = crate::ldraw::source_identity_for(resource_uri, &blob_cids, &digest, file.size); Ok(( blob_cids, digest, file.size, file.mime_type.to_string(), source_identity, ))}
// ---------------------------------------------------------------------------// Actor profile + identity projections// ---------------------------------------------------------------------------
/// Upsert a `space.polymodel.actor.profile` record (singleton, rkey = "self")./// Denormalizes the profile fields for the polymodel profileView union variant/// and stores the raw record for the required `record` field. Handle is not/// stored here (it is identity-level, not part of the record); it lives in/// `identities`, refreshed by identity events.async fn project_profile_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { return Ok(()); }; let rec = match decode::<Profile>(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!( "INSERT INTO profiles \ (did, display_name, description, avatar_json, default_license, pronouns, \ printers_json, links_json, record_json, indexed_at) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \ ON CONFLICT(did) DO UPDATE SET \ display_name = excluded.display_name, description = excluded.description, \ avatar_json = excluded.avatar_json, default_license = excluded.default_license, \ pronouns = excluded.pronouns, printers_json = excluded.printers_json, \ links_json = excluded.links_json, record_json = excluded.record_json, \ indexed_at = excluded.indexed_at", e.did, display_name, description, avatar_json, default_license, pronouns, printers_json, links_json, record_json, indexed_at, ) .execute(&mut **tx) .await?; Ok(())}
/// Delete a polymodel profile row (profile record removed / repo tombstoned).async fn project_profile_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { sqlx::query!("DELETE FROM profiles WHERE did = ?", e.did) .execute(&mut **tx) .await?; Ok(())}
/// Refresh a DID→handle mapping from an identity event. A `None` handle means the/// DID has dropped its handle; it is stored as-is so reads reflect current state.async fn project_identity_upsert( tx: &mut Transaction<'_, Sqlite>, e: &IdentityProjection,) -> anyhow::Result<()> { let updated_at = now_nanos(); sqlx::query!( "INSERT INTO identities (did, handle, updated_at) VALUES (?, ?, ?) \ ON CONFLICT(did) DO UPDATE SET handle = excluded.handle, \ updated_at = excluded.updated_at", e.did, e.handle, updated_at, ) .execute(&mut **tx) .await?; Ok(())}
// ---------------------------------------------------------------------------// Social records (two-step delete + conditional count maintenance)// ---------------------------------------------------------------------------
#[derive(sqlx::FromRow)]struct SubjectRow { subject_uri: String,}
async fn project_like_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { return Ok(()); }; let rec = match decode::<Like>(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 = 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 (?, ?, ?, ?, ?)", e.did, e.rkey, cid, subject, created_at, ) .execute(&mut **tx) .await?;
if res.rows_affected() == 1 { sqlx::query!( "INSERT INTO content_stats (uri, like_count, save_count, tag_count) VALUES (?, 1, 0, 0) \ ON CONFLICT(uri) DO UPDATE SET like_count = like_count + 1", subject, ) .execute(&mut **tx) .await?; } Ok(())}
async fn project_follow_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { return Ok(()); }; let rec = match decode::<Follow>(data, &e.collection) { Ok(r) => r, Err(err) => { tracing::warn!(error = %err, "follow record decode failed"); return Ok(()); } }; let Some(cid) = e.cid.clone() else { return Ok(()); }; let followed = rec.subject.as_str().to_string(); let created_at = rec .created_at .as_ref() .timestamp_nanos_opt() .unwrap_or_else(now_nanos);
sqlx::query!( "INSERT OR IGNORE INTO follows (follower_did, followed_did, rkey, cid, created_at) \ VALUES (?, ?, ?, ?, ?)", e.did, followed, e.rkey, cid, created_at, ) .execute(&mut **tx) .await?; Ok(())}
async fn project_follow_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { sqlx::query!( "DELETE FROM follows WHERE follower_did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; Ok(())}
async fn project_like_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let old: Option<SubjectRow> = sqlx::query_as!( SubjectRow, "SELECT subject_uri FROM likes WHERE did = ? AND rkey = ?", e.did, e.rkey ) .fetch_optional(&mut **tx) .await?;
let res = sqlx::query!( "DELETE FROM likes WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?;
if res.rows_affected() == 1 && let Some(row) = old { sqlx::query!( "UPDATE content_stats SET like_count = MAX(0, like_count - 1) WHERE uri = ?", row.subject_uri, ) .execute(&mut **tx) .await?; } Ok(())}
async fn project_save_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { return Ok(()); }; let rec = match decode::<Save>(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 = 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) \ VALUES (?, ?, ?, ?, ?, ?)", e.did, e.rkey, cid, subject, note, created_at, ) .execute(&mut **tx) .await?;
if res.rows_affected() == 1 { sqlx::query!( "INSERT INTO content_stats (uri, like_count, save_count, tag_count) VALUES (?, 0, 1, 0) \ ON CONFLICT(uri) DO UPDATE SET save_count = save_count + 1", subject, ) .execute(&mut **tx) .await?; } Ok(())}
async fn project_save_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let old: Option<SubjectRow> = sqlx::query_as!( SubjectRow, "SELECT subject_uri FROM saves WHERE did = ? AND rkey = ?", e.did, e.rkey ) .fetch_optional(&mut **tx) .await?;
let res = sqlx::query!( "DELETE FROM saves WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?;
if res.rows_affected() == 1 && let Some(row) = old { sqlx::query!( "UPDATE content_stats SET save_count = MAX(0, save_count - 1) WHERE uri = ?", row.subject_uri, ) .execute(&mut **tx) .await?; } Ok(())}
async fn project_tag_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { return Ok(()); }; let rec = match decode::<Tag>(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 = 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) \ VALUES (?, ?, ?, ?, ?, ?)", e.did, e.rkey, cid, subject, tag, created_at, ) .execute(&mut **tx) .await?;
if res.rows_affected() == 1 { sqlx::query!( "INSERT INTO content_stats (uri, like_count, save_count, tag_count) VALUES (?, 0, 0, 1) \ ON CONFLICT(uri) DO UPDATE SET tag_count = tag_count + 1", subject, ) .execute(&mut **tx) .await?; } Ok(())}
async fn project_tag_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let old: Option<SubjectRow> = sqlx::query_as!( SubjectRow, "SELECT subject_uri FROM tags WHERE did = ? AND rkey = ?", e.did, e.rkey ) .fetch_optional(&mut **tx) .await?;
let res = sqlx::query!("DELETE FROM tags WHERE did = ? AND rkey = ?", e.did, e.rkey) .execute(&mut **tx) .await?;
if res.rows_affected() == 1 && let Some(row) = old { sqlx::query!( "UPDATE content_stats SET tag_count = MAX(0, tag_count - 1) WHERE uri = ?", row.subject_uri, ) .execute(&mut **tx) .await?; } Ok(())}
async fn project_listitem_create( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { return Ok(()); }; let rec = match decode::<Listitem>(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 = 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!( "INSERT OR IGNORE INTO listitems (did, rkey, cid, list_uri, subject_uri, created_at) \ VALUES (?, ?, ?, ?, ?, ?)", e.did, e.rkey, cid, list_uri, subject_uri, created_at, ) .execute(&mut **tx) .await?; Ok(())}
async fn project_list_upsert( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { let Some(data) = &e.record else { return Ok(()); }; let rec = match decode::<List>(data, &e.collection) { Ok(r) => r, Err(err) => { tracing::warn!(error = %err, "list record decode failed"); return Ok(()); } }; let Some(cid) = e.cid.clone() else { return Ok(()); }; let uri = format!("at://{}/{}/{}", e.did, e.collection, e.rkey); let name = rec.name.to_string(); let description = rec.description.as_ref().map(|s| s.to_string()); let purpose = rec.purpose.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 = serde_json::to_string(data)?;
sqlx::query!( "INSERT INTO lists (did, rkey, uri, cid, name, description, purpose, created_at, indexed_at, record_json) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \ ON CONFLICT(did, rkey) DO UPDATE SET \ uri = excluded.uri, cid = excluded.cid, name = excluded.name, \ description = excluded.description, purpose = excluded.purpose, \ created_at = excluded.created_at, indexed_at = excluded.indexed_at, \ record_json = excluded.record_json", e.did, e.rkey, uri, cid, name, description, purpose, created_at, indexed_at, record_json, ) .execute(&mut **tx) .await?; Ok(())}
async fn project_list_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { sqlx::query!( "DELETE FROM lists WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; Ok(())}
async fn project_listitem_delete( tx: &mut Transaction<'_, Sqlite>, e: &ProjectionEvent,) -> anyhow::Result<()> { sqlx::query!( "DELETE FROM listitems WHERE did = ? AND rkey = ?", e.did, e.rkey ) .execute(&mut **tx) .await?; Ok(())}
#[cfg(test)]mod tests { //! These tests exercise the projection against a real (in-memory) SQLite //! database with migrations applied. Production SQL is compile-time checked //! via `query!`; these assertions use runtime queries so the committed //! `.sqlx` cache stays focused on production SQL.
use super::*; use serde_json::json; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use std::str::FromStr;
/// A fresh in-memory SQLite pool with migrations applied. /// /// `max_connections(1)` so every query hits the same `:memory:` database. async fn test_db() -> SqlitePool { let options = SqliteConnectOptions::from_str("sqlite::memory:").expect("parse in-memory url"); let pool = SqlitePoolOptions::new() .max_connections(1) .connect_with(options) .await .expect("connect in-memory pool"); sqlx::migrate!("./migrations") .run(&pool) .await .expect("apply migrations"); pool }
fn thing_event( seq: u64, did: &str, rkey: &str, name: &str, models: &[&str], ) -> ProjectionEvent { ProjectionEvent { seq, did: did.to_string(), collection: "space.polymodel.library.thing".to_string(), rkey: rkey.to_string(), action: "create".to_string(), 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": format!("at://{did}/space.polymodel.library.model/{m}"), "cid": "c"})).collect::<Vec<_>>(), })).unwrap()), cid: Some("thingcid".to_string()), } }
fn like_event(seq: u64, did: &str, rkey: &str, subject: &str) -> ProjectionEvent { ProjectionEvent { seq, did: did.to_string(), collection: "space.polymodel.graph.like".to_string(), rkey: rkey.to_string(), action: "create".to_string(), record: Some( to_data(&json!({ "subject": {"uri": subject, "cid": "c"}, "createdAt": "2024-02-01T00:00:00.000Z", })) .unwrap(), ), cid: Some("likecid".to_string()), } }
fn delete_event(seq: u64, did: &str, collection: &str, rkey: &str) -> ProjectionEvent { ProjectionEvent { seq, did: did.to_string(), collection: collection.to_string(), rkey: rkey.to_string(), action: "delete".to_string(), record: None, cid: None, } }
#[derive(sqlx::FromRow)] struct CountRow { count: i64, }
#[derive(sqlx::FromRow)] struct OptValue { value: Option<String>, }
async fn stat(pool: &SqlitePool, uri: &str, column: &str) -> i64 { // `column` is a hardcoded test literal (like_count / save_count), never // user input, so AssertSqlSafe is appropriate here. Production queries // use static `query!` literals. let sql = format!("SELECT COALESCE({column}, 0) AS count FROM content_stats WHERE uri = ?"); sqlx::query_as::<_, CountRow>(sqlx::AssertSqlSafe(sql.as_str())) .bind(uri) .fetch_optional(pool) .await .expect("stat query") .map(|r| r.count) .unwrap_or(0) }
async fn likes_table_count(pool: &SqlitePool, subject: &str) -> i64 { sqlx::query_as::<_, CountRow>("SELECT COUNT(*) AS count FROM likes WHERE subject_uri = ?") .bind(subject) .fetch_one(pool) .await .expect("likes count") .count }
async fn junction_count(pool: &SqlitePool, thing_uri: &str) -> i64 { sqlx::query_as::<_, CountRow>( "SELECT COUNT(*) AS count FROM thing_models WHERE thing_uri = ?", ) .bind(thing_uri) .fetch_one(pool) .await .expect("junction count") .count }
async fn fts_uris(pool: &SqlitePool, term: &str) -> Vec<String> { #[derive(sqlx::FromRow)] struct UriRow { uri: String, } sqlx::query_as::<_, UriRow>( "SELECT t.uri FROM things_fts \ JOIN things t ON t.rowid = things_fts.rowid \ WHERE things_fts MATCH ?", ) .bind(term) .fetch_all(pool) .await .expect("fts query") .into_iter() .map(|r| r.uri) .collect() }
async fn cursor(pool: &SqlitePool) -> Option<u64> { sqlx::query_as::<_, OptValue>("SELECT value FROM projection_state WHERE key = 'cursor'") .fetch_one(pool) .await .expect("cursor query") .value .and_then(|v| v.parse().ok()) }
#[tokio::test] async fn thing_create_projects_row_junction_stats_and_fts() { 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", &["m1", "m2"]); project_event(&db, &ProjectionInput::Record(e)) .await .expect("project thing");
assert_eq!(junction_count(&db, uri).await, 2); assert_eq!(stat(&db, uri, "like_count").await, 0); // stats row exists, zeroed assert!(fts_uris(&db, "cube").await.contains(&uri.to_string())); assert!( fts_uris(&db, "calibration") .await .contains(&uri.to_string()) ); assert_eq!(cursor(&db).await, Some(10)); }
#[tokio::test] async fn thing_update_rebuilds_junction() { let db = test_db().await; let did = "did:plc:abc"; let uri = "at://did:plc:abc/space.polymodel.library.thing/r1";
// Create with two models. project_event( &db, &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", &["m3"]); upd.action = "update".to_string(); project_event(&db, &ProjectionInput::Record(upd)) .await .unwrap();
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://did:plc:abc/space.polymodel.library.model/m3'", ) .bind(uri) .fetch_one(&db) .await .unwrap() .count; assert_eq!(has_m3, 1); }
#[tokio::test] async fn thing_delete_cascades() { let db = test_db().await; let did = "did:plc:abc"; let uri = "at://did:plc:abc/space.polymodel.library.thing/r1"; project_event( &db, &ProjectionInput::Record(thing_event(1, did, "r1", "Cube", &["m1"])), ) .await .unwrap(); assert!(!fts_uris(&db, "cube").await.is_empty());
project_event( &db, &ProjectionInput::Record(delete_event(2, did, "space.polymodel.library.thing", "r1")), ) .await .unwrap();
assert_eq!(junction_count(&db, uri).await, 0); assert_eq!(stat(&db, uri, "like_count").await, 0); // stats row removed assert!(fts_uris(&db, "cube").await.is_empty()); // FTS delete trigger fired }
#[tokio::test] async fn like_create_increments_count() { let db = test_db().await; let subject = "at://did:plc:author/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(like_event(1, "did:plc:a", "k1", subject)), ) .await .unwrap(); project_event( &db, &ProjectionInput::Record(like_event(2, "did:plc:b", "k2", subject)), ) .await .unwrap();
assert_eq!(likes_table_count(&db, subject).await, 2); assert_eq!(stat(&db, subject, "like_count").await, 2); }
#[tokio::test] async fn like_delete_decrements_count_two_step() { let db = test_db().await; let subject = "at://did:plc:author/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(like_event(1, "did:plc:a", "k1", subject)), ) .await .unwrap(); project_event( &db, &ProjectionInput::Record(delete_event( 2, "did:plc:a", "space.polymodel.graph.like", "k1", )), ) .await .unwrap();
assert_eq!(likes_table_count(&db, subject).await, 0); assert_eq!(stat(&db, subject, "like_count").await, 0); }
#[tokio::test] 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/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(delete_event( 1, "did:plc:a", "space.polymodel.graph.like", "nope", )), ) .await .expect("delete absent like must succeed"); assert_eq!(stat(&db, subject, "like_count").await, 0); }
#[tokio::test] 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/space.polymodel.library.thing/t1"; project_event( &db, &ProjectionInput::Record(like_event(1, "did:plc:a", "k1", subject)), ) .await .unwrap(); project_event( &db, &ProjectionInput::Record(like_event(2, "did:plc:a", "k1", subject)), ) .await .unwrap();
assert_eq!(likes_table_count(&db, subject).await, 1); assert_eq!(stat(&db, subject, "like_count").await, 1); }
#[tokio::test] async fn cursor_advances_with_each_event() { let db = test_db().await; project_event( &db, &ProjectionInput::Record(thing_event(7, "did:plc:a", "r1", "Cube", &[])), ) .await .unwrap(); assert_eq!(cursor(&db).await, Some(7)); project_event( &db, &ProjectionInput::Record(like_event( 99, "did:plc:b", "k1", "at://did:plc:author/space.polymodel.library.thing/x", )), ) .await .unwrap(); assert_eq!(cursor(&db).await, Some(99)); }
#[tokio::test] async fn save_create_and_delete_maintain_count() { let db = test_db().await; 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( 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)) .await .unwrap(); assert_eq!(stat(&db, subject, "save_count").await, 1);
project_event( &db, &ProjectionInput::Record(delete_event( 2, "did:plc:a", "space.polymodel.graph.save", "s1", )), ) .await .unwrap(); assert_eq!(stat(&db, subject, "save_count").await, 0); }
#[tokio::test] async fn thing_update_keeps_fts_consistent() { // Regression for UPSERT vs INSERT OR REPLACE: updating a thing must fire // the AFTER UPDATE FTS trigger so the old term is removed and the new one // added, with no orphaned FTS row. INSERT OR REPLACE leaked one orphan per // update because REPLACE is DELETE+new-rowid (UPDATE trigger never fires // and, with recursive_triggers off, the DELETE trigger doesn't either). let db = test_db().await; let did = "did:plc:abc"; let uri = "at://did:plc:abc/space.polymodel.library.thing/r1";
let create = ProjectionEvent { seq: 1, did: did.to_string(), collection: "space.polymodel.library.thing".to_string(), rkey: "r1".to_string(), action: "create".to_string(), 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())) .await .unwrap(); assert_eq!(fts_uris(&db, "cube").await, vec![uri.to_string()]);
let mut update = create.clone(); update.seq = 2; update.action = "update".to_string(); update.cid = Some("cid2".to_string()); 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();
// Old term must be gone (no orphan), new term present, single live row. assert!( fts_uris(&db, "cube").await.is_empty(), "stale FTS entry for the old term leaked across update" ); assert_eq!(fts_uris(&db, "sphere").await, vec![uri.to_string()]); let live = sqlx::query_as::<_, CountRow>( "SELECT COUNT(*) AS count FROM things WHERE did = ? AND rkey = ?", ) .bind(did) .bind("r1") .fetch_one(&db) .await .unwrap() .count; assert_eq!(live, 1); }
#[tokio::test] async fn ordinary_ldraw_projection_rebuilds_resolver_on_replay_update_and_delete() { use crate::ldraw::resolve::SqliteResolver; use crate::ldraw::{CanonicalPath, LdrawResourceResolver, project_part_path}; use base64::Engine as _; use sha2::{Digest, Sha256};
let db = test_db().await; let did = "did:plc:ordinary"; let thing_uri = format!("at://{did}/space.polymodel.library.thing/thing"); let model_uri = format!("at://{did}/space.polymodel.library.model/model"); let part_uri = format!("at://{did}/space.polymodel.library.part/part"); let bytes = b"0 FILE main.ldr\\n"; let digest = Sha256::digest(bytes); let cid = "bafkreihcysctvh6hxc4gcx2tuspbmx5lv2vmejmm4uzjgdykbbhgjpop34"; let part_record = |name: &str| { to_data(&json!({ "name": name, "format": "LDraw", "file": { "mimeType": "application/x-ldraw", "size": bytes.len(), "digest": { "$bytes": base64::engine::general_purpose::STANDARD.encode(digest) }, "chunks": [{ "blob": { "$type": "blob", "ref": { "$link": cid }, "mimeType": "application/x-ldraw", "size": bytes.len() }, "offset": 0, "size": bytes.len() }] }, "createdAt": "2024-01-01T00:00:00.000Z" })) .unwrap() }; let event = |seq, collection: &str, rkey: &str, action: &str, record, cid: Option<&str>| { ProjectionEvent { seq, did: did.to_owned(), collection: collection.to_owned(), rkey: rkey.to_owned(), action: action.to_owned(), record, cid: cid.map(str::to_owned), } };
project_event( &db, &ProjectionInput::Record(event( 1, "space.polymodel.library.thing", "thing", "create", Some( to_data(&json!({ "name": "Ordinary LDraw Thing", "license": "MIT", "createdAt": "2024-01-01T00:00:00.000Z", "models": [{ "uri": model_uri, "cid": "bafyreicid" }] })) .unwrap(), ), Some("bafyrething"), )), ) .await .unwrap(); project_event( &db, &ProjectionInput::Record(event( 2, "space.polymodel.library.model", "model", "create", Some( to_data(&json!({ "name": "Main model", "createdAt": "2024-01-01T00:00:00.000Z", "parts": [{ "uri": part_uri, "cid": "bafyreicid" }] })) .unwrap(), ), Some("bafyremodel"), )), ) .await .unwrap(); let create_part = event( 3, "space.polymodel.library.part", "part", "create", Some(part_record("main.ldr")), Some("bafyrepart"), ); project_event(&db, &ProjectionInput::Record(create_part.clone())) .await .unwrap();
let resource_uri: String = sqlx::query_scalar( "SELECT resource_uri FROM ldraw_resources WHERE provenance = 'ordinary-part'", ) .fetch_one(&db) .await .unwrap(); assert_eq!(resource_uri, part_uri); let resolver = SqliteResolver::new(db.clone()); let project = AtUri::new_owned(&thing_uri).unwrap(); let part = AtUri::new_owned(&part_uri).unwrap(); let path = CanonicalPath::parse("models/main.ldr").unwrap(); let resolved = resolver.resolve(&project, &path).await.unwrap(); assert_eq!(resolved.source.resource_uri.as_ref(), part_uri); assert_eq!(resolved.target.blob_cids, vec![cid.to_owned()]);
// Two distinct Parts may expose the same project/root/path. The source // rows must both commit, and the resolver must report an ambiguity rather // than poisoning the transaction or silently choosing one source. let duplicate_part_uri = format!("at://{did}/space.polymodel.library.part/duplicate"); project_event( &db, &ProjectionInput::Record(event( 4, "space.polymodel.library.part", "duplicate", "create", Some(part_record("main.ldr")), Some("bafyreduplicate"), )), ) .await .unwrap(); project_event( &db, &ProjectionInput::Record(event( 5, "space.polymodel.library.part", "duplicate", "update", Some(part_record("main.ldr")), Some("bafyreduplicateupdated"), )), ) .await .unwrap(); project_event( &db, &ProjectionInput::Record(event( 6, "space.polymodel.library.model", "model", "update", Some( to_data(&json!({ "name": "Main model", "createdAt": "2024-01-01T00:00:00.000Z", "parts": [ { "uri": part_uri, "cid": "bafyreipart" }, { "uri": duplicate_part_uri, "cid": "bafyreduplicateupdated" } ] })) .unwrap(), ), Some("bafyremodelupdated"), )), ) .await .unwrap(); let duplicate_count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM ldraw_resources WHERE provenance = 'ordinary-part'", ) .fetch_one(&db) .await .unwrap(); assert_eq!(duplicate_count, 2); assert!(matches!( resolver.resolve(&project, &path).await, Err(crate::ldraw::ResolveError::Collision { .. }) ));
// A second Thing can share the same Part source. The source row remains // singular while each project gets its own visible membership/path. let second_did = "did:plc:other"; let second_thing_uri = format!("at://{second_did}/space.polymodel.library.thing/thing2"); project_event( &db, &ProjectionInput::Record(ProjectionEvent { seq: 4, did: second_did.to_owned(), collection: "space.polymodel.library.thing".to_owned(), rkey: "thing2".to_owned(), action: "create".to_owned(), record: Some( to_data(&json!({ "name": "Shared LDraw Thing", "license": "MIT", "createdAt": "2024-01-01T00:00:00.000Z", "models": [{ "uri": model_uri, "cid": "bafyreicid" }] })) .unwrap(), ), cid: Some("bafyrething2".to_owned()), }), ) .await .unwrap(); assert_eq!( sqlx::query_scalar::<_, i64>( "SELECT COUNT(*) FROM ldraw_resource_memberships WHERE provenance = 'ordinary-part'", ) .fetch_one(&db) .await .unwrap(), 4 ); assert_eq!( sqlx::query_scalar::<_, i64>( "SELECT COUNT(*) FROM ldraw_resource_memberships WHERE provenance = 'ordinary-part' AND project_uri = ?", ) .bind(&thing_uri) .fetch_one(&db) .await .unwrap(), 2 ); let second_project = AtUri::new_owned(&second_thing_uri).unwrap(); let external_path = project_part_path(&second_project, &part, "ignored.ldr").unwrap(); assert!(external_path.as_str().starts_with("external/")); assert!( resolver .resolve(&second_project, &external_path) .await .is_ok() ); let source_metadata: (String, String, String) = sqlx::query_as( "SELECT project_uri, root, canonical_path FROM ldraw_resources WHERE resource_uri = ?", ) .bind(&part_uri) .fetch_one(&db) .await .unwrap(); assert_eq!(source_metadata.0, thing_uri); assert_eq!(source_metadata.1, "models"); assert_eq!(source_metadata.2, "models/main.ldr");
// Replaying the same event is idempotent; changing the ordinary record // replaces the derived path and source metadata rather than accumulating rows. project_event(&db, &ProjectionInput::Record(create_part.clone())) .await .unwrap(); let mut update = create_part.clone(); update.seq = 7; update.action = "update".to_owned(); update.cid = Some("bafyrepartupdated".to_owned()); update.record = Some(part_record("updated.ldr")); project_event(&db, &ProjectionInput::Record(update)) .await .unwrap(); assert_eq!( sqlx::query_scalar::<_, i64>( "SELECT COUNT(*) FROM ldraw_resources WHERE provenance = 'ordinary-part'", ) .fetch_one(&db) .await .unwrap(), 2 ); assert!(resolver.resolve(&project, &path).await.is_ok()); let updated_path = CanonicalPath::parse("models/updated.ldr").unwrap(); assert!(resolver.resolve(&project, &updated_path).await.is_ok()); assert!( resolver .resolve(&second_project, &external_path) .await .is_ok() );
// A stale replay cannot restore the old source identity or path. let mut stale_replay = create_part.clone(); stale_replay.seq = 3; project_event(&db, &ProjectionInput::Record(stale_replay)) .await .unwrap(); assert!(resolver.resolve(&project, &updated_path).await.is_ok());
// A malformed newer update invalidates the source and all memberships, // so previously verified bytes cannot remain readable. project_event( &db, &ProjectionInput::Record(event( 8, "space.polymodel.library.part", "part", "update", Some(to_data(&json!({ "name": "malformed" })).unwrap()), Some("bafyrepartmalformed"), )), ) .await .unwrap(); assert!(resolver.resolve(&project, &updated_path).await.is_err()); assert!( resolver .resolve(&second_project, &external_path) .await .is_err() ); assert_eq!( sqlx::query_scalar::<_, i64>( "SELECT COUNT(*) FROM ldraw_resource_memberships WHERE provenance = 'ordinary-part'", ) .fetch_one(&db) .await .unwrap(), 2 );
project_event( &db, &ProjectionInput::Record(delete_event(9, did, "space.polymodel.library.part", "part")), ) .await .unwrap(); project_event( &db, &ProjectionInput::Record(delete_event( 10, did, "space.polymodel.library.part", "duplicate", )), ) .await .unwrap(); assert_eq!( sqlx::query_scalar::<_, i64>( "SELECT COUNT(*) FROM ldraw_resources WHERE provenance = 'ordinary-part'", ) .fetch_one(&db) .await .unwrap(), 0 ); }
#[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); }
#[tokio::test] async fn list_create_update_delete_projects_metadata() { let db = test_db().await; let did = "did:plc:listmaker"; let create = ProjectionEvent { seq: 20, did: did.to_string(), collection: "space.polymodel.graph.list".to_string(), rkey: "l1".to_string(), action: "create".to_string(), record: Some( to_data(&json!({ "name": "Fixture jigs", "description": "Useful shop fixtures", "purpose": "curation", "createdAt": "2024-01-01T00:00:00.000Z" })) .unwrap(), ), cid: Some("bafyreilistone".to_string()), };
project_event(&db, &ProjectionInput::Record(create)) .await .unwrap(); let uri = format!("at://{did}/space.polymodel.graph.list/l1"); let row: (String, String, Option<String>) = sqlx::query_as("SELECT uri, name, purpose FROM lists WHERE did = ? AND rkey = ?") .bind(did) .bind("l1") .fetch_one(&db) .await .unwrap(); assert_eq!(row.0, uri); assert_eq!(row.1, "Fixture jigs"); assert_eq!(row.2.as_deref(), Some("curation"));
let update = ProjectionEvent { seq: 21, did: did.to_string(), collection: "space.polymodel.graph.list".to_string(), rkey: "l1".to_string(), action: "update".to_string(), record: Some( to_data(&json!({ "name": "Workbench jigs", "createdAt": "2024-01-02T00:00:00.000Z" })) .unwrap(), ), cid: Some("bafyreilisttwo".to_string()), }; project_event(&db, &ProjectionInput::Record(update)) .await .unwrap(); let name: String = sqlx::query_scalar("SELECT name FROM lists WHERE did = ? AND rkey = ?") .bind(did) .bind("l1") .fetch_one(&db) .await .unwrap(); assert_eq!(name, "Workbench jigs");
let delete = ProjectionEvent { seq: 22, did: did.to_string(), collection: "space.polymodel.graph.list".to_string(), rkey: "l1".to_string(), action: "delete".to_string(), record: None, cid: None, }; project_event(&db, &ProjectionInput::Record(delete)) .await .unwrap(); let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM lists WHERE did = ?") .bind(did) .fetch_one(&db) .await .unwrap(); assert_eq!(count, 0); }}