import { type ExpressionBuilder, type Generated, type Insertable, type Selectable, sql } from 'kysely' // prettier-ignore

import type * as Webhooks from '../../internal/Webhooks.js'
import type * as Db from '../Db.js'
import * as WebhookQueueCompletions from './webhookQueueCompletions.js'
import type * as db_Schema from '../Schema.js'

/** Columns of the `webhook_queue_events` table, derived from `Schema.WebhookQueueEvent`. */
export type Table = Omit<
  db_Schema.WebhookQueueEvent,
  'attemptCount' | 'attemptingAt' | 'expiresAt' | 'nextAttemptAt' | 'observedAt' | 'status'
> & {
  /** Delivery attempts claimed so far; database default 0. */
  attemptCount: Generated<number>
  /** ISO timestamp of the current claim; database default null. */
  attemptingAt: Generated<string | null>
  /** Legacy dedupe deadline; unwritten since completions took over. */
  expiresAt: Generated<string | null>
  /** ISO retry due time; database default null (due immediately). */
  nextAttemptAt: Generated<string | null>
  /** ISO head-observation stamp; database default null (replays carry none). */
  observedAt: Generated<string | null>
  /** Delivery-obligation lifecycle status; database default `pending`. */
  status: Generated<db_Schema.WebhookQueueEvent['status']>
}

/** A stored webhook Queue event. */
export type Record = Selectable<Table>

/**
 * Reads one pending envelope. The status filter only ever excludes rows
 * completed before the completions split; completions now leave the table.
 */
export async function get(
  db: Db.Db,
  reference: Webhooks.QueueReference,
): Promise<Webhooks.QueueEvent | null> {
  return (
    (await db.kysely
      .selectFrom('webhook_queue_events')
      .selectAll()
      .where('eventId', '=', reference.eventId)
      .where('status', '=', 'pending')
      .where('subscriptionId', '=', reference.subscriptionId)
      .executeTakeFirst()) ?? null
  )
}

/**
 * Stages envelopes as fresh obligations. Unexpired completions and any
 * existing row suppress the replay; a replay stages only when neither exists,
 * which a completion past its dedupe window guarantees for completed work.
 */
export async function ensure(db: Db.Db, rows: readonly Insertable<Table>[]): Promise<number> {
  if (rows.length === 0) return 0
  const values = sql.join(
    rows.map(
      (row) =>
        sql`(${row.subscriptionId}::text, ${row.eventId}::text, ${JSON.stringify(row.envelope)}::jsonb, ${row.createdAt}::text, ${row.nextAttemptAt ?? null}::text, ${row.observedAt ?? null}::text)`,
    ),
  )
  // The VALUES source is raw because kysely has no table constructor for it,
  // but both real tables stay in the builder so WithSchemaPlugin qualifies
  // them (previews isolate by schema; raw identifiers would escape it). The
  // completion check must ride the insert: a separate read would let a
  // completion commit between the statements and re-stage just-finished work.
  // `column1..6` are Postgres's default VALUES names, in the row order above;
  // the window comparison uses each row's own staging time, not a shared clock.
  // Conflicts only ever hit pending rows (or pre-split terminal rows during
  // the drain), and both suppress the replay; a legitimate post-expiry replay
  // meets no conflict because its completed row was deleted.
  const result = await db.kysely
    .insertInto('webhook_queue_events')
    .columns(['subscriptionId', 'eventId', 'envelope', 'createdAt', 'nextAttemptAt', 'observedAt'])
    .expression((eb) =>
      eb
        .selectFrom(sql`(VALUES ${values})`.as('v'))
        .select([
          sql`v.column1`.as('subscriptionId'),
          sql`v.column2`.as('eventId'),
          sql`v.column3`.as('envelope'),
          sql`v.column4`.as('createdAt'),
          sql`v.column5`.as('nextAttemptAt'),
          sql`v.column6`.as('observedAt'),
        ])
        .where(({ exists, not, selectFrom }) =>
          not(
            exists(
              selectFrom('webhook_queue_completions')
                .select('eventId')
                .where('subscriptionId', '=', sql<string>`v.column1`)
                .where('eventId', '=', sql<string>`v.column2`)
                .where('expiresAt', '>', sql<string>`v.column4`),
            ),
          ),
        ),
    )
    .onConflict((oc) => oc.columns(['subscriptionId', 'eventId']).doNothing())
    .executeTakeFirst()
  return Number(result.numInsertedOrUpdatedRows ?? 0n)
}

