import { describe, expect, mock, test } from 'bun:test'
import type {
	ClaimedWebhookDelivery,
	NewWebhookDeliveryEventRow,
	NewWebhookDeliveryOutboxRow,
	WebhookEntry,
} from './db'
import type { WebhookDeliveryWorkerPersistence } from './delivery'
import type { WebhookTransport } from './webhook-url'

interface CapturedRow {
	deliveryId: string
	eventId: string
	payloadJson?: string
	subscriptionSnapshotJson?: string
}

interface CapturedEvent {
	eventId: string
	payloadJson: string
}

let captured: Array<{ event: CapturedEvent; rows: readonly CapturedRow[]; ensureEvent: boolean }> = []
let failCall = 0
let calls = 0
const persistedDeliveryIds = new Set<string>()
const persistedEventIds = new Set<string>()

mock.module('@wispplace/observability', () => ({
	createLogger: () => ({
		debug: () => undefined,
		error: () => undefined,
		info: () => undefined,
		warn: () => undefined,
	}),
}))

mock.module('./db', () => ({}))

mock.module('./redis', () => ({ publishWebhookEvent: async () => ({ published: false, dropped: false }) }))

const captureEnqueue = async (
	event: NewWebhookDeliveryEventRow,
	rows: readonly NewWebhookDeliveryOutboxRow[],
	ensureEvent = true,
) => {
	calls++
	captured.push({ event, rows, ensureEvent })
	if (calls === failCall) throw new Error('temporary database failure')
	if (ensureEvent) persistedEventIds.add(event.eventId)
	let enqueued = 0
	for (const row of rows) {
		if (!persistedDeliveryIds.has(row.deliveryId)) {
			persistedDeliveryIds.add(row.deliveryId)
			enqueued++
		}
	}
	return { enqueued, deduplicated: rows.length - enqueued }
}

const enqueueOptions = {
	enqueueOutbox: captureEnqueue,
	subscriptionFingerprint: () => 'fingerprint',
}

const { enqueueWebhookDeliveries, WebhookDeliveryWorker } = await import('./delivery')

const reset = () => {
	captured = []
	failCall = 0
	calls = 0
	persistedDeliveryIds.clear()
	persistedEventIds.clear()
}

const entries = (count: number): WebhookEntry[] =>
	Array.from({ length: count }, (_, index) => ({
		ownerDid: `did:plc:owner${index}`,
		rkey: `hook-${index}`,
		record: {
			$type: 'place.wisp.v2.wh' as const,
			scope: { aturi: 'at://did:plc:subject' },
			url: 'https://receiver.example/hook',
			createdAt: '2025-01-01T00:00:00.000Z',
		},
	}))

const event = (record: unknown = { text: 'hello' }) => ({
	relay: 'wss://jetstream.example/subscribe',
	timeUs: 1_700_000_000_000_000,
	rev: '3lq6x5f2abcde',
	operation: 'create' as const,
	did: 'did:plc:subject',
	collection: 'app.bsky.feed.post',
	rkey: 'post',
	record,
})

