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

/** Columns of the `webhook_queue_completions` table, derived from `Schema.WebhookQueueCompletion`. */
export type Table = db_Schema.WebhookQueueCompletion

/** A terminal delivery-obligation outcome. */
export type TerminalStatus = db_Schema.WebhookQueueCompletion['status']

/**
 * Reads the unexpired completion for a reference, if any. Rows are written by
 * the queue-event terminal transition and only ever read for dedupe.
 */
export async function get(
  db: Db.Db,
  reference: Webhooks.QueueReference,
  now: string,
): Promise<Table | null> {
  return (
    (await db.kysely
      .selectFrom('webhook_queue_completions')
      .selectAll()
      .where('eventId', '=', reference.eventId)
      .where('subscriptionId', '=', reference.subscriptionId)
      .where('expiresAt', '>', now)
      .executeTakeFirst()) ?? null
  )
}

// Per-statement delete bound: prune runs every minute, so a backlog drains in
// a few ticks instead of one unbounded DELETE churning the origin.
const pruneBatchLimit = 50_000

/** Deletes expired completions, bounded per call; expiry is filtered at query time, so pruning is space-only. */
export async function pruneExpired(db: Db.Db, now: string): Promise<number> {
  const result = await db.kysely
    .deleteFrom('webhook_queue_completions')
    // Repeated on the outer delete: the row recheck under concurrency re-runs
    // only this predicate, so a completion renewed mid-statement survives.
    .where('expiresAt', '<=', now)
    .where(({ eb, refTuple, selectFrom }) =>
      eb(
        refTuple('subscriptionId', 'eventId'),
        'in',
        selectFrom('webhook_queue_completions')
          .select(['subscriptionId', 'eventId'])
          .where('expiresAt', '<=', now)
          .limit(pruneBatchLimit)
          .$asTuple('subscriptionId', 'eventId'),
      ),
    )
    .executeTakeFirst()
  return Number(result.numDeletedRows)
}