/**
* Environment configuration for firehose-service.
*
* Configuration is resolved once at startup. Validation errors intentionally do
* not echo raw values because endpoint URLs can contain credentials.
*/
import {
DEFAULT_CACHE_INVALIDATION_STREAM,
DEFAULT_REVALIDATE_DLQ_STREAM,
DEFAULT_REVALIDATE_STREAM,
resolveCacheInvalidationStreamMaxLen,
} from '@wispplace/constants'
export type ConfigEnv = Readonly<Record<string, string | undefined>>
export interface FirehoseConfig {
// Database
databaseUrl: string
// Firehose
firehoseService: string
firehoseServiceSecondary: string | undefined
firehoseMaxConcurrency: number
firehoseMaxPendingEvents: number
firehoseDrainGraceMs: number
plcDirectoryUrl: string | undefined
// S3 storage (write destination)
s3Bucket: string
s3Region: string
s3Endpoint: string | undefined
s3ForcePathStyle: boolean
s3Prefix: string
allowDiskStorage: boolean
awsAccessKeyId: string | undefined
awsSecretAccessKey: string | undefined
// Health check server
healthPort: number
// Redis revalidation queue
redisUrl: string | undefined
revalidateStream: string
revalidateGroup: string
revalidateStreamMaxLen: number
revalidateDedupeTtlSeconds: number
/** Stream receiving poison messages after bounded revalidation attempts. */
revalidateDlqStream: string
revalidateDlqStreamMaxLen: number
cacheInvalidationStream: string
cacheInvalidationStreamMaxLen: number
// Leader election and the independent OS-process supervisor
leaderElection: boolean
leadershipSupervisorEnabled: boolean
supervisorPath: string | undefined
leaderTtlMs: number
leaderRenewIntervalMs: number
leaderPollIntervalMs: number
cursorSaveIntervalMs: number
// Mode
isDbFillOnly: boolean
isBackfill: boolean
backfillConcurrency: number
hydrantUrl: string | undefined
}
const DEFAULT_DATABASE_URL = 'postgres://postgres:postgres@localhost:5432/wisp'
const DEFAULT_FIREHOSE_SERVICE = 'wss://bsky.network'
const DEFAULT_HEALTH_PORT = 3001
// canonical revalidate stream names come from @wispplace/constants
const DEFAULT_REVALIDATE_GROUP = 'firehose-service'
const DEFAULT_LEADER_TTL_MS = 30_000
const DEFAULT_LEADER_RENEW_INTERVAL_MS = 10_000
const DEFAULT_LEADER_POLL_INTERVAL_MS = 5_000
const DEFAULT_CURSOR_SAVE_INTERVAL_MS = 5_000
const STREAM_NAME_PATTERN = /^[A-Za-z0-9:_-]{1,128}$/
const INTEGER_PATTERN = /^(0|[1-9][0-9]*)$/
const S3_BUCKET_PATTERN =
/^(?=.{3,63}$)(?![0-9]+(?:\.[0-9]+){3}$)(?!.*\.\.)(?!.*[.-]$)[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$/
const S3_REGION_PATTERN = /^[A-Za-z0-9-]{1,64}$/
function hasUnsafeConfigText(value: string, forbidBackslash = false): boolean {
return Array.from(value).some((character) => {
const codePoint = character.codePointAt(0) ?? 0
return (forbidBackslash && character === '\\') || codePoint <= 0x1f || codePoint === 0x7f || character.trim() === ''
})
}
function resolveBoundedInteger(value: string | undefined, fallback: number, minimum: number, maximum: number): number {
if (value === undefined || !INTEGER_PATTERN.test(value)) return fallback
const parsed = Number(value)
if (!Number.isSafeInteger(parsed) || parsed < minimum || parsed > maximum) return fallback
return parsed
}
function resolveStrictBoolean(value: string | undefined, fallback: boolean, field: string): boolean {
if (value === undefined) return fallback
if (value === 'true') return true
if (value === 'false') return false
throw new Error(`Invalid ${field}`)
}
function validateS3Bucket(value: string | undefined): string {
if (value === undefined || value === '') return ''
if (!S3_BUCKET_PATTERN.test(value)) throw new Error('Invalid S3_BUCKET')
return value
}
function validateS3Region(value: string | undefined): string {
const region = value ?? 'us-east-1'
if (!S3_REGION_PATTERN.test(region)) throw new Error('Invalid S3_REGION')
return region
}
function validateS3Prefix(value: string | undefined): string {
const prefix = value ?? 'sites/'
const segments = prefix.split('/')
if (
prefix.length > 512 ||
prefix.startsWith('/') ||
hasUnsafeConfigText(prefix, true) ||
segments.some((segment) => segment === '.' || segment === '..')
) {
throw new Error('Invalid S3_PREFIX')
}
return prefix
}
function validateS3Credentials(accessKeyId: string | undefined, secretAccessKey: string | undefined): void {
const hasAccessKey = Boolean(accessKeyId)
const hasSecretKey = Boolean(secretAccessKey)
if (hasAccessKey !== hasSecretKey) {
throw new Error('AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY must be set together')
}
if ((accessKeyId && hasUnsafeConfigText(accessKeyId)) || (secretAccessKey && hasUnsafeConfigText(secretAccessKey))) {
throw new Error('Invalid AWS credentials')
}
}
function requireBoundedName(value: string | undefined, fallback: string, field: string): string {
const result = value ?? fallback
if (!STREAM_NAME_PATTERN.test(result)) throw new Error(`Invalid ${field}`)
return result
}
function parseUrl(value: string, field: string): URL {
try {
return new URL(value)
} catch {
throw new Error(`Invalid ${field}`)
}
}
function hasForbiddenUrlParts(url: URL): boolean {
return Boolean(url.username || url.password || url.search || url.hash)
}
function isLocalDevelopmentHost(hostname: string): boolean {
return (
hostname === 'localhost' ||
hostname === '127.0.0.1' ||
hostname === '::1' ||
hostname === '[::1]' ||
hostname.endsWith('.localhost')
)
}
function validateRelayUrl(value: string, field: string, production: boolean): string {
const url = parseUrl(value, field)
if ((url.protocol !== 'ws:' && url.protocol !== 'wss:') || !url.hostname || hasForbiddenUrlParts(url)) {
throw new Error(`Invalid ${field}`)
}
if (production && url.protocol !== 'wss:') throw new Error(`Invalid ${field}`)
return value
}
function validateSecureHttpUrl(value: string | undefined, field: string, production: boolean): string | undefined {
if (!value) return undefined
const url = parseUrl(value, field)
const allowLocalHttp = !production && url.protocol === 'http:' && isLocalDevelopmentHost(url.hostname)
if ((!allowLocalHttp && url.protocol !== 'https:') || !url.hostname || hasForbiddenUrlParts(url)) {
throw new Error(`Invalid ${field}`)
}
return value
}
function validateRedisUrl(value: string | undefined): string | undefined {
if (!value) return undefined
const url = parseUrl(value, 'REDIS_URL')
if ((url.protocol !== 'redis:' && url.protocol !== 'rediss:') || !url.hostname || url.search || url.hash) {
throw new Error('Invalid REDIS_URL')
}
return value
}
function validateS3Endpoint(value: string | undefined, production: boolean): string | undefined {
if (!value) return undefined
const url = parseUrl(value, 'S3_ENDPOINT')
const allowLocalHttp = !production && url.protocol === 'http:' && isLocalDevelopmentHost(url.hostname)
if ((!allowLocalHttp && url.protocol !== 'https:') || !url.hostname || hasForbiddenUrlParts(url)) {
throw new Error('Invalid S3_ENDPOINT')
}
return value
}
function validateDatabaseUrl(value: string, production: boolean): string {
if (production && !value) throw new Error('DATABASE_URL is required in production')
const url = parseUrl(value, 'DATABASE_URL')
if ((url.protocol !== 'postgres:' && url.protocol !== 'postgresql:') || !url.hostname) {
throw new Error('Invalid DATABASE_URL')
}
return value
}
interface EnvironmentContext {
production: boolean
developmentOrTest: boolean
}
type ResolvedStorageConfig = Pick<
FirehoseConfig,
| 's3Bucket'
| 's3Region'
| 's3Endpoint'
| 's3ForcePathStyle'
| 's3Prefix'
| 'allowDiskStorage'
| 'awsAccessKeyId'
| 'awsSecretAccessKey'
>
type ResolvedLeaderConfig = Pick<
FirehoseConfig,
| 'leaderElection'
| 'leadershipSupervisorEnabled'
| 'supervisorPath'
| 'leaderTtlMs'
| 'leaderRenewIntervalMs'
| 'leaderPollIntervalMs'
| 'cursorSaveIntervalMs'
>
type ResolvedFirehoseConfig = Pick<
FirehoseConfig,
| 'firehoseService'
| 'firehoseServiceSecondary'
| 'firehoseMaxConcurrency'
| 'firehoseMaxPendingEvents'
| 'firehoseDrainGraceMs'
| 'plcDirectoryUrl'
>
type ResolvedQueueConfig = Pick<
FirehoseConfig,
| 'redisUrl'
| 'revalidateStream'
| 'revalidateGroup'
| 'revalidateStreamMaxLen'
| 'revalidateDedupeTtlSeconds'
| 'revalidateDlqStream'
| 'revalidateDlqStreamMaxLen'
| 'cacheInvalidationStream'
| 'cacheInvalidationStreamMaxLen'
>
type ResolvedModeConfig = Pick<FirehoseConfig, 'isDbFillOnly' | 'isBackfill' | 'backfillConcurrency' | 'hydrantUrl'>
function resolveEnvironmentContext(env: ConfigEnv): EnvironmentContext {
return {
production: env.NODE_ENV === 'production',
developmentOrTest: env.NODE_ENV === 'development' || env.NODE_ENV === 'test',
}
}
function validateDiskStorageMode(allowDiskStorage: boolean, context: EnvironmentContext): void {
if (allowDiskStorage && !context.developmentOrTest) {
throw new Error('FIREHOSE_ALLOW_DISK_STORAGE is only allowed in development or test')
}
}
function validateStoragePresence(
s3Bucket: string,
allowDiskStorage: boolean,
s3Endpoint: string | undefined,
awsAccessKeyId: string | undefined,
awsSecretAccessKey: string | undefined,
): void {
if (s3Bucket) return
if (!allowDiskStorage) {
throw new Error('S3_BUCKET is required unless FIREHOSE_ALLOW_DISK_STORAGE is enabled for development or test')
}
if (s3Endpoint || awsAccessKeyId || awsSecretAccessKey) {
throw new Error('S3_BUCKET is required when S3 endpoint or credentials are configured')
}
}
function resolveStorageConfig(env: ConfigEnv, context: EnvironmentContext): ResolvedStorageConfig {
const allowDiskStorage = resolveStrictBoolean(
env.FIREHOSE_ALLOW_DISK_STORAGE,
env.NODE_ENV === 'test',
'FIREHOSE_ALLOW_DISK_STORAGE',
)
validateDiskStorageMode(allowDiskStorage, context)
const s3Bucket = validateS3Bucket(env.S3_BUCKET)
const s3Region = validateS3Region(env.S3_REGION)
const s3Endpoint = validateS3Endpoint(env.S3_ENDPOINT, context.production)
const s3Prefix = validateS3Prefix(env.S3_PREFIX)
const awsAccessKeyId = env.AWS_ACCESS_KEY_ID
const awsSecretAccessKey = env.AWS_SECRET_ACCESS_KEY
validateS3Credentials(awsAccessKeyId, awsSecretAccessKey)
validateStoragePresence(s3Bucket, allowDiskStorage, s3Endpoint, awsAccessKeyId, awsSecretAccessKey)
return {
s3Bucket,
s3Region,
s3Endpoint,
s3ForcePathStyle: resolveStrictBoolean(env.S3_FORCE_PATH_STYLE, true, 'S3_FORCE_PATH_STYLE'),
s3Prefix,
allowDiskStorage,
awsAccessKeyId,
awsSecretAccessKey,
}
}
function resolveLeaderRenewInterval(leaderTtlMs: number, configuredRenew: number): number {
// Renew often enough to leave room for bounded Redis connect/command
// latency and one or more transient failures. The lease deadline below is
// independent of this value, so a malformed or overly large setting must
// never make us wait most of the TTL before noticing a lost lease.
const maximumSafeRenewInterval = Math.max(100, Math.floor(leaderTtlMs / 4))
return Math.min(configuredRenew, maximumSafeRenewInterval)
}
function validateLeaderRedisRequirement(
production: boolean,
leaderElection: boolean,
redisUrl: string | undefined,
): void {
if (production && !leaderElection) throw new Error('LEADER_ELECTION must be enabled in production')
if ((production || leaderElection) && !redisUrl) {
throw new Error('REDIS_URL is required for durable firehose operation')
}
}
function resolveLeaderConfig(env: ConfigEnv, production: boolean, redisUrl: string | undefined): ResolvedLeaderConfig {
const leaderTtlMs = resolveBoundedInteger(env.LEADER_TTL_MS, DEFAULT_LEADER_TTL_MS, 1_000, 300_000)
const configuredRenew = resolveBoundedInteger(
env.LEADER_RENEW_INTERVAL_MS,
DEFAULT_LEADER_RENEW_INTERVAL_MS,
100,
299_999,
)
const leaderElection = resolveStrictBoolean(env.LEADER_ELECTION, production, 'LEADER_ELECTION')
validateLeaderRedisRequirement(production, leaderElection, redisUrl)
const leadershipSupervisorEnabled = resolveStrictBoolean(
env.LEADERSHIP_SUPERVISOR_ENABLED ?? env.FIREHOSE_SUPERVISOR_ENABLED ?? env.SUPERVISOR_ENABLED,
leaderElection,
'LEADERSHIP_SUPERVISOR_ENABLED',
)
if (production && leaderElection && !leadershipSupervisorEnabled) {
throw new Error('LEADERSHIP_SUPERVISOR_ENABLED must be enabled in production')
}
const supervisorPath =
env.FIREHOSE_SUPERVISOR_PATH || env.LEADERSHIP_SUPERVISOR_PATH || env.SUPERVISOR_PATH || undefined
return {
leaderElection,
leadershipSupervisorEnabled,
supervisorPath,
leaderTtlMs,
leaderRenewIntervalMs: resolveLeaderRenewInterval(leaderTtlMs, configuredRenew),
leaderPollIntervalMs: resolveBoundedInteger(
env.LEADER_POLL_INTERVAL_MS,
DEFAULT_LEADER_POLL_INTERVAL_MS,
100,
60_000,
),
cursorSaveIntervalMs: resolveBoundedInteger(
env.CURSOR_SAVE_INTERVAL_MS,
DEFAULT_CURSOR_SAVE_INTERVAL_MS,
100,
300_000,
),
}
}
function resolveFirehoseConfig(env: ConfigEnv, production: boolean): ResolvedFirehoseConfig {
const secondary = env.FIREHOSE_SERVICE_SECONDARY
return {
firehoseService: validateRelayUrl(env.FIREHOSE_SERVICE ?? DEFAULT_FIREHOSE_SERVICE, 'FIREHOSE_SERVICE', production),
firehoseServiceSecondary: secondary
? validateRelayUrl(secondary, 'FIREHOSE_SERVICE_SECONDARY', production)
: undefined,
firehoseMaxConcurrency: resolveBoundedInteger(env.FIREHOSE_MAX_CONCURRENCY, 5, 1, 20),
firehoseMaxPendingEvents: resolveBoundedInteger(env.FIREHOSE_MAX_PENDING_EVENTS, 10_000, 1, 100_000),
firehoseDrainGraceMs: resolveBoundedInteger(env.FIREHOSE_DRAIN_GRACE_MS, 30_000, 0, 300_000),
plcDirectoryUrl: validateSecureHttpUrl(env.WISP_PLC_DIRECTORY_URL, 'WISP_PLC_DIRECTORY_URL', production),
}
}
function resolveQueueConfig(env: ConfigEnv): ResolvedQueueConfig {
return {
redisUrl: validateRedisUrl(env.REDIS_URL),
revalidateStream: requireBoundedName(
env.WISP_REVALIDATE_STREAM,
DEFAULT_REVALIDATE_STREAM,
'WISP_REVALIDATE_STREAM',
),
revalidateGroup: requireBoundedName(env.WISP_REVALIDATE_GROUP, DEFAULT_REVALIDATE_GROUP, 'WISP_REVALIDATE_GROUP'),
revalidateStreamMaxLen: resolveBoundedInteger(env.WISP_REVALIDATE_STREAM_MAXLEN, 10_000, 1, 1_000_000),
revalidateDedupeTtlSeconds: resolveBoundedInteger(env.WISP_REVALIDATE_DEDUPE_TTL_SECONDS, 60, 1, 86_400),
revalidateDlqStream: requireBoundedName(
env.WISP_REVALIDATE_DLQ_STREAM,
DEFAULT_REVALIDATE_DLQ_STREAM,
'WISP_REVALIDATE_DLQ_STREAM',
),
revalidateDlqStreamMaxLen: resolveBoundedInteger(env.WISP_REVALIDATE_DLQ_STREAM_MAXLEN, 10_000, 1, 1_000_000),
cacheInvalidationStream: requireBoundedName(
env.WISP_CACHE_INVALIDATION_STREAM,
DEFAULT_CACHE_INVALIDATION_STREAM,
'WISP_CACHE_INVALIDATION_STREAM',
),
cacheInvalidationStreamMaxLen: resolveCacheInvalidationStreamMaxLen(env.WISP_CACHE_INVALIDATION_STREAM_MAXLEN),
}
}
function resolveModeConfig(env: ConfigEnv, args: readonly string[], production: boolean): ResolvedModeConfig {
const isDbFillOnly = args.includes('--db-fill-only') || env.DB_FILL_ONLY === 'true'
return {
isDbFillOnly,
isBackfill: args.includes('--backfill') || isDbFillOnly || env.BACKFILL === 'true',
backfillConcurrency: resolveBoundedInteger(env.BACKFILL_CONCURRENCY, 5, 1, 20),
hydrantUrl: validateSecureHttpUrl(env.HYDRANT_URL, 'HYDRANT_URL', production),
}
}
/**
* Resolve configuration without reading global process state. This boundary is
* exported for deterministic validation tests and never logs raw values.
*/
export function resolveConfig(env: ConfigEnv = {}, args: readonly string[] = []): FirehoseConfig {
const context = resolveEnvironmentContext(env)
const databaseUrl = validateDatabaseUrl(
env.DATABASE_URL ?? (context.production ? '' : DEFAULT_DATABASE_URL),
context.production,
)
const storage = resolveStorageConfig(env, context)
const queue = resolveQueueConfig(env)
const leader = resolveLeaderConfig(env, context.production, queue.redisUrl)
const firehose = resolveFirehoseConfig(env, context.production)
const mode = resolveModeConfig(env, args, context.production)
return {
databaseUrl,
...firehose,
...storage,
healthPort: resolveBoundedInteger(env.HEALTH_PORT, DEFAULT_HEALTH_PORT, 1, 65_535),
...queue,
...leader,
...mode,
}
}
export const config = resolveConfig(process.env, process.argv)