//! DB upsert functions for the AppView indexer.
//!
//! These are pure: given a [`PgPool`] and a parsed Jetstream (or PDS-pushed)
//! `CommitOp`, each function performs one idempotent write against the
//! `posts` / `likes` / `reposts` / `follows` tables.
//!
//! Every write is idempotent: UPSERT for create/update and DELETE for
//! removals, so re-applying the same event multiple times (e.g. on a Jetstream
//! reconnect that replays recent messages) is safe.

use anyhow::{anyhow, Result};
use at_firehose::{CommitOp, JetstreamEvent};
use serde_json::Value;
use sqlx::{
    encode::IsNull,
    error::BoxDynError,
    PgPool, Postgres, Type,
};
#[cfg(test)]
use std::time::Duration;
#[cfg(test)]
use tokio::time::timeout;

/// Collections whose `create`/`delete` events we persist to the `posts` table.
#[allow(dead_code)]
pub const POST_COLLECTIONS: &[&str] = &["app.twi.post", "app.bsky.feed.post"];

/// Build a `at://did/collection/rkey` URI from the event's `did` + the
/// commit op's `rkey` (falling back to the trailing component of `path`).
#[allow(dead_code)]
pub fn build_uri(did: &str, collection: &str, op: &CommitOp) -> Option<String> {
    let rkey = op
        .rkey
        .clone()
        .or_else(|| {
            op.path
                .as_deref()
                .and_then(|p| p.rsplit('/').next().map(str::to_string))
        })?;
    Some(format!("at://{did}/{collection}/{rkey}"))
}

/// Best-effort extraction of the collection NSID from a Jetstream commit
/// payload. Jetstream single-op events have it at `commit.collection`;
/// firehose-style batched events have it per-op under `collection`.
pub fn extract_collection(commit: &Value) -> Option<String> {
    if let Some(c) = commit.get("collection").and_then(|v| v.as_str()) {
        return Some(c.to_string());
    }
    if let Some(ops) = commit.get("ops").and_then(|v| v.as_array()) {
        if let Some(c) = ops
            .first()
            .and_then(|o| o.get("collection"))
            .and_then(|v| v.as_str())
        {
            return Some(c.to_string());
        }
    }
    None
}

/// Convert the wire-format `commit` value into a flat list of
/// [`CommitOp`]s. Accepts both the Jetstream single-op shape (where the
/// commit object itself carries `operation`/`rkey`/`record`/...) and the
/// firehose batched shape (where `commit.ops` is an array of ops).
///
/// The wire field name is `operation`, which we map to `CommitOp::action`
/// because that's the name the at-firehose crate settled on.
pub fn commit_op_from_jetstream_value(commit: &Value) -> Vec<CommitOp> {
    if let Some(ops) = commit.get("ops").and_then(|v| v.as_array()) {
        return ops.iter().filter_map(parse_single_op).collect();
    }
    parse_single_op(commit).into_iter().collect()
}

fn parse_single_op(op: &Value) -> Option<CommitOp> {
    let action = op
        .get("operation")
        .or_else(|| op.get("action"))
        .and_then(|v| v.as_str())?
        .to_string();
    let rkey = op.get("rkey").and_then(|v| v.as_str()).map(String::from);
    let path = op.get("path").and_then(|v| v.as_str()).map(String::from);
    let cid = op.get("cid").and_then(|v| v.as_str()).map(String::from);
    let record = op.get("record").cloned();
    Some(CommitOp {
        action,
        rkey,
        path,
        cid,
        record,
    })
}

/// Parse an ISO-8601 timestamp string from a record's `createdAt`. We treat
/// any parse failure as "now" rather than dropping the record — the event
/// itself is still useful; we just lose the client's timestamp.
pub fn parse_created_at(s: Option<&str>) -> chrono::DateTime<chrono::Utc> {
    match s.and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) {
        Some(dt) => dt.with_timezone(&chrono::Utc),
        None => chrono::Utc::now(),
    }
}

// -- cursor ----------------------------------------------------------------

const CURSOR_ROW_ID: i32 = 1;

/// Read the persisted jetstream cursor (microseconds since epoch). Returns 0
/// if the row is missing for any reason (e.g. fresh DB before the migration
/// inserted it).
#[allow(dead_code)]
pub async fn cursor_get(db: &PgPool) -> Result<i64> {
    let row: Option<(i64,)> = sqlx::query_as(
        "SELECT cursor FROM jetstream_cursor WHERE id = $1",
    )
    .bind(CURSOR_ROW_ID)
    .fetch_optional(db)
    .await?;
    Ok(row.map(|(c,)| c).unwrap_or(0))
}

/// Advance the persisted cursor, but never to a lower value.
pub async fn cursor_advance(db: &PgPool, new_value: i64) -> Result<()> {
    sqlx::query(
        r#"UPDATE jetstream_cursor
           SET cursor = GREATEST(cursor, $1), updated_at = now()
           WHERE id = $2"#,
    )
    .bind(new_value)
    .bind(CURSOR_ROW_ID)
    .execute(db)
    .await?;
    Ok(())
}

// -- posts -----------------------------------------------------------------

/// A nullable JSONB column wrapper. We store `embed` as the full
/// AT-Protocol embed object verbatim — the alternative (normalising into
/// a separate `embeds` table) would cost an extra round trip per post
/// and require schema migrations for every new embed variant.
///
/// `Option<Value>` already implements `Encode<Json>` for `serde_json::Value`,
/// so we only need the Decode/Type pair to handle SQL NULL → `None`.
#[derive(Debug, Clone, Default)]
pub struct EmbedColumn(pub Option<Value>);

impl From<Option<Value>> for EmbedColumn {
    fn from(v: Option<Value>) -> Self {
        EmbedColumn(v)
    }
}

impl From<Value> for EmbedColumn {
    fn from(v: Value) -> Self {
        EmbedColumn(Some(v))
    }
}

impl<'r> sqlx::Decode<'r, Postgres> for EmbedColumn {
    fn decode(
        value: <Postgres as sqlx::Database>::ValueRef<'r>,
    ) -> Result<Self, sqlx::error::BoxDynError> {
        let v: Option<Value> = <Option<Value> as sqlx::Decode<Postgres>>::decode(value)?;
        Ok(EmbedColumn(v))
    }
}

