import type { ExpressionBuilder, Selectable } from 'kysely'
import { sql } from 'kysely'

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

const subscriptionLimitLockClass = 1464353612

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

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

/** Reads one live subscription by id for an internal Queue delivery. */
export async function getById(
  db: Db.Db,
  id: string,
  now: string,
): Promise<Webhooks.Subscription | null> {
  const row = await db.kysely
    .selectFrom('webhook_subscriptions')
    .selectAll()
    .where('id', '=', id)
    .where(live(now))
    .executeTakeFirst()
  return row ? toSubscription(row) : null
}

/** Maps an owner union to its stored `ownerId` column value. */
export function ownerId(owner: Webhooks.Owner): string {
  return owner.type === 'api_key' ? owner.orgId : owner.payer
}

/** Maps a stored row to the domain subscription (null columns → absent fields). */
export function toSubscription(row: Record): Webhooks.Subscription {
  return {
    chainId: row.chainId,
    ...(row.context === null ? {} : { context: row.context }),
    createdAt: row.createdAt,
    destination: row.destination,
    ...(row.environment === null ? {} : { environment: row.environment }),
    eventType: row.eventType,
    ...(row.expiresAt === null ? {} : { expiresAt: row.expiresAt }),
    failureCount: row.failureCount,
    filters: row.filters,
    id: row.id,
    ...(row.lastDeliveryAt === null ? {} : { lastDeliveryAt: row.lastDeliveryAt }),
    owner:
      row.ownerType === 'api_key'
        ? { orgId: row.ownerId, type: 'api_key' }
        : { payer: row.ownerId, type: 'mpp' },
    ...(row.projectId === null ? {} : { projectId: row.projectId }),
    secret: row.secret,
    status: row.status,
    updatedAt: row.updatedAt,
  }
}

/**
 * Inserts a subscription row from its domain value.
 *
 * @param db - The database.
 * @param subscription - The subscription to insert.
 * @param options - Initial cursor options.
 */
export async function insert(
  db: Db.Db,
  subscription: Webhooks.Subscription,
  options: insert.Options = {},
): Promise<void> {
  await db.kysely
    .insertInto('webhook_subscriptions')
    .values({
      chainId: subscription.chainId,
      context: subscription.context ?? null,
      createdAt: subscription.createdAt,
      destination: subscription.destination,
      environment: subscription.environment ?? null,
      eventType: subscription.eventType,
      expiresAt: subscription.expiresAt ?? null,
      failureCount: subscription.failureCount,
      filters: subscription.filters,
      id: subscription.id,
      lastDeliveryAt: subscription.lastDeliveryAt ?? null,
      ownerId: ownerId(subscription.owner),
      ownerType: subscription.owner.type,
      pollerCursor: options.pollerCursor ?? null,
      projectId: subscription.projectId ?? null,
      secret: subscription.secret,
      status: subscription.status,
      updatedAt: subscription.updatedAt,
    })
    .execute()
}

export declare namespace insert {
  /** Options for {@link insert}. */
  type Options = {
    /** Initial cursor committed with the subscription. */
    pollerCursor?: string | undefined
  }
}

/**
 * Inserts a subscription when its owner remains below the configured cap.
 * An owner-scoped transaction lock serializes the count and insert.
 */
export async function insertWithinLimit(
  db: Db.Db,
  subscription: Webhooks.Subscription,
  options: insertWithinLimit.Options,
): Promise<boolean> {
  return db.transaction(async (tx) => {
    const owner = `${subscription.owner.type}:${ownerId(subscription.owner)}`
    await sql`select pg_advisory_xact_lock(${subscriptionLimitLockClass}, hashtext(${owner}))`.execute(tx.kysely) // prettier-ignore
    const existing = await count(tx, subscription.owner, options.now)
    if (existing >= options.maxPerOwner) return false
    await insert(
      tx,
      subscription,
      options.pollerCursor === undefined ? {} : { pollerCursor: options.pollerCursor },
    )
    return true
  })
}

export declare namespace insertWithinLimit {
  /** Options for an atomic capped insert. */
  type Options = {
    /** Maximum live subscriptions the owner may hold. */
    maxPerOwner: number
    /** ISO timestamp used to exclude expired rows from the count. */
    now: string
    /** Initial cursor committed with the subscription. */
    pollerCursor?: string | undefined
  }
}

/**
 * Reads an owner's subscription by id; expired rows never match (query-time
 * expiry, same read semantics as the old store TTL).
 *
 * @param db - The database.
 * @param owner - The authenticated owner.
 * @param id - The subscription id (`wh_…`).
 * @param now - ISO timestamp expiry is evaluated against.
 * @returns The subscription, or `null` when absent (or not the owner's).
 */
