//! SQLite read queries and generated-view builders for the PM-26 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 via
//! [`to_data`]. 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).
//! - Ranked views (`getFeed?algorithm=hot`, `searchThings`) use an offset cursor
//!   (stringified); a non-numeric cursor is rejected with 400. Ranking is
//!   deterministic: hot scores tie-break on `rkey` desc; search is BM25.

use std::cmp::Ordering;

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

use super::error::{AppResult, db, internal, invalid_request, not_found};
use super::state::AppState;

// ---------------------------------------------------------------------------
// 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 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_millis(millis: i64) -> Datetime {
    let dt = chrono::DateTime::<chrono::Utc>::from_timestamp_millis(millis)
        .unwrap_or_else(chrono::Utc::now);
    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"))?;
    let value: serde_json::Value =
        serde_json::from_str(raw).map_err(|e| internal(format!("record_json parse: {e}")))?;
    to_data(&value).map_err(|e| internal(format!("record encode: {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),
    };
    Ok(Actor {
        did: did_of(did)?,
        handle: handle_of(
            handle.ok_or_else(|| internal(format!("missing handle for actor {did}")))?,
        )?,
        display_name: display_name.map(s),
        avatar: avatar_url(did, avatar_json.as_deref())?,
        description: None,
        follower_count: None,
        following_count: None,
        pronouns: None,
        thing_count: None,
        viewer: None,
        extra_data: None,
    })
}

async fn viewer_state(
    pool: &SqlitePool,
    viewer_did: Option<&str>,
    subject_uri: &str,
) -> AppResult<ViewerState> {
    let Some(did) = viewer_did else {
        return Ok(ViewerState::default());
    };
    #[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,
    })
}

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

async fn thing_view(
    state: &AppState,
    row: ThingRow,
    viewer_did: Option<&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: s(row.cid),
        author,
        record: record_data(row.record_json.as_deref())?,
        indexed_at: datetime_millis(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) -> AppResult<ThingViewBasic> {
    let record = thing_record(row.record_json.as_deref())?;
    Ok(ThingViewBasic {
        uri: at_uri(&row.uri)?,
        cid: s(row.cid),
        name: s(row.name),
        author: actor_for_did(state, &row.did).await?,
        indexed_at: datetime_millis(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: ViewerState::default(),
        extra_data: None,
    })
}

async fn model_view(
    state: &AppState,
    row: ModelRow,
    viewer_did: Option<&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: s(row.cid),
        author,
        record: record_data(row.record_json.as_deref())?,
        indexed_at: datetime_millis(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<&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: s(row.cid),
        author,
        record: record_data(row.record_json.as_deref())?,
        indexed_at: datetime_millis(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). Shared by getThing and
/// getList's per-subject hydration.
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(super) async fn get_thing(
    state: &AppState,
    uri: &str,
    viewer_did: Option<&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).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<&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).await?);
    }
    Ok(views)
}

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

pub(super) async fn get_model(
    state: &AppState,
    uri: &str,
    viewer_did: Option<&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).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).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).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>,
) -> AppResult<FeedView> {
    let mut items = Vec::with_capacity(rows.len());
    for row in rows {
        items.push(FeedItem {
            thing: thing_basic_view(state, row).await?,
            extra_data: None,
        });
    }
    Ok(FeedView {
        cursor: cursor.map(s),
        items,
        extra_data: None,
    })
}

pub(super) async fn author_things(
    state: &AppState,
    did: &str,
    limit: i64,
    cursor: Option<&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).await
}

pub(super) async fn feed_recent(
    state: &AppState,
    limit: i64,
    cursor: Option<&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).await
}

/// 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,
) -> 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).await
}

