/**
 * TOCTOU-safe affiliate payout ledger debit — shared by withdraw, fraud hold, and admin release.
 *
 * R4 audit-hardened money path — preserve byte-faithfully; do not casually refactor.
 */

import { and, eq, lt, sql } from 'drizzle-orm'
import type { Transaction, TransactionalDatabase } from '@platform-modules/db'
import { appendEntry, debitWithRead, ledgerEntries, type LedgerSchema } from '@platform-modules/ledger'
import {
  recomputeWithdrawable,
  walletVesting,
  type VestingSchema,
  type WalletVesting,
} from '@platform-modules/ledger/vesting'
import { InsufficientBalanceError, InvalidPayoutAmountError } from './errors.js'
import {
  affiliateEntriesTable,
  affiliateEnrollmentsTable,
  affiliatePayoutsTable,
  type AffiliateSchema,
} from './schema.js'
import { computeOwnerEligibleAndPaid } from './maturity-sweep.js'
import { buildIdempotencyKey } from './accrual.js'

export interface PayoutDestination {
  accountId: string
  [k: string]: string
}

export interface PayoutExecutor {
  /**
   * CONTRACT (hard): execute MUST be idempotent on payoutId — the host adapter keys
   * its provider idempotency (Stripe idempotencyKey, …) on payoutId, so a crash-retry
   * of the SAME payout row never double-pays. The outside-the-tx boundary depends on it.
   */
  execute(req: {
    payoutId: string
    amountMinor: bigint
    destination: PayoutDestination
  }): Promise<
    | { ok: true; externalRefs: Record<string, string> }
    | { ok: false; code: string; error: string }
  >
}

export type SettlePayoutInput = {
  payoutId: string
}

export type SettlePayoutResult =
  | { ok: true; ledgerEntryId: string | null; externalRefs: Record<string, string> }
  | { ok: false; code: string; error: string; restored: boolean }

/** Sentinel when payout ledger row is missing or amount does not match. */
export class PayoutLedgerMismatchError extends Error {
  readonly code = 'PAYOUT_LEDGER_MISMATCH' as const
  readonly _affiliateError = 'PayoutLedgerMismatchError' as const

  constructor() {
    super('PAYOUT_LEDGER_MISMATCH')
    this.name = 'PayoutLedgerMismatchError'
  }
}

export function isPayoutLedgerMismatchError(e: unknown): e is PayoutLedgerMismatchError {
  return (
    typeof e === 'object' &&
    e !== null &&
    (e as { _affiliateError?: unknown })._affiliateError === 'PayoutLedgerMismatchError'
  )
}

function throwInsufficientForPayout(
  amountMinor: bigint,
  withdrawableMinor: bigint,
  maturedMinor: bigint,
): never {
  if (withdrawableMinor <= 0n || amountMinor > withdrawableMinor) {
    throw new InsufficientBalanceError('INSUFFICIENT_BALANCE', {
      requiredMinor: amountMinor,
      availableMinor: withdrawableMinor,
      bucket: 'withdrawable',
    })
  }
  throw new InsufficientBalanceError('INSUFFICIENT_BALANCE', {
    requiredMinor: amountMinor,
    availableMinor: maturedMinor,
    bucket: 'matured',
  })
}

/**
 * Lock wallet_vesting owner row, verify withdrawable+matured ceilings, append redemption
 * ledger row via core debitWithRead, decrement maturedMinor, recompute withdrawableMinor,
 * and link payout FK.
 *
 * Idempotent: if the ledger row already exists, balance is not double-debited.
 */
export async function debitPayoutInTx<
  S extends AffiliateSchema & LedgerSchema & VestingSchema,
