import {
	encryptWebhookSecret,
	WEBHOOK_SECRET_ENVELOPE_PREFIX,
	type WebhookSecretEncryptionKeyring,
} from '@wispplace/atproto-utils'
import type { SQL } from 'bun'

const WEBHOOK_SECRET_MIGRATION_LOCK_KEY = 0x5749535057454e43n // "WISPWENC"
const DEFAULT_BATCH_SIZE = 100
const DEFAULT_MAX_BATCHES = 1000

export const WEBHOOK_SECRET_ENCRYPTION_MIGRATION_ERROR = 'webhook_secret_encryption_migration_unavailable'

/** Deliberately generic: database values and driver details must never reach logs or API errors. */
export class WebhookSecretEncryptionMigrationError extends Error {
	constructor() {
		super(WEBHOOK_SECRET_ENCRYPTION_MIGRATION_ERROR)
		this.name = 'WebhookSecretEncryptionMigrationError'
	}
}

export interface LegacyWebhookSecretRow {
	readonly did: string
	readonly name: string
	readonly token: string
}

export interface WebhookSecretEnvelopeMigrationResult {
	readonly encrypted: number
	readonly legacyRemaining: number
	readonly encryptionAvailable: boolean
}

export interface WebhookSecretEnvelopeMigrationOptions {
	/** Bounded so one migration transaction cannot lock an unbounded table. */
	readonly batchSize?: number
	/** Bounded work per startup; a later startup safely resumes remaining rows. */
	readonly maxBatches?: number
	/** Test/operational hook. It receives counts only, never a secret. */
	readonly afterBatchCommitted?: (progress: { encrypted: number; batches: number }) => Promise<void> | void
}

/**
 * Transactional storage abstraction used to make crash/resume behavior testable.
 * The production adapter below binds every call to one advisory-lock-owning
 * reserved PostgreSQL connection.
 */
export interface WebhookSecretEnvelopeMigrationStore {
	transaction<T>(operation: () => Promise<T>): Promise<T>
	selectLegacyBatch(limit: number): Promise<ReadonlyArray<LegacyWebhookSecretRow>>
	compareAndSwap(row: LegacyWebhookSecretRow, envelope: string): Promise<boolean>
	countLegacy(): Promise<number>
}

function migrationFailure(): never {
	throw new WebhookSecretEncryptionMigrationError()
}

const boundedPositiveInteger = (value: number | undefined, fallback: number, maximum: number): number => {
	const result = value ?? fallback
	if (!Number.isSafeInteger(result) || result < 1 || result > maximum) migrationFailure()
	return result
}

/**
 * Convert legacy plaintext rows using a transactional store. Each committed
 * batch has a compare-and-swap update, so a crash after a commit is safe to
 * resume: committed rows are skipped and uncommitted rows stay plaintext.
 */
export const migrateLegacyWebhookSecretRows = async (
	store: WebhookSecretEnvelopeMigrationStore,
	keyring: WebhookSecretEncryptionKeyring | null,
	options: WebhookSecretEnvelopeMigrationOptions = {},
): Promise<WebhookSecretEnvelopeMigrationResult> => {
	const batchSize = boundedPositiveInteger(options.batchSize, DEFAULT_BATCH_SIZE, DEFAULT_BATCH_SIZE)
	const maxBatches = boundedPositiveInteger(options.maxBatches, DEFAULT_MAX_BATCHES, DEFAULT_MAX_BATCHES)

	if (!keyring) {
		const legacyRemaining = await store.transaction(async () => await store.countLegacy())
		if (!Number.isSafeInteger(legacyRemaining) || legacyRemaining < 0) migrationFailure()
		return { encrypted: 0, legacyRemaining, encryptionAvailable: false }
	}

	let encrypted = 0
	for (let batches = 0; batches < maxBatches; batches++) {
		const batch = await store.transaction(async () => {
			const rows = await store.selectLegacyBatch(batchSize)
			if (rows.length > batchSize) migrationFailure()

			let updated = 0
			for (const row of rows) {
				if (typeof row.did !== 'string' || typeof row.name !== 'string' || typeof row.token !== 'string') {
					migrationFailure()
				}
				const envelope = encryptWebhookSecret(row.token, keyring)
				if (await store.compareAndSwap(row, envelope)) updated++
			}
			return { selected: rows.length, updated }
		})

		encrypted += batch.updated
		if (batch.selected === 0 || batch.updated === 0) break
		await options.afterBatchCommitted?.({ encrypted, batches: batches + 1 })
	}

	const legacyRemaining = await store.transaction(async () => await store.countLegacy())
	if (!Number.isSafeInteger(legacyRemaining) || legacyRemaining < 0) migrationFailure()
	return { encrypted, legacyRemaining, encryptionAvailable: true }
}

