//! Item version management: release creation, listing, and download tracking.

use sqlx::PgPool;

use super::models::{DbVersion, VersionS3KeyRow};
use super::{ItemId, UserId, VersionId};
use crate::error::Result;

/// Create a new version for an item, marking it as the current release.
///
/// Wrapped in a transaction so the UPDATE (clearing the old `is_current`
/// flag) and INSERT (setting the new version as current) either both
/// succeed or both roll back.
#[tracing::instrument(skip_all)]
#[allow(clippy::too_many_arguments)]
pub async fn create_version(
    pool: &PgPool,
    item_id: ItemId,
    version_number: &str,
    changelog: Option<&str>,
    file_url: Option<&str>,
    file_size_bytes: Option<i64>,
    file_name: Option<&str>,
    label: Option<&str>,
) -> Result<DbVersion> {
    let mut tx = pool.begin().await?;

    // Unset current on older version numbers (versions with the same number stay current)
    sqlx::query!(
        "UPDATE versions SET is_current = false WHERE item_id = $1 AND version_number != $2",
        item_id as ItemId,
        version_number,
    )
    .execute(&mut *tx)
    .await?;

    // Create new version as current
    let version = sqlx::query_as!(
        DbVersion,
        r#"
        INSERT INTO versions (item_id, version_number, changelog, file_url, file_size_bytes, file_name, is_current, label)
        VALUES ($1, $2, $3, $4, $5, $6, true, $7)
        RETURNING id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
                  file_url, file_size_bytes, file_name, download_count, is_current,
                  created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
                  scan_status AS "scan_status: super::FileScanStatus", label
        "#,
        item_id as ItemId,
        version_number,
        changelog,
        file_url,
        file_size_bytes,
        file_name,
        label,
    )
    .fetch_one(&mut *tx)
    .await?;

    tx.commit().await?;

    Ok(version)
}

/// Hard cap on rows returned by the non-paginated version listing. Items
/// with this many versions are exceptional; if we ever hit the cap a warning
/// fires so we can promote the caller to cursor pagination. Real pagination
/// is deferred (Phase 6/8), this constant just makes the truncation loud.
pub const VERSIONS_LIST_HARD_CAP: i64 = 5000;

/// List all versions for an item, newest first.
///
/// Capped at `VERSIONS_LIST_HARD_CAP` as a safety limit. Hitting the cap is
/// logged at WARN so we notice before a real user gets silently truncated.
#[tracing::instrument(skip_all)]
pub async fn get_versions_by_item(pool: &PgPool, item_id: ItemId) -> Result<Vec<DbVersion>> {
    let versions = sqlx::query_as!(
        DbVersion,
        r#"
        SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
               file_url, file_size_bytes, file_name, download_count, is_current,
               created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
               scan_status AS "scan_status: super::FileScanStatus", label
        FROM versions WHERE item_id = $1 ORDER BY created_at DESC LIMIT $2
        "#,
        item_id as ItemId,
        VERSIONS_LIST_HARD_CAP,
    )
    .fetch_all(pool)
    .await?;

    if versions.len() as i64 == VERSIONS_LIST_HARD_CAP {
        tracing::warn!(
            %item_id, cap = VERSIONS_LIST_HARD_CAP,
            "get_versions_by_item hit hard cap; promote caller to cursor pagination"
        );
    }

    Ok(versions)
}

/// Batch-load versions for multiple items, grouped by item_id.
#[tracing::instrument(skip_all)]
pub async fn get_versions_by_items(
    pool: &PgPool,
    item_ids: &[ItemId],
) -> Result<std::collections::HashMap<ItemId, Vec<DbVersion>>> {
    let versions = sqlx::query_as!(
        DbVersion,
        r#"
        SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
               file_url, file_size_bytes, file_name, download_count, is_current,
               created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
               scan_status AS "scan_status: super::FileScanStatus", label
        FROM versions WHERE item_id = ANY($1) ORDER BY item_id, created_at DESC
        "#,
        item_ids as &[ItemId],
    )
    .fetch_all(pool)
    .await?;

    let mut map: std::collections::HashMap<ItemId, Vec<DbVersion>> =
        std::collections::HashMap::new();
    for v in versions {
        map.entry(v.item_id).or_default().push(v);
    }
    Ok(map)
}

/// Atomically increment the download counter for a version.
#[tracing::instrument(skip_all)]
pub async fn increment_download_count(pool: &PgPool, version_id: VersionId) -> Result<()> {
    sqlx::query!(
        "UPDATE versions SET download_count = download_count + 1 WHERE id = $1",
        version_id as VersionId,
    )
    .execute(pool)
    .await?;

    Ok(())
}