>(
  tx: Transaction<S>,
  userId: string,
  payoutId: string,
  amountMinor: bigint,
): Promise<string> {
  if (amountMinor <= 0n) {
    throw new InvalidPayoutAmountError(amountMinor)
  }

  const idempotencyKey = buildIdempotencyKey('redemption', 'affiliate_payout', payoutId)

  const { inserted, id } = await debitWithRead(
    tx,
    {
      key: idempotencyKey,
      reason: 'redemption',
      ref: {
        sourceType: 'affiliate_payout',
        sourceId: payoutId,
        userId,
      },
      lock: {
        table: walletVesting,
        where: eq(walletVesting.ownerId, userId),
      },
    },
    async (locked: WalletVesting[]) => {
      const row = locked[0]
      const withdrawableMinor = row?.withdrawableMinor ?? 0n
      const maturedMinor = row?.maturedMinor ?? 0n

      if (
        withdrawableMinor <= 0n ||
        amountMinor > withdrawableMinor ||
        amountMinor > maturedMinor
      ) {
        throwInsufficientForPayout(amountMinor, withdrawableMinor, maturedMinor)
      }

      return {
        delta: -amountMinor,
        apply: async (applyTx) => {
          await applyTx
            .update(walletVesting)
            .set({
              maturedMinor: sql`${walletVesting.maturedMinor} - ${amountMinor}`,
              updatedAt: new Date(),
            })
            .where(eq(walletVesting.ownerId, userId))
        },
      }
    },
  )

  let entryId: string

  if (inserted && id !== null) {
    entryId = id

    await tx.insert(affiliateEntriesTable).values({
      entryId: id,
      ownerId: userId,
      entryType: 'redemption',
      sourceType: 'affiliate_payout',
      sourceId: payoutId,
      referralId: null,
      resolvedPct: null,
      memo: payoutId,
    })

    await tx
      .update(affiliatePayoutsTable)
      .set({ ledgerEntryId: id })
      .where(eq(affiliatePayoutsTable.id, payoutId))

    await recomputeWithdrawable(tx, { ownerId: userId }, async () => {
      const { eligibleMinor, paidMinor } = await computeOwnerEligibleAndPaid(tx, userId)
      return { eligibleMinor, paidMinor }
    })
  } else {
    const [existing] = await tx
      .select({ id: ledgerEntries.id })
      .from(ledgerEntries)
      .where(eq(ledgerEntries.idempotencyKey, idempotencyKey))
      .limit(1)

    if (!existing) {
      throw new Error(`PAYOUT_LEDGER_MISSING: ${payoutId}`)
    }

    entryId = existing.id

    await tx
      .update(affiliatePayoutsTable)
      .set({ ledgerEntryId: entryId })
      .where(eq(affiliatePayoutsTable.id, payoutId))
  }

  return entryId
}

/** Debit at most once — no-op when ledger_entry_id is already set. */
export async function ensurePayoutLedgerDebit<
  S extends AffiliateSchema & LedgerSchema & VestingSchema,
>(
  tx: Transaction<S>,
  payout: { id: string; userId: string; amountAgorot: bigint; ledgerEntryId: string | null },
): Promise<string> {
  if (payout.ledgerEntryId) return payout.ledgerEntryId
  return debitPayoutInTx(tx, payout.userId, payout.id, payout.amountAgorot)
}

/**
 * Admin settlement guard: idempotently ensure the redemption debit exists, then verify the
 * recorded ledger entry matches this payout — entryType 'redemption', sourceType 'affiliate_payout',
 * sourceId = payoutId, and the audit `-delta` equals `-amountAgorot` (bigint). Any mismatch throws
 * PayoutLedgerMismatchError. Returns the ledgerEntryId.
 */
export async function verifyPayoutReadyForSettlement<
  S extends AffiliateSchema & LedgerSchema & VestingSchema,