impl<'q> sqlx::Encode<'q, Postgres> for EmbedColumn {
    fn encode_by_ref(
        &self,
        buf: &mut <Postgres as sqlx::Database>::ArgumentBuffer<'q>,
    ) -> Result<IsNull, BoxDynError> {
        match &self.0 {
            Some(v) => <&Value as sqlx::Encode<Postgres>>::encode_by_ref(&v, buf),
            None => Ok(IsNull::Yes),
        }
    }
}

impl Type<Postgres> for EmbedColumn {
    fn type_info() -> <Postgres as sqlx::Database>::TypeInfo {
        <Value as Type<Postgres>>::type_info()
    }
    fn compatible(ty: &<Postgres as sqlx::Database>::TypeInfo) -> bool {
        <Value as Type<Postgres>>::compatible(ty)
    }
}

#[derive(Debug)]
 pub struct PostRow {
    pub uri: String,
    pub did: String,
    pub handle: String,
    pub rkey: String,
    pub collection: String,
    pub text: String,
    pub cid: String,
    pub parent_uri: Option<String>,
    pub root_uri: Option<String>,
    pub embed: Option<Value>,
    pub langs: Option<Vec<String>>,
    pub created_at: chrono::DateTime<chrono::Utc>,
    /// Resolved author avatar CID from the `profiles` cache. Populated
    /// at `upsert_post` time so the PostCard can render an avatar
    /// without a per-row PDS round trip. NULL for users whose profile
    /// hasn't been pushed yet.
    pub avatar_cid: Option<String>,
}

impl PostRow {
    /// Extract a PostRow from the Jetstream / PDS event.
    /// `handle` is not present in Jetstream events and resolving it would
    /// require a PLC directory lookup; we store an empty placeholder and
    /// expect a future handle-sync job to backfill it.
    ///
    /// `embed` is captured verbatim from the record — the UI renders it
    /// by sniffing for `$type` (`app.bsky.embed.images` / `.external` /
    /// `.record`). Keeping it as raw JSON means we don't have to mirror
    /// every embed variant in Rust.
    ///
    /// `pds_handle` is the optional handle forwarded by the PDS through
    /// the `/internal/ingest-commit` payload. Local-PDS users have
    /// `did:key:` DIDs that no PLC directory can resolve, so the PDS is
    /// the only authoritative source for their handle. Pass `Some(handle)`
    /// when you have it; pass `None` (e.g. Jetstream path) and the
    /// `upsert_post` COALESCE guard ensures the empty value won't
    /// clobber a backfilled handle from `firehose::handle_identity`.
    pub fn from_record(
        did: &str,
        rkey: &str,
        collection: &str,
        cid: &str,
        record: &Value,
        pds_handle: Option<&str>,
    ) -> Self {
        let text = record
            .get("text")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();
        let created_at = parse_created_at(
            record.get("createdAt").and_then(|v| v.as_str()),
        );
        let reply = record.get("reply");
        let parent_uri = reply
            .and_then(|r| r.get("parent"))
            .and_then(|p| p.get("uri"))
            .and_then(|u| u.as_str())
            .map(str::to_string);
        let root_uri = reply
            .and_then(|r| r.get("root"))
            .and_then(|p| p.get("uri"))
            .and_then(|u| u.as_str())
            .map(str::to_string);
        let embed = record.get("embed").cloned().filter(|v| !v.is_null());
        let langs = record
            .get("langs")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_str().map(str::to_string))
                    .collect::<Vec<_>>()
            });
        let uri = format!("at://{did}/{collection}/{rkey}");
        let handle = pds_handle
            .map(|h| h.trim().to_string())
            .filter(|h| !h.is_empty())
            .unwrap_or_default();
        Self {
            uri,
            did: did.to_string(),
            handle,
            rkey: rkey.to_string(),
            collection: collection.to_string(),
            text,
            cid: cid.to_string(),
            parent_uri,
            root_uri,
            embed,
            langs,
            created_at,
            // Avatar CID is populated later by the upsert path via a
            // `SELECT avatar_cid FROM profiles WHERE did = $1` lookup,
            // so a freshly indexed post starts at None. (The lookup
            // happens in `upsert_post_with_avatar` below.)
            avatar_cid: None,
        }
    }
}

/// Insert or update a post row keyed by URI. Idempotent.
///
/// `row.avatar_cid` is filled in-place with the current profile-avatar
/// CID for the row's author (from the `profiles` cache, NULL if the
/// profile hasn't been pushed yet). The ON CONFLICT clause uses
/// `COALESCE(EXCLUDED, posts)` so a backfill on a re-indexed post
/// won't overwrite an avatar we already had.
///
/// IMPORTANT: `indexed_at` is NOT touched on conflict. We deliberately
/// preserve the original insert time so the `(indexed_at, uri)` keyset
/// pagination order is stable across Jetstream replays / PDS re-syncs.
pub async fn upsert_post(db: &PgPool, row: &mut PostRow) -> Result<()> {
    if row.avatar_cid.is_none() {
        // Look up the latest avatar CID for this author from the
        // profiles cache (populated by the PDS push path on profile
        // updates). NULL if the profile hasn't been ingested yet —
        // the post will display as the initial-letter avatar until
        // the user uploads one.
        row.avatar_cid = sqlx::query_scalar::<_, Option<String>>(
            "SELECT avatar_cid FROM profiles WHERE did = $1",
        )
        .bind(&row.did)
        .fetch_optional(db)
        .await?
        .flatten();
    }
    sqlx::query(
        r#"INSERT INTO posts
            (uri, did, handle, rkey, collection, text, cid,
             parent_uri, root_uri, embed, langs, created_at,
             avatar_cid)
           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
           ON CONFLICT (uri) DO UPDATE SET
              text       = EXCLUDED.text,
              cid        = EXCLUDED.cid,
              handle     = COALESCE(NULLIF(EXCLUDED.handle,''), posts.handle),
              parent_uri = EXCLUDED.parent_uri,
              root_uri   = EXCLUDED.root_uri,
              embed      = EXCLUDED.embed,
              langs      = EXCLUDED.langs,
              created_at = EXCLUDED.created_at,
              avatar_cid = COALESCE(EXCLUDED.avatar_cid, posts.avatar_cid)"#,
    )
    .bind(&row.uri)
    .bind(&row.did)
    .bind(&row.handle)
    .bind(&row.rkey)
    .bind(&row.collection)
    .bind(&row.text)
    .bind(&row.cid)
    .bind(&row.parent_uri)
    .bind(&row.root_uri)
    .bind(EmbedColumn(row.embed.clone()))
    .bind(&row.langs)
    .bind(row.created_at)
    .bind(&row.avatar_cid)
    .execute(db)
    .await?;
    Ok(())
}

