/**
 * Maturity sweeps — pending→matured and matured→withdrawable.
 *
 * Maturity sweep (sweepMaturation Phase-2) + withdrawable sweep.
 * Host-coupled purchase/deal facts route through read-only MaturityHostStore (spec §3A Wave-5).
 * Pending→matured promotion consumes ledger/vesting promoteEntries (design §3.4).
 */
import { sql, eq, and, or, isNull, inArray } from 'drizzle-orm'
import type { Transaction } from '@platform-modules/db'
import { ledgerEntries, type LedgerSchema } from '@platform-modules/ledger'
import {
  promoteEntries,
  recomputeWithdrawable,
  setEntryWithdrawableAt,
  ledgerEntryVesting,
  walletVesting,
  type VestingSchema,
} from '@platform-modules/ledger/vesting'
import { computeMatureAt } from './accrual.js'
import { affiliateEntriesTable, type AffiliateSchema } from './schema.js'

export interface RedemptionFactRef {
  sourceType: string
  sourceId: string
}

export interface RedemptionFact {
  sourceType: string
  sourceId: string
  kind: 'coupon' | 'physical'
  paidAt: Date
  expiresAt: Date | null
  redeemedAt: Date | null
  purchaseCreatedAt: Date
}

export interface MaturityHostStore {
  /** Read-only; joins host purchases/deals; maps host deal_type → normalized kind. */
  getRedemptionFacts(refs: RedemptionFactRef[]): Promise<RedemptionFact[]>

  getReferralSettings(): Promise<{ holdDays: number; disputeWindowDays: number }>
}

export type WithdrawableAtInput = {
  kind: 'coupon' | 'physical'
  redeemedAt: Date | null
  purchaseCreatedAt: Date
}

/**
 * Compute withdrawable_at for a credit ledger entry from normalized redemption facts.
 *
 * - coupon + redeemed → redeemedAt + disputeWindowDays
 * - coupon + unredeemed → null (not withdrawable yet)
 * - physical → purchaseCreatedAt + disputeWindowDays
 */
export function computeWithdrawableAt(
  input: WithdrawableAtInput,
  disputeWindowDays: number,
): Date | null {
  const windowMs = disputeWindowDays * 86_400_000

  if (input.kind === 'coupon') {
    if (input.redeemedAt) {
      return new Date(input.redeemedAt.getTime() + windowMs)
    }
    return null
  }

  return new Date(input.purchaseCreatedAt.getTime() + windowMs)
}

export type SweepMaturationResult = { promoted: number; recomputed: number; sweptAt: Date }

type CandidateEarn = {
  entryId: string
  ownerId: string
  amountMinor: bigint
  sourceType: string
  sourceId: string
}

type MatchedClawback = {
  entryId: string
  amountMinor: bigint
  sourceType: string
  sourceId: string
}

function factKey(sourceType: string, sourceId: string): string {
  return `${sourceType}:${sourceId}`
}

function toBigint(value: string | bigint | number): bigint {
  if (typeof value === 'bigint') return value
  return BigInt(value)
}

/** Clawback magnitude for netting — reference stores negative; spine delta follows host convention. */
function clawbackMagnitude(delta: bigint): bigint {
  return delta < 0n ? -delta : delta
}

function matchesClawbackSource(
  earnSourceType: string,
  earnSourceId: string,
  clawbackSourceType: string,
  clawbackSourceId: string,
): boolean {
  if (clawbackSourceType !== earnSourceType) return false
  const prefix = `clawback:${earnSourceId}`
  return clawbackSourceId === prefix || clawbackSourceId.startsWith(`${prefix}:`)
}

function matchClawbacks(earns: CandidateEarn[], clawbackRows: MatchedClawback[]): MatchedClawback[] {
  const matched: MatchedClawback[] = []
  const seen = new Set<string>()

  for (const earn of earns) {
    for (const cb of clawbackRows) {
      if (seen.has(cb.entryId)) continue
      if (
        matchesClawbackSource(earn.sourceType, earn.sourceId, cb.sourceType, cb.sourceId)
      ) {
        matched.push(cb)
        seen.add(cb.entryId)
      }
    }
  }

  return matched
}

