//! SQLite read queries and generated-view builders for the read endpoints.
//!
//! Records come from the `record_json` column (written by the projection); the
//! required `record` field on thing/model/part views is built from it. Author
//! handles come from `identities` (refreshed by identity events) and profile
//! display fields from `profiles`. Authored image/blob fields are deserialized
//! through generated lexicon types and resolved to Bluesky CDN URLs in appview
//! response fields; no fake CDN/media data is synthesized.
//!
//! Cursor contract: clients treat `cursor` as opaque.
//! - Chronological feeds (`getAuthorThings`, `getFeed?algorithm=recent`) keyset
//!   on the record's TID `rkey` (descending); the returned cursor is the last
//!   item's `rkey`, and the next page uses `rkey < cursor` (a NULL cursor — no
//!   cursor — returns the first page).
//! - `getFeed?algorithm=hot` uses an opaque versioned keyset cursor over a hot
//!   ranking snapshot. The cursor pins the ranking timestamp and last returned
//!   `(score, rkey, uri)` boundary so later pages use the same total ordering.
//! - `searchThings` uses a stringified offset cursor for BM25 results; a
//!   non-numeric cursor is rejected with 400.

use std::cmp::Ordering;

use jacquard::identity::PublicResolver;
use jacquard::identity::resolver::IdentityResolver;
use jacquard::oauth::client::OAuthSession;
use jacquard_common::deps::smol_str::SmolStr;
use jacquard_common::types::cid::Cid;
use jacquard_common::types::string::{AtUri, Datetime, Did, Handle, UriValue};
use jacquard_common::types::{blob::BlobRef, value::Data};
use polymodel_api::space_polymodel::actor::ProfileListView;
use polymodel_api::space_polymodel::library::{
    Actor, ActorViewerState, FeedItem, FeedView, File, Image, ImageView, ModelView, PartView,
    ThingView, ThingViewBasic, ViewerState, model::Model, part::Part, thing::Thing,
};
use sqlx::SqlitePool;

use super::error::{AppError, AppResult, db, internal, invalid_request, not_found};
use super::state::AppState;
use crate::oauth::SqliteAuthStore;

const HOT_POSITIVE_BATCH: i64 = 256;

// ---------------------------------------------------------------------------
// value-construction helpers
// ---------------------------------------------------------------------------

fn s(v: String) -> SmolStr {
    SmolStr::from(v)
}

fn at_uri(uri: &str) -> AppResult<AtUri> {
    AtUri::new_owned(uri).map_err(|e| invalid_request(format!("invalid at-uri {uri:?}: {e}")))
}

fn cid_of(cid: &str) -> AppResult<Cid> {
    Cid::new_owned(cid.as_bytes()).map_err(|e| internal(format!("invalid cid {cid:?}: {e}")))
}

fn did_of(did: &str) -> AppResult<Did> {
    Did::new_owned(did).map_err(|e| internal(format!("invalid did {did:?}: {e}")))
}
fn handle_of(handle: String) -> AppResult<Handle> {
    Handle::new_owned(handle).map_err(|e| internal(format!("invalid handle: {e}")))
}
fn datetime_nanos(nanos: i64) -> Datetime {
    let dt = chrono::DateTime::<chrono::Utc>::from_timestamp_nanos(nanos);
    Datetime::new(dt.fixed_offset())
}

/// Build the required `record: Data` field from a stored `record_json` blob.
fn record_data(record_json: Option<&str>) -> AppResult<Data> {
    let raw = record_json.ok_or_else(|| internal("content row missing record_json"))?;
    serde_json::from_str::<Data>(raw).map_err(|e| internal(format!("record_json parse: {e}")))
}

fn file_data(file_json: Option<&str>) -> AppResult<File> {
    serde_json::from_str(file_json.ok_or_else(|| internal("part row missing file_json"))?)
        .map_err(|e| internal(format!("file_json parse: {e}")))
}
fn smol_vec(json: Option<&str>) -> Option<Vec<SmolStr>> {
    let arr: Vec<String> = serde_json::from_str(json?).ok()?;
    let v: Vec<SmolStr> = arr.into_iter().map(s).collect();
    if v.is_empty() { None } else { Some(v) }
}

fn uri_vec(json: Option<&str>) -> Option<Vec<UriValue>> {
    let arr: Vec<String> = serde_json::from_str(json?).ok()?;
    arr.into_iter()
        .map(|u| UriValue::new_owned(u).ok())
        .collect::<Option<Vec<_>>>()
        .filter(|v| !v.is_empty())
}
fn blob_cdn_url(repo_did: &str, blob_ref: &BlobRef) -> AppResult<UriValue> {
    let cid = blob_ref.blob().r#ref.as_str();
    UriValue::new_owned(format!(
        "https://cdn.bsky.app/img/feed_thumbnail/plain/{repo_did}/{cid}@jpeg"
    ))
    .map_err(|e| internal(format!("invalid generated image CDN URL: {e}")))
}

fn avatar_cdn_url(repo_did: &str, blob_ref: &BlobRef) -> AppResult<UriValue> {
    let cid = blob_ref.blob().r#ref.as_str();
    UriValue::new_owned(format!(
        "https://cdn.bsky.app/img/avatar/plain/{repo_did}/{cid}@jpeg"
    ))
    .map_err(|e| internal(format!("invalid generated avatar CDN URL: {e}")))
}

fn image_view(repo_did: &str, image: Image) -> AppResult<ImageView> {
    let url = blob_cdn_url(repo_did, &image.image)?;
    Ok(ImageView {
        alt: Some(image.alt),
        aspect_ratio: Some(image.aspect_ratio),
        fullsize: url.clone(),
        thumb: url,
        extra_data: None,
    })
}

fn image_views_from_images(
    repo_did: &str,
    images: Option<Vec<Image>>,
) -> AppResult<Option<Vec<ImageView>>> {
    images
        .unwrap_or_default()
        .into_iter()
        .map(|image| image_view(repo_did, image))
        .collect::<AppResult<Vec<_>>>()
        .map(|views| if views.is_empty() { None } else { Some(views) })
}

fn thing_record(record_json: Option<&str>) -> AppResult<Thing> {
    serde_json::from_str(record_json.ok_or_else(|| internal("thing row missing record_json"))?)
        .map_err(|e| internal(format!("thing record_json typed parse: {e}")))
}

