//! 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 serde_json::Value;
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; `None` for delete events.
    pub record: Option<Value>,
    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)]
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.clone(),
                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 = record.to_string();
            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.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", "space.polymodel.graph.listitem") => {
            project_listitem_create(tx, event).await?;
        }
        ("delete", "space.polymodel.graph.listitem") => {
            project_listitem_delete(tx, event).await?;
        }
        // graph.list lives in Hydrant's fjall store; no SQLite projection.
        // account events are dropped by ProjectionInput::from_hydrant.
        _ => {
            tracing::debug!(action, collection, "no projection rule for event");
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Field extraction helpers (records arrive as serde_json::Value).
// ---------------------------------------------------------------------------

fn s_str(v: &Value, key: &str) -> Option<String> {
    v.get(key).and_then(|x| x.as_str()).map(String::from)
}

/// URI from a single strongRef field, e.g. `derivedFrom: { uri, cid }`.
fn ref_uri(v: &Value, key: &str) -> Option<String> {
    v.get(key)
        .and_then(|r| r.get("uri"))
        .and_then(|u| u.as_str())
        .map(String::from)
}

/// URIs from an array of strongRefs, e.g. `models: [{ uri, cid }, ...]`.
fn ref_uris(v: &Value, key: &str) -> Vec<String> {
    v.get(key)
        .and_then(|x| x.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|el| el.get("uri").and_then(|u| u.as_str()).map(String::from))
                .collect()
        })
        .unwrap_or_default()
}

fn str_array(v: &Value, key: &str) -> Vec<String> {
    v.get(key)
        .and_then(|x| x.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|el| el.as_str().map(String::from))
                .collect()
        })
        .unwrap_or_default()
}

fn parse_nanos(v: &Value, key: &str) -> Option<i64> {
    chrono::DateTime::parse_from_rfc3339(v.get(key)?.as_str()?)
        .ok()
        .and_then(|dt| dt.timestamp_nanos_opt())
}

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(rec) = &e.record else {
        return Ok(());
    };
    let uri = at_uri(&e.did, &e.collection, &e.rkey);
    // Required content fields: skip the event gracefully if absent (malformed
    // record) rather than storing a blank row.
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let Some(name) = s_str(rec, "name") else {
        return Ok(());
    };
    let summary = s_str(rec, "summary");
    let license = s_str(rec, "license");
    let instructions = s_str(rec, "instructions");
    let cover_json = rec.get("cover").map(|v| v.to_string());
    let derived_from_uri = ref_uri(rec, "derivedFrom");

    let tags = str_array(rec, "tags");
    let 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 = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos);
    let indexed_at = now_nanos();
    // The full raw record, so read endpoints can populate the required `record`
    // field on thingView directly from SQLite (and the write path gets
    // read-your-own-write, since Hydrant has no public single-record write API).
    let record_json = rec.to_string();

    // 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_uri) in ref_uris(rec, "models").into_iter().enumerate() {
        sqlx::query!(
            "INSERT OR REPLACE INTO thing_models (thing_uri, model_uri, position) \
             VALUES (?, ?, ?)",
            uri,
            model_uri,
            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(rec) = &e.record else {
        return Ok(());
    };
    let uri = at_uri(&e.did, &e.collection, &e.rkey);
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let Some(name) = s_str(rec, "name") else {
        return Ok(());
    };
    let summary = s_str(rec, "summary");
    let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos);
    let indexed_at = now_nanos();
    let record_json = rec.to_string();

    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_uri) in ref_uris(rec, "parts").into_iter().enumerate() {
        sqlx::query!(
            "INSERT OR REPLACE INTO model_parts (model_uri, part_uri, position) \
             VALUES (?, ?, ?)",
            uri,
            part_uri,
            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(rec) = &e.record else {
        return Ok(());
    };
    let uri = at_uri(&e.did, &e.collection, &e.rkey);
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let Some(name) = s_str(rec, "name") else {
        return Ok(());
    };
    let format = s_str(rec, "format");
    let file_json = rec.get("file").map(|v| v.to_string());
    let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos);
    let indexed_at = now_nanos();
    let record_json = rec.to_string();

    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(rec) = &e.record else {
        return Ok(());
    };
    let display_name = s_str(rec, "displayName");
    let description = s_str(rec, "description");
    let avatar_json = rec.get("avatar").map(|v| v.to_string());
    let default_license = s_str(rec, "defaultLicense");
    let pronouns = s_str(rec, "pronouns");
    let printers_json = rec
        .get("printers")
        .and_then(|v| v.as_array())
        .map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string()));
    let links_json = rec
        .get("links")
        .and_then(|v| v.as_array())
        .map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string()));
    let record_json = rec.to_string();
    let 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(rec) = &e.record else {
        return Ok(());
    };
    let Some(subject) = ref_uri(rec, "subject") else {
        return Ok(());
    };
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let created_at = parse_nanos(rec, "createdAt").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_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(rec) = &e.record else {
        return Ok(());
    };
    let Some(subject) = ref_uri(rec, "subject") else {
        return Ok(());
    };
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let note = s_str(rec, "note");
    let created_at = parse_nanos(rec, "createdAt").unwrap_or_else(now_nanos);

    let 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(rec) = &e.record else {
        return Ok(());
    };
    let Some(subject) = ref_uri(rec, "subject") else {
        return Ok(());
    };
    let Some(tag) = s_str(rec, "tag") else {
        return Ok(());
    };
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let created_at = parse_nanos(rec, "createdAt").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(rec) = &e.record else {
        return Ok(());
    };
    let Some(list_uri) = ref_uri(rec, "list") else {
        return Ok(());
    };
    let Some(subject_uri) = ref_uri(rec, "subject") else {
        return Ok(());
    };
    let Some(cid) = e.cid.clone() else {
        return Ok(());
    };
    let created_at = parse_nanos(rec, "createdAt").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_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(json!({
                "name": name,
                "summary": "a thing summary",
                "license": "MIT", "tags": ["cube", "calibration"],
                "createdAt": "2024-01-15T12:00:00.000Z",
                "models": models.iter().map(|m| json!({"uri": m, "cid": "c"})).collect::<Vec<_>>(),
            })),
            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(json!({
                "subject": {"uri": subject, "cid": "c"},
                "createdAt": "2024-02-01T00:00:00.000Z",
            })),
            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", &["at://m1", "at://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", &["at://m1", "at://m2"])),
        )
        .await
        .unwrap();
        assert_eq!(junction_count(&db, uri).await, 2);

        // Update: same rkey, different models. Old junction rows must be replaced.
        let mut upd = thing_event(2, did, "r1", "Cube v2", &["at://m3"]);
        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://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", &["at://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/.../thing";
        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/.../thing";
        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/.../thing";
        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/.../thing";
        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://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/.../thing";
        let save = ProjectionEvent {
            seq: 1,
            did: "did:plc:a".to_string(),
            collection: "space.polymodel.graph.save".to_string(),
            rkey: "s1".to_string(),
            action: "create".to_string(),
            record: Some(json!({
                "subject": {"uri": subject, "cid": "c"},
                "note": "useful",
                "createdAt": "2024-03-01T00:00:00.000Z",
            })),
            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(json!({
                "name": "Calibration Cube",
                "license": "MIT", "tags": ["cube"],
                "createdAt": "2024-01-01T00:00:00.000Z",
                "models": [],
            })),
            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(json!({
            "name": "Friendly Sphere",
            "license": "MIT", "tags": ["sphere"],
            "createdAt": "2024-01-02T00:00:00.000Z",
            "models": [],
        }));
        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);
    }
}