/// Delete a post by URI. Idempotent (returns Ok even if the row is gone).
pub async fn delete_post(db: &PgPool, uri: &str) -> Result<()> {
    sqlx::query("DELETE FROM posts WHERE uri = $1")
        .bind(uri)
        .execute(db)
        .await?;
    Ok(())
}

// -- likes -----------------------------------------------------------------

/// Insert or ignore a like row keyed by URI. The partial unique index
/// on `(did, post_uri)` lets us reject double-likes at the DB level
/// instead of relying on caller discipline. We also maintain the
/// denormalized `posts.like_count` so the post endpoint doesn't have
/// to `COUNT(*)` over the entire `likes` table on every read.
pub async fn upsert_like(
    db: &PgPool,
    did: &str,
    rkey: &str,
    cid: Option<&str>,
    record: Option<&Value>,
) -> Result<()> {
    let uri = format!("at://{did}/app.bsky.feed.like/{rkey}");
    let _cid = cid.unwrap_or("");
    let post_uri = record
        .and_then(|r| r.get("subject"))
        .and_then(|s| s.get("uri"))
        .and_then(|u| u.as_str())
        .unwrap_or("")
        .to_string();
    let post_cid = record
        .and_then(|r| r.get("subject"))
        .and_then(|s| s.get("cid"))
        .and_then(|u| u.as_str())
        .unwrap_or("")
        .to_string();
    let created_at = parse_created_at(
        record.and_then(|r| r.get("createdAt")).and_then(|v| v.as_str()),
    );
    let mut tx = db.begin().await?;
    let inserted = sqlx::query(
        r#"INSERT INTO likes (uri, did, post_uri, post_cid, created_at)
           VALUES ($1, $2, $3, $4, $5)
           ON CONFLICT (did, post_uri) DO NOTHING
           RETURNING uri"#,
    )
    .bind(&uri)
    .bind(did)
    .bind(&post_uri)
    .bind(&post_cid)
    .bind(created_at)
    .fetch_optional(&mut *tx)
    .await?;
    if inserted.is_some() && !post_uri.is_empty() {
        // Increment the denormalized counter so the post endpoint can
        // serve like_count from `posts` without scanning `likes`.
        sqlx::query(
            "UPDATE posts SET like_count = like_count + 1 WHERE uri = $1",
        )
        .bind(&post_uri)
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(())
}

pub async fn delete_like(db: &PgPool, did: &str, rkey: &str) -> Result<()> {
    let uri = format!("at://{did}/app.bsky.feed.like/{rkey}");
    let mut tx = db.begin().await?;
    let row: Option<(String,)> = sqlx::query_as(
        "DELETE FROM likes WHERE uri = $1 RETURNING post_uri",
    )
    .bind(&uri)
    .fetch_optional(&mut *tx)
    .await?;
    if let Some((post_uri,)) = row {
        if !post_uri.is_empty() {
            sqlx::query(
                "UPDATE posts SET like_count = GREATEST(like_count - 1, 0) WHERE uri = $1",
            )
            .bind(&post_uri)
            .execute(&mut *tx)
            .await?;
        }
    }
    tx.commit().await?;
    Ok(())
}

// -- reposts ---------------------------------------------------------------

pub async fn upsert_repost(
    db: &PgPool,
    did: &str,
    rkey: &str,
    cid: Option<&str>,
    record: Option<&Value>,
) -> Result<()> {
    let uri = format!("at://{did}/app.bsky.feed.repost/{rkey}");
    let _cid = cid.unwrap_or("");
    let post_uri = record
        .and_then(|r| r.get("subject"))
        .and_then(|s| s.get("uri"))
        .and_then(|u| u.as_str())
        .unwrap_or("")
        .to_string();
    let post_cid = record
        .and_then(|r| r.get("subject"))
        .and_then(|s| s.get("cid"))
        .and_then(|u| u.as_str())
        .unwrap_or("")
        .to_string();
    let created_at = parse_created_at(
        record.and_then(|r| r.get("createdAt")).and_then(|v| v.as_str()),
    );
    let mut tx = db.begin().await?;
    let inserted = sqlx::query(
        r#"INSERT INTO reposts (uri, did, post_uri, post_cid, created_at)
           VALUES ($1, $2, $3, $4, $5)
           ON CONFLICT (did, post_uri) DO NOTHING
           RETURNING uri"#,
    )
    .bind(&uri)
    .bind(did)
    .bind(&post_uri)
    .bind(&post_cid)
    .bind(created_at)
    .fetch_optional(&mut *tx)
    .await?;
    if inserted.is_some() && !post_uri.is_empty() {
        sqlx::query(
            "UPDATE posts SET repost_count = repost_count + 1 WHERE uri = $1",
        )
        .bind(&post_uri)
        .execute(&mut *tx)
        .await?;
    }
    tx.commit().await?;
    Ok(())
}

pub async fn delete_repost(db: &PgPool, did: &str, rkey: &str) -> Result<()> {
    let uri = format!("at://{did}/app.bsky.feed.repost/{rkey}");
    let mut tx = db.begin().await?;
    let row: Option<(String,)> = sqlx::query_as(
        "DELETE FROM reposts WHERE uri = $1 RETURNING post_uri",
    )
    .bind(&uri)
    .fetch_optional(&mut *tx)
    .await?;
    if let Some((post_uri,)) = row {
        if !post_uri.is_empty() {
            sqlx::query(
                "UPDATE posts SET repost_count = GREATEST(repost_count - 1, 0) WHERE uri = $1",
            )
            .bind(&post_uri)
            .execute(&mut *tx)
            .await?;
        }
    }
    tx.commit().await?;
    Ok(())
}

// -- follows ---------------------------------------------------------------

pub async fn upsert_follow(
    db: &PgPool,
    follower_did: &str,
    subject_did: &str,
    record: Option<&Value>,
) -> Result<()> {
    let created_at = parse_created_at(
        record.and_then(|r| r.get("createdAt")).and_then(|v| v.as_str()),
    );
    sqlx::query(
        r#"INSERT INTO follows (follower_did, subject_did, created_at)
           VALUES ($1, $2, $3)
           ON CONFLICT (follower_did, subject_did) DO UPDATE SET
             created_at = EXCLUDED.created_at,
             indexed_at = now()"#,
    )
    .bind(follower_did)
    .bind(subject_did)
    .bind(created_at)
    .execute(db)
    .await?;
    Ok(())
}

pub async fn delete_follow(
    db: &PgPool,
    follower_did: &str,
    subject_did: &str,
) -> Result<()> {
    sqlx::query(
        "DELETE FROM follows WHERE follower_did = $1 AND subject_did = $2",
    )
    .bind(follower_did)
    .bind(subject_did)
    .execute(db)
    .await?;
    Ok(())
}

/// Extract the subject DID from a follow record (`{ "subject": "did:..."}`).
pub fn follow_subject_did(record: Option<&Value>) -> Option<String> {
    record?
        .get("subject")?
        .as_str()
        .map(|s| s.to_string())
}

// -- test harness ----------------------------------------------------------

#[cfg(test)]
async fn try_test_db() -> Option<PgPool> {
    let url = std::env::var("DATABASE_URL_APPVIEW").ok()?;
    match timeout(Duration::from_secs(2), sqlx::PgPool::connect(&url)).await {
        Ok(Ok(pool)) => match sqlx::migrate!("../../migrations/appview")
            .run(&pool)
            .await
        {
            Ok(()) => Some(pool),
            Err(_) => None,
        },
        _ => None,
    }
}

/// Route a `JetstreamEvent::kind == "commit"` event to the right upsert /
/// delete function. Returns `Ok(false)` if the event was recognized but
/// not actionable (e.g. unsupported collection), `Ok(true)` if it was
/// applied, `Err(_)` if DB write failed.
pub async fn apply_commit(
    db: &PgPool,
    ev: &JetstreamEvent,
) -> Result<bool> {
    let commit = match &ev.commit {
        Some(c) => c,
        None => return Ok(false),
    };
    let collection = match extract_collection(commit) {
        Some(c) => c,
        None => return Ok(false),
    };
    let ops = commit_op_from_jetstream_value(commit);
    if ops.is_empty() {
        return Ok(false);
    }
    let mut applied = false;
    for op in &ops {
        match collection.as_str() {
            "app.twi.post" | "app.bsky.feed.post" => {
                if op.action == "create" {
                    let rkey = op
                        .rkey
                        .clone()
                        .or_else(|| {
                            op.path
                                .as_deref()
                                .and_then(|p| p.rsplit('/').next().map(str::to_string))
                        })
                        .ok_or_else(|| {
                            anyhow!("create op missing rkey for post")
                        })?;
                    let cid = op.cid.clone().unwrap_or_default();
                    let record = op.record.clone().unwrap_or(Value::Null);
                    // Jetstream `commit` events don't carry the
                    // handle — leave it empty so the upsert
                    // COALESCE guard preserves the row's existing
                    // (or backfilled-from-identity) handle.
                    let mut row = PostRow::from_record(
                        &ev.did,
                        &rkey,
                        &collection,
                        &cid,
                        &record,
                        None,
                    );
                    upsert_post(db, &mut row).await?;
                    applied = true;
                } else if op.action == "delete" {
                    let rkey = op
                        .rkey
                        .clone()
                        .or_else(|| {
                            op.path
                                .as_deref()
                                .and_then(|p| p.rsplit('/').next().map(str::to_string))
                        })
                        .ok_or_else(|| {
                            anyhow!("delete op missing rkey for post")
                        })?;
                    let uri = format!(
                        "at://{}/{}/{}",
                        ev.did, collection, rkey
                    );
                    delete_post(db, &uri).await?;
                    applied = true;
                }
            }
            "app.bsky.feed.like" => {
                let rkey = op
                    .rkey
                    .clone()
                    .or_else(|| {
                        op.path
                            .as_deref()
                            .and_then(|p| p.rsplit('/').next().map(str::to_string))
                    });
                let rkey = match rkey {
                    Some(r) => r,
                    None => {
                        tracing::warn!("like op missing rkey; skipping");
                        continue;
                    }
                };
                if op.action == "create" {
                    upsert_like(
                        db,
                        &ev.did,
                        &rkey,
                        op.cid.as_deref(),
                        op.record.as_ref(),
                    )
                    .await?;
                } else if op.action == "delete" {
                    delete_like(db, &ev.did, &rkey).await?;
                }
                applied = true;
            }
            "app.bsky.feed.repost" => {
                let rkey = op
                    .rkey
                    .clone()
                    .or_else(|| {
                        op.path
                            .as_deref()
                            .and_then(|p| p.rsplit('/').next().map(str::to_string))
                    });
                let rkey = match rkey {
                    Some(r) => r,
                    None => {
                        tracing::warn!("repost op missing rkey; skipping");
                        continue;
                    }
                };
                if op.action == "create" {
                    upsert_repost(
                        db,
                        &ev.did,
                        &rkey,
                        op.cid.as_deref(),
                        op.record.as_ref(),
                    )
                    .await?;
                } else if op.action == "delete" {
                    delete_repost(db, &ev.did, &rkey).await?;
                }
                applied = true;
            }
            "app.bsky.graph.follow" => {
                let subject_did = match op.action.as_str() {
                    "create" => match follow_subject_did(op.record.as_ref()) {
                        Some(s) => s,
                        None => {
                            tracing::warn!(
                                "follow create missing subject; skipping"
                            );
                            continue;
                        }
                    },
                    "delete" => {
                        // Jetstream delete on follows carries no record
                        // value, so we can't know which subject was
                        // unfollowed. The PDS-driven internal ingest path
                        // handles this — it knows the subject from its
                        // own snapshot.
                        tracing::warn!(
                            "follow delete via Jetstream lacks subject; \
                             route through /internal/ingest-commit instead"
                        );
                        continue;
                    }
                    _ => continue,
                };
                if op.action == "create" {
                    upsert_follow(
                        db,
                        &ev.did,
                        &subject_did,
                        op.record.as_ref(),
                    )
                    .await?;
                } else if op.action == "delete" {
                    delete_follow(db, &ev.did, &subject_did).await?;
                }
                applied = true;
            }
            "app.bsky.actor.profile" => {
                // Jetstream carries profile records as plain
                // commit ops (no separate collection). We treat
                // any rkey — usually `self`, but spec allows
                // rkey-rotation — as the user's authoritative
                // profile and upsert into the `profiles` cache.
                //
                // The Jetstream `commit` envelope doesn't carry
                // the handle; we look it up from the `posts`
                // table (backfilled there by the `identity`
                // event stream). Empty is fine — the next
                // handle_sync pass will populate it.
                if op.action == "create" {
                    let record = match op.record.clone() {
                        Some(r) if !r.is_null() => r,
                        _ => continue,
                    };
                    let handle: String = sqlx::query_scalar(
                        "SELECT handle FROM posts \
                          WHERE did = $1 AND handle <> '' \
                          ORDER BY indexed_at DESC LIMIT 1",
                    )
                    .bind(&ev.did)
                    .fetch_optional(db)
                    .await?
                    .unwrap_or_default();
                    upsert_profile(db, &ev.did, &handle, &record).await?;
                    applied = true;
                } else if op.action == "delete" {
                    sqlx::query("DELETE FROM profiles WHERE did = $1")
                        .bind(&ev.did)
                        .execute(db)
                        .await?;
                    applied = true;
                }
            }
            _ => {
                // Unrecognised collection — ignore (may happen when Jetstream
                // sends something we didn't subscribe to).
            }
        }
    }
    Ok(applied)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn commit_op_from_jetstream_value_single_create() {
        let commit = json!({
            "operation": "create",
            "collection": "app.bsky.feed.post",
            "rkey": "3k2abc",
            "cid": "bafyreicid",
            "record": {"text": "hi", "createdAt": "2026-07-01T12:00:00Z"}
        });
        let ops = commit_op_from_jetstream_value(&commit);
        assert_eq!(ops.len(), 1);
        assert_eq!(ops[0].action, "create");
        assert_eq!(ops[0].rkey.as_deref(), Some("3k2abc"));
        assert_eq!(ops[0].cid.as_deref(), Some("bafyreicid"));
        assert_eq!(
            ops[0].record.as_ref().unwrap()["text"].as_str(),
            Some("hi")
        );
    }

    #[test]
    fn commit_op_from_jetstream_value_batched_ops() {
        let commit = json!({
            "ops": [
                {"action": "create", "collection": "app.bsky.feed.like", "rkey": "1",
                 "cid": "c1", "record": {"subject": {"uri": "at://x/y/z", "cid": "cz"},
                                           "createdAt": "2026-07-01T12:00:00Z"}},
                {"action": "delete", "collection": "app.bsky.feed.like", "rkey": "0"}
            ]
        });
        let ops = commit_op_from_jetstream_value(&commit);
        assert_eq!(ops.len(), 2);
        assert_eq!(ops[0].action, "create");
        assert_eq!(ops[0].rkey.as_deref(), Some("1"));
        assert_eq!(ops[1].action, "delete");
        assert_eq!(ops[1].rkey.as_deref(), Some("0"));
    }

    #[test]
    fn commit_op_from_jetstream_value_missing_action_returns_empty() {
        let commit = json!({ "collection": "app.bsky.feed.post", "rkey": "x" });
        let ops = commit_op_from_jetstream_value(&commit);
        assert!(ops.is_empty());
    }

    #[test]
    fn extract_collection_top_level_and_batched() {
        let top = json!({"operation":"create","collection":"app.twi.post","rkey":"r"});
        assert_eq!(extract_collection(&top).as_deref(), Some("app.twi.post"));

        let batched = json!({"ops": [{"collection":"app.bsky.feed.like"}]});
        assert_eq!(
            extract_collection(&batched).as_deref(),
            Some("app.bsky.feed.like")
        );

        let none = json!({});
        assert!(extract_collection(&none).is_none());
    }

    #[test]
    fn build_uri_uses_rkey_or_path() {
        let op_rkey = CommitOp {
            action: "create".into(),
            rkey: Some("rk".into()),
            path: None,
            cid: None,
            record: None,
        };
        assert_eq!(
            build_uri("did:plc:a", "app.twi.post", &op_rkey).as_deref(),
            Some("at://did:plc:a/app.twi.post/rk")
        );

        let op_path = CommitOp {
            action: "delete".into(),
            rkey: None,
            path: Some("app.twi.post/last".into()),
            cid: None,
            record: None,
        };
        assert_eq!(
            build_uri("did:plc:a", "app.twi.post", &op_path).as_deref(),
            Some("at://did:plc:a/app.twi.post/last")
        );
    }

    #[test]
    fn follow_subject_did_extracts() {
        let rec = json!({ "subject": "did:plc:b", "createdAt": "2026-01-01T00:00:00Z" });
        assert_eq!(
            follow_subject_did(Some(&rec)).as_deref(),
            Some("did:plc:b")
        );
        assert!(follow_subject_did(None).is_none());
    }

    #[test]
    fn from_record_captures_embed() {
        let rec = json!({
            "text": "look at this",
            "createdAt": "2026-07-01T12:00:00Z",
            "embed": {
                "$type": "app.bsky.embed.images",
                "images": [
                    {"alt": "a cat", "image": {"$type": "blob", "ref": {"$link": "bafy"}}}
                ]
            }
        });
        let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
        let embed = row.embed.expect("embed must be captured");
        assert_eq!(embed["$type"], "app.bsky.embed.images");
        assert_eq!(embed["images"][0]["alt"], "a cat");
    }

    #[test]
    fn from_record_captures_external_embed() {
        let rec = json!({
            "text": "see link",
            "createdAt": "2026-07-01T12:00:00Z",
            "embed": {
                "$type": "app.bsky.embed.external",
                "external": {
                    "uri": "https://example.com",
                    "title": "Example",
                    "description": "An example"
                }
            }
        });
        let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
        let embed = row.embed.expect("embed must be captured");
        assert_eq!(embed["$type"], "app.bsky.embed.external");
        assert_eq!(embed["external"]["uri"], "https://example.com");
    }

    #[test]
    fn from_record_omits_embed_when_missing() {
        let rec = json!({
            "text": "no embed here",
            "createdAt": "2026-07-01T12:00:00Z"
        });
        let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
        assert!(row.embed.is_none());
    }

    #[test]
    fn from_record_captures_reply_uris() {
        let rec = json!({
            "text": "a reply",
            "createdAt": "2026-07-01T12:00:00Z",
            "reply": {
                "parent": {"uri": "at://did:plc:b/app.twi.post/p", "cid": "cp"},
                "root":   {"uri": "at://did:plc:b/app.twi.post/r", "cid": "cr"}
            }
        });
        let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
        assert_eq!(row.parent_uri.as_deref(), Some("at://did:plc:b/app.twi.post/p"));
        assert_eq!(row.root_uri.as_deref(), Some("at://did:plc:b/app.twi.post/r"));
    }

    fn fake_event(collection: &str, action: &str, rkey: &str) -> JetstreamEvent {
        let commit = if action == "create" {
            json!({
                "operation": action,
                "collection": collection,
                "rkey": rkey,
                "cid": "bafyreicid",
                "record": {
                    "text": "hi there",
                    "createdAt": "2026-07-01T12:00:00Z"
                }
            })
        } else {
            json!({
                "operation": action,
                "collection": collection,
                "rkey": rkey
            })
        };
        JetstreamEvent {
            did: "did:plc:test".into(),
            time_us: 1_700_000_000_000_000,
            kind: "commit".into(),
            commit: Some(commit),
            identity: None,
            account: None,
        }
    }

    #[tokio::test]
    async fn upsert_and_delete_post_round_trip() {
        let Some(db) = try_test_db().await else {
            eprintln!("appview DB unavailable; skipping");
            return;
        };
        // Clean any previous row in this test's slot.
        let _ = sqlx::query("DELETE FROM posts WHERE did = 'did:plc:test'")
            .execute(&db)
            .await
            .unwrap();

        let ev = fake_event("app.twi.post", "create", "abc123");
        let applied = apply_commit(&db, &ev).await.unwrap();
        assert!(applied);
        let (uri, text): (String, String) = sqlx::query_as(
            "SELECT uri, text FROM posts WHERE uri = $1",
        )
        .bind("at://did:plc:test/app.twi.post/abc123")
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(uri, "at://did:plc:test/app.twi.post/abc123");
        assert_eq!(text, "hi there");

        // Verify that the (currently null) embed column is readable —
        // we want to fail loud if the column is missing from the
        // schema rather than silently falling back to text-only posts.
        let embed_json: Option<serde_json::Value> = sqlx::query_scalar(
            "SELECT embed FROM posts WHERE uri = $1",
        )
        .bind("at://did:plc:test/app.twi.post/abc123")
        .fetch_one(&db)
        .await
        .unwrap();
        assert!(
            embed_json.is_none(),
            "embed should be null for plain-text posts, got: {embed_json:?}"
        );

        // Apply again — must be idempotent (no error, still one row).
        let _ = apply_commit(&db, &ev).await.unwrap();
        let count: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM posts WHERE uri = $1",
        )
        .bind("at://did:plc:test/app.twi.post/abc123")
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(count, 1);

        // Delete it.
        let del = fake_event("app.twi.post", "delete", "abc123");
        apply_commit(&db, &del).await.unwrap();
        let remaining: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM posts WHERE uri = $1",
        )
        .bind("at://did:plc:test/app.twi.post/abc123")
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(remaining, 0);
    }

    /// Seed a post with an `app.bsky.embed.images` embed and confirm
    /// the JSONB column round-trips the embed object back as JSON.
    #[tokio::test]
    async fn upsert_post_stores_embed_jsonb() {
        let Some(db) = try_test_db().await else {
            eprintln!("appview DB unavailable; skipping");
            return;
        };
        // Clean any previous row in this test's slot.
        let _ = sqlx::query("DELETE FROM posts WHERE did = 'did:plc:embed'")
            .execute(&db)
            .await
            .unwrap();

        let record = json!({
            "text": "with an image",
            "createdAt": "2026-07-01T12:00:00Z",
            "embed": {
                "$type": "app.bsky.embed.images",
                "images": [
                    {"alt": "first alt", "image": {"$type": "blob", "ref": {"$link": "bafy1"}}},
                    {"alt": "second alt", "image": {"$type": "blob", "ref": {"$link": "bafy2"}}}
                ]
            }
        });
        let mut row = PostRow::from_record(
            "did:plc:embed",
            "embedkey",
            "app.twi.post",
            "cid-embed",
            &record,
            None,
        );
        upsert_post(&db, &mut row).await.unwrap();

        let embed: serde_json::Value = sqlx::query_scalar(
            "SELECT embed FROM posts WHERE uri = $1",
        )
        .bind("at://did:plc:embed/app.twi.post/embedkey")
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(embed["$type"], "app.bsky.embed.images");
        let imgs = embed["images"].as_array().expect("images array");
        assert_eq!(imgs.len(), 2);
        assert_eq!(imgs[0]["alt"], "first alt");
        assert_eq!(imgs[1]["alt"], "second alt");

        // Cleanup.
        let _ = sqlx::query("DELETE FROM posts WHERE did = 'did:plc:embed'")
            .execute(&db)
            .await;
    }

    #[tokio::test]
    async fn upsert_and_delete_follow_round_trip() {
        let Some(db) = try_test_db().await else {
            eprintln!("appview DB unavailable; skipping");
            return;
        };
        // Wipe test data so the test is order-independent.
        let _ = sqlx::query(
            "DELETE FROM follows WHERE follower_did = $1 AND subject_did = $2",
        )
        .bind("did:plc:test")
        .bind("did:plc:b")
        .execute(&db)
        .await
        .unwrap();

        let ev = JetstreamEvent {
            did: "did:plc:test".into(),
            time_us: 1_700_000_000_000_000,
            kind: "commit".into(),
            commit: Some(json!({
                "operation": "create",
                "collection": "app.bsky.graph.follow",
                "rkey": "frk1",
                "cid": "bafyfollow",
                "record": {
                    "subject": "did:plc:b",
                    "createdAt": "2026-01-01T00:00:00Z"
                }
            })),
            identity: None,
            account: None,
        };
        apply_commit(&db, &ev).await.unwrap();
        let (count,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM follows WHERE follower_did = $1 AND subject_did = $2",
        )
        .bind("did:plc:test")
        .bind("did:plc:b")
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(count, 1);

        // Idempotent re-apply.
        apply_commit(&db, &ev).await.unwrap();
        let (count,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM follows WHERE follower_did = $1 AND subject_did = $2",
        )
        .bind("did:plc:test")
        .bind("did:plc:b")
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(count, 1);

        // Delete via the internal API (not via Jetstream — Jetstream
        // delete on follows doesn't carry the subject).
        delete_follow(&db, "did:plc:test", "did:plc:b").await.unwrap();
        let (count,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM follows WHERE follower_did = $1 AND subject_did = $2",
        )
        .bind("did:plc:test")
        .bind("did:plc:b")
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(count, 0);
    }
}

