//! 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 domain-neutral 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>,
}

impl ProjectionEvent {
    /// Convert a Hydrant [`hydrant::control::Event`] into a projection event.
    ///
    /// Returns `Ok(None)` for non-record events (identity/account) which are not
    /// projected into SQLite.
    pub fn from_hydrant(event: &hydrant::control::Event) -> anyhow::Result<Option<Self>> {
        let Some(rec) = event.record.as_ref() else {
            return Ok(None);
        };
        Ok(Some(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()),
        }))
    }
}

/// Project a single event into SQLite, advancing the cursor on success.
pub async fn project_event(db: &SqlitePool, event: &ProjectionEvent) -> anyhow::Result<()> {
    let mut tx = db.begin().await?;
    let action = event.action.as_str();
    let collection = event.collection.as_str();

    match (action, collection) {
        ("create" | "update", "space.polymodel.library.thing") => {
            project_thing_upsert(&mut tx, event).await?;
        }
        ("delete", "space.polymodel.library.thing") => {
            project_thing_delete(&mut tx, event).await?;
        }
        ("create" | "update", "space.polymodel.library.model") => {
            project_model_upsert(&mut tx, event).await?;
        }
        ("delete", "space.polymodel.library.model") => {
            project_model_delete(&mut tx, event).await?;
        }
        ("create" | "update", "space.polymodel.library.part") => {
            project_part_upsert(&mut tx, event).await?;
        }
        ("delete", "space.polymodel.library.part") => {
            project_part_delete(&mut tx, event).await?;
        }
        ("create", "space.polymodel.graph.like") => {
            project_like_create(&mut tx, event).await?;
        }
        ("delete", "space.polymodel.graph.like") => {
            project_like_delete(&mut tx, event).await?;
        }
        ("create", "space.polymodel.graph.save") => {
            project_save_create(&mut tx, event).await?;
        }
        ("delete", "space.polymodel.graph.save") => {
            project_save_delete(&mut tx, event).await?;
        }
        ("create", "space.polymodel.graph.tag") => {
            project_tag_create(&mut tx, event).await?;
        }
        ("delete", "space.polymodel.graph.tag") => {
            project_tag_delete(&mut tx, event).await?;
        }
        ("create", "space.polymodel.graph.listitem") => {
            project_listitem_create(&mut tx, event).await?;
        }
        ("delete", "space.polymodel.graph.listitem") => {
            project_listitem_delete(&mut tx, event).await?;
        }
        // graph.list lives in Hydrant's fjall store; no SQLite projection.
        // account/identity events are filtered out by from_hydrant.
        _ => {
            tracing::debug!(action, collection, "no projection rule for event");
        }
    }

    super::db::save_cursor(&mut tx, event.seq).await?;
    tx.commit().await?;
    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_millis(v: &Value, key: &str) -> Option<i64> {
    chrono::DateTime::parse_from_rfc3339(v.get(key)?.as_str()?)
        .ok()
        .map(|dt| dt.timestamp_millis())
}

fn now_millis() -> i64 {
    chrono::Utc::now().timestamp_millis()
}

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_millis(rec, "createdAt").unwrap_or_else(now_millis);
    let indexed_at = now_millis();

    // 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) \
         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",
        e.did,
        e.rkey,
        uri,
        cid,
        name,
        summary,
        license,
        tags_json,
        tags_text,
        instructions,
        cover_json,
        derived_from_uri,
        created_at,
        indexed_at,
    )
    .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_millis(rec, "createdAt").unwrap_or_else(now_millis);
    let indexed_at = now_millis();

    sqlx::query!(
        "INSERT INTO models \
         (did, rkey, uri, cid, name, summary, created_at, indexed_at) \
         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",
        e.did,
        e.rkey,
        uri,
        cid,
        name,
        summary,
        created_at,
        indexed_at,
    )
    .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_millis(rec, "createdAt").unwrap_or_else(now_millis);
    let indexed_at = now_millis();

    sqlx::query!(
        "INSERT INTO parts \
         (did, rkey, uri, cid, name, format, file_json, created_at, indexed_at) \
         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",
        e.did,
        e.rkey,
        uri,
        cid,
        name,
        format,
        file_json,
        created_at,
        indexed_at,
    )
    .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(())
}

// ---------------------------------------------------------------------------
// 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 created_at = parse_millis(rec, "createdAt").unwrap_or_else(now_millis);

    let res = sqlx::query!(
        "INSERT OR IGNORE INTO likes (did, rkey, subject_uri, created_at) VALUES (?, ?, ?, ?)",
        e.did,
        e.rkey,
        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 note = s_str(rec, "note");
    let created_at = parse_millis(rec, "createdAt").unwrap_or_else(now_millis);

    let res = sqlx::query!(
        "INSERT OR IGNORE INTO saves (did, rkey, subject_uri, note, created_at) \
         VALUES (?, ?, ?, ?, ?)",
        e.did,
        e.rkey,
        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 created_at = parse_millis(rec, "createdAt").unwrap_or_else(now_millis);

    let res = sqlx::query!(
        "INSERT OR IGNORE INTO tags (did, rkey, subject_uri, tag, created_at) \
         VALUES (?, ?, ?, ?, ?)",
        e.did,
        e.rkey,
        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 created_at = parse_millis(rec, "createdAt").unwrap_or_else(now_millis);

    // No denormalized count for listitems; list membership is queried directly.
    sqlx::query!(
        "INSERT OR IGNORE INTO listitems (did, rkey, list_uri, subject_uri, created_at) \
         VALUES (?, ?, ?, ?, ?)",
        e.did,
        e.rkey,
        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",
                "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, &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,
            &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, &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, &thing_event(1, did, "r1", "Cube", &["at://m1"]))
            .await
            .unwrap();
        assert!(!fts_uris(&db, "cube").await.is_empty());

        project_event(
            &db,
            &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, &like_event(1, "did:plc:a", "k1", subject))
            .await
            .unwrap();
        project_event(&db, &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, &like_event(1, "did:plc:a", "k1", subject))
            .await
            .unwrap();
        project_event(
            &db,
            &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,
            &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, &like_event(1, "did:plc:a", "k1", subject))
            .await
            .unwrap();
        project_event(&db, &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, &thing_event(7, "did:plc:a", "r1", "Cube", &[]))
            .await
            .unwrap();
        assert_eq!(cursor(&db).await, Some(7));
        project_event(&db, &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, &save).await.unwrap();
        assert_eq!(stat(&db, subject, "save_count").await, 1);

        project_event(
            &db,
            &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",
                "tags": ["cube"],
                "createdAt": "2024-01-01T00:00:00.000Z",
                "models": [],
            })),
            cid: Some("cid1".to_string()),
        };
        project_event(&db, &create).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",
            "tags": ["sphere"],
            "createdAt": "2024-01-02T00:00:00.000Z",
            "models": [],
        }));
        project_event(&db, &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);
    }
}