/** Bound clawback candidates to locked ready-earns (matched_clawbacks JOIN parity). */
function clawbackCandidateFilter(earns: CandidateEarn[]) {
  const sourceTypes = [...new Set(earns.map((e) => e.sourceType))]
  const exactSourceIds = earns.map((e) => `clawback:${e.sourceId}`)
  const prefixStartsWith = earns.map((e) =>
    sql`starts_with(${affiliateEntriesTable.sourceId}, ${`clawback:${e.sourceId}:`})`,
  )

  return and(
    eq(affiliateEntriesTable.entryType, 'refund_clawback'),
    isNull(affiliateEntriesTable.consumedAt),
    inArray(affiliateEntriesTable.sourceType, sourceTypes),
    or(inArray(affiliateEntriesTable.sourceId, exactSourceIds), ...prefixStartsWith),
  )
}

function netPerOwner(
  earns: CandidateEarn[],
  clawbacks: MatchedClawback[],
): { earnEntryIds: { entryId: string; ownerId: string }[]; perOwnerDeltas: { ownerId: string; deltaMinor: bigint }[] } {
  const earnById = new Map<string, CandidateEarn>()
  for (const earn of earns) {
    earnById.set(earn.entryId, earn)
  }
  const dedupedEarns = [...earnById.values()]

  const earnEntryIds = dedupedEarns.map((e) => ({ entryId: e.entryId, ownerId: e.ownerId }))

  const ownerEarnSum = new Map<string, bigint>()
  for (const earn of dedupedEarns) {
    ownerEarnSum.set(earn.ownerId, (ownerEarnSum.get(earn.ownerId) ?? 0n) + earn.amountMinor)
  }

  const ownerClawbackSum = new Map<string, bigint>()
  for (const cb of clawbacks) {
    const earn = dedupedEarns.find((e) =>
      matchesClawbackSource(e.sourceType, e.sourceId, cb.sourceType, cb.sourceId),
    )
    if (!earn) continue
    ownerClawbackSum.set(earn.ownerId, (ownerClawbackSum.get(earn.ownerId) ?? 0n) + cb.amountMinor)
  }

  const perOwnerDeltas = [...ownerEarnSum.keys()].map((ownerId) => ({
    ownerId,
    deltaMinor: (ownerEarnSum.get(ownerId) ?? 0n) - (ownerClawbackSum.get(ownerId) ?? 0n),
  }))

  return { earnEntryIds, perOwnerDeltas }
}

/**
 * Maturation sweep: promotes pending ledger_entry_vesting rows to matured.
 *
 * NO-OP gate: return early if nothing is ready.
 * Phase 1: Recompute mature_at for coupon rows where coupon has since been redeemed (via host facts).
 * Phase 2: SELECT+net clawbacks in JS → promoteEntries → mark clawbacks consumed (design §3.4).
 */
export async function sweepMaturation<
  S extends AffiliateSchema & LedgerSchema & VestingSchema,