fn model_record(record_json: Option<&str>) -> AppResult<Model> {
    serde_json::from_str(record_json.ok_or_else(|| internal("model row missing record_json"))?)
        .map_err(|e| internal(format!("model record_json typed parse: {e}")))
}

fn part_record(record_json: Option<&str>) -> AppResult<Part> {
    serde_json::from_str(record_json.ok_or_else(|| internal("part row missing record_json"))?)
        .map_err(|e| internal(format!("part record_json typed parse: {e}")))
}

fn avatar_url(repo_did: &str, avatar_json: Option<&str>) -> AppResult<Option<UriValue>> {
    let Some(raw) = avatar_json else {
        return Ok(None);
    };
    let blob_ref: BlobRef =
        serde_json::from_str(raw).map_err(|e| internal(format!("avatar_json parse: {e}")))?;
    Ok(Some(avatar_cdn_url(repo_did, &blob_ref)?))
}
// ---------------------------------------------------------------------------
// rows
// ---------------------------------------------------------------------------

#[allow(dead_code)] // query-result row; mirrors projected columns
#[derive(sqlx::FromRow)]
struct ThingRow {
    did: String,
    rkey: String,
    uri: String,
    cid: String,
    name: String,
    summary: Option<String>,
    license: String,
    tags_json: Option<String>,
    cover_json: Option<String>,
    derived_from_uri: Option<String>,
    created_at: i64,
    indexed_at: i64,
    record_json: Option<String>,
    like_count: i64,
    save_count: i64,
    model_count: i64,
    part_count: i64,
}
#[allow(dead_code)] // query-result row; mirrors projected columns
#[derive(sqlx::FromRow)]
struct ModelRow {
    did: String,
    rkey: String,
    uri: String,
    cid: String,
    name: String,
    summary: Option<String>,
    created_at: i64,
    indexed_at: i64,
    record_json: Option<String>,
    part_count: i64,
}

#[allow(dead_code)] // query-result row; mirrors projected columns
#[derive(sqlx::FromRow)]
struct PartRow {
    did: String,
    rkey: String,
    uri: String,
    cid: String,
    name: String,
    format: Option<String>,
    file_json: String,
    created_at: i64,
    indexed_at: i64,
    record_json: Option<String>,
}

#[allow(dead_code)] // query-result row; mirrors projected columns
#[derive(sqlx::FromRow)]
struct ProfileRow {
    did: String,
    display_name: Option<String>,
    description: Option<String>,
    avatar_json: Option<String>,
    default_license: Option<String>,
    pronouns: Option<String>,
    printers_json: Option<String>,
    links_json: Option<String>,
    record_json: String,
    indexed_at: i64,
}

// ---------------------------------------------------------------------------
// author + viewer
// ---------------------------------------------------------------------------

async fn actor_for_did(state: &AppState, did: &str) -> AppResult<Actor> {
    #[derive(sqlx::FromRow)]
    struct AuthorRow {
        handle: Option<String>,
        display_name: Option<String>,
        avatar_json: Option<String>,
    }
    let row = db(sqlx::query_as!(
        AuthorRow,
        r#"SELECT (SELECT handle FROM identities WHERE did = ?) AS "handle?",
                      (SELECT display_name FROM profiles WHERE did = ?) AS "display_name?",
                      (SELECT avatar_json FROM profiles WHERE did = ?) AS "avatar_json?""#,
        did,
        did,
        did,
    )
    .fetch_optional(&state.pool)
    .await)?;
    let (handle, display_name, avatar_json) = match row {
        Some(r) => (r.handle, r.display_name, r.avatar_json),
        None => (None, None, None),
    };
    let did_typed = did_of(did)?;
    // Read-your-own-writes / not-yet-indexed authors: fall back to resolving the
    // handle (local cache, then DID document) instead of 500-ing the read.
    let handle = match handle {
        Some(h) => h,
        None => resolve_handle(state, &did_typed).await?,
    };
    let stats = actor_follow_stats(&state.pool, did, None).await?;
    Ok(Actor {
        did: did_typed,
        handle: handle_of(handle)?,
        display_name: display_name.map(s),
        avatar: avatar_url(did, avatar_json.as_deref())?,
        description: None,
        follower_count: Some(stats.follower_count),
        following_count: Some(stats.following_count),
        pronouns: None,
        thing_count: None,
        viewer: None,
        extra_data: None,
    })
}

struct ActorFollowStats {
    follower_count: i64,
    following_count: i64,
    viewer: ActorViewerState,
}

async fn actor_follow_stats(
    pool: &SqlitePool,
    actor_did: &str,
    viewer_did: Option<Did<&str>>,
) -> AppResult<ActorFollowStats> {
    #[derive(sqlx::FromRow)]
    struct CountRow {
        count: i64,
    }
    let follower_count = db(sqlx::query_as!(
        CountRow,
        r#"SELECT COUNT(*) AS "count!: i64" FROM follows WHERE followed_did = ?"#,
        actor_did,
    )
    .fetch_one(pool)
    .await)?
    .count;
    let following_count = db(sqlx::query_as!(
        CountRow,
        r#"SELECT COUNT(*) AS "count!: i64" FROM follows WHERE follower_did = ?"#,
        actor_did,
    )
    .fetch_one(pool)
    .await)?
    .count;

    let Some(viewer_did) = viewer_did else {
        return Ok(ActorFollowStats {
            follower_count,
            following_count,
            viewer: ActorViewerState::default(),
        });
    };
    let viewer_did = viewer_did.as_str();
    #[derive(sqlx::FromRow)]
    struct RkeyRow {
        rkey: String,
    }
    let following = db(sqlx::query_as!(
        RkeyRow,
        "SELECT rkey FROM follows WHERE follower_did = ? AND followed_did = ?",
        viewer_did,
        actor_did,
    )
    .fetch_optional(pool)
    .await)?;
    let followed_by = db(sqlx::query_as!(
        RkeyRow,
        "SELECT rkey FROM follows WHERE follower_did = ? AND followed_did = ?",
        actor_did,
        viewer_did,
    )
    .fetch_optional(pool)
    .await)?;
    let following = following
        .map(|row| {
            at_uri(&format!(
                "at://{viewer_did}/space.polymodel.graph.follow/{}",
                row.rkey
            ))
        })
        .transpose()?;
    let followed_by = followed_by
        .map(|row| {
            at_uri(&format!(
                "at://{actor_did}/space.polymodel.graph.follow/{}",
                row.rkey
            ))
        })
        .transpose()?;

    Ok(ActorFollowStats {
        follower_count,
        following_count,
        viewer: ActorViewerState {
            following,
            followed_by,
            extra_data: None,
            ..ActorViewerState::default()
        },
    })
}