/// Backfill the `posts.handle` column for every row belonging to
/// `did`. Used by the Jetstream `identity` handler when Jetstream
/// tells us a DID's handle has changed — every existing post row
/// needs the new value.
///
/// **Race-safety**: this only writes if the row's current handle is
/// empty OR doesn't match, so concurrent PDS pushes (which carry
/// the same handle) and concurrent identity replays don't fight.
/// The `WHERE handle IS DISTINCT FROM $1` makes the update a
/// no-op when the value is already correct, which Postgres treats
/// cheaply.
///
/// Returns the number of rows updated.
pub async fn backfill_handle(
    db: &PgPool,
    did: &str,
    new_handle: &str,
) -> Result<u64> {
    let res = sqlx::query(
        "UPDATE posts SET handle = $1 WHERE did = $2 AND handle IS DISTINCT FROM $1",
    )
    .bind(new_handle)
    .bind(did)
    .execute(db)
    .await?;
    Ok(res.rows_affected())
}

/// Upsert a `profiles` row for `did`. The caller has just ingested
/// the profile record body (decoded CBOR), and the denormalised
/// counts are computed here (a single `SELECT COUNT(*)` over each
/// side-table — cheap with the existing PK indexes on `posts.did` and
/// `follows.{follower,subject}_did`).
pub async fn upsert_profile(
    db: &PgPool,
    did: &str,
    handle: &str,
    record: &Value,
) -> Result<()> {
    let display_name = record
        .get("displayName")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let description = record
        .get("description")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let avatar_cid = blob_link_of(record, "avatar");
    let banner_cid = blob_link_of(record, "banner");

    // Denormalised counts. Cheap with the existing PKs.
    let post_count: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM posts WHERE did = $1 \
         AND collection IN ('app.twi.post','app.bsky.feed.post')",
    )
    .bind(did)
    .fetch_one(db)
    .await?;
    let follower_count: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM follows WHERE subject_did = $1",
    )
    .bind(did)
    .fetch_one(db)
    .await?;
    let following_count: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM follows WHERE follower_did = $1",
    )
    .bind(did)
    .fetch_one(db)
    .await?;

    sqlx::query(
        r#"INSERT INTO profiles
              (did, handle, display_name, description,
               avatar_cid, banner_cid,
               post_count, follower_count, following_count, indexed_at)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now())
           ON CONFLICT (did) DO UPDATE SET
              handle        = EXCLUDED.handle,
              display_name  = EXCLUDED.display_name,
              description   = EXCLUDED.description,
              avatar_cid    = EXCLUDED.avatar_cid,
              banner_cid    = EXCLUDED.banner_cid,
              post_count    = EXCLUDED.post_count,
              follower_count= EXCLUDED.follower_count,
              following_count= EXCLUDED.following_count,
              indexed_at    = now()"#,
    )
    .bind(did)
    .bind(handle)
    .bind(display_name)
    .bind(description)
    .bind(avatar_cid)
    .bind(banner_cid)
    .bind(post_count)
    .bind(follower_count)
    .bind(following_count)
    .execute(db)
    .await?;
    Ok(())
}

