import type { Selectable } from 'kysely'

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_deliveries` table, derived from `Schema.WebhookDelivery`. */
export type Table = db_Schema.WebhookDelivery

/** A stored webhook-delivery row. */
export type Record = Selectable<Table>

/** A failed delivery joined to its organization-owned subscription. */
export type OrganizationFailure = Pick<
  Record,
  'attempt' | 'createdAt' | 'error' | 'id' | 'responseStatus' | 'subscriptionId'
>

/** Maps a stored row to the domain delivery (null columns → absent fields). */
export function toDelivery(row: Record): Webhooks.Delivery {
  return {
    attempt: row.attempt,
    createdAt: row.createdAt,
    envelope: row.envelope,
    ...(row.error === null ? {} : { error: row.error }),
    eventId: row.eventId,
    id: row.id,
    requestUrl: row.requestUrl,
    ...(row.responseMs === null ? {} : { responseMs: row.responseMs }),
    ...(row.responseStatus === null ? {} : { responseStatus: row.responseStatus }),
    status: row.status,
    subscriptionId: row.subscriptionId,
  }
}

/**
 * Inserts a delivery row from its domain value.
 *
 * @param db - The database.
 * @param delivery - The delivery to insert.
 * @param expiresAt - Retention deadline (ISO); the row is invisible past it.
 */
export async function insert(
  db: Db.Db,
  delivery: Webhooks.Delivery,
  expiresAt: string,
): Promise<void> {
  await db.kysely
    .insertInto('webhook_deliveries')
    .values({
      attempt: delivery.attempt,
      createdAt: delivery.createdAt,
      envelope: delivery.envelope,
      error: delivery.error ?? null,
      eventId: delivery.eventId,
      expiresAt,
      id: delivery.id,
      requestUrl: delivery.requestUrl,
      responseMs: delivery.responseMs ?? null,
      responseStatus: delivery.responseStatus ?? null,
      status: delivery.status,
      subscriptionId: delivery.subscriptionId,
    })
    .execute()
}

/**
 * Lists a subscription's unexpired deliveries, newest first, with keyset
 * paging (`id < cursor` — ids embed a zero-padded timestamp, so lexical order
 * is chronological).
 *
 * @param db - The database.
 * @param subscriptionId - The owning subscription id (`wh_…`).
 * @param options - Paging options plus the retention timestamp.
 * @returns The deliveries.
 */
export async function list(
  db: Db.Db,
  subscriptionId: string,
  options: list.Options,
): Promise<Webhooks.Delivery[]> {
  let query = db.kysely
    .selectFrom('webhook_deliveries')
    .selectAll()
    .where('subscriptionId', '=', subscriptionId)
    .where('expiresAt', '>', options.now)
    .orderBy('id', 'desc')
  if (options.cursor) query = query.where('id', '<', options.cursor)
  if (options.offset) query = query.offset(options.offset)
  if (options.limit !== undefined) query = query.limit(options.limit)
  return (await query.execute()).map(toDelivery)
}

/** Lists recent failed webhook deliveries across an organization. */
export function listFailuresByOrg(
  db: Db.Db,
  orgId: string,
  options: listFailuresByOrg.Options,
): Promise<OrganizationFailure[]> {
  return db.kysely
    .selectFrom('webhook_deliveries')
    .innerJoin(
      'webhook_subscriptions',
      'webhook_subscriptions.id',
      'webhook_deliveries.subscriptionId',
    )
    .select([
      'webhook_deliveries.attempt',
      'webhook_deliveries.createdAt',
      'webhook_deliveries.error',
      'webhook_deliveries.id',
      'webhook_deliveries.responseStatus',
      'webhook_deliveries.subscriptionId',
    ])
    .where('webhook_subscriptions.ownerId', '=', orgId)
    .where('webhook_subscriptions.ownerType', '=', 'api_key')
    .where('webhook_deliveries.createdAt', '>=', options.from)
    .where('webhook_deliveries.expiresAt', '>', options.now)
    .where('webhook_deliveries.status', '=', 'failed')
    .orderBy('webhook_deliveries.createdAt', 'desc')
    .orderBy('webhook_deliveries.id', 'desc')
    .limit(options.limit)
    .execute()
}

export declare namespace listFailuresByOrg {
  /** Bounds for the organization failure scan. */
  type Options = {
    /** Window start (ISO 8601), inclusive. */
    from: string
    /** Maximum failures to return. */
    limit: number
    /** ISO timestamp retention is evaluated against. */
    now: string
  }
}

/** Counts failed webhook deliveries across an organization in a time window. */
export async function countFailuresByOrg(
  db: Db.Db,
  orgId: string,
  options: countFailuresByOrg.Options,
): Promise<number> {
  const row = await db.kysely
    .selectFrom('webhook_deliveries')
    .innerJoin(
      'webhook_subscriptions',
      'webhook_subscriptions.id',
      'webhook_deliveries.subscriptionId',
    )
    .select((eb) => eb.fn.countAll().as('count'))
    .where('webhook_subscriptions.ownerId', '=', orgId)
    .where('webhook_subscriptions.ownerType', '=', 'api_key')
    .where('webhook_deliveries.createdAt', '>=', options.from)
    .where('webhook_deliveries.expiresAt', '>', options.now)
    .where('webhook_deliveries.status', '=', 'failed')
    .executeTakeFirstOrThrow()
  return Number(row.count)
}

export declare namespace countFailuresByOrg {
  /** Bounds for the organization failure count. */
  type Options = {
    /** Window start (ISO 8601), inclusive. */
    from: string
    /** ISO timestamp retention is evaluated against. */
    now: string
  }
}

export declare namespace list {
  /** Options for {@link list}. */
  type Options = {
    /** Return deliveries whose id sorts before this one (older entries). */
    cursor?: string | undefined
    /** Maximum deliveries to return. */
    limit?: number | undefined
    /** ISO timestamp retention is evaluated against. */
    now: string
    /** Rows to skip from the head (positional pagination; exclusive with `cursor`). */
    offset?: number | undefined
  }
}

/**
 * Counts a subscription's unexpired deliveries.
 *
 * @param db - The database.
 * @param subscriptionId - The owning subscription id (`wh_…`).
 * @param now - ISO timestamp retention is evaluated against.
 * @returns The count.
 */
export async function count(db: Db.Db, subscriptionId: string, now: string): Promise<number> {
  const row = await db.kysely
    .selectFrom('webhook_deliveries')
    .select((eb) => eb.fn.countAll().as('count'))
    .where('subscriptionId', '=', subscriptionId)
    .where('expiresAt', '>', now)
    .executeTakeFirstOrThrow()
  return Number(row.count)
}

/**
 * Reads a single unexpired delivery scoped to its subscription.
 *
 * @param db - The database.
 * @param subscriptionId - The owning subscription id (`wh_…`).
 * @param deliveryId - The delivery id (`whd_…`).
 * @param now - ISO timestamp retention is evaluated against.
 * @returns The delivery, or `null` when absent.
 */
export async function get(
  db: Db.Db,
  subscriptionId: string,
  deliveryId: string,
  now: string,
): Promise<Webhooks.Delivery | null> {
  const row = await db.kysely
    .selectFrom('webhook_deliveries')
    .selectAll()
    .where('subscriptionId', '=', subscriptionId)
    .where('id', '=', deliveryId)
    .where('expiresAt', '>', now)
    .executeTakeFirst()
  return row ? toDelivery(row) : 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 deliveries past their retention deadline. Called best-effort from
 * the poller tick.
 *
 * @param db - The database.
 * @param now - ISO timestamp retention is evaluated against.
 * @returns The number of rows deleted.
 */
export async function pruneExpired(db: Db.Db, now: string): Promise<number> {
  const result = await db.kysely
    .deleteFrom('webhook_deliveries')
    // Repeated on the outer delete: the row recheck under concurrency re-runs
    // only this predicate, so a row renewed mid-statement survives.
    .where('expiresAt', '<=', now)
    .where('id', 'in', (eb) =>
      eb
        .selectFrom('webhook_deliveries')
        .select('id')
        .where('expiresAt', '<=', now)
        .limit(pruneBatchLimit),
    )
    .executeTakeFirst()
  return Number(result.numDeletedRows)
}