pub(super) async fn feed_hot(
    state: &AppState,
    limit: i64,
    cursor: Option<&str>,
) -> AppResult<FeedView> {
    let offset = parse_offset(cursor)?;
    // Candidate window: the most recent things. Hot score is computed in Rust so
    // the formula (engagement / (age_hours + 2)^1.5) does not depend on SQLite
    // math functions, and ranking is deterministic for a fixed `now`.
    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
               ORDER BY t.created_at DESC LIMIT 500"#,
        )
        .fetch_all(&state.pool)
        .await,
    )?;
    let now = chrono::Utc::now().timestamp_millis();
    let mut scored: Vec<(f64, ThingRow)> =
        rows.into_iter().map(|r| (hot_score(&r, now), r)).collect();
    // score desc, tie-break rkey desc (deterministic, reproducible page boundary).
    scored.sort_by(|a, b| {
        b.0.partial_cmp(&a.0)
            .unwrap_or(Ordering::Equal)
            .then_with(|| b.1.rkey.cmp(&a.1.rkey))
    });
    let page: Vec<ThingRow> = scored
        .into_iter()
        .skip(offset as usize)
        .take(limit as usize + 1)
        .map(|(_, r)| r)
        .collect();
    let cursor = if page.len() as i64 > limit {
        Some((offset + limit).to_string())
    } else {
        None
    };
    let page = page.into_iter().take(limit as usize).collect();
    feed_view(state, page, cursor).await
}

fn hot_score(row: &ThingRow, now_millis: i64) -> f64 {
    let age_hours = ((now_millis - row.created_at).max(0) as f64) / 3_600_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>,
) -> 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 = if rows.len() as i64 > limit {
        Some((offset + limit).to_string())
    } else {
        None
    };
    let page: Vec<ThingRow> = rows.into_iter().take(limit as usize).collect();
    feed_view(state, page, cursor).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)
        }
    }
}

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

pub(super) async fn list_things(
    state: &AppState,
    list_uri: &str,
    limit: i64,
    cursor: Option<&str>,
) -> AppResult<(Vec<ThingViewBasic>, Option<String>)> {
    let offset = parse_offset(cursor)?;
    // Over-fetch subjects, then hydrate only those resolving to a thing; non-thing
    // subjects are filtered out (listitems.subject_uri may reference any record).
    let fetch = limit + 1;
    #[derive(sqlx::FromRow)]
    struct ListItemRow {
        subject_uri: String,
    }
    let rows = db(
        sqlx::query_as!(
            ListItemRow,
            r#"SELECT subject_uri FROM listitems WHERE list_uri = ? ORDER BY created_at, did, rkey LIMIT ? OFFSET ?"#,
            list_uri,
            fetch,
            offset,
        )
        .fetch_all(&state.pool)
        .await,
    )?;
    // Another page exists when the raw subject over-fetch exceeded the limit,
    // independent of how many subjects hydrated to things.
    let cursor = if rows.len() as i64 > limit {
        Some((offset + limit).to_string())
    } else {
        None
    };
    let mut items = Vec::with_capacity(rows.len().min(limit as usize));
    for li in rows.into_iter().take(limit as usize) {
        if let Some(thing) = thing_basic_by_uri(state, &li.subject_uri).await? {
            items.push(thing);
        } else {
            tracing::debug!(subject_uri = %li.subject_uri, "list item is not a thing; skipped");
        }
    }
    Ok((items, cursor))
}

async fn thing_basic_by_uri(state: &AppState, uri: &str) -> AppResult<Option<ThingViewBasic>> {
    match fetch_thing_by_uri(state, uri).await? {
        Some(row) => Ok(Some(thing_basic_view(state, row).await?)),
        None => Ok(None),
    }
}

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

pub(super) struct ProfileStats {
    pub thing_count: i64,
    pub total_likes: 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,
    })
}

pub(super) async fn handle_for_did(state: &AppState, did: &str) -> AppResult<Option<String>> {
    #[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.
async fn resolve_handle(state: &AppState, did: &Did) -> AppResult<String> {
    if let Some(h) = handle_for_did(state, did.as_ref()).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 handle_str = resolve_handle(state, did).await?;
    let record_value: serde_json::Value = 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: to_data(&record_value).map_err(|e| internal(format!("record encode: {e}")))?,
        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),
        indexed_at: Some(datetime_millis(row.indexed_at)),
        extra_data: None,
    }))
}