>(
  tx: Transaction<S>,
  payout: { id: string; userId: string; amountAgorot: bigint; ledgerEntryId: string | null },
): Promise<string> {
  const ledgerEntryId = await ensurePayoutLedgerDebit(tx, payout)

  const [row] = await tx
    .select({
      entryType: affiliateEntriesTable.entryType,
      sourceType: affiliateEntriesTable.sourceType,
      sourceId: affiliateEntriesTable.sourceId,
      delta: ledgerEntries.delta,
    })
    .from(affiliateEntriesTable)
    .innerJoin(ledgerEntries, eq(affiliateEntriesTable.entryId, ledgerEntries.id))
    .where(eq(affiliateEntriesTable.entryId, ledgerEntryId))
    .limit(1)

  if (
    !row ||
    row.entryType !== 'redemption' ||
    row.sourceType !== 'affiliate_payout' ||
    row.sourceId !== payout.id ||
    row.delta !== -payout.amountAgorot
  ) {
    throw new PayoutLedgerMismatchError()
  }

  return ledgerEntryId
}

/**
 * Restore wallet_vesting.maturedMinor after a failed/cancelled payout that already debited,
 * then recompute withdrawableMinor. Caller MUST mark the payout failed/cancelled in the same
 * tx before recompute so paidMinor excludes the released payout.
 * Idempotent on adjustment:affiliate_payout_reversal:<payoutId> (canonical key via buildIdempotencyKey; matches the fold + host-bridge reversal so cross-writer reversals dedup, never collide on affiliate_entries_idem_uq).
 */
export async function restorePayoutDebitInTx<
  S extends AffiliateSchema & LedgerSchema & VestingSchema,
>(
  tx: Transaction<S>,
  userId: string,
  payoutId: string,
  amountMinor: bigint,
): Promise<boolean> {
  if (amountMinor <= 0n) {
    throw new InvalidPayoutAmountError(amountMinor)
  }

  const reversalKey = buildIdempotencyKey('adjustment', 'affiliate_payout_reversal', payoutId)

  const { inserted, id } = await appendEntry(tx, {
    key: reversalKey,
    delta: amountMinor,
    reason: 'adjustment',
    ref: {
      sourceType: 'affiliate_payout_reversal',
      sourceId: payoutId,
      userId,
    },
  })

  if (!inserted || id === null) return false

  await tx
    .select()
    .from(walletVesting)
    .where(eq(walletVesting.ownerId, userId))
    .for('update')

  await tx
    .update(walletVesting)
    .set({
      maturedMinor: sql`${walletVesting.maturedMinor} + ${amountMinor}`,
      updatedAt: new Date(),
    })
    .where(eq(walletVesting.ownerId, userId))

  await tx.insert(affiliateEntriesTable).values({
    entryId: id,
    ownerId: userId,
    entryType: 'adjustment',
    sourceType: 'affiliate_payout_reversal',
    sourceId: payoutId,
    memo: `payout_failure_restore:${payoutId}`,
  })

  await recomputeWithdrawable(tx, { ownerId: userId }, async () => {
    const { eligibleMinor, paidMinor } = await computeOwnerEligibleAndPaid(tx, userId)
    return { eligibleMinor, paidMinor }
  })

  return true
}

type ClaimResult =
  | { kind: 'not_found' }
  | {
      kind: 'already_paid'
      ledgerEntryId: string | null
      stripeTransferId: string | null
      stripePayoutId: string | null
    }
  | { kind: 'in_progress' }
  | { kind: 'terminal' }
  | { kind: 'not_approved' }
  | { kind: 'invalid_amount' }
  | { kind: 'no_destination' }
  | {
      kind: 'claimed'
      ledgerEntryId: string
      userId: string
      amountAgorot: bigint
      destination: PayoutDestination
    }

function paidExternalRefs(
  stripeTransferId: string | null,
  stripePayoutId: string | null,
): Record<string, string> {
  return {
    ...(stripeTransferId ? { transferId: stripeTransferId } : {}),
    ...(stripePayoutId ? { payoutId: stripePayoutId } : {}),
  }
}

/**
 * Money-out boundary (R4 audit-hardened): claim tx → executor OUTSIDE tx → persist/restore tx.
 * Idempotent on payoutId — already-paid rows return prior success without re-executing.
 */
