//! Project member and revenue split operations.

use sqlx::PgPool;

use super::ProjectRole;
use super::id_types::{ProjectId, TipId, TransactionId, UserId};
use super::models::{DbProjectMember, DbProjectMemberWithUser, DbRevenueSplit, DbSplitExportRow};
use crate::error::{AppError, Result};

// ── Project Members ──

/// Take a row lock on the parent `projects` row so every revenue-split mutation
/// for the project serializes, including the first two members added
/// concurrently, which member-row locks alone would not block (audit Run 13
/// Conc TOCTOU). Callers must already be inside a transaction.
async fn lock_project_for_splits(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    project_id: ProjectId,
) -> Result<()> {
    sqlx::query("SELECT 1 FROM projects WHERE id = $1 FOR UPDATE")
        .bind(project_id)
        .execute(&mut **tx)
        .await?;
    Ok(())
}

/// Add a member to a project with a revenue split percentage (0-100).
///
/// The sum of all member splits (excluding the owner's implicit remainder)
/// must not exceed 100.
#[tracing::instrument(skip(pool))]
pub(crate) async fn add_project_member(
    pool: &PgPool,
    project_id: ProjectId,
    user_id: UserId,
    role: ProjectRole,
    split_percent: i16,
    added_by: UserId,
) -> Result<DbProjectMember> {
    // Reject negative and out-of-range splits before any DB write. The Run #6
    // audit caught the missing lower bound, negative percentages flow through
    // `compute_splits` and record negative obligations.
    if !(0..=100).contains(&split_percent) {
        return Err(AppError::BadRequest(format!(
            "split_percent must be between 0 and 100 (got {split_percent})"
        )));
    }

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

    // Serialize every split mutation for this project on the parent row. Locking
    // only the member rows does NOT block a concurrent INSERT of a new member
    // when none exist yet, so two "first members" could each pass the cap check
    // and race the total past 100% (audit Run 13 Conc TOCTOU). The projects row
    // always exists, so `FOR UPDATE` on it serializes all adds/updates.
    lock_project_for_splits(&mut tx, project_id).await?;

    // Subtract the existing row's split (if this is an upsert) before the cap
    // check; otherwise a legitimate update is rejected as "> 100%" whenever
    // current_split + new_split crosses 100 even if the post-update total wouldn't.
    let existing_split: (Option<i16>,) = sqlx::query_as(
        "SELECT split_percent FROM project_members WHERE project_id = $1 AND user_id = $2",
    )
    .bind(project_id)
    .bind(user_id)
    .fetch_optional(&mut *tx)
    .await?
    .map_or((None,), |r: (i16,)| (Some(r.0),));
    let existing = existing_split.0.unwrap_or(0) as i64;

    let current_total: (Option<i64>,) = sqlx::query_as(
        "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
    )
    .bind(project_id)
    .fetch_one(&mut *tx)
    .await?;

    let total = current_total.0.unwrap_or(0);
    let new_total = total - existing + split_percent as i64;
    if new_total > 100 {
        return Err(AppError::BadRequest(format!(
            "Total split would be {new_total}%, exceeding 100%"
        )));
    }

    let member = sqlx::query_as::<_, DbProjectMember>(
        r"
        INSERT INTO project_members (project_id, user_id, role, split_percent, added_by)
        VALUES ($1, $2, $3, $4, $5)
        ON CONFLICT (project_id, user_id) DO UPDATE
            SET role = EXCLUDED.role,
                split_percent = EXCLUDED.split_percent
        RETURNING *
        ",
    )
    .bind(project_id)
    .bind(user_id)
    .bind(role)
    .bind(split_percent)
    .bind(added_by)
    .fetch_one(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(member)
}

/// Remove a member from a project.
#[tracing::instrument(skip(pool))]
pub(crate) async fn remove_project_member(
    pool: &PgPool,
    project_id: ProjectId,
    user_id: UserId,
) -> Result<bool> {
    let result = sqlx::query("DELETE FROM project_members WHERE project_id = $1 AND user_id = $2")
        .bind(project_id)
        .bind(user_id)
        .execute(pool)
        .await?;

    Ok(result.rows_affected() > 0)
}

/// Get all members of a project with user info, ordered by split descending.
#[tracing::instrument(skip(pool))]
pub(crate) async fn get_project_members(
    pool: &PgPool,
    project_id: ProjectId,
) -> Result<Vec<DbProjectMemberWithUser>> {
    let members = sqlx::query_as::<_, DbProjectMemberWithUser>(
        r"
        SELECT pm.id, pm.project_id, pm.user_id, pm.role, pm.split_percent, pm.added_at,
               pm.accepted_at,
               u.username, u.display_name, u.stripe_account_id, u.stripe_charges_enabled,
               u.settlement_currency
        FROM project_members pm
        JOIN users u ON u.id = pm.user_id
        WHERE pm.project_id = $1
        ORDER BY pm.split_percent DESC
        ",
    )
    .bind(project_id)
    .fetch_all(pool)
    .await?;

    Ok(members)
}

/// Accept a split invitation. Returns false if there was nothing pending.
///
/// Idempotent by the `accepted_at IS NULL` guard: a double-click cannot move an
/// acceptance timestamp forward, which matters because that timestamp is the
/// line between sales the owner kept and sales this collaborator shares in.
#[tracing::instrument(skip(pool))]
pub(crate) async fn accept_split_invitation(
    pool: &PgPool,
    project_id: ProjectId,
    user_id: UserId,
) -> Result<bool> {
    let result = sqlx::query(
        "UPDATE project_members SET accepted_at = NOW() \
         WHERE project_id = $1 AND user_id = $2 AND accepted_at IS NULL",
    )
    .bind(project_id)
    .bind(user_id)
    .execute(pool)
    .await?;
    Ok(result.rows_affected() > 0)
}

/// Decline a split invitation, removing the membership and freeing the reserved
/// percentage. Only ever touches a row that is still pending, so it cannot undo
/// an acceptance.
#[tracing::instrument(skip(pool))]
pub(crate) async fn decline_split_invitation(
    pool: &PgPool,
    project_id: ProjectId,
    user_id: UserId,
) -> Result<bool> {
    let result = sqlx::query(
        "DELETE FROM project_members \
         WHERE project_id = $1 AND user_id = $2 AND accepted_at IS NULL",
    )
    .bind(project_id)
    .bind(user_id)
    .execute(pool)
    .await?;
    Ok(result.rows_affected() > 0)
}

/// A split invitation awaiting this creator's answer.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct PendingSplitInvitation {
    pub project_id: ProjectId,
    pub project_title: String,
    pub project_slug: crate::db::Slug,
    pub owner_username: String,
    pub split_percent: i16,
    pub added_at: chrono::DateTime<chrono::Utc>,
    /// The currency the project sells in, which is its owner's. If it differs
    /// from the invited creator's own, their share converts at their payout and
    /// they carry that cost, which is the whole reason this screen exists.
    pub project_currency: crate::currency::SettlementCurrency,
}