/// Record that a user downloaded a specific version (idempotent).
#[tracing::instrument(skip_all)]
pub async fn record_user_download(
    pool: &PgPool,
    user_id: UserId,
    item_id: ItemId,
    version_id: VersionId,
) -> Result<()> {
    // Explicit conflict target, the table's PRIMARY KEY is
    // (user_id, item_id, version_id), but `ON CONFLICT DO NOTHING` without
    // a target would silently swallow conflicts on ANY future constraint
    // we add (a unique index on downloaded_at, say). Naming the target
    // means a new constraint surfaces as an error rather than a no-op.
    sqlx::query!(
        r#"
        INSERT INTO user_downloads (user_id, item_id, version_id)
        VALUES ($1, $2, $3)
        ON CONFLICT (user_id, item_id, version_id) DO NOTHING
        "#,
        user_id as UserId,
        item_id as ItemId,
        version_id as VersionId,
    )
    .execute(pool)
    .await?;

    Ok(())
}

/// Get the latest version ID a user has downloaded for an item, if any.
#[tracing::instrument(skip_all)]
pub async fn get_user_latest_download(
    pool: &PgPool,
    user_id: UserId,
    item_id: ItemId,
) -> Result<Option<VersionId>> {
    let row = sqlx::query_scalar!(
        r#"
        SELECT ud.version_id AS "version_id: VersionId" FROM user_downloads ud
        JOIN versions v ON v.id = ud.version_id
        WHERE ud.user_id = $1 AND ud.item_id = $2
        ORDER BY v.created_at DESC
        LIMIT 1
        "#,
        user_id as UserId,
        item_id as ItemId,
    )
    .fetch_optional(pool)
    .await?;

    Ok(row)
}

/// Fetch a version by primary key. Returns `None` if not found.
#[tracing::instrument(skip_all)]
pub async fn get_version_by_id(pool: &PgPool, version_id: VersionId) -> Result<Option<DbVersion>> {
    let version = sqlx::query_as!(
        DbVersion,
        r#"
        SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
               file_url, file_size_bytes, file_name, download_count, is_current,
               created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
               scan_status AS "scan_status: super::FileScanStatus", label
        FROM versions WHERE id = $1
        "#,
        version_id as VersionId,
    )
    .fetch_optional(pool)
    .await?;

    Ok(version)
}

/// Collect all S3 keys for versions owned by a user.
///
/// Returns s3_key, file_name, version_number, item title, and project slug.
/// Only includes versions that have an S3 key.
#[tracing::instrument(skip_all)]
pub async fn get_user_version_s3_keys(
    pool: &PgPool,
    user_id: super::UserId,
) -> Result<Vec<VersionS3KeyRow>> {
    let rows = sqlx::query_as!(
        VersionS3KeyRow,
        r#"
        SELECT v.s3_key, v.file_name, v.version_number AS "version_number!", i.title AS "item_title!",
               p.id AS "project_id!: super::ProjectId", p.slug AS "project_slug!: super::Slug", v.file_size_bytes
        FROM versions v
        JOIN items i ON v.item_id = i.id
        JOIN projects p ON i.project_id = p.id
        WHERE p.user_id = $1 AND v.s3_key IS NOT NULL
        ORDER BY p.slug, i.sort_order, v.created_at DESC
        LIMIT $2
        "#,
        user_id as UserId,
        VERSIONS_LIST_HARD_CAP,
    )
    .fetch_all(pool)
    .await?;

    if rows.len() as i64 == VERSIONS_LIST_HARD_CAP {
        tracing::warn!(
            %user_id, cap = VERSIONS_LIST_HARD_CAP,
            "get_user_version_s3_keys (account export) hit hard cap; some files will be omitted from export"
        );
    }

    Ok(rows)
}

