//! Item CRUD: creation, listing, text body updates, and ownership lookups.
//!
//! Bulk and structural operations (move, bulk_*, duplicate) live in the
//! `bulk` submodule and are re-exported flat so call sites still see
//! `db::items::bulk_publish` etc.

mod bulk;
mod media;

pub use bulk::*;
pub use media::*;

use sqlx::PgPool;

use super::enums::{AiTier, ItemType};
use super::models::{DbItem, ItemS3KeyRow};
use super::{ItemId, MtThreadId, PriceCents, ProjectId, UserId};
use crate::error::Result;

/// Per-viewer display flags for an item's store page (wishlisted / in-cart /
/// number of the viewer's collections containing it). Display-only, none of
/// these gate access.
#[derive(Debug, Default)]
pub struct ViewerItemFlags {
    pub is_wishlisted: bool,
    pub in_cart: bool,
    pub collection_count: i64,
}

/// Fetch a viewer's display flags for an item in ONE query instead of three
/// sequential round-trips (wishlist EXISTS, cart EXISTS, collections COUNT).
/// Fewer round-trips AND fewer pooled connections per item-page render, the
/// collapse the public item page wants (vs. fanning the three out concurrently,
/// which would add connection pressure on a hot public path).
#[tracing::instrument(skip_all)]
pub async fn get_viewer_item_flags(
    pool: &PgPool,
    user_id: UserId,
    item_id: ItemId,
) -> Result<ViewerItemFlags> {
    let (is_wishlisted, in_cart, collection_count) = sqlx::query_as::<_, (bool, bool, i64)>(
        r"
        SELECT
            EXISTS(SELECT 1 FROM wishlists  WHERE user_id = $1 AND item_id = $2),
            EXISTS(SELECT 1 FROM cart_items WHERE user_id = $1 AND item_id = $2),
            (SELECT COUNT(*) FROM collections c
               JOIN collection_items ci ON ci.collection_id = c.id
               WHERE c.user_id = $1 AND ci.item_id = $2)
        ",
    )
    .bind(user_id)
    .bind(item_id)
    .fetch_one(pool)
    .await?;

    Ok(ViewerItemFlags {
        is_wishlisted,
        in_cart,
        collection_count,
    })
}

/// Insert a new item into a project and return the created row.
///
/// Auto-generates a URL-safe slug from the title. If the slug collides with
/// an existing item in the same project, appends a counter suffix.
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all)]
/// Find an existing untitled, unpublished wizard draft for a project + type.
///
/// The single-type-card item wizard used to `create_item` a fresh "Untitled"
/// row on every GET, so a prefetch or a re-visit piled up orphan drafts. The
/// wizard now reuses the most recent such draft (newest first) instead of
/// minting another (Run 21). A draft the creator actually names/publishes no
/// longer matches, so real items are never recycled.
#[tracing::instrument(skip_all)]
pub async fn find_untitled_wizard_draft(
    pool: &PgPool,
    project_id: ProjectId,
    item_type: ItemType,
) -> Result<Option<DbItem>> {
    let item = sqlx::query_as::<_, DbItem>(
        r"
        SELECT * FROM items
        WHERE project_id = $1 AND item_type = $2
          AND title = 'Untitled' AND is_public = false AND deleted_at IS NULL
        ORDER BY created_at DESC
        LIMIT 1
        ",
    )
    .bind(project_id)
    .bind(item_type)
    .fetch_optional(pool)
    .await?;
    Ok(item)
}

#[allow(clippy::too_many_arguments)]
pub async fn create_item(
    pool: &PgPool,
    project_id: ProjectId,
    title: &str,
    description: Option<&str>,
    price_cents: PriceCents,
    item_type: ItemType,
    ai_tier: AiTier,
    ai_disclosure: Option<&str>,
) -> Result<DbItem> {
    // Slug uniqueness is enforced by the per-project unique index: insert the
    // bare slug, and on a 23505 collision append an incrementing suffix and
    // retry (`base`, `base-2`, `base-3`, ...). The retry IS the dedup, a
    // pre-insert `SELECT EXISTS` check would just be a redundant round-trip on
    // the common (no-collision) path and a TOCTOU on the racy one.
    let base = crate::helpers::slugify(title).to_string();
    crate::helpers::insert_with_unique_slug(&base, |slug| async move {
        let slug = super::validated_types::Slug::from_trusted(slug);
        sqlx::query_as::<_, DbItem>(
            r"
            INSERT INTO items (project_id, title, description, price_cents, item_type, slug, ai_tier, ai_disclosure)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            RETURNING *
            ",
        )
        .bind(project_id)
        .bind(title)
        .bind(description)
        .bind(price_cents)
        .bind(item_type)
        .bind(&slug)
        .bind(ai_tier)
        .bind(ai_disclosure)
        .fetch_one(pool)
        .await
        .map_err(Into::into)
    })
    .await
}

/// Fetch an item by primary key. Returns `None` if not found.
#[tracing::instrument(skip_all)]
pub async fn get_item_by_id(pool: &PgPool, id: ItemId) -> Result<Option<DbItem>> {
    let item = sqlx::query_as::<_, DbItem>("SELECT * FROM items WHERE id = $1")
        .bind(id)
        .fetch_optional(pool)
        .await?;

    Ok(item)
}