export async function settlePayout<S extends AffiliateSchema & LedgerSchema & VestingSchema>(
  db: TransactionalDatabase<S>,
  executor: PayoutExecutor,
  input: SettlePayoutInput,
): Promise<SettlePayoutResult> {
  const claim = await db.transaction(async (tx) => {
    const [row] = await tx
      .select({
        status: affiliatePayoutsTable.status,
        userId: affiliatePayoutsTable.userId,
        amountAgorot: affiliatePayoutsTable.amountAgorot,
        enrollmentId: affiliatePayoutsTable.enrollmentId,
        ledgerEntryId: affiliatePayoutsTable.ledgerEntryId,
        stripeTransferId: affiliatePayoutsTable.stripeTransferId,
        stripePayoutId: affiliatePayoutsTable.stripePayoutId,
        stripeAccountId: affiliateEnrollmentsTable.stripeAccountId,
      })
      .from(affiliatePayoutsTable)
      .innerJoin(
        affiliateEnrollmentsTable,
        eq(affiliatePayoutsTable.enrollmentId, affiliateEnrollmentsTable.id),
      )
      .where(eq(affiliatePayoutsTable.id, input.payoutId))
      .for('update')
      .limit(1)

    if (!row) {
      return { kind: 'not_found' as const }
    }

    if (row.status === 'paid') {
      return {
        kind: 'already_paid' as const,
        ledgerEntryId: row.ledgerEntryId,
        stripeTransferId: row.stripeTransferId,
        stripePayoutId: row.stripePayoutId,
      }
    }

    if (row.status === 'processing') {
      return { kind: 'in_progress' as const }
    }

    if (row.status === 'failed' || row.status === 'cancelled') {
      return { kind: 'terminal' as const }
    }

    if (row.status === 'requested') {
      return { kind: 'not_approved' as const }
    }

    if (row.status !== 'approved') {
      return { kind: 'not_found' as const }
    }

    if (row.amountAgorot <= 0n) {
      return { kind: 'invalid_amount' as const }
    }

    if (!row.stripeAccountId) {
      return { kind: 'no_destination' as const }
    }

    const destination: PayoutDestination = { accountId: row.stripeAccountId }

    const ledgerEntryId = await verifyPayoutReadyForSettlement(tx, {
      id: input.payoutId,
      userId: row.userId,
      amountAgorot: row.amountAgorot,
      ledgerEntryId: row.ledgerEntryId,
    })

    await tx
      .update(affiliatePayoutsTable)
      .set({ status: 'processing', processingAt: new Date() })
      .where(eq(affiliatePayoutsTable.id, input.payoutId))

    return {
      kind: 'claimed' as const,
      ledgerEntryId,
      userId: row.userId,
      amountAgorot: row.amountAgorot,
      destination,
    }
  })

  if (claim.kind === 'not_found') {
    return { ok: false, code: 'PAYOUT_NOT_FOUND', error: 'payout row missing', restored: false }
  }

  if (claim.kind === 'already_paid') {
    return {
      ok: true,
      ledgerEntryId: claim.ledgerEntryId,
      externalRefs: paidExternalRefs(claim.stripeTransferId, claim.stripePayoutId),
    }
  }

  if (claim.kind === 'in_progress') {
    return {
      ok: false,
      code: 'PAYOUT_IN_PROGRESS',
      error: 'payout already processing',
      restored: false,
    }
  }

  if (claim.kind === 'terminal') {
    return {
      ok: false,
      code: 'PAYOUT_TERMINAL',
      error: 'payout row is terminal — retry requires a new payout',
      restored: false,
    }
  }

  if (claim.kind === 'not_approved') {
    return {
      ok: false,
      code: 'PAYOUT_NOT_APPROVED',
      error: 'payout is not approved for settlement',
      restored: false,
    }
  }

  if (claim.kind === 'invalid_amount') {
    return {
      ok: false,
      code: 'PAYOUT_INVALID_AMOUNT',
      error: 'payout amount must be positive',
      restored: false,
    }
  }

  if (claim.kind === 'no_destination') {
    return {
      ok: false,
      code: 'PAYOUT_NO_DESTINATION',
      error: 'enrollment has no payout destination',
      restored: false,
    }
  }

  const amountMinor = claim.amountAgorot

  const execResult = await executor.execute({
    payoutId: input.payoutId,
    amountMinor,
    destination: claim.destination,
  })

  if (!execResult.ok) {
    const restored = await db.transaction(async (tx) => {
      await tx
        .update(affiliatePayoutsTable)
        .set({
          status: 'failed',
          failedAt: new Date(),
          failureReason: execResult.error,
        })
        .where(
          and(
            eq(affiliatePayoutsTable.id, input.payoutId),
            eq(affiliatePayoutsTable.status, 'processing'),
          ),
        )
      return restorePayoutDebitInTx(tx, claim.userId, input.payoutId, amountMinor)
    })

    return {
      ok: false,
      code: execResult.code,
      error: execResult.error,
      restored,
    }
  }

  const transferId = execResult.externalRefs.transferId ?? null
  const stripePayoutId = execResult.externalRefs.payoutId ?? null

  const persist = await db.transaction(async (tx) => {
    const [updated] = await tx
      .update(affiliatePayoutsTable)
      .set({
        status: 'paid',
        paidAt: new Date(),
        stripeTransferId: transferId,
        stripePayoutId,
      })
      .where(
        and(
          eq(affiliatePayoutsTable.id, input.payoutId),
          eq(affiliatePayoutsTable.status, 'processing'),
        ),
      )
      .returning({ id: affiliatePayoutsTable.id })

    if (updated) {
      return { kind: 'paid' as const }
    }

    await tx
      .update(affiliatePayoutsTable)
      .set({
        stripeTransferId: transferId,
        stripePayoutId,
      })
      .where(eq(affiliatePayoutsTable.id, input.payoutId))

    return { kind: 'swept' as const }
  })

  if (persist.kind === 'swept') {
    return {
      ok: false,
      code: 'PAYOUT_SWEPT_DURING_EXECUTION',
      error: 'reconcile: transfer landed after stuck-sweep restored debit',
      restored: true,
    }
  }

  return {
    ok: true,
    ledgerEntryId: claim.ledgerEntryId,
    externalRefs: execResult.externalRefs,
  }
}