const countFromRow = (value: unknown): number => {
	const numberValue = typeof value === 'number' ? value : typeof value === 'string' ? Number(value) : Number.NaN
	if (!Number.isSafeInteger(numberValue) || numberValue < 0) migrationFailure()
	return numberValue
}

/**
 * Encrypt all legacy plaintext webhook tokens under a distinct, session-scoped
 * PostgreSQL advisory lock. This runs only after the normal schema migration
 * created `webhook_secrets`; it intentionally does not share or race that
 * migration lock.
 */
export const migrateWebhookSecretEnvelopes = async (
	database: SQL,
	keyring: WebhookSecretEncryptionKeyring | null,
	options: WebhookSecretEnvelopeMigrationOptions = {},
): Promise<WebhookSecretEnvelopeMigrationResult> => {
	let reserved: Awaited<ReturnType<SQL['reserve']>> | undefined
	let lockHeld = false
	let connectionMustClose = false
	let failed = false
	let result: WebhookSecretEnvelopeMigrationResult | undefined

	try {
		reserved = await database.reserve()
		await reserved`SET lock_timeout = '30s'`
		await reserved`SELECT pg_advisory_lock(${WEBHOOK_SECRET_MIGRATION_LOCK_KEY}::bigint)`
		lockHeld = true

		const legacyPrefix = `${WEBHOOK_SECRET_ENVELOPE_PREFIX}%`
		const store: WebhookSecretEnvelopeMigrationStore = {
			transaction: async <T>(operation: () => Promise<T>): Promise<T> => {
				// Use the advisory-lock-owning reserved connection itself. Calling
				// SQL.begin() on a generic client could reserve a different session.
				await reserved!`BEGIN`
				try {
					const value = await operation()
					await reserved!`COMMIT`
					return value
				} catch (error) {
					try {
						await reserved!`ROLLBACK`
					} catch {
						// The caller receives the same generic migration failure below.
					}
					throw error
				}
			},
			selectLegacyBatch: async (limit) => {
				const rows = await reserved!<Array<LegacyWebhookSecretRow>>`
					SELECT did, name, token
					FROM webhook_secrets
					WHERE token NOT LIKE ${legacyPrefix}
					ORDER BY did ASC, name ASC
					LIMIT ${limit}
					FOR UPDATE SKIP LOCKED
				`
				return rows
			},
			compareAndSwap: async (row, envelope) => {
				const rows = await reserved!<Array<{ did: string }>>`
					UPDATE webhook_secrets
					SET token = ${envelope}
					WHERE did = ${row.did}
						AND name = ${row.name}
						AND token = ${row.token}
					RETURNING did
				`
				return rows.length === 1
			},
			countLegacy: async () => {
				const rows = await reserved!<Array<{ count: number | string }>>`
					SELECT COUNT(*) AS count
					FROM webhook_secrets
					WHERE token NOT LIKE ${legacyPrefix}
				`
				return countFromRow(rows[0]?.count)
			},
		}

		result = await migrateLegacyWebhookSecretRows(store, keyring, options)
	} catch {
		failed = true
	} finally {
		if (reserved) {
			if (lockHeld) {
				try {
					const rows = await reserved<Array<{ unlocked: boolean }>>`
						SELECT pg_advisory_unlock(${WEBHOOK_SECRET_MIGRATION_LOCK_KEY}::bigint) AS unlocked
					`
					if (!rows[0]?.unlocked) {
						failed = true
						connectionMustClose = true
					}
				} catch {
					failed = true
					connectionMustClose = true
				}
			}

			if (!connectionMustClose) {
				try {
					await reserved`SET lock_timeout = DEFAULT`
				} catch {
					failed = true
					connectionMustClose = true
				}
			}

			if (connectionMustClose) {
				try {
					await reserved.close({ timeout: 0 })
				} catch {
					failed = true
				}
			} else {
				try {
					reserved.release()
				} catch {
					failed = true
				}
			}
		}
	}

	if (failed || !result) return migrationFailure()
	return result
}