/// Fetch multiple public items by id in one query. Replaces a per-id
/// `get_item_by_id` loop (N+1) where the caller only wants the public rows.
/// Order is not guaranteed.
#[tracing::instrument(skip_all)]
pub async fn get_public_items_by_ids(pool: &PgPool, ids: &[ItemId]) -> Result<Vec<DbItem>> {
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    let items =
        sqlx::query_as::<_, DbItem>("SELECT * FROM items WHERE id = ANY($1) AND is_public = true")
            .bind(ids)
            .fetch_all(pool)
            .await?;

    Ok(items)
}

/// Fetch titles for a batch of item IDs. Returns (item_id, title) pairs.
#[tracing::instrument(skip_all)]
pub async fn get_item_titles_batch(pool: &PgPool, ids: &[ItemId]) -> Result<Vec<(ItemId, String)>> {
    if ids.is_empty() {
        return Ok(vec![]);
    }
    let rows: Vec<(ItemId, String)> =
        sqlx::query_as("SELECT id, title FROM items WHERE id = ANY($1)")
            .bind(ids)
            .fetch_all(pool)
            .await?;

    Ok(rows)
}

/// Fetch project_id for a batch of item IDs. Returns (item_id, project_id) pairs.
///
/// Used by bulk operations to verify all items belong to the same project in one query.
#[tracing::instrument(skip_all)]
pub async fn get_item_project_ids_batch(
    pool: &PgPool,
    ids: &[ItemId],
) -> Result<Vec<(ItemId, super::ProjectId)>> {
    if ids.is_empty() {
        return Ok(vec![]);
    }
    let rows: Vec<(ItemId, super::ProjectId)> =
        sqlx::query_as("SELECT id, project_id FROM items WHERE id = ANY($1)")
            .bind(ids)
            .fetch_all(pool)
            .await?;

    Ok(rows)
}

/// List all items in a project, ordered by sort_order then newest.
///
/// Capped at 500 as a safety limit.
#[tracing::instrument(skip_all)]
pub async fn get_items_by_project(pool: &PgPool, project_id: ProjectId) -> Result<Vec<DbItem>> {
    let items = sqlx::query_as::<_, DbItem>(
        // No LIMIT: one project's live item set is naturally bounded; the old
        // flat cap silently truncated dashboard/exports (audit Run 17 Perf).
        "SELECT * FROM items WHERE project_id = $1 AND deleted_at IS NULL ORDER BY sort_order, created_at DESC",
    )
    .bind(project_id)
    .fetch_all(pool)
    .await?;

    Ok(items)
}

/// List all items across all projects owned by a user, newest first.
///
/// Capped at 500 as a safety limit.
#[tracing::instrument(skip_all)]
pub async fn get_items_by_user(pool: &PgPool, user_id: UserId) -> Result<Vec<DbItem>> {
    /// Page size for the accumulating fetch.
    const PAGE: i64 = 1_000;
    /// Hard cap so even a huge catalog can't load an unbounded result set.
    const MAX_ITEMS: usize = 100_000;

    // Previously a flat `LIMIT 500` silently truncated the result, a creator with
    // more than 500 items exported only their newest 500 with no indication
    // (data loss in the export). Page through instead, releasing the connection
    // between pages, up to a sane cap (Perf, Run 9).
    let mut all = Vec::new();
    let mut offset = 0i64;
    loop {
        let page = sqlx::query_as::<_, DbItem>(
            r"
            SELECT i.* FROM items i
            JOIN projects p ON i.project_id = p.id
            WHERE p.user_id = $1 AND i.deleted_at IS NULL
            ORDER BY i.created_at DESC
            LIMIT $2 OFFSET $3
            ",
        )
        .bind(user_id)
        .bind(PAGE)
        .bind(offset)
        .fetch_all(pool)
        .await?;
        let n = page.len();
        all.extend(page);
        offset += n as i64;
        if (n as i64) < PAGE || all.len() >= MAX_ITEMS {
            break;
        }
    }

    Ok(all)
}

/// Count a user's (non-deleted) items without materializing the rows. For
/// callers that only need the total, this avoids fetching up to 500 full rows
/// just to `.len()` them (ultra-fuzz Run 6 R6-Perf-M4).
#[tracing::instrument(skip_all)]
pub async fn count_items_by_user(pool: &PgPool, user_id: UserId) -> Result<i64> {
    let count: i64 = sqlx::query_scalar(
        r"
        SELECT COUNT(*) FROM items i
        JOIN projects p ON i.project_id = p.id
        WHERE p.user_id = $1 AND i.deleted_at IS NULL
        ",
    )
    .bind(user_id)
    .fetch_one(pool)
    .await?;
    Ok(count)
}