/**
 * Reclaim payout rows stuck in processing after a worker crash mid-executor call.
 * sweepStuckAffiliatePayouts (R4 audit-hardened) — preserve byte-faithfully.
 *
 * ACCEPTED residual window (spec §5): if the transfer succeeded but the worker crashed
 * before persisting the ref, this restores the debit while funds already moved — accepted trade-off.
 */
export async function sweepStuckPayouts<S extends AffiliateSchema & LedgerSchema & VestingSchema>(
  db: TransactionalDatabase<S>,
  olderThanMinutes = 10,
): Promise<{ reclaimed: number }> {
  const stuck = await db
    .select({
      id: affiliatePayoutsTable.id,
      userId: affiliatePayoutsTable.userId,
      amountAgorot: affiliatePayoutsTable.amountAgorot,
    })
    .from(affiliatePayoutsTable)
    .where(
      and(
        eq(affiliatePayoutsTable.status, 'processing'),
        lt(
          affiliatePayoutsTable.processingAt,
          sql`now() - (${olderThanMinutes} * interval '1 minute')`,
        ),
      ),
    )

  let reclaimed = 0

  for (const row of stuck) {
    await db.transaction(async (tx) => {
      const [updated] = await tx
        .update(affiliatePayoutsTable)
        .set({
          status: 'failed',
          failedAt: new Date(),
          failureReason: 'stuck_processing_sweep',
        })
        .where(
          and(
            eq(affiliatePayoutsTable.id, row.id),
            eq(affiliatePayoutsTable.status, 'processing'),
          ),
        )
        .returning({ id: affiliatePayoutsTable.id })

      if (!updated) return

      await restorePayoutDebitInTx(tx, row.userId, row.id, row.amountAgorot)
      reclaimed += 1
    })
  }

  return { reclaimed }
}