/// Pull a blob-ref `$link` out of a profile record's field.
/// Accepts both the modern shape
/// (`{ $type: "blob", ref: { $link: "..." } }`)
/// and the legacy shape (`{ $link: "..." }`) for robustness.
fn blob_link_of(record: &Value, field: &str) -> Option<String> {
    let v = record.get(field)?;
    // Try `ref.$link` first, then flat `$link`.
    if let Some(link) = v.get("ref").and_then(|r| r.get("$link")).and_then(|s| s.as_str()) {
        return Some(link.to_string());
    }
    v.get("$link").and_then(|s| s.as_str()).map(str::to_string)
}

#[cfg(test)]
mod profile_tests {
    use super::*;
    use serde_json::json;

    /// Open the appview DB used by integration tests, running
    /// migrations first. Returns `None` when no DB is reachable so
    /// the test can `eprintln!` and bail (no panic).
    async fn try_test_db() -> Option<PgPool> {
        let url = std::env::var("DATABASE_URL_APPVIEW").ok()?;
        match timeout(
            Duration::from_secs(2),
            sqlx::PgPool::connect(&url),
        )
        .await
        {
            Ok(Ok(pool)) => match sqlx::migrate!("../../migrations/appview").run(&pool).await {
                Ok(()) => Some(pool),
                Err(_) => None,
            },
            _ => None,
        }
    }