/// Every split invitation this creator has not yet answered.
#[tracing::instrument(skip(pool))]
pub(crate) async fn get_pending_invitations(
    pool: &PgPool,
    user_id: UserId,
) -> Result<Vec<PendingSplitInvitation>> {
    let rows = sqlx::query_as::<_, PendingSplitInvitation>(
        r"
        SELECT pm.project_id, p.title AS project_title, p.slug AS project_slug,
               owner.username AS owner_username, pm.split_percent, pm.added_at,
               owner.settlement_currency AS project_currency
        FROM project_members pm
        JOIN projects p ON p.id = pm.project_id
        JOIN users owner ON owner.id = p.user_id
        WHERE pm.user_id = $1 AND pm.accepted_at IS NULL
        ORDER BY pm.added_at DESC
        LIMIT 100
        ",
    )
    .bind(user_id)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Get the total split percentage allocated to members (excluding the owner).
///
/// Counts pending invitations too. The percentage is reserved the moment it is
/// offered, so an owner cannot promise the same revenue twice while the first
/// collaborator has not answered.
#[tracing::instrument(skip(pool))]
pub(crate) async fn get_total_split_percent(pool: &PgPool, project_id: ProjectId) -> Result<i64> {
    let row: (Option<i64>,) = sqlx::query_as(
        "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
    )
    .bind(project_id)
    .fetch_one(pool)
    .await?;

    Ok(row.0.unwrap_or(0))
}

/// Update a member's split percentage.
#[allow(dead_code)]
#[tracing::instrument(skip(pool))]
pub(crate) async fn update_member_split(
    pool: &PgPool,
    project_id: ProjectId,
    user_id: UserId,
    new_split_percent: i16,
) -> Result<()> {
    if !(0..=100).contains(&new_split_percent) {
        return Err(AppError::BadRequest(format!(
            "split_percent must be between 0 and 100 (got {new_split_percent})"
        )));
    }
    let mut tx = pool.begin().await?;

    // Serialize on the parent project row (see `add_project_member`). The prior
    // `SUM(...) FOR UPDATE` below was invalid, Postgres rejects FOR UPDATE with
    // an aggregate, so this function 500'd on every call (audit Run 13); the lock
    // now lives on the projects row and the SUM runs plain.
    lock_project_for_splits(&mut tx, project_id).await?;

    let current: Option<DbProjectMember> =
        sqlx::query_as("SELECT * FROM project_members WHERE project_id = $1 AND user_id = $2")
            .bind(project_id)
            .bind(user_id)
            .fetch_optional(&mut *tx)
            .await?;

    let current_member_split = current.map_or(0, |m| m.split_percent as i64);

    let total_row: (Option<i64>,) = sqlx::query_as(
        "SELECT SUM(split_percent)::BIGINT FROM project_members WHERE project_id = $1",
    )
    .bind(project_id)
    .fetch_one(&mut *tx)
    .await?;

    let total = total_row.0.unwrap_or(0);
    let new_total = total - current_member_split + new_split_percent as i64;
    if new_total > 100 {
        return Err(AppError::BadRequest(format!(
            "Total split would be {new_total}%, exceeding 100%"
        )));
    }

    sqlx::query(
        "UPDATE project_members SET split_percent = $3 WHERE project_id = $1 AND user_id = $2",
    )
    .bind(project_id)
    .bind(user_id)
    .bind(new_split_percent)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(())
}

// ── Revenue Splits ──

/// Record revenue splits for a completed tip.
#[tracing::instrument(skip(pool))]
pub(crate) async fn create_tip_splits(
    pool: &PgPool,
    tip_id: TipId,
    splits: &[(UserId, i64, i16)], // (recipient_id, amount_cents, split_percent)
) -> Result<()> {
    if splits.is_empty() {
        return Ok(());
    }
    let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
    let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
    let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
    // ON CONFLICT DO NOTHING (against the partial revenue_splits_tip_recipient_key,
    // migration 163): one split row per recipient per tip, so a crash-recovery
    // redelivery that re-runs `record_tip_splits` is a no-op rather than a
    // double-credit (Run 20 Payments chronic). The WHERE predicate matches the
    // partial index so PostgreSQL infers it as the arbiter.
    sqlx::query(
        r"
        INSERT INTO revenue_splits (tip_id, recipient_id, amount_cents, split_percent, status, currency)
        SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending',
               (SELECT currency FROM tips WHERE id = $1)
        ON CONFLICT (tip_id, recipient_id) WHERE tip_id IS NOT NULL DO NOTHING
        ",
    )
    .bind(tip_id)
    .bind(&recipient_ids)
    .bind(&amounts)
    .bind(&percents)
    .execute(pool)
    .await?;
    Ok(())
}

/// Record revenue splits for a completed transaction (item purchase).
#[tracing::instrument(skip(pool))]
pub(crate) async fn create_transaction_splits(
    pool: &PgPool,
    transaction_id: TransactionId,
    splits: &[(UserId, i64, i16)], // (recipient_id, amount_cents, split_percent)
) -> Result<()> {
    if splits.is_empty() {
        return Ok(());
    }
    let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
    let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
    let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
    // ON CONFLICT DO NOTHING (against revenue_splits_tx_recipient_key): a
    // crash-recovery finalize re-run records no duplicate splits.
    sqlx::query(
        r"
        INSERT INTO revenue_splits (transaction_id, recipient_id, amount_cents, split_percent, status, currency)
        SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending',
               (SELECT currency FROM transactions WHERE id = $1)
        ON CONFLICT (transaction_id, recipient_id) DO NOTHING
        ",
    )
    .bind(transaction_id)
    .bind(&recipient_ids)
    .bind(&amounts)
    .bind(&percents)
    .execute(pool)
    .await?;
    Ok(())
}

/// Get all revenue splits for a recipient, most recent first.
#[allow(dead_code)]
#[tracing::instrument(skip(pool))]
pub(crate) async fn get_splits_for_recipient(
    pool: &PgPool,
    recipient_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<DbRevenueSplit>> {
    let splits = sqlx::query_as::<_, DbRevenueSplit>(
        r"
        SELECT * FROM revenue_splits
        WHERE recipient_id = $1
        ORDER BY created_at DESC
        LIMIT $2 OFFSET $3
        ",
    )
    .bind(recipient_id)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await?;

    Ok(splits)
}

/// Total split revenue owed to a recipient (all completed splits).
#[tracing::instrument(skip(pool))]
pub(crate) async fn total_split_revenue(
    pool: &PgPool,
    recipient_id: UserId,
) -> Result<crate::currency::MoneyByCurrency> {
    // Grouped, because this is the one place a creator genuinely holds money in
    // someone else's currency: a split is denominated in the paying project's
    // currency, which is its owner's, not the recipient's.
    let rows: Vec<(crate::currency::SettlementCurrency, Option<i64>)> = sqlx::query_as(
        "SELECT currency, SUM(amount_cents)::BIGINT FROM revenue_splits \
         WHERE recipient_id = $1 GROUP BY currency",
    )
    .bind(recipient_id)
    .fetch_all(pool)
    .await?;

    Ok(crate::currency::MoneyByCurrency::from_rows(
        rows.into_iter().map(|(c, cents)| (c, cents.unwrap_or(0))),
    ))
}

/// Count of split records for a recipient.
#[tracing::instrument(skip(pool))]
pub(crate) async fn count_splits_for_recipient(pool: &PgPool, recipient_id: UserId) -> Result<i64> {
    let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM revenue_splits WHERE recipient_id = $1")
        .bind(recipient_id)
        .fetch_one(pool)
        .await?;

    Ok(row.0)
}

/// Get all splits involving a user (as owner or recipient) for CSV export.
/// Returns splits where the user is either:
/// - The recipient (collaborator receiving a share), or
/// - The seller/tip recipient (owner who owes collaborators)
#[tracing::instrument(skip(pool))]
/// One page of a user's revenue splits for CSV export, newest first.
///
/// Paginated so the splits export streams in bounded batches instead of loading
/// every split in one query (ultra-fuzz Run 4 S1). Stable `(created_at, id)`
/// ordering keeps OFFSET batches consistent.
pub(crate) async fn get_splits_for_export_page(
    pool: &PgPool,
    user_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<DbSplitExportRow>> {
    let rows = sqlx::query_as::<_, DbSplitExportRow>(
        r"
        SELECT rs.id, rs.recipient_id, rs.amount_cents, rs.split_percent, rs.created_at,
               CASE WHEN rs.transaction_id IS NOT NULL THEN 'sale' ELSE 'tip' END AS source_type,
               u.username AS recipient_username
        FROM revenue_splits rs
        JOIN users u ON u.id = rs.recipient_id
        LEFT JOIN transactions t ON t.id = rs.transaction_id
        LEFT JOIN tips tip ON tip.id = rs.tip_id
        WHERE rs.recipient_id = $1
           OR COALESCE(t.seller_id, tip.recipient_id) = $1
        ORDER BY rs.created_at DESC, rs.id DESC
        LIMIT $2 OFFSET $3
        ",
    )
    .bind(user_id)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// Total split obligations owed by a project owner (splits on their transactions/tips).
#[tracing::instrument(skip(pool))]
pub(crate) async fn total_split_obligations(pool: &PgPool, owner_id: UserId) -> Result<i64> {
    let row: (Option<i64>,) = sqlx::query_as(
        r"
        SELECT SUM(rs.amount_cents)::BIGINT
        FROM revenue_splits rs
        LEFT JOIN transactions t ON t.id = rs.transaction_id
        LEFT JOIN tips tip ON tip.id = rs.tip_id
        WHERE COALESCE(t.seller_id, tip.recipient_id) = $1
        ",
    )
    .bind(owner_id)
    .fetch_one(pool)
    .await?;

    Ok(row.0.unwrap_or(0))
}