From 56c586ded3f8aba512e7023852f7025947c3f3b4 Mon Sep 17 00:00:00 2001 From: Orual Date: Tue, 4 Aug 2026 13:32:08 -0400 Subject: [PATCH] PM-80: harden projection and renderer recovery --- ...a7ff671ec85cfdf99ce2d2954d9058f1fc2f3.json | 12 + ...6df3ebea3419eea4fa549aece6dee86951b0.json} | 8 +- ...5dd6e3a5e577fab77fca888fa6561d3b4f379.json | 12 + ...5472d2fbd59a08bf4da00e6fb1f93770ca78d.json | 12 - ...de13d80ff20abdeb35ae2b030e1f79fd81af2.json | 12 + ...16b14c92bdb38e2ab74ab4c47cfccd86ed083.json | 12 + ...9a121241a97dde9c833a63a5019a57fee1ad9.json | 12 + ...b2f17fdeb5062c6508d94c503e15b0cb3f7c6.json | 12 - ...454745d09f4a14cede190566f4c6019de3d95.json | 12 + ...8424b23dc0f5f203dd7a9a0bf57acf17dff87.json | 26 ++ ...ac74add7908d5e84e06ff09f9e46a065ec44a.json | 12 - .../011_pm80_projection_memberships.sql | 75 ++++++ src/indexing/db.rs | 3 +- src/indexing/projection.rs | 226 +++++++++++++++--- src/ldraw/mod.rs | 2 +- src/ldraw/paths.rs | 185 ++++++++++++++ src/ldraw/resolve.rs | 7 +- 17 files changed, 564 insertions(+), 76 deletions(-) create mode 100644 .sqlx/query-0501daa78b69311ce9930970042a7ff671ec85cfdf99ce2d2954d9058f1fc2f3.json rename .sqlx/{query-d87395649e34a00258562514babc61492a5433ad04a26e0f7d6b4ac843dd0dcc.json => query-0572fe522555a3399aa7d0b6c3416df3ebea3419eea4fa549aece6dee86951b0.json} (81%) create mode 100644 .sqlx/query-1c9d5bf76f91eb1a6486cb4b94b5dd6e3a5e577fab77fca888fa6561d3b4f379.json delete mode 100644 .sqlx/query-267ae501eae19fca4681bec219a5472d2fbd59a08bf4da00e6fb1f93770ca78d.json create mode 100644 .sqlx/query-2f77e15bd412042a6be9b272373de13d80ff20abdeb35ae2b030e1f79fd81af2.json create mode 100644 .sqlx/query-35b90c9506f2c9eeaa2a898008416b14c92bdb38e2ab74ab4c47cfccd86ed083.json create mode 100644 .sqlx/query-83398b1cc4795a74500e76bc2689a121241a97dde9c833a63a5019a57fee1ad9.json delete mode 100644 .sqlx/query-8d0411b95d5c5447bbc69b1d8d2b2f17fdeb5062c6508d94c503e15b0cb3f7c6.json create mode 100644 .sqlx/query-bab1ca94529787f458b772d6005454745d09f4a14cede190566f4c6019de3d95.json create mode 100644 .sqlx/query-d278ac69e0828cf012cd34f6a838424b23dc0f5f203dd7a9a0bf57acf17dff87.json delete mode 100644 .sqlx/query-f929e6ebccd3ad9bc57f9c33d62ac74add7908d5e84e06ff09f9e46a065ec44a.json create mode 100644 migrations/011_pm80_projection_memberships.sql diff --git a/.sqlx/query-0501daa78b69311ce9930970042a7ff671ec85cfdf99ce2d2954d9058f1fc2f3.json b/.sqlx/query-0501daa78b69311ce9930970042a7ff671ec85cfdf99ce2d2954d9058f1fc2f3.json new file mode 100644 index 0000000..698706a --- /dev/null +++ b/.sqlx/query-0501daa78b69311ce9930970042a7ff671ec85cfdf99ce2d2954d9058f1fc2f3.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM ldraw_resource_memberships WHERE provenance = ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "0501daa78b69311ce9930970042a7ff671ec85cfdf99ce2d2954d9058f1fc2f3" +} diff --git a/.sqlx/query-d87395649e34a00258562514babc61492a5433ad04a26e0f7d6b4ac843dd0dcc.json b/.sqlx/query-0572fe522555a3399aa7d0b6c3416df3ebea3419eea4fa549aece6dee86951b0.json similarity index 81% rename from .sqlx/query-d87395649e34a00258562514babc61492a5433ad04a26e0f7d6b4ac843dd0dcc.json rename to .sqlx/query-0572fe522555a3399aa7d0b6c3416df3ebea3419eea4fa549aece6dee86951b0.json index 1c6c9f1..9897dc2 100644 --- a/.sqlx/query-d87395649e34a00258562514babc61492a5433ad04a26e0f7d6b4ac843dd0dcc.json +++ b/.sqlx/query-0572fe522555a3399aa7d0b6c3416df3ebea3419eea4fa549aece6dee86951b0.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "SELECT resource_uri, owner_did, sha256, ordered_blob_cids, root, canonical_path, mime_type, byte_length\n FROM ldraw_resources\n WHERE project_uri = ?", + "query": "SELECT r.resource_uri, r.owner_did, r.sha256, r.ordered_blob_cids, m.root, m.canonical_path, r.mime_type, r.byte_length\n FROM ldraw_resource_memberships m\n JOIN ldraw_resources r ON r.resource_uri = m.resource_uri\n WHERE m.project_uri = ?", "describe": { "columns": [ { @@ -53,7 +53,7 @@ "type_info": "Text", "origin": { "Table": { - "table": "ldraw_resources", + "table": "ldraw_resource_memberships", "name": "root" } } @@ -64,7 +64,7 @@ "type_info": "Text", "origin": { "Table": { - "table": "ldraw_resources", + "table": "ldraw_resource_memberships", "name": "canonical_path" } } @@ -106,5 +106,5 @@ false ] }, - "hash": "d87395649e34a00258562514babc61492a5433ad04a26e0f7d6b4ac843dd0dcc" + "hash": "0572fe522555a3399aa7d0b6c3416df3ebea3419eea4fa549aece6dee86951b0" } diff --git a/.sqlx/query-1c9d5bf76f91eb1a6486cb4b94b5dd6e3a5e577fab77fca888fa6561d3b4f379.json b/.sqlx/query-1c9d5bf76f91eb1a6486cb4b94b5dd6e3a5e577fab77fca888fa6561d3b4f379.json new file mode 100644 index 0000000..c71f586 --- /dev/null +++ b/.sqlx/query-1c9d5bf76f91eb1a6486cb4b94b5dd6e3a5e577fab77fca888fa6561d3b4f379.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO projection_state (key, value, updated_at) VALUES ('cursor', ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at WHERE CAST(excluded.value AS INTEGER) > CAST(projection_state.value AS INTEGER)", + "describe": { + "columns": [], + "parameters": { + "Right": 2 + }, + "nullable": [] + }, + "hash": "1c9d5bf76f91eb1a6486cb4b94b5dd6e3a5e577fab77fca888fa6561d3b4f379" +} diff --git a/.sqlx/query-267ae501eae19fca4681bec219a5472d2fbd59a08bf4da00e6fb1f93770ca78d.json b/.sqlx/query-267ae501eae19fca4681bec219a5472d2fbd59a08bf4da00e6fb1f93770ca78d.json deleted file mode 100644 index 586a737..0000000 --- a/.sqlx/query-267ae501eae19fca4681bec219a5472d2fbd59a08bf4da00e6fb1f93770ca78d.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - "describe": { - "columns": [], - "parameters": { - "Right": 16 - }, - "nullable": [] - }, - "hash": "267ae501eae19fca4681bec219a5472d2fbd59a08bf4da00e6fb1f93770ca78d" -} diff --git a/.sqlx/query-2f77e15bd412042a6be9b272373de13d80ff20abdeb35ae2b030e1f79fd81af2.json b/.sqlx/query-2f77e15bd412042a6be9b272373de13d80ff20abdeb35ae2b030e1f79fd81af2.json new file mode 100644 index 0000000..4d51865 --- /dev/null +++ b/.sqlx/query-2f77e15bd412042a6be9b272373de13d80ff20abdeb35ae2b030e1f79fd81af2.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "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", + "describe": { + "columns": [], + "parameters": { + "Right": 4 + }, + "nullable": [] + }, + "hash": "2f77e15bd412042a6be9b272373de13d80ff20abdeb35ae2b030e1f79fd81af2" +} diff --git a/.sqlx/query-35b90c9506f2c9eeaa2a898008416b14c92bdb38e2ab74ab4c47cfccd86ed083.json b/.sqlx/query-35b90c9506f2c9eeaa2a898008416b14c92bdb38e2ab74ab4c47cfccd86ed083.json new file mode 100644 index 0000000..d8451cc --- /dev/null +++ b/.sqlx/query-35b90c9506f2c9eeaa2a898008416b14c92bdb38e2ab74ab4c47cfccd86ed083.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "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, licence = excluded.licence, notices_json = excluded.notices_json, updated_at = excluded.updated_at WHERE ldraw_resources.provenance = excluded.provenance", + "describe": { + "columns": [], + "parameters": { + "Right": 16 + }, + "nullable": [] + }, + "hash": "35b90c9506f2c9eeaa2a898008416b14c92bdb38e2ab74ab4c47cfccd86ed083" +} diff --git a/.sqlx/query-83398b1cc4795a74500e76bc2689a121241a97dde9c833a63a5019a57fee1ad9.json b/.sqlx/query-83398b1cc4795a74500e76bc2689a121241a97dde9c833a63a5019a57fee1ad9.json new file mode 100644 index 0000000..feecca6 --- /dev/null +++ b/.sqlx/query-83398b1cc4795a74500e76bc2689a121241a97dde9c833a63a5019a57fee1ad9.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO ldraw_verification (resource_uri, source_identity, state, diagnostic, updated_at) VALUES (?, ?, 'verified', NULL, ?) ON CONFLICT(resource_uri) DO UPDATE SET source_identity = excluded.source_identity, state = excluded.state, diagnostic = excluded.diagnostic, updated_at = excluded.updated_at", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "83398b1cc4795a74500e76bc2689a121241a97dde9c833a63a5019a57fee1ad9" +} diff --git a/.sqlx/query-8d0411b95d5c5447bbc69b1d8d2b2f17fdeb5062c6508d94c503e15b0cb3f7c6.json b/.sqlx/query-8d0411b95d5c5447bbc69b1d8d2b2f17fdeb5062c6508d94c503e15b0cb3f7c6.json deleted file mode 100644 index 90992d0..0000000 --- a/.sqlx/query-8d0411b95d5c5447bbc69b1d8d2b2f17fdeb5062c6508d94c503e15b0cb3f7c6.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "INSERT INTO projection_state (key, value, updated_at) VALUES ('cursor', ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", - "describe": { - "columns": [], - "parameters": { - "Right": 2 - }, - "nullable": [] - }, - "hash": "8d0411b95d5c5447bbc69b1d8d2b2f17fdeb5062c6508d94c503e15b0cb3f7c6" -} diff --git a/.sqlx/query-bab1ca94529787f458b772d6005454745d09f4a14cede190566f4c6019de3d95.json b/.sqlx/query-bab1ca94529787f458b772d6005454745d09f4a14cede190566f4c6019de3d95.json new file mode 100644 index 0000000..1f7fcad --- /dev/null +++ b/.sqlx/query-bab1ca94529787f458b772d6005454745d09f4a14cede190566f4c6019de3d95.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT OR IGNORE INTO ldraw_resource_memberships (resource_uri, project_uri, root, canonical_path, provenance) VALUES (?, ?, ?, ?, ?)", + "describe": { + "columns": [], + "parameters": { + "Right": 5 + }, + "nullable": [] + }, + "hash": "bab1ca94529787f458b772d6005454745d09f4a14cede190566f4c6019de3d95" +} diff --git a/.sqlx/query-d278ac69e0828cf012cd34f6a838424b23dc0f5f203dd7a9a0bf57acf17dff87.json b/.sqlx/query-d278ac69e0828cf012cd34f6a838424b23dc0f5f203dd7a9a0bf57acf17dff87.json new file mode 100644 index 0000000..e503467 --- /dev/null +++ b/.sqlx/query-d278ac69e0828cf012cd34f6a838424b23dc0f5f203dd7a9a0bf57acf17dff87.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "SELECT seq FROM projection_revisions WHERE did = ? AND collection = ? AND rkey = ?", + "describe": { + "columns": [ + { + "name": "seq", + "ordinal": 0, + "type_info": "Integer", + "origin": { + "Table": { + "table": "projection_revisions", + "name": "seq" + } + } + } + ], + "parameters": { + "Right": 3 + }, + "nullable": [ + false + ] + }, + "hash": "d278ac69e0828cf012cd34f6a838424b23dc0f5f203dd7a9a0bf57acf17dff87" +} diff --git a/.sqlx/query-f929e6ebccd3ad9bc57f9c33d62ac74add7908d5e84e06ff09f9e46a065ec44a.json b/.sqlx/query-f929e6ebccd3ad9bc57f9c33d62ac74add7908d5e84e06ff09f9e46a065ec44a.json deleted file mode 100644 index b58bbb9..0000000 --- a/.sqlx/query-f929e6ebccd3ad9bc57f9c33d62ac74add7908d5e84e06ff09f9e46a065ec44a.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "INSERT INTO ldraw_verification (resource_uri, source_identity, state, diagnostic, updated_at) VALUES (?, ?, 'verified', NULL, ?)", - "describe": { - "columns": [], - "parameters": { - "Right": 3 - }, - "nullable": [] - }, - "hash": "f929e6ebccd3ad9bc57f9c33d62ac74add7908d5e84e06ff09f9e46a065ec44a" -} diff --git a/migrations/011_pm80_projection_memberships.sql b/migrations/011_pm80_projection_memberships.sql new file mode 100644 index 0000000..db22d00 --- /dev/null +++ b/migrations/011_pm80_projection_memberships.sql @@ -0,0 +1,75 @@ +-- PM-80: preserve project-scoped membership for shared ordinary Part resources. +-- ldraw_resources remains the PM-76 source/byte identity table; this junction +-- carries each project/path membership so one Part may resolve independently in +-- every Thing that references it. +CREATE TABLE ldraw_resource_memberships ( + resource_uri TEXT NOT NULL, + project_uri TEXT NOT NULL, + root TEXT NOT NULL, + canonical_path TEXT NOT NULL, + provenance TEXT NOT NULL, + PRIMARY KEY (resource_uri, project_uri, root, canonical_path) +); +CREATE INDEX idx_ldraw_resource_memberships_project_path + ON ldraw_resource_memberships(project_uri, root, canonical_path); +CREATE INDEX idx_ldraw_resource_memberships_resource + ON ldraw_resource_memberships(resource_uri); + +-- Migration 009 placed project/path uniqueness on the source table. That +-- uniqueness belongs on the membership junction now: two source records may +-- legitimately share one visible path, where resolver collision handling can +-- reject the ambiguous lookup without rejecting either source row at write time. +CREATE TABLE ldraw_resources_pm80 ( + resource_uri TEXT PRIMARY KEY, + owner_did TEXT NOT NULL, + project_uri TEXT NOT NULL, + record_cid TEXT NOT NULL, + rkey TEXT NOT NULL, + root TEXT NOT NULL, + canonical_path TEXT NOT NULL, + mime_type TEXT NOT NULL, + byte_length INTEGER NOT NULL CHECK (byte_length >= 0), + sha256 BLOB NOT NULL CHECK (length(sha256) = 32), + ordered_blob_cids TEXT NOT NULL, + provenance TEXT NOT NULL, + licence TEXT NOT NULL, + notices_json TEXT NOT NULL, + snapshot_id TEXT, + legacy INTEGER NOT NULL DEFAULT 0 CHECK (legacy IN (0, 1)), + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + UNIQUE (resource_uri, sha256, byte_length, ordered_blob_cids) +); +INSERT INTO ldraw_resources_pm80 + SELECT resource_uri, owner_did, project_uri, record_cid, rkey, root, + canonical_path, mime_type, byte_length, sha256, ordered_blob_cids, + provenance, licence, notices_json, snapshot_id, legacy, created_at, + updated_at + FROM ldraw_resources; +DROP TABLE ldraw_resources; +ALTER TABLE ldraw_resources_pm80 RENAME TO ldraw_resources; +CREATE INDEX idx_ldraw_resources_owner ON ldraw_resources(owner_did); +CREATE INDEX idx_ldraw_resources_project_path + ON ldraw_resources(project_uri, root, canonical_path); +CREATE INDEX idx_ldraw_resources_source + ON ldraw_resources(resource_uri, sha256, byte_length); + +-- Existing PM-76 rows already carry their project membership on the source row. +-- Backfill it so upgrading databases has the same resolver behavior as fresh +-- projections; future PM-76 writers remain compatible through the source-row +-- fallback in the resolver. +INSERT OR IGNORE INTO ldraw_resource_memberships + (resource_uri, project_uri, root, canonical_path, provenance) +SELECT resource_uri, project_uri, root, canonical_path, provenance +FROM ldraw_resources; + +-- Hydrant's cursor orders events globally, but replay/eager paths can deliver a +-- record out of order. Keep the newest source event per record key so an older +-- update/delete cannot overwrite a newer projected identity or derived bytes. +CREATE TABLE projection_revisions ( + did TEXT NOT NULL, + collection TEXT NOT NULL, + rkey TEXT NOT NULL, + seq INTEGER NOT NULL CHECK (seq >= 0), + PRIMARY KEY (did, collection, rkey) +); diff --git a/src/indexing/db.rs b/src/indexing/db.rs index 6e933f7..da73170 100644 --- a/src/indexing/db.rs +++ b/src/indexing/db.rs @@ -78,7 +78,8 @@ pub async fn save_cursor(tx: &mut Transaction<'_, Sqlite>, cursor: u64) -> anyho .expect("current timestamp must fit in i64 nanoseconds"); sqlx::query!( "INSERT INTO projection_state (key, value, updated_at) VALUES ('cursor', ?, ?) \ - ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", + ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at \ + WHERE CAST(excluded.value AS INTEGER) > CAST(projection_state.value AS INTEGER)", cursor.to_string(), now ) diff --git a/src/indexing/projection.rs b/src/indexing/projection.rs index 92bdd61..a124d9c 100644 --- a/src/indexing/projection.rs +++ b/src/indexing/projection.rs @@ -107,22 +107,97 @@ impl ProjectionInput { /// 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 seq = match input { + let should_apply = match input { ProjectionInput::Record(event) => { - project_record(&mut tx, event).await?; - clear_pending(&mut tx, event).await?; - event.seq + event_is_current( + &mut tx, + event.seq, + &event.did, + &event.collection, + &event.rkey, + ) + .await? } ProjectionInput::Identity(event) => { - project_identity_upsert(&mut tx, event).await?; - event.seq + event_is_current(&mut tx, event.seq, &event.did, "#identity", "self").await? } }; - super::db::save_cursor(&mut tx, seq).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 { + 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. /// @@ -358,12 +433,14 @@ async fn project_thing_upsert( e: &ProjectionEvent, ) -> anyhow::Result<()> { let Some(data) = &e.record else { + invalidate_thing(tx, e).await?; return Ok(()); }; let rec = match decode::(data, &e.collection) { Ok(r) => r, Err(err) => { - tracing::warn!(error = %err, "thing record decode failed"); + tracing::warn!(error = %err, "thing record decode failed; invalidating projection"); + invalidate_thing(tx, e).await?; return Ok(()); } }; @@ -478,6 +555,27 @@ async fn project_thing_upsert( 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, @@ -504,12 +602,14 @@ async fn project_model_upsert( e: &ProjectionEvent, ) -> anyhow::Result<()> { let Some(data) = &e.record else { + invalidate_model(tx, e).await?; return Ok(()); }; let rec = match decode::(data, &e.collection) { Ok(r) => r, Err(err) => { - tracing::warn!(error = %err, "model record decode failed"); + tracing::warn!(error = %err, "model record decode failed; invalidating projection"); + invalidate_model(tx, e).await?; return Ok(()); } }; @@ -588,12 +688,14 @@ async fn project_part_upsert( e: &ProjectionEvent, ) -> anyhow::Result<()> { let Some(data) = &e.record else { + invalidate_part(tx, e).await?; return Ok(()); }; let rec = match decode::(data, &e.collection) { Ok(r) => r, Err(err) => { - tracing::warn!(error = %err, "part record decode failed"); + tracing::warn!(error = %err, "part record decode failed; invalidating projection"); + invalidate_part(tx, e).await?; return Ok(()); } }; @@ -640,7 +742,25 @@ async fn project_part_upsert( Ok(()) } -async fn project_part_delete( +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<()> { @@ -654,6 +774,13 @@ async fn project_part_delete( 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 @@ -676,6 +803,12 @@ async fn rebuild_ordinary_ldraw_resources(tx: &mut Transaction<'_, Sqlite>) -> a ) .execute(&mut **tx) .await?; + sqlx::query!( + "DELETE FROM ldraw_resource_memberships WHERE provenance = ?", + DERIVED_LDRAW_PROVENANCE, + ) + .execute(&mut **tx) + .await?; sqlx::query!( "DELETE FROM ldraw_resources WHERE provenance = ?", DERIVED_LDRAW_PROVENANCE, @@ -696,13 +829,35 @@ async fn rebuild_ordinary_ldraw_resources(tx: &mut Transaction<'_, Sqlite>) -> a .fetch_all(&mut **tx) .await?; - let mut seen_parts = HashSet::new(); - let mut seen_paths = HashSet::new(); + let mut seen_sources = HashSet::new(); + let mut seen_memberships = HashSet::new(); for row in rows { - // A part may be linked by several models. The PM-76 resource identity is - // the authored part URI, so retain the first deterministic project link - // rather than creating conflicting rows for one source resource. - if !seen_parts.insert(row.uri.clone()) { + 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, "ordinary LDraw path derivation failed"); + continue; + }; + if !seen_memberships.insert(( + row.uri.clone(), + row.project_uri.clone(), + path.as_str().to_owned(), + )) { + continue; + } + if seen_sources.contains(&row.uri) { + sqlx::query!( + "INSERT OR IGNORE 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?; continue; } let file: File = match serde_json::from_str(&row.file_json) { @@ -742,18 +897,6 @@ async fn rebuild_ordinary_ldraw_resources(tx: &mut Transaction<'_, Sqlite>) -> a continue; } - let canonical_path = if row.name.contains('/') { - format!("models/{}", row.name.trim_start_matches("models/")) - } else { - format!("models/{}", row.name) - }; - let Ok(path) = crate::ldraw::CanonicalPath::parse(&canonical_path) else { - continue; - }; - if !seen_paths.insert((row.project_uri.clone(), path.as_str().to_owned())) { - continue; - } - let source_uri = AtUri::new_owned(&row.uri)?; let digest = digest.to_vec(); let source_identity = crate::ldraw::source_identity_for(&source_uri, &blob_cids, &digest, file.size); @@ -766,7 +909,13 @@ async fn rebuild_ordinary_ldraw_resources(tx: &mut Transaction<'_, Sqlite>) -> a (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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + 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, 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, @@ -786,16 +935,31 @@ async fn rebuild_ordinary_ldraw_resources(tx: &mut Transaction<'_, Sqlite>) -> a ) .execute(&mut **tx) .await?; + sqlx::query!( + "INSERT OR IGNORE 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?; sqlx::query!( "INSERT INTO ldraw_verification \ (resource_uri, source_identity, state, diagnostic, updated_at) \ - VALUES (?, ?, 'verified', NULL, ?)", + VALUES (?, ?, 'verified', NULL, ?) \ + 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?; + seen_sources.insert(row.uri); } Ok(()) } diff --git a/src/ldraw/mod.rs b/src/ldraw/mod.rs index 25d510a..a0699c9 100644 --- a/src/ldraw/mod.rs +++ b/src/ldraw/mod.rs @@ -4,5 +4,5 @@ pub mod paths; pub mod resolve; pub mod verification; -pub use paths::{CanonicalPath, RootId}; +pub use paths::{CanonicalPath, RootId, project_part_path}; pub use resolve::{LdrawResourceResolver, ResolveError, source_identity_for}; diff --git a/src/ldraw/paths.rs b/src/ldraw/paths.rs index f051b10..a45bec3 100644 --- a/src/ldraw/paths.rs +++ b/src/ldraw/paths.rs @@ -1,7 +1,116 @@ use std::fmt; +use jacquard_common::types::string::AtUri; use unicode_normalization::UnicodeNormalization; +/// Build the visible path for a Part membership. +/// +/// A bare Part name retains the historical `models/` root. Once the author +/// supplies a relative path, it is preserved as-is (after `CanonicalPath` +/// validation). Parts owned by another authority are put below an explicit +/// source namespace. Jacquard validates the URI, collection, and rkey before +/// this function reaches the string/SQL boundary. +pub fn project_part_path( + project: &AtUri, + part: &AtUri, + declared_name: &str, +) -> Result { + if project.authority().as_str() == part.authority().as_str() { + let declared = if declared_name.contains('/') { + declared_name.to_owned() + } else { + format!("models/{declared_name}") + }; + return CanonicalPath::parse(&declared); + } + + let Some(collection) = part.collection() else { + return Err(PathDiagnostic::SourceIdentity); + }; + let Some(rkey) = part.rkey() else { + return Err(PathDiagnostic::SourceIdentity); + }; + if collection.as_str() != "space.polymodel.library.part" { + return Err(PathDiagnostic::SourceIdentity); + } + let path = format!( + "external/{}/{}", + encode_identity_segment(part.authority().as_str()), + encode_identity_segment(rkey.as_str()), + ); + CanonicalPath::parse(&path) +} + +/// Decode an external Part path into its authority and rkey. +/// +/// The rkey is deliberately returned unchanged: Jacquard's validated rkey is +/// already a safe relative path segment. This is a pure inverse of the +/// authority encoding and is useful for route/path round-trip tests. +pub fn decode_external_part_path(path: &CanonicalPath) -> Option<(String, String)> { + let segments = path.segments().collect::>(); + if segments.first().copied()? != "external" || segments.len() != 3 { + return None; + } + let authority = decode_identity_segment(segments[1])?; + let rkey = decode_identity_segment(segments[2])?; + jacquard_common::types::recordkey::Rkey::::new_owned(rkey.clone()).ok()?; + Some((authority, rkey)) +} + +/// Escape one authority component with a self-delimiting, ASCII-only codec. +/// +/// `x` is the only prefix and `_hh` is the only escape form. The unescaped +/// alphabet excludes `_`, `%`, `.`, and path separators, so escaped-looking +/// input, dot segments, Unicode normalization, and URL decoding cannot alias +/// an encoded identity. +fn encode_identity_segment(value: &str) -> String { + let mut out = String::from("x"); + for byte in value.as_bytes() { + if byte.is_ascii_alphanumeric() || *byte == b'-' || *byte == b'~' { + out.push(*byte as char); + } else { + out.push('_'); + out.push(hex_digit(byte >> 4)); + out.push(hex_digit(byte & 0x0f)); + } + } + out +} + +fn decode_identity_segment(value: &str) -> Option { + let bytes = value.strip_prefix('x')?.as_bytes(); + let mut decoded = Vec::with_capacity(bytes.len()); + let mut index = 0; + while index < bytes.len() { + match bytes[index] { + b'_' if index + 2 < bytes.len() => { + let high = from_hex(bytes[index + 1])?; + let low = from_hex(bytes[index + 2])?; + decoded.push(high << 4 | low); + index += 3; + } + byte if byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'~' => { + decoded.push(byte); + index += 1; + } + _ => return None, + } + } + String::from_utf8(decoded).ok() +} + +fn hex_digit(value: u8) -> char { + char::from(b"0123456789abcdef"[value as usize]) +} + +fn from_hex(value: u8) -> Option { + match value { + b'0'..=b'9' => Some(value - b'0'), + b'a'..=b'f' => Some(value - b'a' + 10), + _ => None, + } +} + /// A validated, NFC-preserving virtual filename. It is never interpreted as a URL. #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] pub struct CanonicalPath(String); @@ -87,6 +196,7 @@ pub enum PathDiagnostic { ControlOrColon, NotNfc, Segment, + SourceIdentity, } #[derive(Clone, Debug, Eq, Hash, PartialEq)] @@ -155,6 +265,81 @@ mod tests { assert_eq!(CanonicalPath::parse("café").unwrap().as_str(), "café"); } + #[test] + fn authority_path_codec_round_trips_and_is_injective() { + let project = AtUri::new_owned("at://did:plc:owner/space.polymodel.library.thing/project") + .unwrap(); + let mut paths = Vec::new(); + for (authority, rkey) in [ + ("did:plc:alpha", "same_rkey"), + ("did:plc:beta", "same_rkey"), + ("did:plc:alpha", "same-rkey"), + ("did:plc:alpha", "a.b"), + ("did:plc:alpha", "a~b:c"), + ] { + let part = AtUri::new_owned(&format!( + "at://{authority}/space.polymodel.library.part/{rkey}" + )) + .unwrap(); + let path = project_part_path(&project, &part, "ignored.ldr").unwrap(); + assert!(path.as_str().starts_with("external/x")); + assert!(!path.as_str().starts_with('/')); + assert!(!path.as_str().contains("..")); + let (decoded_authority, decoded_rkey) = decode_external_part_path(&path).unwrap(); + assert_eq!(decoded_authority, authority); + assert_eq!(decoded_rkey, rkey); + paths.push(path); + } + for (index, path) in paths.iter().enumerate() { + assert!(paths[index + 1..].iter().all(|other| other != path)); + } + } + + #[test] + fn same_authority_preserves_declared_project_relative_path() { + let project = AtUri::new_owned("at://did:plc:owner/space.polymodel.library.thing/project") + .unwrap(); + let part = AtUri::new_owned( + "at://did:plc:owner/space.polymodel.library.part/part", + ) + .unwrap(); + assert_eq!( + project_part_path(&project, &part, "models/sub/main.ldr") + .unwrap() + .as_str(), + "models/sub/main.ldr" + ); + assert_eq!( + project_part_path(&project, &part, "main.ldr") + .unwrap() + .as_str(), + "models/main.ldr" + ); + } + + #[test] + fn external_codec_rejects_path_aliases_and_invalid_rkeys() { + let project = AtUri::new_owned("at://did:plc:owner/space.polymodel.library.thing/project") + .unwrap(); + let part = AtUri::new_owned( + "at://did:plc:external/space.polymodel.library.part/x", + ) + .unwrap(); + let valid = project_part_path(&project, &part, "x.ldr").unwrap(); + assert_eq!( + decode_external_part_path(&valid).unwrap(), + ("did:plc:external".to_owned(), "x".to_owned()) + ); + let malformed_escape = CanonicalPath::parse("external/x_zz/x").unwrap(); + assert!(decode_external_part_path(&malformed_escape).is_none()); + assert!(CanonicalPath::parse("external/xdid_3aplc/a%2Fb").is_err()); + assert!(CanonicalPath::parse("external/xdid_3aplc/../x").is_err()); + assert!(AtUri::::new_owned( + "at://did:plc:owner/space.polymodel.library.part/cafe\u{301}" + ) + .is_err()); + } + #[test] fn precedence_and_ascii_case_policy_are_stable() { let path = CanonicalPath::parse("parts/3001.dat").unwrap(); diff --git a/src/ldraw/resolve.rs b/src/ldraw/resolve.rs index 37bd3da..0b171b1 100644 --- a/src/ldraw/resolve.rs +++ b/src/ldraw/resolve.rs @@ -118,9 +118,10 @@ impl SqliteResolver { path: &CanonicalPath, ) -> Result { let rows = sqlx::query!( - r#"SELECT resource_uri, owner_did, sha256, ordered_blob_cids, root, canonical_path, mime_type, byte_length - FROM ldraw_resources - WHERE project_uri = ?"#, + r#"SELECT r.resource_uri, r.owner_did, r.sha256, r.ordered_blob_cids, m.root, m.canonical_path, r.mime_type, r.byte_length + FROM ldraw_resource_memberships m + JOIN ldraw_resources r ON r.resource_uri = m.resource_uri + WHERE m.project_uri = ?"#, project.as_ref(), ).fetch_all(&self.pool).await.map_err(|_| ResolveError::Unavailable { diagnostic: AvailabilityDiagnostic::SourceUnavailable })?; let mut entries = Vec::with_capacity(rows.len()); -- knotmirror 0.0.1