    #[test]
    fn blob_link_of_modern_shape() {
        let rec = json!({
            "avatar": {
                "$type": "blob",
                "ref":    { "$link": "bafyavatar" },
                "mimeType": "image/png",
                "size": 1234
            }
        });
        assert_eq!(blob_link_of(&rec, "avatar").as_deref(), Some("bafyavatar"));
    }

    #[test]
    fn blob_link_of_legacy_flat_link() {
        let rec = json!({ "banner": { "$link": "bafybanner" } });
        assert_eq!(blob_link_of(&rec, "banner").as_deref(), Some("bafybanner"));
    }

    #[test]
    fn blob_link_of_missing_field() {
        let rec = json!({ "displayName": "x" });
        assert_eq!(blob_link_of(&rec, "avatar"), None);
    }

    #[tokio::test]
    async fn upsert_profile_round_trip() {
        let Some(db) = try_test_db().await else {
            eprintln!("appview DB unavailable; skipping");
            return;
        };
        // Use a unique DID per test run so we don't collide with the
        // migration backfill (which seeded a row for every distinct
        // DID in `posts`).
        let did = format!(
            "did:plc:profile_test_{}",
            uuid::Uuid::new_v4().simple()
        );
        let handle = format!("user.{}.test", uuid::Uuid::new_v4().simple());

        // Seed a couple of posts so post_count is non-zero.
        for rkey in &["p1", "p2"] {
            sqlx::query(
                r#"INSERT INTO posts
                     (uri, did, handle, rkey, collection, text, cid,
                      parent_uri, root_uri, langs, created_at)
                   VALUES ($1,$2,$3,$4,'app.twi.post','seed','bafy',NULL,NULL,NULL, now())
                   ON CONFLICT (uri) DO NOTHING"#,
            )
            .bind(format!("at://{did}/app.twi.post/{rkey}"))
            .bind(&did)
            .bind(&handle)
            .bind(rkey)
            .execute(&db)
            .await
            .unwrap();
        }

        let record = json!({
            "displayName": "Alice",
            "description": "tester",
            "avatar": { "$type": "blob", "ref": { "$link": "bafyavatar" } },
            "banner": { "$type": "blob", "ref": { "$link": "bafybanner" } }
        });
        upsert_profile(&db, &did, &handle, &record).await.unwrap();

        let row: (
            String,             // handle
            Option<String>,     // display_name
            Option<String>,     // description
            Option<String>,     // avatar_cid
            Option<String>,     // banner_cid
            i64,                // post_count
            i64,                // follower_count
            i64,                // following_count
        ) = sqlx::query_as(
            "SELECT handle, display_name, description, avatar_cid, banner_cid, \
                    post_count, follower_count, following_count \
               FROM profiles WHERE did = $1",
        )
        .bind(&did)
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(row.0, handle);
        assert_eq!(row.1.as_deref(), Some("Alice"));
        assert_eq!(row.2.as_deref(), Some("tester"));
        assert_eq!(row.3.as_deref(), Some("bafyavatar"));
        assert_eq!(row.4.as_deref(), Some("bafybanner"));
        assert_eq!(row.5, 2, "post_count must reflect seeded posts");

        // Update: change display name, drop banner — verify replace
        // semantics (NULL fields overwrite, not coalesce).
        let record2 = json!({ "displayName": "Alice 2" });
        upsert_profile(&db, &did, &handle, &record2).await.unwrap();
        let name: Option<String> = sqlx::query_scalar(
            "SELECT display_name FROM profiles WHERE did = $1",
        )
        .bind(&did)
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(name.as_deref(), Some("Alice 2"));

        // Cleanup.
        let _ = sqlx::query("DELETE FROM profiles WHERE did = $1")
            .bind(&did)
            .execute(&db)
            .await;
        let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
            .bind(&did)
            .execute(&db)
            .await;
    }