/// Count items per project for all projects owned by a user.
///
/// Returns `(project_id, count)` tuples. Used by the CLI to avoid N+1 queries.
#[tracing::instrument(skip_all)]
pub async fn count_items_by_user_projects(
    pool: &PgPool,
    user_id: UserId,
) -> Result<Vec<(ProjectId, i64)>> {
    let rows: Vec<(ProjectId, i64)> = sqlx::query_as(
        r"
        SELECT i.project_id, COUNT(*) AS cnt
        FROM items i
        JOIN projects p ON i.project_id = p.id
        WHERE p.user_id = $1 AND i.deleted_at IS NULL
        GROUP BY i.project_id
        ",
    )
    .bind(user_id)
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// List only public items in a project, ordered by sort_order then newest.
///
/// Capped at 500 as a safety limit.
#[tracing::instrument(skip_all)]
pub async fn get_public_items_by_project(
    pool: &PgPool,
    project_id: ProjectId,
) -> Result<Vec<DbItem>> {
    let items = sqlx::query_as::<_, DbItem>(
        "SELECT * FROM items WHERE project_id = $1 AND is_public = true AND listed = true ORDER BY sort_order, created_at DESC",
    )
    .bind(project_id)
    .fetch_all(pool)
    .await?;

    Ok(items)
}

/// Partially update an item's fields (COALESCE keeps existing values when `None`).
///
/// `publish_at` uses a double-Option: `None` = no change, `Some(None)` = clear schedule,
/// `Some(Some(dt))` = set schedule.
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all)]
pub async fn update_item(
    pool: &PgPool,
    id: ItemId,
    user_id: UserId,
    title: Option<&str>,
    description: Option<&str>,
    price_cents: Option<PriceCents>,
    item_type: Option<ItemType>,
    is_public: Option<bool>,
    pwyw_enabled: Option<bool>,
    pwyw_min_cents: Option<PriceCents>,
    publish_at: Option<Option<chrono::DateTime<chrono::Utc>>>,
    web_only: Option<bool>,
    ai_tier: Option<AiTier>,
    ai_disclosure: Option<Option<&str>>,
) -> Result<DbItem> {
    // Flatten the double-Option: if outer is None, pass current DB value (via SQL CASE).
    // $10 = whether to update publish_at, $11 = the new value (NULL to clear).
    let update_publish_at = publish_at.is_some();
    let publish_at_value = publish_at.flatten();

    // ai_disclosure uses the same double-Option pattern as publish_at:
    // None = no change, Some(None) = clear, Some(Some(text)) = set.
    let update_ai_disclosure = ai_disclosure.is_some();
    let ai_disclosure_value = ai_disclosure.flatten();

    let item = sqlx::query_as::<_, DbItem>(
        r"
        UPDATE items
        SET title = COALESCE($3, title),
            description = COALESCE($4, description),
            price_cents = COALESCE($5, price_cents),
            item_type = COALESCE($6, item_type),
            is_public = CASE WHEN removed_by_admin AND $7 = true THEN false ELSE COALESCE($7, is_public) END,
            pwyw_enabled = COALESCE($8, pwyw_enabled),
            pwyw_min_cents = COALESCE($9, pwyw_min_cents),
            publish_at = CASE WHEN $10 THEN $11 ELSE publish_at END,
            web_only = COALESCE($12, web_only),
            ai_tier = COALESCE($13, ai_tier),
            ai_disclosure = CASE WHEN $14 THEN $15 ELSE ai_disclosure END
        WHERE id = $1
          AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
        RETURNING *
        ",
    )
    .bind(id)
    .bind(user_id)
    .bind(title)
    .bind(description)
    .bind(price_cents)
    .bind(item_type)
    .bind(is_public)
    .bind(pwyw_enabled)
    .bind(pwyw_min_cents)
    .bind(update_publish_at)
    .bind(publish_at_value)
    .bind(web_only)
    .bind(ai_tier)
    .bind(update_ai_disclosure)
    .bind(ai_disclosure_value)
    .fetch_one(pool)
    .await?;

    Ok(item)
}

/// Publish all items whose scheduled publish time has passed.
///
/// Atomically sets `is_public = true` and clears `publish_at`, returning the
/// newly published items so the caller can send release announcements.
#[tracing::instrument(skip_all)]
pub async fn publish_scheduled_items(pool: &PgPool) -> Result<Vec<DbItem>> {
    let items = sqlx::query_as::<_, DbItem>(
        r"
        UPDATE items
        SET is_public = true, publish_at = NULL, updated_at = NOW()
        WHERE publish_at IS NOT NULL AND publish_at <= NOW() AND is_public = false AND removed_by_admin = false
        RETURNING *
        ",
    )
    .fetch_all(pool)
    .await?;

    Ok(items)
}

/// Soft-delete an item (sets deleted_at, recoverable for 7 days).
#[tracing::instrument(skip_all)]
pub async fn delete_item(pool: &PgPool, id: ItemId, user_id: UserId) -> Result<()> {
    sqlx::query(
        r"
        UPDATE items SET deleted_at = NOW(), is_public = false
        WHERE id = $1 AND deleted_at IS NULL
          AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
        ",
    )
    .bind(id)
    .bind(user_id)
    .execute(pool)
    .await?;

    Ok(())
}