/** Due, unclaimed pending work: shared by {@link claim} and {@link due}. */
function whereClaimable(
  eb: ExpressionBuilder<Db.Database, 'webhook_queue_events'>,
  options: { now: string; staleBefore: string },
) {
  return eb.and([
    eb('status', '=', 'pending'),
    eb.or([eb('nextAttemptAt', 'is', null), eb('nextAttemptAt', '<=', options.now)]),
    eb.or([eb('attemptingAt', 'is', null), eb('attemptingAt', '<=', options.staleBefore)]),
  ])
}

/** Claims one due pending obligation with a single atomic update. */
export async function claim(
  db: Db.Db,
  reference: Webhooks.QueueReference,
  options: claim.Options,
): Promise<claim.Result> {
  const claimed = await db.kysely
    .updateTable('webhook_queue_events')
    .set((eb) => ({ attemptCount: eb('attemptCount', '+', 1), attemptingAt: options.now }))
    .where('subscriptionId', '=', reference.subscriptionId)
    .where('eventId', '=', reference.eventId)
    .where((eb) => whereClaimable(eb, options))
    .returningAll()
    .executeTakeFirst()
  if (claimed) return { record: claimed, type: 'claimed' }

  // Queue first, completions second: a complete() landing between the reads
  // then still surfaces its completion, so a finished obligation never reads
  // as missing, the one claim outcome alerts treat as a fault.
  const existing = await db.kysely
    .selectFrom('webhook_queue_events')
    .select(['attemptingAt', 'nextAttemptAt', 'status'])
    .where('subscriptionId', '=', reference.subscriptionId)
    .where('eventId', '=', reference.eventId)
    .executeTakeFirst()
  if (!existing) {
    // An unexpired completion reads as terminal so claim-outcome metrics keep
    // their meaning; an expired one reads as missing, like a pruned row.
    const completed = await WebhookQueueCompletions.get(db, reference, options.now)
    if (completed) return { status: completed.status, type: 'terminal' }
    return { type: 'missing' }
  }
  // Reachable only for rows completed before the completions split.
  if (existing.status !== 'pending') return { status: existing.status, type: 'terminal' }
  if (existing.attemptingAt !== null && existing.attemptingAt > options.staleBefore)
    return { type: 'attempting' }
  return { nextAttemptAt: existing.nextAttemptAt, type: 'scheduled' }
}

export declare namespace claim {
  /** Claim clock bounds. */
  type Options = {
    /** Current ISO timestamp; becomes the claim marker. */
    now: string
    /** ISO cutoff below which an existing claim counts as stale. */
    staleBefore: string
  }

  /** Atomic claim outcome. */
  type Result =
    | { type: 'attempting' }
    | { record: Record; type: 'claimed' }
    | { type: 'missing' }
    | { nextAttemptAt: string | null; type: 'scheduled' }
    | { status: WebhookQueueCompletions.TerminalStatus; type: 'terminal' }
}

/**
 * Marks a claimed obligation terminal: one atomic statement deletes the queue
 * row and writes the insert-only completion that starts the dedupe window.
 * The `attemptingAt` predicate fences stale claimants.
 */
export async function complete(
  db: Db.Db,
  reference: Webhooks.QueueReference,
  options: complete.Options,
): Promise<boolean> {
  const result = await db.kysely
    .with('done', (qb) =>
      qb
        .deleteFrom('webhook_queue_events')
        .where('subscriptionId', '=', reference.subscriptionId)
        .where('eventId', '=', reference.eventId)
        .where('status', '=', 'pending')
        .where('attemptingAt', '=', options.claimedAt)
        .returning(['subscriptionId', 'eventId']),
    )
    .insertInto('webhook_queue_completions')
    .columns(['subscriptionId', 'eventId', 'status', 'expiresAt'])
    .expression((eb) =>
      eb
        .selectFrom('done')
        .select((s) => [
          'subscriptionId',
          'eventId',
          s.val(options.status).as('status'),
          s.val(options.expiresAt).as('expiresAt'),
        ]),
    )
    // Re-completing a replayed obligation: the latest delivery outcome wins
    // and retention restarts, matching pre-split replay semantics; doNothing
    // would also report zero rows and fail a fence that won.
    .onConflict((oc) =>
      oc.columns(['subscriptionId', 'eventId']).doUpdateSet((eb) => ({
        expiresAt: eb.ref('excluded.expiresAt'),
        status: eb.ref('excluded.status'),
      })),
    )
    .executeTakeFirst()
  return result.numInsertedOrUpdatedRows === 1n
}