/// Update a version's S3 key, file size, and file name in one query.
///
/// `expected_old_s3_key` guards against a lost-update race: two concurrent
/// confirms can both pass the idempotency gate (reading the same prior
/// `version.s3_key`) and both succeed in incrementing storage; without this
/// guard, the second's UPDATE silently overwrites the first's, leaking S3
/// objects and double-charging storage.
///
/// `Ok(None)` means the row exists but the `s3_key` no longer matches the
/// expected value, the caller is responsible for the rollback (refund the
/// storage increment, delete the new S3 object).
#[tracing::instrument(skip_all)]
pub async fn update_version_file<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    version_id: VersionId,
    expected_old_s3_key: Option<&str>,
    s3_key: &str,
    file_size_bytes: Option<i64>,
    file_name: Option<&str>,
) -> Result<Option<DbVersion>> {
    let version = sqlx::query_as!(
        DbVersion,
        r#"
        UPDATE versions
        SET s3_key = $2, file_size_bytes = $3, file_name = $4
        WHERE id = $1
          AND s3_key IS NOT DISTINCT FROM $5
        RETURNING id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
                  file_url, file_size_bytes, file_name, download_count, is_current,
                  created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
                  scan_status AS "scan_status: super::FileScanStatus", label
        "#,
        version_id as VersionId,
        s3_key,
        file_size_bytes,
        file_name,
        expected_old_s3_key,
    )
    .fetch_optional(executor)
    .await?;

    Ok(version)
}

/// Delete a version by ID, decrementing the owning user's storage counter
/// and enqueuing its S3 object for durable deletion in the same transaction.
///
/// Both the storage refund and the S3-delete enqueue must succeed together
/// with the row delete, otherwise a future caller of this function could
/// forget either step and leak storage credit or orphan an S3 object.
/// `delete_version_row_only` exists for cases where the caller has already
/// handled both side effects (e.g. cascading item delete that batches them).
#[tracing::instrument(skip_all)]
pub async fn delete_version(pool: &PgPool, version_id: VersionId) -> Result<()> {
    // Look up the owning user so we can refund storage. Ownership is stable for
    // a version's lifetime; its size + key, by contrast, can change under a
    // concurrent replace-confirm, so those come from the DELETE's RETURNING
    // below, never a pre-tx read (Run #18 Storage B5).
    let owner_id: Option<super::UserId> = sqlx::query_scalar!(
        r#"
        SELECT p.user_id AS "user_id!: super::UserId"
        FROM versions v
        JOIN items i ON v.item_id = i.id
        JOIN projects p ON i.project_id = p.id
        WHERE v.id = $1
        "#,
        version_id as VersionId,
    )
    .fetch_optional(pool)
    .await?;

    let mut tx = pool.begin().await?;

    // DELETE ... RETURNING so the refund + S3 enqueue act on the row's ACTUAL
    // state at delete time. A replace-confirm that commits between a pre-tx read
    // and this DELETE would otherwise make us refund the OLD size and enqueue
    // the OLD key while leaking the new one. RETURNING also gives us the
    // rows-affected discipline for free: a concurrent double-delete finds no row
    // and refunds nothing (Run #12 LOW + Run #18 Storage B5).
    let deleted: Option<(Option<String>, Option<i64>)> = sqlx::query!(
        "DELETE FROM versions WHERE id = $1 RETURNING s3_key, file_size_bytes",
        version_id as VersionId,
    )
    .fetch_optional(&mut *tx)
    .await?
    .map(|r| (r.s3_key, r.file_size_bytes));

    if let Some((s3_key, file_size_bytes)) = deleted {
        if let Some(user_id) = owner_id
            && let Some(size) = file_size_bytes
            && size > 0
        {
            crate::db::creator_tiers::decrement_storage_used(&mut *tx, user_id, size).await?;
        }

        // Enqueue the S3 delete inside the SAME tx as the row delete + refund.
        // After commit the row is gone (so the key is non-live and the deletion
        // worker can act), and a crash between commit and a post-commit enqueue
        // can no longer orphan the object, all three effects are atomic, as the
        // doc comment promises.
        if let Some(s3_key) = s3_key {
            crate::db::pending_s3_deletions::enqueue_deletions(
                &mut *tx,
                &[(s3_key, "main".to_string())],
                "version_delete",
            )
            .await?;
        }
    }

    tx.commit().await?;

    Ok(())
}

/// Sum all version file sizes for a given item (for storage decrement on item delete).
#[tracing::instrument(skip_all)]
pub async fn sum_file_sizes_for_item(pool: &PgPool, item_id: super::ItemId) -> Result<i64> {
    // SUM over many bigints widens to NUMERIC in Postgres; clamp on both
    // sides (>=0 and <=i64::MAX) before casting back to BIGINT, without
    // GREATEST(0, ...), a corrupt-negative row could propagate a negative
    // total that later under-flows storage accounting.
    let total: i64 = sqlx::query_scalar!(
        r#"SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size_bytes), 9223372036854775807))::BIGINT, 0) AS "total!" FROM versions WHERE item_id = $1 AND file_size_bytes IS NOT NULL"#,
        item_id as ItemId,
    )
    .fetch_one(pool)
    .await?;

    Ok(total)
}