/**
 * End-to-end integration test against a real PDS and Jetstream.
 *
 * Flow:
 *   1. Spin up a local delivery server to receive webhook POSTs.
 *   2. Connect JetstreamClient watching the test DID.
 *   3. Create place.wisp.v2.wh on PDS → Jetstream delivers it → lands in local DB.
 *   4. Create app.bsky.feed.post → direct match fires → delivery #1.
 *   5. Create app.bsky.feed.like at the post → backlink fires → delivery #2.
 *   6. Delete all created records. Print timing.
 *
 * Env vars (set in .env or prefix the command):
 *   TEST_PDS_HANDLE    default: testacc.sharkgirl.pet
 *   TEST_PDS_PASSWORD  required
 *   TEST_PDS_URL       optional override for PDS base URL
 *   JETSTREAM_URL      optional override
 *   TEST_DELIVERY_PORT default: 19876
 */

import { createHmac, randomBytes } from 'node:crypto'
import { encryptWebhookSecret, parseWebhookSecretEncryptionKeyring } from '@wispplace/atproto-utils'
import type { Main as WhRecord } from '@wispplace/lexicons/types/place/wisp/v2/wh'
import {
	db,
	deleteWebhookRecord,
	findBacklinkWebhooks,
	findWebhooksForDid,
	getWebhookSecretToken,
	upsertWebhookRecord,
} from '../src/lib/db'
import { deliverWebhook, WebhookDeliveryWorker } from '../src/lib/delivery'
import { JetstreamClient } from '../src/lib/jetstream'
import { matchWebhooks } from '../src/lib/matcher'

// ---------------------------------------------------------------------------
// Local-only benchmark transport setup
// ---------------------------------------------------------------------------

// This benchmark intentionally receives deliveries over a local HTTP listener.
// The pinned URL layer requires both gates; do not copy these settings to a
// production process.
process.env.NODE_ENV = 'development'
process.env.WISP_ALLOW_LOCALHOST_FETCH = '1'
// Keep direct test rows encrypted too. An operator-supplied key wins; otherwise
// this one-process benchmark uses an ephemeral key that is never printed.
process.env.WEBHOOK_SECRET_ENCRYPTION_KEY ??= randomBytes(32).toString('base64url')

// ---------------------------------------------------------------------------
// Config
// ---------------------------------------------------------------------------

const HANDLE = process.env.TEST_PDS_HANDLE ?? 'testacc.sharkgirl.pet'
const PASSWORD = process.env.TEST_PDS_PASSWORD
const JETSTREAM_URL = process.env.JETSTREAM_URL ?? 'wss://jetstream2.us-east.bsky.network/subscribe'
const DELIVERY_PORT = parseInt(process.env.TEST_DELIVERY_PORT ?? '19876', 10)
const EVENT_TIMEOUT_MS = 30_000

if (!PASSWORD) {
	console.error('TEST_PDS_PASSWORD is required')
	process.exit(1)
}

// ---------------------------------------------------------------------------
// PDS helpers
// ---------------------------------------------------------------------------

async function derivePdsUrl(handle: string): Promise<string> {
	if (process.env.TEST_PDS_URL) return process.env.TEST_PDS_URL.replace(/\/$/, '')
	// For subdomain handles like testacc.sharkgirl.pet, try the parent domain
	const parts = handle.split('.')
	if (parts.length >= 2) {
		const candidate = `https://${parts.slice(-2).join('.')}`
		try {
			const res = await fetch(`${candidate}/xrpc/com.atproto.server.describeServer`, {
				signal: AbortSignal.timeout(5_000),
			})
			if (res.ok) return candidate
		} catch {}
	}
	throw new Error(`Could not derive PDS URL for ${handle}. Set TEST_PDS_URL explicitly.`)
}

interface Session {
	did: string
	accessJwt: string
}

async function createSession(pdsUrl: string, identifier: string, password: string): Promise<Session> {
	const res = await fetch(`${pdsUrl}/xrpc/com.atproto.server.createSession`, {
		method: 'POST',
		headers: { 'Content-Type': 'application/json' },
		body: JSON.stringify({ identifier, password }),
		signal: AbortSignal.timeout(10_000),
	})
	if (!res.ok) throw new Error(`createSession failed: ${res.status} ${await res.text()}`)
	return res.json() as Promise<Session>
}