export declare namespace complete {
  /** Fenced terminal transition. */
  type Options = {
    /** Claim marker returned by {@link claim}; fences stale claimants. */
    claimedAt: string
    /** ISO retention deadline for the dedupe completion. */
    expiresAt: string
    /** Terminal outcome. */
    status: WebhookQueueCompletions.TerminalStatus
  }
}

/** Releases a claim and schedules the next attempt; fenced like {@link complete}. */
export async function schedule(
  db: Db.Db,
  reference: Webhooks.QueueReference,
  options: schedule.Options,
): Promise<boolean> {
  const result = await db.kysely
    .updateTable('webhook_queue_events')
    .set({ attemptingAt: null, nextAttemptAt: options.nextAttemptAt })
    .where('subscriptionId', '=', reference.subscriptionId)
    .where('eventId', '=', reference.eventId)
    .where('status', '=', 'pending')
    .where('attemptingAt', '=', options.claimedAt)
    .executeTakeFirst()
  return result.numUpdatedRows === 1n
}

export declare namespace schedule {
  /** Fenced retry scheduling. */
  type Options = {
    /** Claim marker returned by {@link claim}; fences stale claimants. */
    claimedAt: string
    /** ISO due time of the next attempt. */
    nextAttemptAt: string
  }
}

/** Lists due pending references for the sweeper, most overdue first. */
export async function due(db: Db.Db, options: due.Options): Promise<Webhooks.QueueReference[]> {
  return (
    db.kysely
      .selectFrom('webhook_queue_events')
      .select(['eventId', 'subscriptionId'])
      .where((eb) => whereClaimable(eb, options))
      // Matches the pending partial index order so the scan needs no sort.
      .orderBy(sql`next_attempt_at nulls first`)
      .limit(options.limit)
      .execute()
  )
}

export declare namespace due {
  /** Sweep bounds. */
  type Options = {
    /** Maximum references to return. */
    limit: number
    /** Current ISO timestamp. */
    now: string
    /** ISO cutoff below which an existing claim counts as stale. */
    staleBefore: string
  }
}

/** Returns the pending-obligation snapshot, split by due-and-unclaimed work. */
export async function pending(db: Db.Db, options: pending.Options): Promise<pending.Result> {
  const row = await db.kysely
    .selectFrom('webhook_queue_events')
    .select((eb) => [
      eb.fn.countAll<number>().as('count'),
      eb.fn.min('createdAt').as('oldestAt'),
      eb.fn
        .countAll<number>()
        .filterWhere((fb) => whereClaimable(fb, options))
        .as('dueCount'),
      eb.fn
        // Null next attempts (migration-backfilled rows) are due immediately;
        // coalesce to staging time so they age instead of vanishing from MIN.
        .min(eb.fn.coalesce('nextAttemptAt', 'createdAt'))
        .filterWhere((fb) => whereClaimable(fb, options))
        .as('oldestDueAt'),
    ])
    .where('status', '=', 'pending')
    .executeTakeFirstOrThrow()
  return {
    count: Number(row.count),
    dueCount: Number(row.dueCount),
    oldestAt: row.oldestAt,
    oldestDueAt: row.oldestDueAt,
  }
}

export declare namespace pending {
  /** Snapshot clock bounds; the read-only subset of {@link claim.Options}. */
  type Options = Pick<claim.Options, 'now' | 'staleBefore'>

  /** Pending-work snapshot. */
  type Result = {
    /** Pending obligations, including in-flight claims and scheduled retries. */
    count: number
    /** Obligations due now with no live claim; the sweeper's backlog. */
    dueCount: number
    /** Oldest pending staging timestamp, or null when none. */
    oldestAt: string | null
    /** Oldest due-and-unclaimed retry time, or null when none; drives the stall alert. */
    oldestDueAt: string | null
  }
}