export async function get(
  db: Db.Db,
  owner: Webhooks.Owner,
  id: string,
  now: string,
  access?: Webhooks.Access,
): Promise<Webhooks.Subscription | null> {
  let query = db.kysely
    .selectFrom('webhook_subscriptions')
    .selectAll()
    .where('id', '=', id)
    .where('ownerType', '=', owner.type)
    .where('ownerId', '=', ownerId(owner))
    .where(live(now))
  if (access) query = query.where(privateResourceAccess(access))
  const row = await query.executeTakeFirst()
  return row ? toSubscription(row) : null
}

/**
 * Lists an owner's live subscriptions, newest first, with keyset paging
 * (`id < cursor` — ids embed a zero-padded timestamp, so lexical order is
 * chronological).
 *
 * @param db - The database.
 * @param owner - The authenticated owner.
 * @param options - Paging options plus the expiry timestamp.
 * @returns The subscriptions.
 */
export async function list(
  db: Db.Db,
  owner: Webhooks.Owner,
  options: list.Options,
): Promise<Webhooks.Subscription[]> {
  let query = db.kysely
    .selectFrom('webhook_subscriptions')
    .selectAll()
    .where('ownerType', '=', owner.type)
    .where('ownerId', '=', ownerId(owner))
    .where(live(options.now))
    .orderBy('id', 'desc')
  if (options.access) query = query.where(privateResourceAccess(options.access))
  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(toSubscription)
}

export declare namespace list {
  /** Options for {@link list}. */
  type Options = {
    /** Private-resource visibility scope. */
    access?: Webhooks.Access | undefined
    /** Return subscriptions older than this id (keyset paging, newest first). */
    cursor?: string | undefined
    /** Maximum subscriptions to return. */
    limit?: number | undefined
    /** ISO timestamp expiry is evaluated against. */
    now: string
    /** Rows to skip from the head (positional pagination; exclusive with `cursor`). */
    offset?: number | undefined
  }
}

/**
 * Counts an owner's live subscriptions.
 *
 * @param db - The database.
 * @param owner - The authenticated owner.
 * @param now - ISO timestamp expiry is evaluated against.
 * @returns The count.
 */
export async function count(
  db: Db.Db,
  owner: Webhooks.Owner,
  now: string,
  access?: Webhooks.Access,
): Promise<number> {
  let query = db.kysely
    .selectFrom('webhook_subscriptions')
    .select((eb) => eb.fn.countAll().as('count'))
    .where('ownerType', '=', owner.type)
    .where('ownerId', '=', ownerId(owner))
    .where(live(now))
  if (access) query = query.where(privateResourceAccess(access))
  const row = await query.executeTakeFirstOrThrow()
  return Number(row.count)
}

/**
 * Applies a column patch to a subscription. Only the provided columns are
 * written; `context: null` clears the stored context.
 *
 * @param db - The database.
 * @param id - The subscription id (`wh_…`).
 * @param set - The columns to write.
 * @returns The updated row's subscription, or `undefined` when absent.
 */
export async function update(
  db: Db.Db,
  id: string,
  set: update.Set,
): Promise<Webhooks.Subscription | undefined> {
  const row = await db.kysely
    .updateTable('webhook_subscriptions')
    .set(set)
    .where('id', '=', id)
    .returningAll()
    .executeTakeFirst()
  return row ? toSubscription(row) : undefined
}

export declare namespace update {
  /** Columns writable by a subscription patch. */
  type Set = {
    /** New human context; null clears it. */
    context?: Webhooks.Context | null | undefined
    /** New delivery destination. */
    destination?: Webhooks.Destination | undefined
    /** New filter predicates. */
    filters?: Webhooks.Subscription['filters'] | undefined
    /** New lifecycle status. */
    status?: Webhooks.Status | undefined
    /** Mutation timestamp (ISO). */
    updatedAt: string
  }
}

/**
 * Deletes an owner's live subscription; delivery rows cascade via the FK.
 *
 * @param db - The database.
 * @param owner - The authenticated owner.
 * @param id - The subscription id (`wh_…`).
 * @param now - ISO timestamp expiry is evaluated against.
 * @returns Whether a row was deleted.
 */
export async function remove(
  db: Db.Db,
  owner: Webhooks.Owner,
  id: string,
  now: string,
  access?: Webhooks.Access,
): Promise<boolean> {
  let query = db.kysely
    .deleteFrom('webhook_subscriptions')
    .where('id', '=', id)
    .where('ownerType', '=', owner.type)
    .where('ownerId', '=', ownerId(owner))
    .where(live(now))
  if (access) query = query.where(privateResourceAccess(access))
  const result = await query.executeTakeFirst()
  return result.numDeletedRows > 0n
}