async function createRecord(
	pdsUrl: string,
	jwt: string,
	repo: string,
	collection: string,
	record: Record<string, unknown>,
	rkey?: string,
): Promise<{ uri: string; cid: string }> {
	const body: Record<string, unknown> = { repo, collection, record }
	if (rkey) body.rkey = rkey
	const res = await fetch(`${pdsUrl}/xrpc/com.atproto.repo.createRecord`, {
		method: 'POST',
		headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${jwt}` },
		body: JSON.stringify(body),
		signal: AbortSignal.timeout(10_000),
	})
	if (!res.ok) throw new Error(`createRecord (${collection}) failed: ${res.status} ${await res.text()}`)
	return res.json() as Promise<{ uri: string; cid: string }>
}

async function deleteRecord(
	pdsUrl: string,
	jwt: string,
	repo: string,
	collection: string,
	rkey: string,
): Promise<void> {
	const res = await fetch(`${pdsUrl}/xrpc/com.atproto.repo.deleteRecord`, {
		method: 'POST',
		headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${jwt}` },
		body: JSON.stringify({ repo, collection, rkey }),
		signal: AbortSignal.timeout(10_000),
	})
	if (!res.ok) console.warn(`  deleteRecord ${collection}/${rkey}: ${res.status}`)
}

// ---------------------------------------------------------------------------
// Local delivery server
// ---------------------------------------------------------------------------

const deliveries: Array<{ ts: number; body: unknown; signature?: string; rawBody: string }> = []
let resolveDelivery: (() => void) | null = null
let expectedDeliveries = 0

const deliveryServer = Bun.serve({
	port: DELIVERY_PORT,
	routes: {
		'/': {
			POST: async (req) => {
				const rawBody = await req.text()
				const signature = req.headers.get('x-webhook-signature') ?? undefined
				const body = JSON.parse(rawBody)
				deliveries.push({ ts: Date.now(), body, signature, rawBody })
				console.log(
					`  [delivery #${deliveries.length}] sig=${signature ?? 'none'} ${JSON.stringify(body).slice(0, 80)}`,
				)
				if (deliveries.length >= expectedDeliveries) resolveDelivery?.()
				return new Response('ok')
			},
		},
	},
	fetch: () => new Response('Not Found', { status: 404 }),
})

function waitForDeliveries(n: number): Promise<void> {
	expectedDeliveries = n
	if (deliveries.length >= n) return Promise.resolve()
	return new Promise<void>((resolve, reject) => {
		resolveDelivery = resolve
		setTimeout(
			() => reject(new Error(`Timed out waiting for ${n} deliveries (got ${deliveries.length})`)),
			EVENT_TIMEOUT_MS,
		)
	})
}

// ---------------------------------------------------------------------------
// Jetstream event handler
// ---------------------------------------------------------------------------

const testDid: { value: string } = { value: '' }
let deliveryWorker: WebhookDeliveryWorker | undefined
let resolveWhRegistered: (() => void) | null = null
const whRegistered = new Promise<void>((resolve) => {
	resolveWhRegistered = resolve
})

async function handleEvent(event: {
	kind: string
	did: string
	commit?: { operation: string; collection: string; rkey: string; record?: unknown; cid?: string }
}) {
	if (event.kind !== 'commit' || !event.commit) return
	const { did } = event
	const { operation: op, collection, rkey, record, cid } = event.commit
	if (op !== 'create' && op !== 'update' && op !== 'delete') return

	if (collection === 'place.wisp.v2.wh') {
		if (op === 'delete') {
			await deleteWebhookRecord(did, rkey)
		} else if (record) {
			const wh = record as WhRecord
			if (wh.scope?.aturi && wh.url) {
				await upsertWebhookRecord(did, rkey, wh)
				console.log(`  [wh registered] ${did}/${rkey} scope=${wh.scope.aturi} backlinks=${wh.scope.backlinks ?? false}`)
				resolveWhRegistered?.()
			}
		}
		return
	}

	// Only process events from the test DID to avoid noise
	if (did !== testDid.value) return

	const directCandidates = await findWebhooksForDid(did)
	const backlinkCandidates = await findBacklinkWebhooks()
	const seen = new Set(directCandidates.map((e) => `${e.ownerDid}/${e.rkey}`))
	const candidates = [...directCandidates]
	for (const entry of backlinkCandidates) {
		const k = `${entry.ownerDid}/${entry.rkey}`
		if (!seen.has(k)) {
			seen.add(k)
			candidates.push(entry)
		}
	}

	if (candidates.length === 0) return

	const matched = matchWebhooks(candidates, did, collection, rkey, op as any, record ?? null)
	for (const entry of matched) {
		await deliverWebhook(entry, did, collection, rkey, op as any, cid, record)
	}
	// Legacy deliverWebhook enqueues. Drain it deterministically so this bench
	// proves local registration and actual delivery, not only outbox insertion.
	await deliveryWorker?.runOnce()
}

