/**
 * Cache invalidation publisher
 *
 * Publishes invalidation messages to Redis pub/sub so the hosting-service
 * can clear its local caches (tiered storage, redirect rules) when a site
 * is updated or deleted via the firehose.
 */

import {
	DEFAULT_CACHE_INVALIDATION_CHANNEL,
	publishCacheInvalidationEvent,
	revalidationQuarantineKey,
	revalidationSiteVersionKey,
} from '@wispplace/constants'
import { createLogger } from '@wispplace/observability'
import Redis from 'ioredis'
import { config } from '../config'
import { isSiteDeleteTombstoneReason } from './revalidate-queue'

const logger = createLogger('firehose-service')

let publisher: Redis | null = null
let publisherReadyTimeoutMs = 5_000
let loggedMissingRedis = false

export interface CacheInvalidationPublisherReadiness {
	readonly status: string
	once(event: string, listener: (...args: unknown[]) => void): unknown
	removeListener(event: string, listener: (...args: unknown[]) => void): unknown
}

/** Wait for the eager ioredis connection before issuing commands with offline queuing disabled. */
export async function waitForCacheInvalidationPublisherReady(
	redis: CacheInvalidationPublisherReadiness,
	timeoutMs: number,
): Promise<void> {
	if (redis.status === 'ready') return

	await new Promise<void>((resolve, reject) => {
		let settled = false
		const finish = (error?: unknown) => {
			if (settled) return
			settled = true
			clearTimeout(timer)
			redis.removeListener('ready', onReady)
			redis.removeListener('error', onError)
			if (error === undefined) resolve()
			else reject(error)
		}
		const onReady = () => finish()
		const onError = (error: unknown) => finish(error)
		const timer = setTimeout(() => finish(new Error(`Redis publisher was not ready within ${timeoutMs}ms`)), timeoutMs)
		redis.once('ready', onReady)
		redis.once('error', onError)
	})
}

function getPublisher(): Redis | null {
	if (!config.redisUrl) {
		if (!loggedMissingRedis) {
			logger.warn('[CacheInvalidation] REDIS_URL not set; cache invalidation publishing disabled')
			loggedMissingRedis = true
		}
		return null
	}

	if (!publisher) {
		logger.info('[CacheInvalidation] Connecting to Redis for publishing')
		const commandTimeout = Math.max(100, Math.min(5_000, Math.floor(config.leaderTtlMs / 4)))
		publisherReadyTimeoutMs = commandTimeout
		publisher = new Redis(config.redisUrl, {
			maxRetriesPerRequest: 0,
			enableReadyCheck: true,
			enableOfflineQueue: false,
			connectTimeout: commandTimeout,
			commandTimeout,
		})

		publisher.on('error', (err) => {
			logger.error('[CacheInvalidation] Redis error', err)
		})

		publisher.on('ready', () => {
			logger.info('[CacheInvalidation] Redis publisher connected')
		})
	}

	return publisher
}

export async function publishCacheInvalidation(
	did: string,
	rkey: string,
	action: 'updating' | 'update' | 'delete' | 'settings',
	token?: string,
): Promise<void> {
	const redis = getPublisher()
	if (!redis) return

	try {
		await waitForCacheInvalidationPublisherReady(redis, publisherReadyTimeoutMs)
		const streamId = await publishCacheInvalidationEvent(
			redis,
			{ did, rkey, action, token },
			DEFAULT_CACHE_INVALIDATION_CHANNEL,
			config.cacheInvalidationStream,
			config.cacheInvalidationStreamMaxLen,
		)
		logger.debug(
			`[CacheInvalidation] Publishing ${action} for ${did}/${rkey} to ${DEFAULT_CACHE_INVALIDATION_CHANNEL} (stream ${streamId})`,
		)
	} catch (err) {
		logger.error('[CacheInvalidation] Failed to publish', err)
	}
}

type RevalidationDedupeCategory = 'delete-tombstone' | 'settings' | 'rewrite-miss' | 'storage-miss' | 'full-repair'

/**
 * Whether direct firehose work is durably covered by the revalidation stream.
 * Only `enqueued` and `deduplicated` permit a caller to advance its own cursor.
 * `capacity` is explicit backpressure and must be retried without dropping work.
 */
export type SiteRevalidationEnqueueResult = 'enqueued' | 'deduplicated' | 'quarantined' | 'capacity' | 'unavailable'

export interface RevalidationQueueClient {
	/** Runs the revalidation enqueue script atomically. */
	eval(script: string, keyCount: number, ...args: string[]): PromiseLike<unknown>
}

// This script deliberately has no MAXLEN. A producer must never remove a
// pending consumer-group entry. Capacity is a hard backpressure boundary; ACK
// plus XDEL is the only normal removal path.
const ENQUEUE_REVALIDATION_SCRIPT = `
if ARGV[7] == '1' then
  local quarantine = redis.call('GET', KEYS[3])
  if quarantine and (ARGV[8] == '' or ARGV[8] <= quarantine) then return {-2, quarantine} end
end

local existing = redis.call('GET', KEYS[1])
if existing then
  local found = redis.call('XRANGE', KEYS[2], existing, existing, 'COUNT', 1)
  if #found == 1 and found[1][1] == existing then return {0, existing} end
  redis.call('DEL', KEYS[1])
end
if redis.call('XLEN', KEYS[2]) >= tonumber(ARGV[1]) then return {-1, ''} end
local id = redis.call('XADD', KEYS[2], '*', 'did', ARGV[2], 'rkey', ARGV[3], 'reason', ARGV[4], 'ts', ARGV[5], 'sourceVersion', ARGV[8])
redis.call('SET', KEYS[1], id, 'EX', ARGV[6])
return {1, id}
`