    /// `apply_commit` must dispatch an `app.bsky.actor.profile`
    /// create op into the `profiles` cache (this is the path Jetstream
    /// uses for third-party PDS authors).
    #[tokio::test]
    async fn apply_commit_indexes_profile_create() {
        let Some(db) = try_test_db().await else {
            eprintln!("appview DB unavailable; skipping");
            return;
        };
        let did = format!(
            "did:plc:profile_commit_{}",
            uuid::Uuid::new_v4().simple()
        );
        let _ = sqlx::query("DELETE FROM profiles WHERE did = $1")
            .bind(&did)
            .execute(&db)
            .await;

        // Seed a post so the indexer can find a known handle.
        sqlx::query(
            r#"INSERT INTO posts
                 (uri, did, handle, rkey, collection, text, cid,
                  parent_uri, root_uri, langs, created_at)
               VALUES ($1,$2,$3,'seed','app.twi.post','hi','bafy',NULL,NULL,NULL, now())
               ON CONFLICT (uri) DO NOTHING"#,
        )
        .bind(format!("at://{did}/app.twi.post/seed"))
        .bind(&did)
        .bind("alice.test")
        .execute(&db)
        .await
        .unwrap();

        let commit = json!({
            "operation": "create",
            "collection": "app.bsky.actor.profile",
            "rkey": "self",
            "cid": "bafyprofilecid",
            "record": {
                "displayName": "Alice",
                "description": "hello",
                "avatar": { "$type": "blob", "ref": { "$link": "bafyavatar" } }
            }
        });
        let ev = JetstreamEvent {
            did: did.clone(),
            time_us: 1_700_000_000_000_000,
            kind: "commit".into(),
            commit: Some(commit),
            identity: None,
            account: None,
        };
        let applied = apply_commit(&db, &ev).await.unwrap();
        assert!(applied);

        let row: (Option<String>, Option<String>, Option<String>) = sqlx::query_as(
            "SELECT display_name, description, avatar_cid \
               FROM profiles WHERE did = $1",
        )
        .bind(&did)
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(row.0.as_deref(), Some("Alice"));
        assert_eq!(row.1.as_deref(), Some("hello"));
        assert_eq!(row.2.as_deref(), Some("bafyavatar"));

        // Delete op should wipe the row.
        let del = json!({
            "operation": "delete",
            "collection": "app.bsky.actor.profile",
            "rkey": "self"
        });
        let ev_del = JetstreamEvent {
            did: did.clone(),
            time_us: 1_700_000_001_000_000,
            kind: "commit".into(),
            commit: Some(del),
            identity: None,
            account: None,
        };
        apply_commit(&db, &ev_del).await.unwrap();
        let remaining: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM profiles WHERE did = $1",
        )
        .bind(&did)
        .fetch_one(&db)
        .await
        .unwrap();
        assert_eq!(remaining, 0, "delete op must remove profile row");

        // Cleanup.
        let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
            .bind(&did)
            .execute(&db)
            .await;
    }
}