import { createLogger } from '@wispplace/observability'
import { RedisClient } from 'bun'
import { config } from '../config'
const logger = createLogger('webhook-service:redis')
export interface WebhookEvent {
ownerDid: string
rkey: string
url: string
eventKind: 'create' | 'update' | 'delete'
eventDid: string
eventCollection: string
eventRkey: string
cid?: string
deliveredAt: string
status: 'ok' | 'failed'
}
export interface RedisPublishResult {
readonly published: boolean
readonly dropped: boolean
}
let publisher: RedisClient | null = null
let connected = false
let closing = false
let reconnectTimer: ReturnType<typeof setTimeout> | null = null
let reconnectAttempts = 0
let loggedMissingRedis = false
let dropped = 0
let inFlight = 0
let lastFailureLogAt = 0
const activePublishes = new Set<Promise<unknown>>()
type RedisClientFactory = (url: string) => RedisClient
const defaultRedisClientFactory: RedisClientFactory = (url) => new RedisClient(url)
let redisClientFactory: RedisClientFactory = defaultRedisClientFactory
/** Test seam for exercising connection/readiness without a live Redis server. */
export function setRedisClientFactoryForTests(factory?: RedisClientFactory): void {
if (publisher || reconnectTimer) throw new Error('Cannot replace an active Redis client')
redisClientFactory = factory ?? defaultRedisClientFactory
}
function incrementDropped(): void {
dropped = Math.min(Number.MAX_SAFE_INTEGER, dropped + 1)
}
function safeLog(message: string): void {
const now = Date.now()
if (now - lastFailureLogAt < 60_000) return
lastFailureLogAt = now
logger.warn(message)
}
function validText(value: unknown, maxLength: number): value is string {
return typeof value === 'string' && value.length > 0 && value.length <= maxLength && !hasControlCharacter(value)
}
function hasControlCharacter(value: string): boolean {
for (let index = 0; index < value.length; index++) {
if ((value.charCodeAt(index) || 0) < 32) return true
}
return false
}
function validEvent(event: WebhookEvent): boolean {
return (
validText(event.ownerDid, 2_048) &&
validText(event.rkey, 512) &&
validText(event.url, 2_048) &&
(event.eventKind === 'create' || event.eventKind === 'update' || event.eventKind === 'delete') &&
validText(event.eventDid, 2_048) &&
validText(event.eventCollection, 253) &&
validText(event.eventRkey, 512) &&
(event.cid === undefined || validText(event.cid, 512)) &&
validText(event.deliveredAt, 128) &&
(event.status === 'ok' || event.status === 'failed')
)
}
function closeClient(client: RedisClient | null): void {
if (!client) return
try {
client.close()
} catch {
// Closing is best-effort and must not expose a Redis error.
}
}
function scheduleReconnect(): void {
if (closing || !config.redisUrl || reconnectTimer || publisher) return
if (reconnectAttempts >= config.redisReconnectMaxAttempts) {
safeLog('[Redis] publisher reconnect limit reached')
return
}
reconnectAttempts++
const exponent = Math.min(reconnectAttempts - 1, 12)
const ceiling = Math.min(config.redisReconnectMaxMs, config.redisReconnectMinMs * 2 ** exponent)
const delay = Math.max(1, Math.floor(Math.random() * ceiling))
reconnectTimer = setTimeout(() => {
reconnectTimer = null
void createPublisher()
}, delay)
}
async function createPublisher(): Promise<void> {
if (closing || !config.redisUrl || publisher) return
let client: RedisClient | undefined
try {
client = redisClientFactory(config.redisUrl)
publisher = client
connected = false
client.onconnect = () => {
if (publisher !== client || closing) return
connected = true
reconnectAttempts = 0
logger.info('[Redis] publisher connected')
}
client.onclose = () => {
if (publisher !== client) return
publisher = null
connected = false
if (!closing) {
safeLog('[Redis] publisher disconnected')
scheduleReconnect()
}
}
// RedisClient is lazy: constructing it does not complete AUTH/PING. Await
// connect so health reflects command readiness rather than object creation.
await client.connect()
if (publisher === client && !closing) connected = true
} catch {
if (publisher === client) {
publisher = null
connected = false
}
closeClient(client ?? null)
safeLog('[Redis] publisher connection failed')
scheduleReconnect()
}
}
async function availablePublisher(): Promise<RedisClient | null> {
if (!config.redisUrl) {
if (!loggedMissingRedis) {
loggedMissingRedis = true
logger.warn('[Redis] REDIS_URL not set; event publishing disabled')
}
return null
}
if (closing) return null
if (publisher && connected) return publisher
if (!publisher && !reconnectTimer) void createPublisher()
return null
}
function withTimeout<T>(operation: Promise<T>, timeoutMs: number): Promise<T> {
let timer: ReturnType<typeof setTimeout> | undefined
return Promise.race([
operation,
new Promise<T>((_resolve, reject) => {
timer = setTimeout(() => reject(new Error('Redis publish timed out')), timeoutMs)
}),
]).finally(() => {
if (timer) clearTimeout(timer)
})
}
/**
* Publish delivery audit metadata without buffering disconnected work. Redis is
* never part of webhook delivery/cursor correctness, so callers receive a bounded
* dropped result instead of an unbounded outage queue.
*/
export async function publishWebhookEvent(event: WebhookEvent): Promise<RedisPublishResult> {
if (!validEvent(event)) {
incrementDropped()
safeLog('[Redis] refusing invalid delivery event')
return { published: false, dropped: true }
}
const body = JSON.stringify(event)
if (Buffer.byteLength(body) > 8 * 1_024) {
incrementDropped()
safeLog('[Redis] refusing oversized delivery event')
return { published: false, dropped: true }
}
const client = await availablePublisher()
if (!client || inFlight >= config.redisPublishMaxInFlight) {
incrementDropped()
return { published: false, dropped: true }
}
inFlight++
const operation = Promise.resolve(client.publish(config.webhookEventsChannel, body))
const tracked = operation.finally(() => {
inFlight--
activePublishes.delete(tracked)
})
activePublishes.add(tracked)
try {
await withTimeout(tracked, config.redisPublishTimeoutMs)
return { published: true, dropped: false }
} catch {
incrementDropped()
safeLog('[Redis] delivery event publish failed')
if (publisher === client) {
publisher = null
connected = false
}
closeClient(client)
scheduleReconnect()
return { published: false, dropped: true }
}
}
export function getRedisPublisherHealth(): {
configured: boolean
connected: boolean
closing: boolean
dropped: number
inFlight: number
} {
return { configured: Boolean(config.redisUrl), connected, closing, dropped, inFlight }
}
export async function closeRedisPublisher(): Promise<void> {
closing = true
if (reconnectTimer) clearTimeout(reconnectTimer)
reconnectTimer = null
const client = publisher
publisher = null
connected = false
closeClient(client)
// Do not await timed-out network operations forever. Their semaphore slots are
// bounded and the client close causes normal implementations to reject them.
void Promise.allSettled([...activePublishes])
}
/** Test-only reset that always closes the previous client before dropping state. */
export function resetRedisPublisherForTests(): void {
if (reconnectTimer) clearTimeout(reconnectTimer)
reconnectTimer = null
const client = publisher
publisher = null
connected = false
closing = false
reconnectAttempts = 0
loggedMissingRedis = false
dropped = 0
lastFailureLogAt = 0
redisClientFactory = defaultRedisClientFactory
closeClient(client)
}