>(
  tx: Transaction<S>,
  host: MaturityHostStore,
): Promise<SweepMaturationResult> {
  const now = new Date()

  const probe = (await tx.execute(sql`
    SELECT 1
    FROM ledger_entry_vesting lev
    INNER JOIN affiliate_entries ae ON ae.entry_id = lev.entry_id
    WHERE lev.mature_at <= NOW()
      AND lev.swept_at IS NULL
      AND ae.entry_type IN ('affiliate_commission', 'referral_reward')
    LIMIT 1
  `)) as { rows: unknown[] }
  if (!probe.rows.length) return { promoted: 0, recomputed: 0, sweptAt: now }

  const settings = await host.getReferralSettings()

  const candidateRows = (await tx.execute(sql`
    SELECT lev.entry_id, ae.source_type, ae.source_id, lev.mature_at
    FROM ledger_entry_vesting lev
    INNER JOIN affiliate_entries ae ON ae.entry_id = lev.entry_id
    WHERE lev.swept_at IS NULL
      AND ae.entry_type IN ('affiliate_commission', 'referral_reward')
  `)) as {
    rows: Array<{
      entry_id: string
      source_type: string
      source_id: string
      mature_at: string | null
    }>
  }

  let recomputed = 0
  if (candidateRows.rows.length > 0) {
    const refs = candidateRows.rows.map((row) => ({
      sourceType: row.source_type,
      sourceId: row.source_id,
    }))
    const facts = await host.getRedemptionFacts(refs)
    const factByKey = new Map(facts.map((f) => [factKey(f.sourceType, f.sourceId), f]))

    for (const row of candidateRows.rows) {
      const fact = factByKey.get(factKey(row.source_type, row.source_id))
      if (!fact || fact.kind !== 'coupon' || !fact.redeemedAt) continue

      const newMatureAt = computeMatureAt(
        {
          kind: fact.kind,
          paidAt: fact.paidAt,
          expiresAt: fact.expiresAt,
          redeemedAt: fact.redeemedAt,
        },
        { holdDays: settings.holdDays },
      )

      const currentMs = row.mature_at ? new Date(row.mature_at).getTime() : null
      if (currentMs === newMatureAt.getTime()) continue

      await tx
        .update(ledgerEntryVesting)
        .set({ matureAt: newMatureAt })
        .where(eq(ledgerEntryVesting.entryId, row.entry_id))
      recomputed++
    }
  }

  const earnResult = (await tx.execute(sql`
    SELECT
      lev.entry_id AS entry_id,
      ae.owner_id AS owner_id,
      le.delta AS amount_minor,
      ae.source_type AS source_type,
      ae.source_id AS source_id
    FROM ledger_entry_vesting lev
    INNER JOIN ledger_entries le ON le.id = lev.entry_id
    INNER JOIN affiliate_entries ae ON ae.entry_id = lev.entry_id
    WHERE lev.mature_at <= ${now.toISOString()}
      AND lev.swept_at IS NULL
      AND le.delta > 0
      AND ae.entry_type IN ('affiliate_commission', 'referral_reward')
    FOR UPDATE OF lev SKIP LOCKED
  `)) as {
    rows: Array<{
      entry_id: string
      owner_id: string
      amount_minor: string | bigint
      source_type: string
      source_id: string
    }>
  }

  const earns: CandidateEarn[] = earnResult.rows.map((row) => ({
    entryId: row.entry_id,
    ownerId: row.owner_id,
    amountMinor: toBigint(row.amount_minor),
    sourceType: row.source_type,
    sourceId: row.source_id,
  }))

  if (earns.length === 0) {
    return { promoted: 0, recomputed, sweptAt: now }
  }

  const clawbackRows = await tx
    .select({
      entryId: affiliateEntriesTable.entryId,
      sourceType: affiliateEntriesTable.sourceType,
      sourceId: affiliateEntriesTable.sourceId,
      delta: ledgerEntries.delta,
    })
    .from(affiliateEntriesTable)
    .innerJoin(ledgerEntries, eq(ledgerEntries.id, affiliateEntriesTable.entryId))
    .where(clawbackCandidateFilter(earns))

  const clawbacksForMatch: MatchedClawback[] = clawbackRows.map((row) => ({
    entryId: row.entryId,
    sourceType: row.sourceType,
    sourceId: row.sourceId,
    amountMinor: clawbackMagnitude(row.delta),
  }))

  const matchedClawbacks = matchClawbacks(earns, clawbacksForMatch)
  const { earnEntryIds, perOwnerDeltas } = netPerOwner(earns, matchedClawbacks)

  await promoteEntries(tx, { earnEntryIds, perOwnerDeltas })

  if (matchedClawbacks.length > 0) {
    const clawbackIds = matchedClawbacks.map((cb) => cb.entryId)
    await tx
      .update(affiliateEntriesTable)
      .set({ consumedAt: now })
      .where(
        and(
          eq(affiliateEntriesTable.entryType, 'refund_clawback'),
          isNull(affiliateEntriesTable.consumedAt),
          inArray(affiliateEntriesTable.entryId, clawbackIds),
        ),
      )
  }

  return {
    promoted: earnEntryIds.length,
    recomputed,
    sweptAt: now,
  }
}

/**
 * Eligible/paid query for one owner — PLAIN reads only (P5 deadlock-avoidance).
 * Keys earn eligibility off withdrawableAt + sweptAt (matured set), never sweptAt IS NULL.
 */
export async function computeOwnerEligibleAndPaid<
  S extends AffiliateSchema & LedgerSchema & VestingSchema,