async fn viewer_state(
    pool: &SqlitePool,
    viewer_did: Option<Did<&str>>,
    subject_uri: &str,
) -> AppResult<ViewerState> {
    let Some(did) = viewer_did else {
        return Ok(ViewerState::default());
    };
    let did = did.as_str();
    #[derive(sqlx::FromRow)]
    struct IdRow {
        rkey: String,
    }
    let like = db(sqlx::query_as!(
        IdRow,
        "SELECT rkey FROM likes WHERE did = ? AND subject_uri = ?",
        did,
        subject_uri
    )
    .fetch_optional(pool)
    .await)?;
    let save = db(sqlx::query_as!(
        IdRow,
        "SELECT rkey FROM saves WHERE did = ? AND subject_uri = ?",
        did,
        subject_uri
    )
    .fetch_optional(pool)
    .await)?;
    if like.is_none() && save.is_none() {
        return Ok(ViewerState::default());
    }
    let like = like
        .map(|r| at_uri(&format!("at://{did}/space.polymodel.graph.like/{}", r.rkey)))
        .transpose()?;
    let save = save
        .map(|r| at_uri(&format!("at://{did}/space.polymodel.graph.save/{}", r.rkey)))
        .transpose()?;
    Ok(ViewerState {
        like,
        save,
        extra_data: None,
    })
}

pub(super) async fn viewer_did(
    session: Option<&OAuthSession<PublicResolver, SqliteAuthStore>>,
) -> Option<Did> {
    match session {
        Some(s) => Some(s.session_info().await.0),
        None => None,
    }
}

// ---------------------------------------------------------------------------
// view builders
// ---------------------------------------------------------------------------

async fn thing_view(
    state: &AppState,
    row: ThingRow,
    viewer_did: Option<Did<&str>>,
) -> AppResult<ThingView> {
    let author = actor_for_did(state, &row.did).await?;
    let viewer = viewer_state(&state.pool, viewer_did, &row.uri).await?;
    let record = thing_record(row.record_json.as_deref())?;
    Ok(ThingView {
        uri: at_uri(&row.uri)?,
        cid: cid_of(&row.cid)?,
        author,
        record: record_data(row.record_json.as_deref())?,
        indexed_at: datetime_nanos(row.indexed_at),
        name: s(row.name),
        summary: row.summary.map(s),
        license: s(row.license),
        tags: smol_vec(row.tags_json.as_deref()),
        like_count: row.like_count,
        save_count: row.save_count,
        model_count: row.model_count,
        part_count: row.part_count,
        viewer,
        cover: image_views_from_images(&row.did, record.cover)?,
        previews: image_views_from_images(&row.did, record.previews)?,
        derived_from: None,
        intended_to_work_with: None,
        extra_data: None,
    })
}

async fn thing_basic_view(
    state: &AppState,
    row: ThingRow,
    viewer_did: Option<Did<&str>>,
) -> AppResult<ThingViewBasic> {
    let record = thing_record(row.record_json.as_deref())?;
    Ok(ThingViewBasic {
        uri: at_uri(&row.uri)?,
        cid: cid_of(&row.cid)?,
        name: s(row.name),
        author: actor_for_did(state, &row.did).await?,
        indexed_at: datetime_nanos(row.indexed_at),
        like_count: row.like_count,
        save_count: row.save_count,
        model_count: row.model_count,
        part_count: row.part_count,
        cover: image_views_from_images(&row.did, record.cover)?,
        previews: image_views_from_images(&row.did, record.previews)?,
        tags: smol_vec(row.tags_json.as_deref()),
        viewer: viewer_state(&state.pool, viewer_did, &row.uri).await?,
        extra_data: None,
    })
}

async fn model_view(
    state: &AppState,
    row: ModelRow,
    viewer_did: Option<Did<&str>>,
) -> AppResult<ModelView> {
    let author = actor_for_did(state, &row.did).await?;
    let viewer = viewer_state(&state.pool, viewer_did, &row.uri).await?;
    let record = model_record(row.record_json.as_deref())?;
    Ok(ModelView {
        uri: at_uri(&row.uri)?,
        cid: cid_of(&row.cid)?,
        author,
        record: record_data(row.record_json.as_deref())?,
        indexed_at: datetime_nanos(row.indexed_at),
        name: s(row.name),
        summary: row.summary.map(s),
        part_count: row.part_count,
        viewer,
        cover: image_views_from_images(&row.did, record.cover)?,
        previews: image_views_from_images(&row.did, record.previews)?,
        extra_data: None,
    })
}

async fn part_view(
    state: &AppState,
    row: PartRow,
    viewer_did: Option<Did<&str>>,
) -> AppResult<PartView> {
    let author = actor_for_did(state, &row.did).await?;
    let viewer = viewer_state(&state.pool, viewer_did, &row.uri).await?;
    let record = part_record(row.record_json.as_deref())?;
    Ok(PartView {
        uri: at_uri(&row.uri)?,
        cid: cid_of(&row.cid)?,
        author,
        record: record_data(row.record_json.as_deref())?,
        indexed_at: datetime_nanos(row.indexed_at),
        name: s(row.name),
        format: row.format.map(s),
        file: file_data(Some(row.file_json.as_str()))?,
        previews: image_views_from_images(&row.did, record.previews)?,
        viewer,
        extra_data: None,
    })
}

// ---------------------------------------------------------------------------
// getThing / getModel lookups
// ---------------------------------------------------------------------------

/// Fetch a thing row by URI (with stats + model count).
async fn fetch_thing_by_uri(state: &AppState, uri: &str) -> AppResult<Option<ThingRow>> {
    db(sqlx::query_as!(
        ThingRow,
        r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                  t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                  t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                  t.record_json,
                  COALESCE(cs.like_count, 0) AS "like_count!: i64",
                  COALESCE(cs.save_count, 0) AS "save_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
           FROM things t LEFT JOIN content_stats cs ON cs.uri = t.uri WHERE t.uri = ?"#,
        uri,
    )
    .fetch_optional(&state.pool)
    .await)
}