// ---------------------------------------------------------------------------
// DB secret helpers (mirrors main-app logic without importing it)
// ---------------------------------------------------------------------------

async function insertTestSecret(did: string, name: string): Promise<string> {
	const token = `wsk_${randomBytes(24).toString('base64url')}`
	const keyring = parseWebhookSecretEncryptionKeyring({
		WEBHOOK_SECRET_ENCRYPTION_KEY: process.env.WEBHOOK_SECRET_ENCRYPTION_KEY,
		WEBHOOK_SECRET_ENCRYPTION_PREVIOUS_KEYS: process.env.WEBHOOK_SECRET_ENCRYPTION_PREVIOUS_KEYS,
	})
	const envelope = encryptWebhookSecret(token, keyring)
	await db`
		INSERT INTO webhook_secrets (did, name, token, created_at)
		VALUES (${did}, ${name}, ${envelope}, NOW())
		ON CONFLICT (did, name) DO UPDATE SET token = EXCLUDED.token, last_rotated_at = NOW()
	`
	return token
}

async function deleteTestSecret(did: string, name: string): Promise<void> {
	await db`DELETE FROM webhook_secrets WHERE did = ${did} AND name = ${name}`
}

// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------

const createdRecords: Array<{ collection: string; rkey: string }> = []

async function run() {
	console.log(`\n=== wisp webhook e2e ===`)
	console.log(`PDS handle : ${HANDLE}`)
	console.log(`Jetstream  : ${JETSTREAM_URL}`)
	console.log(`Delivery   : http://localhost:${DELIVERY_PORT}/`)
	console.log()

	// The explicit loopback flag is still checked by pinnedWebhookFetch against
	// NODE_ENV=development and WISP_ALLOW_LOCALHOST_FETCH=1 above.
	deliveryWorker = new WebhookDeliveryWorker({ allowLoopback: true, requestTimeoutMs: 10_000 })

	// 1. Auth
	const pdsUrl = await derivePdsUrl(HANDLE)
	console.log(`PDS URL    : ${pdsUrl}`)
	const session = await createSession(pdsUrl, HANDLE, PASSWORD!)
	testDid.value = session.did
	console.log(`Test DID   : ${session.did}\n`)

	// 2. Jetstream
	const js = new JetstreamClient({
		url: JETSTREAM_URL,
		wantedDids: [session.did],
		onEvent: handleEvent as any,
		onConnect: () => console.log('[jetstream] connected'),
		onDisconnect: () => console.log('[jetstream] disconnected'),
		onError: (err) => console.error('[jetstream] error:', err.message),
	})
	js.start()
	// Give the WS a moment to connect before creating records
	await Bun.sleep(1_500)

	const deliveryUrl = `http://localhost:${DELIVERY_PORT}/`
	const scopeAturi = `at://${session.did}/app.bsky.feed.post`

	try {
		// 3. Create place.wisp.v2.wh record
		console.log('--- step 1: create place.wisp.v2.wh ---')
		const t0 = Date.now()
		const whRkey = `bench-wh-${Date.now()}`
		const { uri: whUri } = await createRecord(
			pdsUrl,
			session.accessJwt,
			session.did,
			'place.wisp.v2.wh',
			{
				$type: 'place.wisp.v2.wh',
				scope: { $type: 'place.wisp.v2.wh#atUri', aturi: scopeAturi, backlinks: true },
				url: deliveryUrl,
				events: ['create'],
				enabled: true,
				createdAt: new Date().toISOString(),
			},
			whRkey,
		)
		createdRecords.push({ collection: 'place.wisp.v2.wh', rkey: whRkey })
		console.log(`  created ${whUri}`)

		await Promise.race([
			whRegistered,
			new Promise((_, reject) =>
				setTimeout(() => reject(new Error('Timed out waiting for wh to be registered in DB')), EVENT_TIMEOUT_MS),
			),
		])
		console.log(`  registered in DB in ${Date.now() - t0}ms\n`)

		// 4. Create app.bsky.feed.post → direct match
		console.log('--- step 2: create post (expect delivery #1 — direct match) ---')
		const t1 = Date.now()
		const { uri: postUri, cid: postCid } = await createRecord(
			pdsUrl,
			session.accessJwt,
			session.did,
			'app.bsky.feed.post',
			{
				$type: 'app.bsky.feed.post',
				text: 'wisp webhook e2e test post',
				createdAt: new Date().toISOString(),
			},
		)
		const postRkey = postUri.split('/').pop()!
		createdRecords.push({ collection: 'app.bsky.feed.post', rkey: postRkey })
		console.log(`  created ${postUri}`)

		await waitForDeliveries(1)
		console.log(`  delivery #1 received in ${Date.now() - t1}ms\n`)

		// 5. Create app.bsky.feed.like referencing the post → backlink match
		console.log('--- step 3: like post (expect delivery #2 — backlink match) ---')
		const t2 = Date.now()
		const { uri: likeUri } = await createRecord(pdsUrl, session.accessJwt, session.did, 'app.bsky.feed.like', {
			$type: 'app.bsky.feed.like',
			subject: { uri: postUri, cid: postCid },
			createdAt: new Date().toISOString(),
		})
		const likeRkey = likeUri.split('/').pop()!
		createdRecords.push({ collection: 'app.bsky.feed.like', rkey: likeRkey })
		console.log(`  created ${likeUri}`)

		await waitForDeliveries(2)
		console.log(`  delivery #2 received in ${Date.now() - t2}ms\n`)

		// 6. SecretId signing test
		console.log('--- step 4: secretId signing ---')
		const secretName = `e2e-test-${Date.now()}`
		const secretToken = await insertTestSecret(session.did, secretName)
		console.log(`  inserted secret "${secretName}"`)

		// Verify getWebhookSecretToken can look it up
		const lookedUp = await getWebhookSecretToken(session.did, secretName)
		if (lookedUp !== secretToken) throw new Error('getWebhookSecretToken mismatch')
		console.log(`  getWebhookSecretToken ✓`)

		// Create a webhook with secretId, deliver manually, check signature
		const signedWhRkey = `bench-wh-signed-${Date.now()}`
		const signedWh: WhRecord = {
			$type: 'place.wisp.v2.wh',
			scope: { $type: 'place.wisp.v2.wh#atUri', aturi: `at://${session.did}/app.bsky.feed.post` },
			url: deliveryUrl,
			secretId: secretName,
			enabled: true,
			createdAt: new Date().toISOString(),
		}
		await upsertWebhookRecord(session.did, signedWhRkey, signedWh)

		// Deliver a fake event
		const beforeCount = deliveries.length
		await deliverWebhook(
			{ ownerDid: session.did, rkey: signedWhRkey, record: signedWh },
			session.did,
			'app.bsky.feed.post',
			'test-rkey',
			'create',
			undefined,
			{ text: 'signed test' },
		)
		const delivered = await deliveryWorker?.runOnce()
		if (!delivered) throw new Error('Signed webhook was not claimed for delivery')

		// Wait for it
		await new Promise<void>((resolve, reject) => {
			const check = () => {
				if (deliveries.length > beforeCount) resolve()
			}
			check()
			const iv = setInterval(check, 50)
			setTimeout(() => {
				clearInterval(iv)
				reject(new Error('Timed out waiting for signed delivery'))
			}, EVENT_TIMEOUT_MS)
		})

		const signedDelivery = deliveries[deliveries.length - 1]!
		if (!signedDelivery.signature) throw new Error('Expected X-Webhook-Signature header but got none')

		const expected = `sha256=${createHmac('sha256', secretToken).update(signedDelivery.rawBody).digest('hex')}`
		if (signedDelivery.signature !== expected) {
			throw new Error(`Signature mismatch:\n  got:      ${signedDelivery.signature}\n  expected: ${expected}`)
		}
		console.log(`  X-Webhook-Signature ✓  ${signedDelivery.signature.slice(0, 32)}...`)

		// Cleanup secret + wh record
		await deleteWebhookRecord(session.did, signedWhRkey)
		await deleteTestSecret(session.did, secretName)
		console.log(`  cleaned up secret + webhook record\n`)

		console.log(`=== passed ✓  total time ${Date.now() - t0}ms ===`)
	} finally {
		// Cleanup
		console.log('\n--- cleanup ---')
		for (const { collection, rkey } of createdRecords.reverse()) {
			await deleteRecord(pdsUrl, session.accessJwt, session.did, collection, rkey)
			console.log(`  deleted ${collection}/${rkey}`)
		}
		js.destroy()
		await deliveryWorker?.stop()
		deliveryServer.stop(true)
		await db.close()
	}
}

run().catch((err) => {
	console.error('\n=== failed ✗ ===\n', err.message)
	process.exit(1)
})