/// Restore a soft-deleted item.
#[tracing::instrument(skip_all)]
pub async fn restore_item(pool: &PgPool, id: ItemId, user_id: UserId) -> Result<bool> {
    let result = sqlx::query(
        r"
        UPDATE items SET deleted_at = NULL
        WHERE id = $1 AND deleted_at IS NOT NULL
          AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
        ",
    )
    .bind(id)
    .bind(user_id)
    .execute(pool)
    .await?;

    Ok(result.rows_affected() > 0)
}

/// Get soft-deleted items for a project (for the "Recently Deleted" section).
#[tracing::instrument(skip_all)]
pub async fn get_deleted_items_by_project(
    pool: &PgPool,
    project_id: ProjectId,
) -> Result<Vec<DbItem>> {
    let items = sqlx::query_as::<_, DbItem>(
        "SELECT * FROM items WHERE project_id = $1 AND deleted_at IS NOT NULL ORDER BY deleted_at DESC",
    )
    .bind(project_id)
    .fetch_all(pool)
    .await?;

    Ok(items)
}

/// Collect S3 keys from items that are about to be purged (soft-deleted >7 days).
/// Returns all non-null S3 keys (audio, cover, video) so they can be deleted
/// from S3 before the DB rows are removed.
#[tracing::instrument(skip_all)]
pub async fn get_expired_deleted_item_s3_keys(pool: &PgPool) -> Result<Vec<String>> {
    let keys: Vec<String> = sqlx::query_scalar(
        r"
        SELECT k FROM (
            SELECT unnest(ARRAY_REMOVE(ARRAY[audio_s3_key, cover_s3_key, video_s3_key], NULL)) AS k
            FROM items
            WHERE deleted_at IS NOT NULL AND deleted_at < NOW() - INTERVAL '7 days'
              AND (audio_s3_key IS NOT NULL OR cover_s3_key IS NOT NULL OR video_s3_key IS NOT NULL)
        ) sub
        ",
    )
    .fetch_all(pool)
    .await?;

    Ok(keys)
}

/// Collect S3 keys from versions belonging to items about to be purged.
/// Must be called before purge since CASCADE delete destroys version rows.
#[tracing::instrument(skip_all)]
pub async fn get_expired_deleted_item_version_s3_keys(pool: &PgPool) -> Result<Vec<String>> {
    let keys: Vec<String> = sqlx::query_scalar(
        r"
        SELECT v.s3_key
        FROM versions v
        JOIN items i ON v.item_id = i.id
        WHERE i.deleted_at IS NOT NULL AND i.deleted_at < NOW() - INTERVAL '7 days'
          AND v.s3_key IS NOT NULL
        ",
    )
    .fetch_all(pool)
    .await?;

    Ok(keys)
}

/// Sum total file sizes per user for items about to be purged, including version files.
/// Returns (user_id, total_bytes) pairs for storage decrement.
#[tracing::instrument(skip_all)]
pub async fn get_expired_deleted_item_storage_by_user<'e>(
    executor: impl sqlx::PgExecutor<'e>,
) -> Result<Vec<(super::UserId, i64)>> {
    let rows: Vec<(super::UserId, i64)> = sqlx::query_as(
        r"
        SELECT p.user_id,
               COALESCE(SUM(
                   COALESCE(i.audio_file_size_bytes, 0) +
                   COALESCE(i.cover_file_size_bytes, 0) +
                   COALESCE(i.video_file_size_bytes, 0) +
                   COALESCE(ver.version_bytes, 0) +
                   COALESCE(igal.gallery_bytes, 0)
               ), 0)::BIGINT AS total_bytes
        FROM items i
        JOIN projects p ON i.project_id = p.id
        LEFT JOIN LATERAL (
            SELECT COALESCE(SUM(v.file_size_bytes), 0)::BIGINT AS version_bytes
            FROM versions v
            WHERE v.item_id = i.id AND v.file_size_bytes IS NOT NULL
        ) ver ON true
        LEFT JOIN LATERAL (
            SELECT COALESCE(SUM(ii.file_size_bytes), 0)::BIGINT AS gallery_bytes
            FROM item_images ii WHERE ii.item_id = i.id
        ) igal ON true
        WHERE i.deleted_at IS NOT NULL AND i.deleted_at < NOW() - INTERVAL '7 days'
        GROUP BY p.user_id
        ",
    )
    .fetch_all(executor)
    .await?;

    Ok(rows)
}

/// Collect all S3 keys from items belonging to a project (audio, cover, video).
#[tracing::instrument(skip_all)]
pub async fn get_project_item_s3_keys(
    pool: &PgPool,
    project_id: super::ProjectId,
) -> Result<Vec<String>> {
    let keys: Vec<String> = sqlx::query_scalar(
        r"
        SELECT k FROM (
            SELECT unnest(ARRAY_REMOVE(ARRAY[audio_s3_key, cover_s3_key, video_s3_key], NULL)) AS k
            FROM items
            WHERE project_id = $1
              AND (audio_s3_key IS NOT NULL OR cover_s3_key IS NOT NULL OR video_s3_key IS NOT NULL)
        ) sub
        ",
    )
    .bind(project_id)
    .fetch_all(pool)
    .await?;

    Ok(keys)
}