pub(crate) async fn get_thing(
    state: &AppState,
    uri: &str,
    viewer_did: Option<Did<&str>>,
) -> AppResult<(ThingView, Vec<ModelView>)> {
    let row = fetch_thing_by_uri(state, uri)
        .await?
        .ok_or_else(not_found)?;
    let thing = thing_view(state, row, viewer_did.clone()).await?;
    let models = models_for_thing(state, uri, viewer_did).await?;
    Ok((thing, models))
}

async fn models_for_thing(
    state: &AppState,
    thing_uri: &str,
    viewer_did: Option<Did<&str>>,
) -> AppResult<Vec<ModelView>> {
    let rows = db(
        sqlx::query_as!(
            ModelRow,
            r#"SELECT m.did AS "did!", m.rkey AS "rkey!", m.uri AS "uri!", m.cid AS "cid!",
                      m.name AS "name!", m.summary, m.created_at AS "created_at!",
                      m.indexed_at AS "indexed_at!", m.record_json,
                      (SELECT COUNT(*) FROM model_parts mp WHERE mp.model_uri = m.uri) AS "part_count!: i64"
               FROM thing_models tm JOIN models m ON m.uri = tm.model_uri
               WHERE tm.thing_uri = ? ORDER BY tm.position"#,
            thing_uri,
        )
        .fetch_all(&state.pool)
        .await,
    )?;
    let mut views = Vec::with_capacity(rows.len());
    for row in rows {
        views.push(model_view(state, row, viewer_did.clone()).await?);
    }
    Ok(views)
}

pub(crate) struct ModelDetail {
    pub model: ModelView,
    pub parts: Vec<PartView>,
    pub thing: Option<ThingViewBasic>,
}

pub(crate) async fn get_model(
    state: &AppState,
    uri: &str,
    viewer_did: Option<Did<&str>>,
) -> AppResult<ModelDetail> {
    let row = db(
        sqlx::query_as!(
            ModelRow,
            r#"SELECT m.did AS "did!", m.rkey AS "rkey!", m.uri AS "uri!", m.cid AS "cid!",
                      m.name AS "name!", m.summary, m.created_at AS "created_at!",
                      m.indexed_at AS "indexed_at!", m.record_json,
                      (SELECT COUNT(*) FROM model_parts mp WHERE mp.model_uri = m.uri) AS "part_count!: i64"
               FROM models m WHERE m.uri = ?"#,
            uri,
        )
        .fetch_optional(&state.pool)
        .await,
    )?
    .ok_or_else(not_found)?;
    let model = model_view(state, row, viewer_did.clone()).await?;

    let parts = db(sqlx::query_as!(
        PartRow,
        r#"SELECT p.did AS "did!", p.rkey AS "rkey!", p.uri AS "uri!", p.cid AS "cid!",
                      p.name AS "name!", p.format, p.file_json AS "file_json!",
                      p.created_at AS "created_at!", p.indexed_at AS "indexed_at!", p.record_json
               FROM model_parts mp JOIN parts p ON p.uri = mp.part_uri
               WHERE mp.model_uri = ? ORDER BY mp.position"#,
        uri,
    )
    .fetch_all(&state.pool)
    .await)?;
    let mut part_views = Vec::with_capacity(parts.len());
    for row in parts {
        part_views.push(part_view(state, row, viewer_did.clone()).await?);
    }

    let thing = db(
        sqlx::query_as!(
            ThingRow,
            r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                      t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                      t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                      t.record_json,
                      COALESCE(cs.like_count, 0) AS "like_count!: i64",
                      COALESCE(cs.save_count, 0) AS "save_count!: i64",
                      (SELECT COUNT(*) FROM thing_models tm2 WHERE tm2.thing_uri = t.uri) AS "model_count!: i64",
                      (SELECT COUNT(*) FROM thing_models tm2 JOIN model_parts mp ON mp.model_uri = tm2.model_uri WHERE tm2.thing_uri = t.uri) AS "part_count!: i64"
               FROM thing_models tm JOIN things t ON t.uri = tm.thing_uri
               LEFT JOIN content_stats cs ON cs.uri = t.uri
               WHERE tm.model_uri = ? ORDER BY tm.position LIMIT 1"#,
            uri,
        )
        .fetch_optional(&state.pool)
        .await,
    )?;
    let thing = if let Some(row) = thing {
        Some(thing_basic_view(state, row, viewer_did).await?)
    } else {
        None
    };

    Ok(ModelDetail {
        model,
        parts: part_views,
        thing,
    })
}

// ---------------------------------------------------------------------------
// feeds (feedView of thingViewBasic)
// ---------------------------------------------------------------------------

async fn feed_view(
    state: &AppState,
    rows: Vec<ThingRow>,
    cursor: Option<String>,
    viewer_did: Option<Did<&str>>,
) -> AppResult<FeedView> {
    let mut items = Vec::with_capacity(rows.len());
    for row in rows {
        items.push(FeedItem {
            thing: thing_basic_view(state, row, viewer_did.clone()).await?,
            extra_data: None,
        });
    }
    Ok(FeedView {
        cursor: cursor.map(s),
        items,
        extra_data: None,
    })
}

pub(super) fn empty_feed() -> FeedView {
    FeedView {
        cursor: None,
        items: Vec::new(),
        extra_data: None,
    }
}