function privateResourceAccess(access: Webhooks.Access) {
  const fundingRead = access.scopes.some(
    (scope) => scope === Scope.wildcard || scope === 'funding:read',
  )
  return (eb: ExpressionBuilder<Db.Database, 'webhook_subscriptions'>) =>
    eb.or([
      sql<boolean>`${eb.ref('eventType')} not like ${'funding:%'}`,
      ...(fundingRead
        ? [
            eb.and([
              sql<boolean>`${eb.ref('eventType')} like ${'funding:%'}`,
              eb('environment', '=', access.environment),
              ...(access.projectId === undefined ? [] : [eb('projectId', '=', access.projectId)]),
            ]),
          ]
        : []),
    ])
}

/**
 * Lists live `active` subscriptions for a `(chainId, eventType)` stream,
 * served by the partial status index. Each row carries its cursor so callers
 * avoid a per-subscription read.
 *
 * @param db - The database.
 * @param chainId - Chain the stream belongs to.
 * @param eventType - Event type of the stream.
 * @param now - ISO timestamp expiry is evaluated against.
 * @returns Cursor-and-subscription pairs, in id (chronological) order.
 */
export async function listActive(
  db: Db.Db,
  chainId: number,
  eventType: Webhooks.EventType,
  now: string,
): Promise<{ cursor: string | null; subscription: Webhooks.Subscription }[]> {
  const rows = await db.kysely
    .selectFrom('webhook_subscriptions')
    .selectAll()
    .where('chainId', '=', chainId)
    .where('eventType', '=', eventType)
    .where('status', '=', 'active')
    .where(live(now))
    .orderBy('id')
    .execute()
  return rows.map((row) => ({ cursor: row.pollerCursor, subscription: toSubscription(row) }))
}

/**
 * Lists every live `active` subscription on a chain in one read. Callers
 * group the rows by event type so empty streams cost no extra database reads.
 *
 * @param db - The database.
 * @param options - Chain, configured event types, and expiry time.
 * @returns Cursor-and-subscription pairs ordered by event type, then id.
 */
export async function listActiveForChain(
  db: Db.Db,
  options: listActiveForChain.Options,
): Promise<{ cursor: string | null; subscription: Webhooks.Subscription }[]> {
  if (options.eventTypes.length === 0) return []
  const rows = await db.kysely
    .selectFrom('webhook_subscriptions')
    .selectAll()
    .where('chainId', '=', options.chainId)
    .where('eventType', 'in', options.eventTypes)
    .where('status', '=', 'active')
    .where(live(options.now))
    .orderBy('eventType')
    .orderBy('id')
    .execute()
  return rows.map((row) => ({ cursor: row.pollerCursor, subscription: toSubscription(row) }))
}

export declare namespace listActiveForChain {
  /** Filters for the chain-level subscription read. */
  type Options = {
    /** Chain whose subscriptions to read. */
    chainId: number
    /** Event types eligible for the read. */
    eventTypes: readonly Webhooks.EventType[]
    /** ISO timestamp expiry is evaluated against. */
    now: string
  }
}

/** Lists live funding subscriptions visible to one resource owner. */
export async function listActiveForFundingEvent(
  db: Db.Db,
  options: listActiveForFundingEvent.Options,
): Promise<Webhooks.Subscription[]> {
  let query = db.kysely
    .selectFrom('webhook_subscriptions')
    .selectAll()
    .where('chainId', '=', options.chainId)
    .where('environment', '=', options.environment)
    .where('eventType', '=', options.eventType)
    .where('ownerId', '=', options.orgId)
    .where('ownerType', '=', 'api_key')
    .where('status', '=', 'active')
    .where(live(options.now))
  if (options.eventCreatedAt !== undefined)
    query = query.where('createdAt', '<=', options.eventCreatedAt)
  const projectId = options.projectId
  query =
    projectId === undefined
      ? query.where('projectId', 'is', null)
      : query.where((eb) => eb.or([eb('projectId', 'is', null), eb('projectId', '=', projectId)]))
  // Keep subscriptions present until their staged queue events commit.
  return (await query.orderBy('id').forKeyShare().execute()).map(toSubscription)
}

export declare namespace listActiveForFundingEvent {
  /** Ownership scope and chain for a funding resource event. */
  type Options = {
    /** Destination Tempo chain id. */
    chainId: number
    /** Resource API-key environment. */
    environment: 'production' | 'sandbox'
    /** Event timestamp used to keep subscriptions future-only, when supplied. */
    eventCreatedAt?: string | undefined
    /** Funding event type to match. */
    eventType: Extract<Webhooks.EventType, `funding:${string}`>
    /** ISO timestamp used to exclude expired subscriptions. */
    now: string
    /** Owning organization id. */
    orgId: string
    /** Attributed funding resource project, when present. */
    projectId?: string | undefined
  }
}