/// Collect S3 keys from versions belonging to items in a project.
#[tracing::instrument(skip_all)]
pub async fn get_project_version_s3_keys(
    pool: &PgPool,
    project_id: super::ProjectId,
) -> Result<Vec<String>> {
    let keys: Vec<String> = sqlx::query_scalar(
        r"
        SELECT v.s3_key
        FROM versions v
        JOIN items i ON v.item_id = i.id
        WHERE i.project_id = $1 AND v.s3_key IS NOT NULL
        ",
    )
    .bind(project_id)
    .fetch_all(pool)
    .await?;

    Ok(keys)
}

/// Sum total file sizes for everything in a project that charges storage:
/// item audio/cover/video, versions, the item and project gallery carousels,
/// and the project cover image. Used to refund storage on project delete, it
/// must cover every category the upload paths charge, or the delete under-
/// refunds and leaves the creator's counter inflated (Run #18 Storage B2).
#[tracing::instrument(skip_all)]
pub async fn get_project_storage_bytes(pool: &PgPool, project_id: super::ProjectId) -> Result<i64> {
    let total: i64 = sqlx::query_scalar(
        r"
        SELECT (
            COALESCE((
                SELECT SUM(
                    COALESCE(i.audio_file_size_bytes, 0) +
                    COALESCE(i.cover_file_size_bytes, 0) +
                    COALESCE(i.video_file_size_bytes, 0) +
                    COALESCE(ver.version_bytes, 0) +
                    COALESCE(igal.gallery_bytes, 0)
                )::BIGINT
                FROM items i
                LEFT JOIN LATERAL (
                    SELECT COALESCE(SUM(v.file_size_bytes), 0)::BIGINT AS version_bytes
                    FROM versions v
                    WHERE v.item_id = i.id AND v.file_size_bytes IS NOT NULL
                ) ver ON true
                LEFT JOIN LATERAL (
                    SELECT COALESCE(SUM(ii.file_size_bytes), 0)::BIGINT AS gallery_bytes
                    FROM item_images ii WHERE ii.item_id = i.id
                ) igal ON true
                WHERE i.project_id = $1
            ), 0)
            + COALESCE((
                SELECT SUM(pi.file_size_bytes)::BIGINT
                FROM project_images pi WHERE pi.project_id = $1
            ), 0)
            + COALESCE((
                SELECT p.cover_image_size_bytes FROM projects p WHERE p.id = $1
            ), 0)
        )::BIGINT
        ",
    )
    .bind(project_id)
    .fetch_one(pool)
    .await?;

    Ok(total)
}

