//! 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::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::{model::Model, part::Part, thing::Thing},
};
use sqlx::{Sqlite, SqlitePool, Transaction};

/// 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 seq = match input {
        ProjectionInput::Record(event) => {
            project_record(&mut tx, event).await?;
            clear_pending(&mut tx, event).await?;
            event.seq
        }
        ProjectionInput::Identity(event) => {
            project_identity_upsert(&mut tx, event).await?;
            event.seq
        }
    };
    super::db::save_cursor(&mut tx, seq).await?;
    tx.commit().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?;
        }
        ("delete", "space.polymodel.library.thing") => {
            project_thing_delete(tx, event).await?;
        }
        ("create" | "update", "space.polymodel.library.model") => {
            project_model_upsert(tx, event).await?;
        }
        ("delete", "space.polymodel.library.model") => {
            project_model_delete(tx, event).await?;
        }
        ("create" | "update", "space.polymodel.library.part") => {
            project_part_upsert(tx, event).await?;
        }
        ("delete", "space.polymodel.library.part") => {
            project_part_delete(tx, event).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 {
        return Ok(());
    };
    let rec = match decode::<Thing>(data, &e.collection) {
        Ok(r) => r,
        Err(err) => {
            tracing::warn!(error = %err, "thing record decode failed");
            return Ok(());
        }
    };
    let uri = at_uri(&e.did, &e.collection, &e.rkey);
    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 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 {
        return Ok(());
    };
    let rec = match decode::<Model>(data, &e.collection) {
        Ok(r) => r,
        Err(err) => {
            tracing::warn!(error = %err, "model record decode failed");
            return Ok(());
        }
    };
    let uri = at_uri(&e.did, &e.collection, &e.rkey);
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let 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 {
        return Ok(());
    };
    let rec = match decode::<Part>(data, &e.collection) {
        Ok(r) => r,
        Err(err) => {
            tracing::warn!(error = %err, "part record decode failed");
            return Ok(());
        }
    };
    let uri = at_uri(&e.did, &e.collection, &e.rkey);
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let 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 project_part_delete(
    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(())
}

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