/**
 * Reads a subscription's keyset cursor.
 *
 * @param db - The database.
 * @param id - The subscription id (`wh_…`).
 * @returns The cursor, or `null` when unset (or the row is absent).
 */
export async function getCursor(db: Db.Db, id: string): Promise<string | null> {
  const row = await db.kysely
    .selectFrom('webhook_subscriptions')
    .select('pollerCursor')
    .where('id', '=', id)
    .executeTakeFirst()
  return row?.pollerCursor ?? null
}

/**
 * Records a successful delivery: resets the consecutive-failure counter and
 * stamps `lastDeliveryAt`. Skipped when neither field would move, so
 * deliveries stop contending on the row's lock.
 *
 * @param db - The database.
 * @param id - The subscription id (`wh_…`).
 * @param options - Delivery timestamp and stamp-staleness bound.
 * @returns The updated subscription, or `undefined` when nothing was written.
 */
export async function recordSuccess(
  db: Db.Db,
  id: string,
  options: recordSuccess.Options,
): Promise<Webhooks.Subscription | undefined> {
  const { now, staleBefore } = options
  const row = await db.kysely
    .updateTable('webhook_subscriptions')
    .set({ failureCount: 0, lastDeliveryAt: now, updatedAt: now })
    .where('id', '=', id)
    .where((eb) =>
      eb.or([
        eb('failureCount', '>', 0),
        eb('lastDeliveryAt', 'is', null),
        eb('lastDeliveryAt', '<', staleBefore),
      ]),
    )
    .returningAll()
    .executeTakeFirst()
  return row ? toSubscription(row) : undefined
}

export declare namespace recordSuccess {
  /** Clock bounds for one success record. */
  type Options = {
    /** Delivery timestamp (ISO). */
    now: string
    /** Rewrite the stamp only when it predates this (ISO). */
    staleBefore: string
  }
}

/**
 * Records a failed delivery: atomically increments the consecutive-failure
 * counter and disables the subscription once it reaches `maxFailures` (raw
 * fragments use storage column names — they bypass the camelCase mapping).
 * Skipped once disabled at the cap, when nothing would move.
 *
 * @param db - The database.
 * @param id - The subscription id (`wh_…`).
 * @param maxFailures - Consecutive failures before auto-disabling.
 * @param now - Failure timestamp (ISO).
 * @returns The updated subscription, or `undefined` when nothing was written.
 */
export async function recordFailure(
  db: Db.Db,
  id: string,
  maxFailures: number,
  now: string,
): Promise<Webhooks.Subscription | undefined> {
  const row = await db.kysely
    .updateTable('webhook_subscriptions')
    .set({
      failureCount: sql`failure_count + 1`,
      status: sql`CASE WHEN failure_count + 1 >= ${maxFailures} THEN 'disabled' ELSE status END`,
      updatedAt: now,
    })
    .where('id', '=', id)
    // In-flight failures race the disable and would queue on the row lock. A
    // re-enabled subscription keeps its capped counter; the status arm lets
    // its next failure re-disable it.
    .where((eb) => eb.or([eb('failureCount', '<', maxFailures), eb('status', '!=', 'disabled')]))
    .returningAll()
    .executeTakeFirst()
  return row ? toSubscription(row) : undefined
}

// Smaller than the ledger-table bounds: each subscription delete cascades
// into deliveries, queue events, and completions, fanning the statement out.
const pruneBatchLimit = 1_000

/**
 * Deletes expired subscriptions (their delivery rows cascade). Called
 * best-effort on a schedule.
 *
 * @param db - The database.
 * @param now - ISO timestamp expiry is evaluated against.
 * @returns The number of rows deleted.
 */
export async function pruneExpired(db: Db.Db, now: string): Promise<number> {
  // Bounded like the ledger prunes: this delete cascades into deliveries,
  // queue events, and completions, so it fans out the widest of the four.
  const result = await db.kysely
    .deleteFrom('webhook_subscriptions')
    // Repeated on the outer delete: the row recheck under concurrency re-runs
    // only this predicate, so a subscription renewed mid-statement survives.
    .where('expiresAt', '<=', now)
    .where('id', 'in', (eb) =>
      eb
        .selectFrom('webhook_subscriptions')
        .select('id')
        .where('expiresAt', 'is not', null)
        .where('expiresAt', '<=', now)
        .limit(pruneBatchLimit),
    )
    .executeTakeFirst()
  return Number(result.numDeletedRows)
}

/** Live-row predicate: no expiry, or expiry strictly in the future. */
function live(now: string) {
  return (eb: ExpressionBuilder<Db.Database, 'webhook_subscriptions'>) =>
    eb.or([eb('expiresAt', 'is', null), eb('expiresAt', '>', now)])
}