/// Permanently delete items that were soft-deleted more than 7 days ago.
#[tracing::instrument(skip_all)]
pub async fn purge_expired_deleted_items<'e>(executor: impl sqlx::PgExecutor<'e>) -> Result<u64> {
    let result = sqlx::query(
        "DELETE FROM items WHERE deleted_at IS NOT NULL AND deleted_at < NOW() - INTERVAL '7 days'",
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// Update the audio S3 key for an item (defense-in-depth: verifies ownership)
#[tracing::instrument(skip_all)]
pub async fn update_item_audio_s3_key(
    pool: &PgPool,
    item_id: ItemId,
    user_id: UserId,
    s3_key: &str,
) -> Result<DbItem> {
    let item = sqlx::query_as::<_, DbItem>(
        r"
        UPDATE items
        SET audio_s3_key = $2, updated_at = NOW()
        WHERE id = $1
          AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
        RETURNING *
        ",
    )
    .bind(item_id)
    .bind(s3_key)
    .bind(user_id)
    .fetch_one(pool)
    .await?;

    Ok(item)
}

/// Update the cover image S3 key for an item (defense-in-depth: verifies ownership)
#[tracing::instrument(skip_all)]
pub async fn update_item_cover_s3_key(
    pool: &PgPool,
    item_id: ItemId,
    user_id: UserId,
    s3_key: &str,
) -> Result<DbItem> {
    let item = sqlx::query_as::<_, DbItem>(
        r"
        UPDATE items
        SET cover_s3_key = $2, updated_at = NOW()
        WHERE id = $1
          AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
        RETURNING *
        ",
    )
    .bind(item_id)
    .bind(s3_key)
    .bind(user_id)
    .fetch_one(pool)
    .await?;

    Ok(item)
}

/// List all public items across all public projects owned by a user, newest first.
///
/// Used by the creator RSS feed to avoid O(projects) queries.
/// Capped at 50 items (standard RSS feed size).
#[tracing::instrument(skip_all)]
pub async fn get_public_items_by_user(pool: &PgPool, user_id: UserId) -> Result<Vec<DbItem>> {
    let items = sqlx::query_as::<_, DbItem>(
        r"
        SELECT i.* FROM items i
        JOIN projects p ON i.project_id = p.id
        WHERE p.user_id = $1 AND p.is_public = true AND i.is_public = true AND i.listed = true
        ORDER BY i.created_at DESC
        LIMIT 50
        ",
    )
    .bind(user_id)
    .fetch_all(pool)
    .await?;

    Ok(items)
}

/// Get the owner (user_id) of an item through its project
#[tracing::instrument(skip_all)]
pub async fn get_item_owner(pool: &PgPool, item_id: ItemId) -> Result<Option<UserId>> {
    let owner: Option<UserId> = sqlx::query_scalar(
        r"
        SELECT p.user_id
        FROM items i
        JOIN projects p ON i.project_id = p.id
        WHERE i.id = $1
        ",
    )
    .bind(item_id)
    .fetch_optional(pool)
    .await?;

    Ok(owner)
}

/// Pre-flight access check for content streaming/download. Returns item data
/// plus ownership, purchase, subscription, and bundle access.
/// Returns `None` if the item does not exist (or is soft-deleted).
#[derive(Debug)]
pub struct ItemAccessCheck {
    // Ownership
    pub owner_id: UserId,
    // Access flags (only meaningful when a user_id is provided)
    pub has_purchased: bool,
    /// Subscription access as a witness, not a bool: `Some` only when the sealed
    /// gate confirmed an active, in-period subscription. Feed it straight into
    /// `pricing::AccessContext::subscription`.
    pub subscription: Option<crate::db::subscriptions::SubscriptionGate>,
    pub has_bundle_access: bool,
}

/// Ownership + the access flags that are pure transaction/bundle lookups.
/// `has_subscription` is deliberately NOT here: it must come from
/// [`crate::db::subscriptions::has_access`], the single sealed access gate, so
/// the grant predicate (incl. the `current_period_end` clause) cannot be
/// hand-written and drift. (Payments S1 / CHRONIC 2: this query previously
/// inlined the predicate and dropped the period clause on the download path.)
#[derive(Debug, sqlx::FromRow)]
struct ItemAccessPartial {
    owner_id: UserId,
    has_purchased: bool,
    has_bundle_access: bool,
}

#[tracing::instrument(skip_all)]
pub async fn check_item_access(
    pool: &PgPool,
    item_id: ItemId,
    user_id: Option<UserId>,
) -> Result<Option<ItemAccessCheck>> {
    // When no user is provided, access flags are all false
    let uid = user_id.unwrap_or(UserId::nil());
    let partial = sqlx::query_as::<_, ItemAccessPartial>(
        r"
        SELECT
            p.user_id AS owner_id,
            EXISTS(
                SELECT 1 FROM transactions t
                WHERE t.item_id = i.id AND t.buyer_id = $2 AND t.status = 'completed'
            ) AS has_purchased,
            EXISTS(
                SELECT 1 FROM bundle_items bi
                JOIN transactions bt ON bt.item_id = bi.bundle_id
                WHERE bi.item_id = i.id AND bt.buyer_id = $2 AND bt.status = 'completed'
            ) AS has_bundle_access
        FROM items i
        JOIN projects p ON i.project_id = p.id
        WHERE i.id = $1 AND i.deleted_at IS NULL
        ",
    )
    .bind(item_id)
    .bind(uid)
    .fetch_optional(pool)
    .await?;

    let Some(partial) = partial else {
        return Ok(None);
    };

    // Subscription access goes through the one sealed gate, which returns a
    // witness. Anonymous callers (no user_id) can't hold a subscription, so skip
    // the query entirely.
    let subscription = match user_id {
        Some(real_uid) => {
            crate::db::subscriptions::SubscriptionGate::check(
                pool,
                real_uid,
                crate::db::subscriptions::SubscriptionScope::Item(item_id),
            )
            .await?
        }
        None => None,
    };

    Ok(Some(ItemAccessCheck {
        owner_id: partial.owner_id,
        has_purchased: partial.has_purchased,
        subscription,
        has_bundle_access: partial.has_bundle_access,
    }))
}

/// Update the text body content for an item (for articles/essays).
///
/// Recomputes `word_count` and `reading_time_minutes` on every save.
/// Reading time uses ~200 wpm (average adult reading speed) with floor
/// division, clamped to a minimum of 1 minute so the UI never shows "0 min".
#[tracing::instrument(skip_all)]
pub async fn update_item_text(
    pool: &PgPool,
    id: ItemId,
    user_id: UserId,
    body: &str,
) -> Result<DbItem> {
    let word_count = body.split_whitespace().count() as i32;
    let reading_time = (word_count / 200).max(1);

    let item = sqlx::query_as::<_, DbItem>(
        r"
        UPDATE items
        SET body = $3, word_count = $4, reading_time_minutes = $5, updated_at = NOW()
        WHERE id = $1
          AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
        RETURNING *
        ",
    )
    .bind(id)
    .bind(user_id)
    .bind(body)
    .bind(word_count)
    .bind(reading_time)
    .fetch_one(pool)
    .await?;

    Ok(item)
}

/// Update the license key settings on an item.
#[tracing::instrument(skip_all)]
pub async fn update_item_license_settings(
    pool: &PgPool,
    item_id: ItemId,
    user_id: UserId,
    enable_license_keys: bool,
    default_max_activations: Option<i32>,
) -> Result<()> {
    sqlx::query(
        r"
        UPDATE items
        SET enable_license_keys = $3, default_max_activations = $4, updated_at = NOW()
        WHERE id = $1
          AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
        ",
    )
    .bind(item_id)
    .bind(user_id)
    .bind(enable_license_keys)
    .bind(default_max_activations)
    .execute(pool)
    .await?;

    Ok(())
}

/// Update the license text preset and optional custom text on an item.
#[tracing::instrument(skip_all)]
pub async fn update_item_license_text(
    pool: &PgPool,
    item_id: ItemId,
    user_id: UserId,
    license_preset: Option<&str>,
    custom_license_text: Option<&str>,
) -> Result<()> {
    sqlx::query(
        r"
        UPDATE items
        SET license_preset = $3, custom_license_text = $4, updated_at = NOW()
        WHERE id = $1
          AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
        ",
    )
    .bind(item_id)
    .bind(user_id)
    .bind(license_preset)
    .bind(custom_license_text)
    .execute(pool)
    .await?;

    Ok(())
}

/// Increment the denormalized sales_count for an item (called on purchase/claim).
///
/// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
/// can include this in a larger transaction when needed.
#[tracing::instrument(skip_all)]
pub async fn increment_sales_count<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    item_id: ItemId,
) -> Result<()> {
    sqlx::query("UPDATE items SET sales_count = sales_count + 1 WHERE id = $1")
        .bind(item_id)
        .execute(executor)
        .await?;

    Ok(())
}

/// Decrement the denormalized sales_count for an item (called on refund/unclaim).
#[tracing::instrument(skip_all)]
pub async fn decrement_sales_count<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    item_id: ItemId,
) -> Result<()> {
    sqlx::query("UPDATE items SET sales_count = GREATEST(sales_count - 1, 0) WHERE id = $1")
        .bind(item_id)
        .execute(executor)
        .await?;

    Ok(())
}