function getRevalidationDedupeCategory(reason: string): RevalidationDedupeCategory {
	if (isSiteDeleteTombstoneReason(reason)) return 'delete-tombstone'
	if (reason.startsWith('firehose-settings-failed:')) return 'settings'
	if (reason.startsWith('rewrite-miss')) return 'rewrite-miss'
	if (reason.startsWith('storage-miss')) return 'storage-miss'
	return 'full-repair'
}

function revalidationErrorKind(error: unknown): string {
	if (error instanceof Error && error.name) return error.name
	return 'UnknownError'
}

function parseEnqueueResult(value: unknown): { status: number; streamId: string } | null {
	if (!Array.isArray(value) || typeof value[0] !== 'number' || typeof value[1] !== 'string') return null
	return { status: value[0], streamId: value[1] }
}

export async function enqueueSiteRevalidationWithRedis(
	redis: RevalidationQueueClient,
	did: string,
	rkey: string,
	reason: string,
	sourceVersion = '',
): Promise<SiteRevalidationEnqueueResult> {
	const dedupeCategory = getRevalidationDedupeCategory(reason)
	const dedupeKey = `revalidate:site:${dedupeCategory}:${did}:${rkey}`
	try {
		const outcome = parseEnqueueResult(
			await redis.eval(
				ENQUEUE_REVALIDATION_SCRIPT,
				3,
				dedupeKey,
				config.revalidateStream,
				revalidationQuarantineKey(did, rkey),
				config.revalidateStreamMaxLen.toString(),
				did,
				rkey,
				reason,
				Date.now().toString(),
				config.revalidateDedupeTtlSeconds.toString(),
				dedupeCategory === 'settings' ? '0' : '1',
				sourceVersion,
			),
		)
		if (!outcome) {
			logger.error('[Revalidate] Redis enqueue script returned an invalid result', undefined, { did, rkey, reason })
			return 'unavailable'
		}
		if (outcome.status === 0 && outcome.streamId) return 'deduplicated'
		if (outcome.status === -2 && outcome.streamId) return 'quarantined'
		if (outcome.status === 1 && outcome.streamId) {
			logger.info(`[Revalidate] Enqueued ${did}/${rkey} after firehose failure`, { reason, streamId: outcome.streamId })
			return 'enqueued'
		}
		if (outcome.status === -1) {
			logger.warn(`[Revalidate] Queue capacity reached for ${did}/${rkey}`)
			return 'capacity'
		}
		return 'unavailable'
	} catch (err) {
		logger.error('[Revalidate] Failed to enqueue site after firehose failure', undefined, {
			did,
			rkey,
			reason,
			errorKind: revalidationErrorKind(err),
		})
		return 'unavailable'
	}
}

export async function enqueueSiteRevalidation(
	did: string,
	rkey: string,
	reason: string,
	sourceVersion = '',
): Promise<SiteRevalidationEnqueueResult> {
	const redis = getPublisher()
	if (!redis) return 'unavailable'
	return enqueueSiteRevalidationWithRedis(redis, did, rkey, reason, sourceVersion)
}

/**
 * Record a successfully reconciled repo revision and clear only an older DLQ
 * fence. ATProto repo revisions are monotonic TIDs, so relay replays of the
 * same or an older event cannot reopen hosting repair work.
 */
export async function recordSiteReconciliation(did: string, rkey: string, revision: string): Promise<boolean> {
	if (!config.redisUrl) return true
	const redis = getPublisher()
	if (!redis || !/^[a-z2-7]{13}$/.test(revision)) return false
	try {
		await waitForCacheInvalidationPublisherReady(redis, publisherReadyTimeoutMs)
		const result = await redis.eval(
			`local current = redis.call('GET', KEYS[1])
if current and current >= ARGV[1] then return {0, current, 0} end
redis.call('SET', KEYS[1], ARGV[1])
local quarantine = redis.call('GET', KEYS[2])
if quarantine and ARGV[1] > quarantine then
  redis.call('DEL', KEYS[2])
  return {1, ARGV[1], 1}
end
return {1, ARGV[1], 0}`,
			2,
			revalidationSiteVersionKey(did, rkey),
			revalidationQuarantineKey(did, rkey),
			revision,
		)
		return Array.isArray(result) && (result[0] === 0 || result[0] === 1)
	} catch (error) {
		logger.error('[Revalidate] Failed to record successful site reconciliation', undefined, {
			did,
			rkey,
			errorKind: revalidationErrorKind(error),
		})
		return false
	}
}

export async function closeCacheInvalidationPublisher(): Promise<void> {
	if (publisher) {
		const toClose = publisher
		publisher = null
		await toClose.quit()
	}
}