describe('durable normalized webhook fanout', () => {
	test('uses bounded 1000-row chunks with one shared immutable event body', async () => {
		reset()
		const result = await enqueueWebhookDeliveries(entries(1_001), event(), enqueueOptions)

		expect(result).toEqual({ enqueued: 1_001, deduplicated: 0 })
		expect(captured.map((chunk) => chunk.rows.length)).toEqual([1_000, 1])
		expect(captured.map((chunk) => chunk.ensureEvent)).toEqual([true, false])
		expect(new Set(captured.map((chunk) => chunk.event.eventId)).size).toBe(1)
		expect(new Set(captured.map((chunk) => chunk.event.payloadJson)).size).toBe(1)
		for (const chunk of captured) {
			for (const row of chunk.rows) {
				expect(row.eventId).toBe(chunk.event.eventId)
				expect('payloadJson' in row).toBe(false)
				expect('subscriptionSnapshotJson' in row).toBe(false)
			}
		}
	})

	test('does not multiply a near-limit body across 10,000 outbox rows', async () => {
		reset()
		await enqueueWebhookDeliveries(entries(10_000), event({ text: 'x'.repeat(500 * 1024) }), enqueueOptions)

		expect(captured).toHaveLength(10)
		expect(captured.every((chunk) => chunk.rows.length <= 1_000)).toBe(true)
		expect(new Set(captured.map((chunk) => chunk.event.eventId)).size).toBe(1)
		expect(new Set(captured.map((chunk) => chunk.event.payloadJson)).size).toBe(1)
		expect(
			captured
				.flatMap((chunk) => chunk.rows)
				.every((row) => !('payloadJson' in row) && !('subscriptionSnapshotJson' in row)),
		).toBe(true)
	})

	test('replays a partially committed large fanout with stable IDs', async () => {
		reset()
		failCall = 2
		await expect(enqueueWebhookDeliveries(entries(1_001), event(), enqueueOptions)).rejects.toThrow(
			'temporary database failure',
		)
		expect(persistedDeliveryIds).toHaveLength(1_000)

		failCall = 0
		const replay = await enqueueWebhookDeliveries(entries(1_001), event(), enqueueOptions)
		expect(replay).toEqual({ enqueued: 1, deduplicated: 1_000 })
		expect(persistedDeliveryIds).toHaveLength(1_001)
		expect(persistedEventIds).toHaveLength(1)
	})

	function deferred<T>() {
		let resolve: (value: T | PromiseLike<T>) => void = () => undefined
		let reject: (reason?: unknown) => void = () => undefined
		const promise = new Promise<T>((resolvePromise, rejectPromise) => {
			resolve = resolvePromise
			reject = rejectPromise
		})
		return { promise, resolve, reject }
	}

	const claimedRow = (): ClaimedWebhookDelivery => ({
		deliveryId: 'whd_v1_delivery',
		ownerDid: 'did:plc:owner',
		webhookRkey: 'hook',
		targetUrl: 'https://receiver.example/hook',
		signingMode: 'none',
		subscriptionFingerprint: 'fingerprint',
		payloadBody:
			'{"id":"event","event":"create","did":"did:plc:subject","collection":"app.bsky.feed.post","rkey":"post","timestamp":"2025-01-01T00:00:00.000Z"}',
		attemptCount: 0,
		leaseToken: 'lease',
		sourceOperation: 'create',
		sourceTimeUs: 1,
	})

	function workerPersistence(
		claimWebhookDeliveryOutbox: WebhookDeliveryWorkerPersistence['claimWebhookDeliveryOutbox'],
		overrides: Partial<WebhookDeliveryWorkerPersistence> = {},
	): WebhookDeliveryWorkerPersistence {
		return {
			cancelWebhookDeliveryForSubscriptionChange: async () => true,
			claimWebhookDeliveryOutbox,
			getCurrentWebhookSubscription: async () => ({
				url: 'https://receiver.example/hook',
				fingerprint: 'fingerprint',
				signingMode: 'none',
			}),
			getWebhookInlineSecret: async () => null,
			getWebhookSecretToken: async () => null,
			markWebhookDeliverySucceeded: async () => true,
			renewWebhookDeliveryLease: async () => true,
			rescheduleWebhookDelivery: async () => true,
			runWebhookMaintenance: async () => ({
				ran: true,
				outbox: 0,
				eventLogs: 0,
				tombstones: 0,
				backlinkReferences: 0,
				intakeQuarantines: 0,
				deliveryEvents: 0,
			}),
			...overrides,
		}
	}

	test('aborts a hung pinned attempt before shutdown returns and leaves its lease for reclaim', async () => {
		const transportStarted = deferred<void>()
		let aborted = false
		let posts = 0
		let updates = 0
		const hungTransport: WebhookTransport = ({ signal }) => {
			posts++
			transportStarted.resolve()
			return new Promise<Response>((_resolve, reject) => {
				signal.addEventListener(
					'abort',
					() => {
						aborted = true
						reject(signal.reason)
					},
					{ once: true },
				)
			})
		}
		const persistence = workerPersistence(async () => [claimedRow()], {
			markWebhookDeliverySucceeded: async () => {
				updates++
				return true
			},
			rescheduleWebhookDelivery: async () => {
				updates++
				return true
			},
		})
		const worker = new WebhookDeliveryWorker({
			concurrency: 1,
			batchSize: 1,
			requestTimeoutMs: 100,
			resolver: async () => [{ address: '8.8.8.8', family: 4 }],
			transport: hungTransport,
			persistence,
		})
		const pass = worker.runOnce()
		await transportStarted.promise
		expect(await worker.stop(100)).toBe(true)
		await pass
		expect(aborted).toBe(true)
		expect(posts).toBe(1)
		expect(updates).toBe(0)
		expect(await worker.runOnce()).toBe(0)

		// A separate worker can reclaim the unchanged durable row and complete it.
		let reclaimed = 0
		const reclaimer = new WebhookDeliveryWorker({
			concurrency: 1,
			batchSize: 1,
			requestTimeoutMs: 100,
			resolver: async () => [{ address: '8.8.8.8', family: 4 }],
			transport: async () => new Response(null, { status: 204 }),
			persistence: workerPersistence(async () => [claimedRow()], {
				markWebhookDeliverySucceeded: async () => {
					reclaimed++
					return true
				},
			}),
		})
		await reclaimer.runOnce()
		expect(reclaimed).toBe(1)
	})

	test('aborts at deadline, clears heartbeat timers, and waits for an in-flight refresh before DB teardown', async () => {
		const transportStarted = deferred<void>()
		const refreshStarted = deferred<void>()
		const finishRefresh = deferred<boolean>()
		let aborted = false
		let renewals = 0
		let updates = 0
		const persistence = workerPersistence(async () => [claimedRow()], {
			renewWebhookDeliveryLease: async () => {
				renewals++
				if (renewals === 1) return true
				refreshStarted.resolve()
				return finishRefresh.promise
			},
			markWebhookDeliverySucceeded: async () => {
				updates++
				return true
			},
			rescheduleWebhookDelivery: async () => {
				updates++
				return true
			},
		})
		const worker = new WebhookDeliveryWorker({
			concurrency: 1,
			batchSize: 1,
			requestTimeoutMs: 3_000,
			leaseMs: 8_000,
			resolver: async () => [{ address: '8.8.8.8', family: 4 }],
			transport: ({ signal }) => {
				transportStarted.resolve()
				return new Promise<Response>((_resolve, reject) => {
					signal.addEventListener(
						'abort',
						() => {
							aborted = true
							reject(signal.reason)
						},
						{ once: true },
					)
				})
			},
			persistence,
		})
		const pass = worker.runOnce()
		await transportStarted.promise
		// leaseMs is bounded to 8s, so its heartbeat starts before the 3s request deadline.
		await refreshStarted.promise
		expect(await worker.stop(100)).toBe(false)
		expect(aborted).toBe(true)
		expect(updates).toBe(0)
		finishRefresh.resolve(true)
		await pass
		await new Promise((resolve) => setTimeout(resolve, 25))
		expect(renewals).toBe(2)
		expect(updates).toBe(0)
	}, 8_000)
})