pub(super) async fn author_things(
    state: &AppState,
    did: &str,
    limit: i64,
    cursor: Option<&str>,
    viewer_did: Option<Did<&str>>,
) -> AppResult<FeedView> {
    let fetch = limit + 1;
    let rows = db(
        sqlx::query_as!(
            ThingRow,
            r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                      t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                      t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                      t.record_json,
                      COALESCE(cs.like_count, 0) AS "like_count!: i64",
                      COALESCE(cs.save_count, 0) AS "save_count!: i64",
                      (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
               FROM things t LEFT JOIN content_stats cs ON cs.uri = t.uri
               WHERE t.did = ? AND (? IS NULL OR t.rkey < ?) ORDER BY t.rkey DESC LIMIT ?"#,
            did,
            cursor,
            cursor,
            fetch,
        )
        .fetch_all(&state.pool)
        .await,
    )?;
    rkey_paged_feed(state, rows, limit, viewer_did).await
}

pub(crate) async fn feed_recent(
    state: &AppState,
    limit: i64,
    cursor: Option<&str>,
    viewer_did: Option<Did<&str>>,
) -> AppResult<FeedView> {
    let fetch = limit + 1;
    let rows = db(
        sqlx::query_as!(
            ThingRow,
            r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                      t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                      t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                      t.record_json,
                      COALESCE(cs.like_count, 0) AS "like_count!: i64",
                      COALESCE(cs.save_count, 0) AS "save_count!: i64",
                      (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
               FROM things t LEFT JOIN content_stats cs ON cs.uri = t.uri
               WHERE (? IS NULL OR t.rkey < ?) ORDER BY t.rkey DESC LIMIT ?"#,
            cursor,
            cursor,
            fetch,
        )
        .fetch_all(&state.pool)
        .await,
    )?;
    rkey_paged_feed(state, rows, limit, viewer_did).await
}

pub(crate) async fn feed_following(
    state: &AppState,
    viewer_did: Did<&str>,
    limit: i64,
    cursor: Option<&str>,
) -> AppResult<FeedView> {
    let cursor = parse_following_cursor(cursor)?;
    let fetch = limit + 1;
    let rows = db(sqlx::query_as!(
        ThingRow,
        r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                  t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                  t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                  t.record_json,
                  COALESCE(cs.like_count, 0) AS "like_count!: i64",
                  COALESCE(cs.save_count, 0) AS "save_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
           FROM things t
           JOIN follows f ON f.followed_did = t.did
           LEFT JOIN content_stats cs ON cs.uri = t.uri
           WHERE f.follower_did = ?
             AND (? IS NULL OR t.rkey < ? OR (t.rkey = ? AND t.did < ?))
           ORDER BY t.rkey DESC, t.did DESC LIMIT ?"#,
        viewer_did.as_str(),
        cursor.as_ref().map(|c| c.rkey.as_str()),
        cursor.as_ref().map(|c| c.rkey.as_str()),
        cursor.as_ref().map(|c| c.rkey.as_str()),
        cursor.as_ref().map(|c| c.did.as_str()),
        fetch,
    )
    .fetch_all(&state.pool)
    .await)?;
    compound_paged_feed(state, rows, limit, Some(viewer_did)).await
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct FollowingCursor {
    rkey: String,
    did: String,
}

fn parse_following_cursor(cursor: Option<&str>) -> AppResult<Option<FollowingCursor>> {
    let Some(cursor) = cursor else {
        return Ok(None);
    };
    let Some(rest) = cursor.strip_prefix("following:v1:") else {
        return Err(invalid_request("invalid following cursor"));
    };
    let Some((rkey, did)) = rest.split_once(':') else {
        return Err(invalid_request("invalid following cursor"));
    };
    if rkey.is_empty() || did.is_empty() {
        return Err(invalid_request("invalid following cursor"));
    }
    Ok(Some(FollowingCursor {
        rkey: rkey.to_string(),
        did: did.to_string(),
    }))
}

fn following_cursor(row: &ThingRow) -> String {
    format!("following:v1:{}:{}", row.rkey, row.did)
}

/// Reduce an over-fetched (`limit+1`) rkey-ordered page into a feed, returning
/// the last *returned* item's `rkey` as the cursor when another page exists.
/// Using the last returned item (not the over-fetched boundary) keeps
/// `rkey < cursor` from skipping the boundary item on the next page.
async fn rkey_paged_feed(
    state: &AppState,
    mut rows: Vec<ThingRow>,
    limit: i64,
    viewer_did: Option<Did<&str>>,
) -> AppResult<FeedView> {
    let cursor = if rows.len() as i64 > limit {
        rows.get((limit - 1) as usize).map(|r| r.rkey.clone())
    } else {
        None
    };
    rows.truncate(limit as usize);
    feed_view(state, rows, cursor, viewer_did).await
}

async fn compound_paged_feed(
    state: &AppState,
    mut rows: Vec<ThingRow>,
    limit: i64,
    viewer_did: Option<Did<&str>>,
) -> AppResult<FeedView> {
    let cursor = if rows.len() as i64 > limit {
        rows.get((limit - 1) as usize).map(following_cursor)
    } else {
        None
    };
    rows.truncate(limit as usize);
    feed_view(state, rows, cursor, viewer_did).await
}

pub(super) async fn feed_hot(
    state: &AppState,
    limit: i64,
    cursor: Option<&str>,
    viewer_did: Option<Did<&str>>,
) -> AppResult<FeedView> {
    let default_now_nanos = chrono::Utc::now()
        .timestamp_nanos_opt()
        .expect("current timestamp must fit in i64 nanoseconds");
    feed_hot_at(state, limit, cursor, viewer_did, default_now_nanos).await
}

pub(super) async fn feed_hot_at(
    state: &AppState,
    limit: i64,
    cursor: Option<&str>,
    viewer_did: Option<Did<&str>>,
    default_now_nanos: i64,
) -> AppResult<FeedView> {
    let page_state = parse_hot_cursor(cursor, default_now_nanos)?;
    let now = page_state.now_nanos;
    let mut scored = hot_positive_candidates(state, limit, &page_state).await?;

    if scored.len() as i64 <= limit {
        scored.extend(
            hot_zero_engagement_candidates(state, limit + 1 - scored.len() as i64, &page_state)
                .await?,
        );
    }

    // score desc, then rkey desc and uri desc. rkeys are only per-repo unique, so
    // uri completes the total order used by the page boundary.
    sort_hot_scores(&mut scored);
    scored.truncate(limit as usize + 1);
    let cursor = if scored.len() as i64 > limit {
        scored
            .get((limit - 1) as usize)
            .map(|(score, row)| encode_hot_cursor(now, *score, &row.rkey, &row.uri))
    } else {
        None
    };
    scored.truncate(limit as usize);
    let page = scored.into_iter().map(|(_, row)| row).collect();
    feed_view(state, page, cursor, viewer_did).await
}

async fn hot_positive_candidates(
    state: &AppState,
    limit: i64,
    page_state: &HotPageState,
) -> AppResult<Vec<(f64, ThingRow)>> {
    if !page_state.may_include_positive_scores() {
        return Ok(Vec::new());
    }

    let mut scored = Vec::new();
    let mut offset = 0;
    loop {
        // PM-51 keeps ranking exact without a fixed recent window, but this is
        // still a per-request candidate scan. PM-56 tracks replacing it with a
        // materialized/indexed hot-feed ranking table.
        let rows = db(
            sqlx::query_as!(
                ThingRow,
                r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                          t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                          t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                          t.record_json,
                          COALESCE(cs.like_count, 0) AS "like_count!: i64",
                          COALESCE(cs.save_count, 0) AS "save_count!: i64",
                          (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                      (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
                   FROM things t LEFT JOIN content_stats cs ON cs.uri = t.uri
                   WHERE COALESCE(cs.like_count, 0) + COALESCE(cs.save_count, 0) > 0
                   ORDER BY COALESCE(cs.like_count, 0) + COALESCE(cs.save_count, 0) DESC,
                            t.rkey DESC, t.uri DESC
                   LIMIT ? OFFSET ?"#,
                HOT_POSITIVE_BATCH,
                offset,
            )
            .fetch_all(&state.pool)
            .await,
        )?;
        if rows.is_empty() {
            break;
        }

        let batch_len = rows.len() as i64;
        let last_engagement = rows
            .last()
            .map(|row| row.like_count + row.save_count)
            .unwrap_or(0);
        for row in rows {
            let score = hot_score(&row, page_state.now_nanos);
            if page_state.is_after_boundary(score, &row.rkey, &row.uri) {
                scored.push((score, row));
            }
        }
        sort_hot_scores(&mut scored);
        scored.truncate(limit as usize + 1);

        if batch_len < HOT_POSITIVE_BATCH || can_stop_positive_scan(&scored, limit, last_engagement)
        {
            break;
        }
        offset += HOT_POSITIVE_BATCH;
    }
    Ok(scored)
}

async fn hot_zero_engagement_candidates(
    state: &AppState,
    fetch: i64,
    page_state: &HotPageState,
) -> AppResult<Vec<(f64, ThingRow)>> {
    if fetch <= 0 || !page_state.may_include_zero_scores() {
        return Ok(Vec::new());
    }
    let (after_rkey, after_uri) = page_state.zero_score_boundary();
    let rows = db(
        sqlx::query_as!(
            ThingRow,
            r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                      t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                      t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                      t.record_json,
                      COALESCE(cs.like_count, 0) AS "like_count!: i64",
                      COALESCE(cs.save_count, 0) AS "save_count!: i64",
                      (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
               FROM things t LEFT JOIN content_stats cs ON cs.uri = t.uri
               WHERE COALESCE(cs.like_count, 0) + COALESCE(cs.save_count, 0) = 0
                 AND (? IS NULL OR t.rkey < ? OR (t.rkey = ? AND t.uri < ?))
               ORDER BY t.rkey DESC, t.uri DESC
               LIMIT ?"#,
            after_rkey,
            after_rkey,
            after_rkey,
            after_uri,
            fetch,
        )
        .fetch_all(&state.pool)
        .await,
    )?;
    Ok(rows.into_iter().map(|row| (0.0, row)).collect())
}

fn sort_hot_scores(scored: &mut [(f64, ThingRow)]) {
    scored.sort_by(hot_score_order);
}

fn hot_score_order(a: &(f64, ThingRow), b: &(f64, ThingRow)) -> Ordering {
    b.0.partial_cmp(&a.0)
        .unwrap_or(Ordering::Equal)
        .then_with(|| b.1.rkey.cmp(&a.1.rkey))
        .then_with(|| b.1.uri.cmp(&a.1.uri))
}

fn can_stop_positive_scan(scored: &[(f64, ThingRow)], limit: i64, last_engagement: i64) -> bool {
    let Some((worst_score, _)) = scored.get(limit as usize) else {
        return false;
    };
    let max_unseen_score = (last_engagement as f64) / 2.0_f64.powf(1.5);
    max_unseen_score < *worst_score
}

struct HotPageState {
    now_nanos: i64,
    boundary: Option<HotCursorBoundary>,
}

struct HotCursorBoundary {
    score: f64,
    rkey: String,
    uri: String,
}

impl HotPageState {
    fn may_include_positive_scores(&self) -> bool {
        self.boundary
            .as_ref()
            .is_none_or(|boundary| boundary.score > 0.0)
    }

    fn may_include_zero_scores(&self) -> bool {
        self.boundary
            .as_ref()
            .is_none_or(|boundary| boundary.score >= 0.0)
    }

    fn zero_score_boundary(&self) -> (Option<&str>, Option<&str>) {
        match &self.boundary {
            Some(boundary) if boundary.score == 0.0 => {
                (Some(boundary.rkey.as_str()), Some(boundary.uri.as_str()))
            }
            _ => (None, None),
        }
    }

    fn is_after_boundary(&self, score: f64, rkey: &str, uri: &str) -> bool {
        let Some(boundary) = &self.boundary else {
            return true;
        };
        score < boundary.score
            || (score == boundary.score
                && (rkey < boundary.rkey.as_str()
                    || (rkey == boundary.rkey.as_str() && uri < boundary.uri.as_str())))
    }
}

fn parse_hot_cursor(cursor: Option<&str>, default_now_nanos: i64) -> AppResult<HotPageState> {
    let Some(cursor) = cursor else {
        return Ok(HotPageState {
            now_nanos: default_now_nanos,
            boundary: None,
        });
    };

    // The hot cursor pins the ranking timestamp. Without that snapshot value,
    // time-dependent hot scores can shift between requests and make a page
    // boundary skip or duplicate rows.
    let mut parts = cursor.splitn(6, ':');
    let valid_prefix = parts.next() == Some("hot") && parts.next() == Some("v1");
    let Some(now_nanos) = parts.next() else {
        return Err(invalid_hot_cursor());
    };
    let Some(score_bits_hex) = parts.next() else {
        return Err(invalid_hot_cursor());
    };
    let Some(rkey) = parts.next() else {
        return Err(invalid_hot_cursor());
    };
    let Some(uri) = parts.next() else {
        return Err(invalid_hot_cursor());
    };
    if !valid_prefix || rkey.is_empty() || uri.is_empty() {
        return Err(invalid_hot_cursor());
    }
    let now_nanos = now_nanos.parse().map_err(|_| invalid_hot_cursor())?;
    let score_bits = u64::from_str_radix(score_bits_hex, 16).map_err(|_| invalid_hot_cursor())?;
    let score = f64::from_bits(score_bits);
    if !(score.is_finite() && score >= 0.0) {
        return Err(invalid_hot_cursor());
    }
    Ok(HotPageState {
        now_nanos,
        boundary: Some(HotCursorBoundary {
            score,
            rkey: rkey.to_owned(),
            uri: uri.to_owned(),
        }),
    })
}

fn encode_hot_cursor(now_nanos: i64, score: f64, rkey: &str, uri: &str) -> String {
    format!("hot:v1:{now_nanos}:{:016x}:{rkey}:{uri}", score.to_bits())
}

fn invalid_hot_cursor() -> AppError {
    invalid_request("cursor must be an opaque hot feed cursor")
}

fn hot_score(row: &ThingRow, now_nanos: i64) -> f64 {
    let age_hours = ((now_nanos - row.created_at).max(0) as f64) / 3_600_000_000_000.0;
    let engagement = row.like_count as f64 + row.save_count as f64;
    engagement / (age_hours + 2.0).powf(1.5)
}

pub(super) async fn search_things(
    state: &AppState,
    q: &str,
    limit: i64,
    cursor: Option<&str>,
    viewer_did: Option<Did<&str>>,
) -> AppResult<FeedView> {
    let offset = parse_offset(cursor)?;
    let fetch = limit + 1;
    let rows = db(
        sqlx::query_as!(
            ThingRow,
            r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                      t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                      t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                      t.record_json,
                      COALESCE(cs.like_count, 0) AS "like_count!: i64",
                      COALESCE(cs.save_count, 0) AS "save_count!: i64",
                      (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
               FROM things_fts
               JOIN things t ON t.rowid = things_fts.rowid
               LEFT JOIN content_stats cs ON cs.uri = t.uri
               WHERE things_fts MATCH ?
               ORDER BY bm25(things_fts, 10.0, 5.0, 3.0, 1.0)
               LIMIT ? OFFSET ?"#,
            q,
            fetch,
            offset,
        )
        .fetch_all(&state.pool)
        .await,
    )?;
    let cursor = offset_page_cursor(rows.len(), limit, offset);
    let page: Vec<ThingRow> = rows.into_iter().take(limit as usize).collect();
    feed_view(state, page, cursor, viewer_did).await
}

fn parse_offset(cursor: Option<&str>) -> AppResult<i64> {
    match cursor {
        None => Ok(0),
        Some(c) => {
            let offset: i64 = c
                .parse()
                .map_err(|_| invalid_request("cursor must be a non-negative integer offset"))?;
            if offset < 0 {
                return Err(invalid_request(
                    "cursor must be a non-negative integer offset",
                ));
            }
            Ok(offset)
        }
    }
}

fn offset_page_cursor(row_count: usize, limit: i64, offset: i64) -> Option<String> {
    (row_count as i64 > limit).then(|| (offset + limit).to_string())
}

// ---------------------------------------------------------------------------
// getList (thingViewBasic[] from listitems)
// ---------------------------------------------------------------------------

pub(super) async fn list_things(
    state: &AppState,
    list_uri: &str,
    limit: i64,
    cursor: Option<&str>,
    viewer_did: Option<Did<&str>>,
) -> AppResult<(Vec<ThingViewBasic>, Option<String>)> {
    let offset = parse_offset(cursor)?;
    let fetch = limit + 1;
    let rows = db(sqlx::query_as!(
        ThingRow,
        r#"SELECT t.did AS "did!", t.rkey AS "rkey!", t.uri AS "uri!", t.cid AS "cid!",
                  t.name AS "name!", t.summary, t.license AS "license!", t.tags_json, t.cover_json,
                  t.derived_from_uri, t.created_at AS "created_at!", t.indexed_at AS "indexed_at!",
                  t.record_json,
                  COALESCE(cs.like_count, 0) AS "like_count!: i64",
                  COALESCE(cs.save_count, 0) AS "save_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm WHERE tm.thing_uri = t.uri) AS "model_count!: i64",
                  (SELECT COUNT(*) FROM thing_models tm JOIN model_parts mp ON mp.model_uri = tm.model_uri WHERE tm.thing_uri = t.uri) AS "part_count!: i64"
           FROM listitems li
           JOIN things t ON t.uri = li.subject_uri
           LEFT JOIN content_stats cs ON cs.uri = t.uri
           WHERE li.list_uri = ?
           ORDER BY li.created_at, li.did, li.rkey
           LIMIT ? OFFSET ?"#,
        list_uri,
        fetch,
        offset,
    )
    .fetch_all(&state.pool)
    .await)?;
    let cursor = offset_page_cursor(rows.len(), limit, offset);
    let mut items = Vec::with_capacity(rows.len().min(limit as usize));
    for row in rows.into_iter().take(limit as usize) {
        items.push(thing_basic_view(state, row, viewer_did.clone()).await?);
    }
    Ok((items, cursor))
}

// ---------------------------------------------------------------------------
// getProfile (polymodel variant)
// ---------------------------------------------------------------------------

pub(super) struct ProfileStats {
    pub thing_count: i64,
    pub total_likes: i64,
}

#[derive(sqlx::FromRow)]
struct ProfileListRow {
    uri: String,
    cid: String,
    name: String,
    description: Option<String>,
    purpose: Option<String>,
    created_at: i64,
    item_count: i64,
}

async fn profile_for_did(state: &AppState, did: &str) -> AppResult<Option<ProfileRow>> {
    let row = db(sqlx::query_as!(
        ProfileRow,
        r#"SELECT did AS "did!", display_name, description, avatar_json, default_license,
                      pronouns, printers_json, links_json, record_json AS "record_json!",
                      indexed_at AS "indexed_at!"
               FROM profiles WHERE did = ?"#,
        did,
    )
    .fetch_optional(&state.pool)
    .await)?;
    Ok(row)
}

pub(super) async fn profile_stats(state: &AppState, did: &str) -> AppResult<ProfileStats> {
    #[derive(sqlx::FromRow)]
    struct StatsRow {
        thing_count: i64,
        total_likes: i64,
    }
    let row = db(
        sqlx::query_as!(
            StatsRow,
            r#"SELECT COUNT(*) AS "thing_count!: i64",
                      COALESCE((SELECT SUM(cs.like_count) FROM things t2
                       LEFT JOIN content_stats cs ON cs.uri = t2.uri WHERE t2.did = ?), 0) AS "total_likes!: i64"
               FROM things t WHERE t.did = ?"#,
            did,
            did,
        )
        .fetch_one(&state.pool)
        .await,
    )?;
    Ok(ProfileStats {
        thing_count: row.thing_count,
        total_likes: row.total_likes,
    })
}

async fn profile_lists(state: &AppState, did: &str) -> AppResult<Vec<ProfileListView>> {
    let rows = db(sqlx::query_as!(
        ProfileListRow,
        r#"SELECT l.uri AS "uri!", l.cid AS "cid!", l.name AS "name!",
                  l.description, l.purpose, l.created_at AS "created_at!",
                  COUNT(li.subject_uri) AS "item_count!: i64"
           FROM lists l
           LEFT JOIN listitems li ON li.list_uri = l.uri
           WHERE l.did = ?
           GROUP BY l.uri, l.cid, l.name, l.description, l.purpose, l.created_at
           ORDER BY l.created_at DESC, l.rkey DESC"#,
        did,
    )
    .fetch_all(&state.pool)
    .await)?;

    rows.into_iter()
        .map(|row| {
            Ok(ProfileListView {
                uri: at_uri(&row.uri)?,
                cid: jacquard_common::types::string::Cid::new_owned(row.cid.as_bytes())
                    .map_err(|e| internal(format!("invalid list cid: {e}")))?,
                name: s(row.name),
                description: row.description.map(s),
                purpose: row.purpose.map(s),
                item_count: row.item_count,
                created_at: datetime_nanos(row.created_at),
                extra_data: None,
            })
        })
        .collect()
}

pub(super) async fn handle_for_did(state: &AppState, did: &Did) -> AppResult<Option<String>> {
    let did = did.as_str();
    #[derive(sqlx::FromRow)]
    struct HandleRow {
        handle: Option<String>,
    }
    let row = db(sqlx::query_as!(
        HandleRow,
        r#"SELECT handle AS "handle?" FROM identities WHERE did = ?"#,
        did
    )
    .fetch_optional(&state.pool)
    .await)?;
    Ok(row.and_then(|r| r.handle))
}

/// Resolve an actor's handle from the identity projection, falling back to the
/// DID document if no identity event has landed yet. A polymodel-profiled actor
/// always has a resolvable handle — never return an empty string, which would
/// fail handle validation.
pub(super) async fn resolve_handle(state: &AppState, did: &Did) -> AppResult<String> {
    if let Some(h) = handle_for_did(state, did).await? {
        return Ok(h);
    }
    let doc = state
        .resolver
        .resolve_did_doc(did)
        .await
        .map_err(|e| internal(format!("DID document resolution failed: {e}")))?
        .into_owned()
        .map_err(|e| internal(format!("DID document parse failed: {e}")))?;
    doc.also_known_as
        .and_then(|aka| aka.into_iter().next())
        .map(|h| h.to_string())
        .ok_or_else(|| internal("DID document has no handle (alsoKnownAs)"))
}

/// Build the polymodel `profileView` union variant for a DID, or `None` if the
/// actor has no `space.polymodel.actor.profile` record.
pub(super) async fn polymodel_profile_view(
    state: &AppState,
    did: &Did,
) -> AppResult<Option<polymodel_api::space_polymodel::actor::ProfileView>> {
    let Some(row) = profile_for_did(state, did.as_ref()).await? else {
        return Ok(None);
    };
    let stats = profile_stats(state, did.as_ref()).await?;
    let follows = actor_follow_stats(&state.pool, did.as_ref(), None).await?;
    let lists = profile_lists(state, did.as_ref()).await?;
    let handle_str = resolve_handle(state, did).await?;
    let record: Data = serde_json::from_str(&row.record_json)
        .map_err(|e| internal(format!("profile record_json parse: {e}")))?;
    Ok(Some(polymodel_api::space_polymodel::actor::ProfileView {
        did: did.clone(),
        handle: s(handle_str),
        record,
        display_name: row.display_name.map(s),
        description: row.description.map(s),
        avatar: avatar_url(did.as_ref(), row.avatar_json.as_deref())?,
        default_license: row.default_license.map(s),
        pronouns: row.pronouns.map(s),
        printers: smol_vec(row.printers_json.as_deref()),
        links: uri_vec(row.links_json.as_deref()),
        thing_count: Some(stats.thing_count),
        total_likes: Some(stats.total_likes),
        follower_count: Some(follows.follower_count),
        following_count: Some(follows.following_count),
        viewer: Some(follows.viewer),
        lists: if lists.is_empty() { None } else { Some(lists) },
        indexed_at: Some(datetime_nanos(row.indexed_at)),
        extra_data: None,
    }))
}

#[cfg(test)]
mod hot_tests {
    use super::*;

    fn row(rkey: &str, uri: &str) -> ThingRow {
        ThingRow {
            did: "did:plc:test".to_owned(),
            rkey: rkey.to_owned(),
            uri: uri.to_owned(),
            cid: "cid".to_owned(),
            name: "test".to_owned(),
            summary: None,
            license: "CC-BY-4.0".to_owned(),
            tags_json: None,
            cover_json: None,
            derived_from_uri: None,
            created_at: 0,
            indexed_at: 0,
            record_json: None,
            like_count: 0,
            save_count: 0,
            model_count: 0,
            part_count: 0,
        }
    }

    #[test]
    fn positive_scan_bound_stops_only_when_unseen_scores_cannot_enter_page() {
        let page = vec![
            (10.0, row("3c", "at://did:plc:c/thing/3c")),
            (9.0, row("3b", "at://did:plc:b/thing/3b")),
        ];
        assert!(can_stop_positive_scan(&page, 1, 1));
        assert!(!can_stop_positive_scan(&page, 1, 100));
        assert!(!can_stop_positive_scan(&page[..1], 1, 1));
    }

    #[test]
    fn rejects_malformed_hot_cursor_score_bits() {
        let bits = |s: f64| format!("{:016x}", s.to_bits());
        let cursor = |score_bits: String| format!("hot:v1:100:{score_bits}:rkey:at://did:plc:x/x");
        assert!(parse_hot_cursor(Some(&cursor(bits(f64::NAN))), 0).is_err());
        assert!(parse_hot_cursor(Some(&cursor(bits(f64::INFINITY))), 0).is_err());
        assert!(parse_hot_cursor(Some(&cursor(bits(f64::NEG_INFINITY))), 0).is_err());
        assert!(parse_hot_cursor(Some(&cursor(bits(-1.0))), 0).is_err());
        assert!(parse_hot_cursor(Some(&cursor(bits(0.0))), 0).is_ok());
        assert!(parse_hot_cursor(Some(&cursor(bits(12.5))), 0).is_ok());
    }
}