//! 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);
    }
}