>(
  tx: Transaction<S>,
  ownerId: string,
): Promise<{ eligibleMinor: bigint; paidMinor: bigint; maturedMinor: bigint }> {
  const [walletRow] = await tx
    .select({ maturedMinor: walletVesting.maturedMinor })
    .from(walletVesting)
    .where(eq(walletVesting.ownerId, ownerId))
    .limit(1)
  const maturedMinor = walletRow?.maturedMinor ?? 0n

  const eligibleResult = (await tx.execute(sql`
    SELECT COALESCE(SUM(sub.delta), 0) AS eligible_minor
    FROM (
      SELECT le.delta
      FROM ledger_entry_vesting lev
      INNER JOIN ledger_entries le ON le.id = lev.entry_id
      INNER JOIN affiliate_entries ae ON ae.entry_id = lev.entry_id
      LEFT JOIN referrals r ON r.id = ae.referral_id
      WHERE ae.owner_id = ${ownerId}
        AND ae.entry_type IN ('affiliate_commission', 'referral_reward')
        AND lev.withdrawable_at IS NOT NULL
        AND lev.withdrawable_at <= NOW()
        AND lev.swept_at IS NOT NULL
        AND (ae.referral_id IS NULL OR r.quarantined_at IS NULL)
      UNION ALL
      SELECT le.delta
      FROM affiliate_entries ae
      INNER JOIN ledger_entries le ON le.id = ae.entry_id
      WHERE ae.owner_id = ${ownerId}
        AND ae.entry_type = 'refund_clawback'
        AND ae.consumed_at IS NOT NULL
      UNION ALL
      SELECT le.delta
      FROM affiliate_entries ae
      INNER JOIN ledger_entries le ON le.id = ae.entry_id
      WHERE ae.owner_id = ${ownerId}
        AND ae.entry_type = 'redemption'
        AND ae.source_type <> 'affiliate_payout'
    ) sub
  `)) as { rows: Array<{ eligible_minor: string | bigint }> }

  const paidResult = (await tx.execute(sql`
    SELECT COALESCE(SUM(amount_agorot), 0) AS paid_minor
    FROM affiliate_payouts
    WHERE user_id = ${ownerId}
      AND status IN ('requested', 'approved', 'processing', 'paid')
  `)) as { rows: Array<{ paid_minor: string | bigint | number }> }

  const eligibleMinor = toBigint(eligibleResult.rows[0]?.eligible_minor ?? 0)
  const paidMinor = toBigint(paidResult.rows[0]?.paid_minor ?? 0)

  return { eligibleMinor, paidMinor, maturedMinor }
}

/**
 * Withdrawable sweep: compute withdrawable_at anchors then recompute withdrawableMinor.
 */
export async function sweepWithdrawable<
  S extends AffiliateSchema & LedgerSchema & VestingSchema,
>(
  tx: Transaction<S>,
  host: MaturityHostStore,
): Promise<{ updated: number }> {
  const settings = await host.getReferralSettings()

  const dw = Number(settings.disputeWindowDays)
  if (!Number.isInteger(dw) || dw < 0 || dw > 3650) {
    return { updated: 0 }
  }

  const affectedOwners = new Set<string>()

  const anchorCandidates = (await tx.execute(sql`
    SELECT lev.entry_id, ae.source_type, ae.source_id, ae.owner_id
    FROM ledger_entry_vesting lev
    INNER JOIN affiliate_entries ae ON ae.entry_id = lev.entry_id
    WHERE ae.entry_type IN ('affiliate_commission', 'referral_reward')
      AND lev.withdrawable_at IS NULL
  `)) as {
    rows: Array<{
      entry_id: string
      source_type: string
      source_id: string
      owner_id: string
    }>
  }

  if (anchorCandidates.rows.length > 0) {
    const refs = anchorCandidates.rows.map((row) => ({
      sourceType: row.source_type,
      sourceId: row.source_id,
    }))
    const facts = await host.getRedemptionFacts(refs)
    const factByKey = new Map(facts.map((f) => [factKey(f.sourceType, f.sourceId), f]))

    for (const row of anchorCandidates.rows) {
      affectedOwners.add(row.owner_id)
      const fact = factByKey.get(factKey(row.source_type, row.source_id))
      if (!fact) continue

      const withdrawableAt = computeWithdrawableAt(
        {
          kind: fact.kind,
          redeemedAt: fact.redeemedAt,
          purchaseCreatedAt: fact.purchaseCreatedAt,
        },
        dw,
      )

      await setEntryWithdrawableAt(tx, {
        entryId: row.entry_id,
        withdrawableAt,
      })
    }
  }

  const walletOwners = (await tx.execute(sql`
    SELECT owner_id FROM wallet_vesting
  `)) as { rows: Array<{ owner_id: string }> }
  for (const row of walletOwners.rows) {
    affectedOwners.add(row.owner_id)
  }

  let updated = 0
  for (const ownerId of affectedOwners) {
    await recomputeWithdrawable(tx, { ownerId }, async () => {
      const { eligibleMinor, paidMinor } = await computeOwnerEligibleAndPaid(tx, ownerId)
      return { eligibleMinor, paidMinor }
    })
    updated++
  }

  return { updated }
}