/// Atomically mark an item as having had its release announced.
/// Returns false if already announced (prevents duplicate announcements on unpublish/republish).
#[tracing::instrument(skip_all)]
pub async fn mark_release_announced(pool: &PgPool, item_id: ItemId) -> Result<bool> {
    let result = sqlx::query(
        "UPDATE items SET release_announced_at = NOW() WHERE id = $1 AND release_announced_at IS NULL",
    )
    .bind(item_id)
    .execute(pool)
    .await?;

    Ok(result.rows_affected() > 0)
}

/// Set the linked MT thread ID for an item.
#[tracing::instrument(skip_all)]
pub async fn set_mt_thread_id(pool: &PgPool, item_id: ItemId, thread_id: MtThreadId) -> Result<()> {
    sqlx::query("UPDATE items SET mt_thread_id = $2 WHERE id = $1")
        .bind(item_id)
        .bind(thread_id)
        .execute(pool)
        .await?;
    Ok(())
}

/// Collect all S3 keys for items owned by a user (audio + cover + video).
///
/// Returns item title, project slug, and optional S3 keys for audio/cover/video.
/// Only includes items that have at least one S3 key.
#[tracing::instrument(skip_all)]
pub async fn get_user_s3_keys(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemS3KeyRow>> {
    let rows = sqlx::query_as::<_, ItemS3KeyRow>(
        r"
        SELECT i.title, p.id AS project_id, p.slug AS project_slug,
               i.audio_s3_key, i.cover_s3_key, i.video_s3_key,
               i.audio_file_size_bytes, i.cover_file_size_bytes, i.video_file_size_bytes
        FROM items i JOIN projects p ON i.project_id = p.id
        WHERE p.user_id = $1 AND (i.audio_s3_key IS NOT NULL OR i.cover_s3_key IS NOT NULL OR i.video_s3_key IS NOT NULL)
        ORDER BY p.slug, i.sort_order
        ",
        // No LIMIT: this drives per-user S3 storage accounting + cleanup; a cap
        // silently dropped keys for a >500-item user, orphaning their objects
        // (audit Run 17 Perf, same silent-truncation class).
    )
    .bind(user_id)
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// Check whether a user has at least one public item across all their projects.
#[tracing::instrument(skip_all)]
pub async fn has_public_item_by_user(pool: &PgPool, user_id: UserId) -> Result<bool> {
    let exists: bool = sqlx::query_scalar(
        r"
        SELECT EXISTS(
            SELECT 1 FROM items i
            JOIN projects p ON i.project_id = p.id
            WHERE p.user_id = $1 AND i.is_public = true
        )
        ",
    )
    .bind(user_id)
    .fetch_one(pool)
    .await?;

    Ok(exists)
}

/// Increment the play count for an item (called on audio/video stream request).
#[tracing::instrument(skip_all)]
pub async fn increment_play_count(pool: &PgPool, item_id: ItemId) -> Result<()> {
    sqlx::query("UPDATE items SET play_count = play_count + 1 WHERE id = $1")
        .bind(item_id)
        .execute(pool)
        .await?;

    Ok(())
}

/// Record a unique listener and increment unique_play_count if this is the first
/// play by this user. Returns true if new unique listener, false if already counted.
/// Runs in a transaction so the user_plays INSERT and count UPDATE are atomic.
#[tracing::instrument(skip_all)]
pub async fn record_unique_play(
    pool: &PgPool,
    user_id: super::UserId,
    item_id: ItemId,
) -> Result<bool> {
    let mut tx = pool.begin().await?;

    let result = sqlx::query(
        "INSERT INTO user_plays (user_id, item_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
    )
    .bind(user_id)
    .bind(item_id)
    .execute(&mut *tx)
    .await?;

    if result.rows_affected() > 0 {
        sqlx::query("UPDATE items SET unique_play_count = unique_play_count + 1 WHERE id = $1")
            .bind(item_id)
            .execute(&mut *tx)
            .await?;
        tx.commit().await?;
        Ok(true)
    } else {
        tx.commit().await?;
        Ok(false)
    }
}

/// Increment the item-level download count (called alongside per-version increment).
#[tracing::instrument(skip_all)]
pub async fn increment_item_download_count(pool: &PgPool, item_id: ItemId) -> Result<()> {
    sqlx::query("UPDATE items SET download_count = download_count + 1 WHERE id = $1")
        .bind(item_id)
        .execute(pool)
        .await?;

    Ok(())
}

/// Fetch a public item by project ID and slug (for custom domain routing).
#[tracing::instrument(skip_all)]
pub async fn get_item_by_project_and_slug(
    pool: &PgPool,
    project_id: ProjectId,
    slug: &str,
) -> Result<Option<DbItem>> {
    let item = sqlx::query_as::<_, DbItem>(
        "SELECT * FROM items WHERE project_id = $1 AND slug = $2 AND is_public = true",
    )
    .bind(project_id)
    .bind(slug)
    .fetch_optional(pool)
    .await?;

    Ok(item)
}

/// Hide every public item for ALL given creators in one statement, for the
/// post-grace scheduler sweep, replaces N per-user UPDATEs on the lock-held tick
/// (Perf-S4, Run 9). Returns the total rows affected. No-op on an empty slice.
///
/// Stamps `hidden_by_suspension_at` so the matching `unhide_all_items_for_user`
/// restores ONLY what this sweep hid, never a creator's genuine drafts.
#[tracing::instrument(skip_all)]
pub async fn hide_all_items_for_users(pool: &PgPool, user_ids: &[UserId]) -> Result<u64> {
    if user_ids.is_empty() {
        return Ok(0);
    }
    let result = sqlx::query(
        r"
        UPDATE items SET is_public = false, hidden_by_suspension_at = NOW()
        WHERE project_id IN (SELECT id FROM projects WHERE user_id = ANY($1))
          AND is_public = true
        ",
    )
    .bind(user_ids)
    .execute(pool)
    .await?;

    Ok(result.rows_affected())
}

/// Unhide items for a user that the post-grace sweep hid. Used when a creator
/// re-subscribes. Returns the number of items unhidden.
///
/// Restores ONLY items stamped by `hide_all_items_for_users`
/// (`hidden_by_suspension_at IS NOT NULL`) and clears the stamp, a draft the
/// creator never published (`is_public = false` with a NULL stamp) is left
/// private, so a lapsed-then-renewed subscription can't leak unpublished work.
#[tracing::instrument(skip_all)]
pub async fn unhide_all_items_for_user(pool: &PgPool, user_id: UserId) -> Result<u64> {
    let result = sqlx::query(
        r"
        UPDATE items SET is_public = true, hidden_by_suspension_at = NULL
        WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
          AND hidden_by_suspension_at IS NOT NULL
          AND removed_at IS NULL
        ",
    )
    .bind(user_id)
    .execute(pool)
    .await?;

    Ok(result.rows_affected())
}

/// Admin: remove an item (hide from public, record reason). The item stays in the DB
/// and the creator can see it in their dashboard with the removal reason.
#[tracing::instrument(skip_all)]
pub async fn admin_remove_item(pool: &PgPool, item_id: ItemId, reason: &str) -> Result<DbItem> {
    let item = sqlx::query_as::<_, DbItem>(
        r"
        UPDATE items
        SET removed_by_admin = true,
            removal_reason = $2,
            removed_at = NOW(),
            is_public = false
        WHERE id = $1
        RETURNING *
        ",
    )
    .bind(item_id)
    .bind(reason)
    .fetch_one(pool)
    .await?;

    Ok(item)
}

/// Admin: restore a previously removed item. Clears the removal fields
/// but does NOT re-publish (creator must publish manually).
#[tracing::instrument(skip_all)]
pub async fn admin_restore_item(pool: &PgPool, item_id: ItemId) -> Result<DbItem> {
    let item = sqlx::query_as::<_, DbItem>(
        r"
        UPDATE items
        SET removed_by_admin = false,
            removal_reason = NULL,
            removed_at = NULL
        WHERE id = $1
        RETURNING *
        ",
    )
    .bind(item_id)
    .fetch_one(pool)
    .await?;

    Ok(item)
}

/// Count public, listed items (for landing page stats).
#[tracing::instrument(skip_all)]
pub async fn count_public_listed(pool: &PgPool) -> Result<i64> {
    let count = sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM items WHERE is_public = true AND listed = true",
    )
    .fetch_one(pool)
    .await?;

    Ok(count)
}