import {
decryptWebhookSecret,
encryptWebhookSecret,
isValidWebhookSecretId,
parseWebhookSecretEncryptionKeyring,
type WebhookSecretEncryptionKeyring,
} from '@wispplace/atproto-utils'
import { DELETED_SITE_RECORD_CID } from '@wispplace/constants'
import { SQL } from 'bun'
import { probeConnectionWithRetry, resolveConnectionWarmingIntervalMs } from './connection-warming'
import { resolveDatabaseConfiguration } from './database-config'
import { createDatabaseReadCircuit, type DatabaseReadProbeResult } from './database-read-circuit'
import { assessDatabaseReadReplication } from './database-read-replication'
import { isValidHandle, toDomain } from './domain-utils'
import { runDatabaseMigrations } from './migrations'
import { withReservedOAuthLock } from './oauth-lock'
import { createPresentationReadQueries } from './presentation-reads'
import { waitForSiteCacheProjection } from './site-cache-wait'
import { migrateWebhookSecretEnvelopes } from './webhook-secret-encryption'
export { isValidHandle, toDomain } from './domain-utils'
/**
* The primary database client. This remains the backward-compatible strong-
* consistency client for every write and any read that affects security,
* authorization, or a mutation decision.
*/
export const databaseConfiguration = resolveDatabaseConfiguration(process.env)
const createPoolOptions = (pool: typeof databaseConfiguration.primaryPool) => ({
max: pool.max,
idleTimeout: pool.idleTimeoutSeconds,
connectionTimeout: pool.connectionTimeoutSeconds,
maxLifetime: 300,
})
export const db = new SQL(databaseConfiguration.primaryUrl, createPoolOptions(databaseConfiguration.primaryPool))
/**
* How many primary connections are held open between requests.
*
* Two, because the widest fan-out on the primary in one request is the pair of
* lookups at the end of the OAuth callback. A third concurrent query still
* connects on demand; it just is not on the common path.
* See {@link probeConnectionWithRetry} for why this is worth doing at all.
*/
const WARM_PRIMARY_CONNECTIONS = 2
/**
* Hold connections open so a request never pays for establishing one.
*
* The probes run concurrently on purpose: one at a time would keep re-using a
* single pooled connection and leave the rest to lapse.
*/
export const warmPrimaryConnections = async (): Promise<void> => {
await Promise.all(
Array.from({ length: WARM_PRIMARY_CONNECTIONS }, () => probeConnectionWithRetry(() => db`SELECT 1`)),
)
}
export const connectionWarmingIntervalMs = resolveConnectionWarmingIntervalMs(
databaseConfiguration.primaryPool.idleTimeoutSeconds,
)
// Do not create another pool unless the operator explicitly configured a
// different endpoint. The raw replica client stays private so call sites must
// opt into the named eventualRead repository below.
const replicaReadDb = databaseConfiguration.hasSeparateReadPool
? new SQL(databaseConfiguration.readUrl, {
...createPoolOptions(databaseConfiguration.readPool),
connection: {
statement_timeout: databaseConfiguration.readQueryTimeoutMs,
},
})
: undefined
/** True when presentation-only reads have a separately configured pool. */
export const hasSeparateDatabaseReadPool = databaseConfiguration.hasSeparateReadPool
type ReadEndpointProbeRow = {
transaction_read_only: boolean | string
sensitive_data_restricted: boolean | string
write_privileges_restricted: boolean | string
in_recovery: boolean | string
received_lsn: string | null
replayed_lsn: string | null
last_replay_at_ms: number | string | null
receiver_streaming: boolean | string | null
receiver_last_message_receipt_at_ms: number | string | null
observed_at_ms: number | string
}
const asBoolean = (value: boolean | string): boolean => value === true || value === 'true' || value === 't'
const asFiniteNumberOrNull = (value: number | string | null): number | null => {
if (value === null) return null
const parsed = Number(value)
return Number.isFinite(parsed) ? parsed : null
}
const probeReadEndpoint = async (): Promise<DatabaseReadProbeResult> => {
if (!replicaReadDb) throw new Error('No configured database read endpoint')
const query = replicaReadDb<ReadEndpointProbeRow[]>`
SELECT
(current_setting('transaction_read_only', true) = 'on') AS transaction_read_only,
NOT EXISTS (
SELECT 1
FROM pg_catalog.pg_class relation
JOIN pg_catalog.pg_namespace schema ON schema.oid = relation.relnamespace
WHERE schema.nspname = 'public'
AND relation.relkind IN ('r', 'p', 'v', 'm', 'f')
AND relation.relname NOT IN ('domains', 'custom_domains', 'site_cache', 'supporter')
AND (
has_table_privilege(current_user, relation.oid, 'SELECT')
OR has_any_column_privilege(current_user, relation.oid, 'SELECT')
)
) AS sensitive_data_restricted,
NOT EXISTS (
SELECT 1
FROM pg_catalog.pg_class relation
JOIN pg_catalog.pg_namespace schema ON schema.oid = relation.relnamespace
WHERE schema.nspname = 'public'
AND relation.relkind IN ('r', 'p', 'v', 'm', 'f')
AND (
has_table_privilege(current_user, relation.oid, 'INSERT')
OR has_table_privilege(current_user, relation.oid, 'UPDATE')
OR has_table_privilege(current_user, relation.oid, 'DELETE')
OR has_table_privilege(current_user, relation.oid, 'TRUNCATE')
OR has_table_privilege(current_user, relation.oid, 'REFERENCES')
OR has_table_privilege(current_user, relation.oid, 'TRIGGER')
OR has_any_column_privilege(current_user, relation.oid, 'INSERT')
OR has_any_column_privilege(current_user, relation.oid, 'UPDATE')
)
) AS write_privileges_restricted,
pg_is_in_recovery() AS in_recovery,
pg_last_wal_receive_lsn()::TEXT AS received_lsn,
pg_last_wal_replay_lsn()::TEXT AS replayed_lsn,
EXTRACT(EPOCH FROM pg_last_xact_replay_timestamp()) * 1000 AS last_replay_at_ms,
receiver.receiver_streaming AS receiver_streaming,
EXTRACT(EPOCH FROM receiver.last_message_receipt_at) * 1000 AS receiver_last_message_receipt_at_ms,
EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS observed_at_ms
FROM (SELECT 1) AS probe
LEFT JOIN LATERAL public.wisp_replica_receiver_status() AS receiver ON true
`
let timeout: ReturnType<typeof setTimeout> | undefined
const timeoutResult = new Promise<never>((_, reject) => {
timeout = setTimeout(() => {
try {
query.cancel()
} catch {
// The circuit only needs the bounded timeout; never expose a driver error.
}
reject(new Error('Database read endpoint probe timed out'))
}, databaseConfiguration.readProbeTimeoutMs)
})
try {
const rows = await Promise.race([query, timeoutResult])
const row = rows[0]
if (!row) throw new Error('Database read endpoint probe returned no rows')
const inRecovery = asBoolean(row.in_recovery)
const replication = assessDatabaseReadReplication(
{
inRecovery,
receivedLsn: row.received_lsn,
replayedLsn: row.replayed_lsn,
lastReplayAtMs: asFiniteNumberOrNull(row.last_replay_at_ms),
receiverStreaming: row.receiver_streaming === null ? false : asBoolean(row.receiver_streaming),
receiverLastMessageReceiptAtMs: asFiniteNumberOrNull(row.receiver_last_message_receipt_at_ms),
},
asFiniteNumberOrNull(row.observed_at_ms) ?? Date.now(),
databaseConfiguration.readReceiverFreshnessMs,
)
return {
transactionReadOnly: asBoolean(row.transaction_read_only),
sensitiveDataRestricted: asBoolean(row.sensitive_data_restricted),
writePrivilegesRestricted: asBoolean(row.write_privileges_restricted),
inRecovery,
replicationReceiverHealthy: replication.replicationReceiverHealthy,
replayLagMs: replication.replayLagMs,
}
} finally {
if (timeout) clearTimeout(timeout)
}
}
const databaseReadCircuit = createDatabaseReadCircuit({
configured: hasSeparateDatabaseReadPool,
maxReplayLagMs: databaseConfiguration.readMaxReplayLagMs,
probeIntervalMs: databaseConfiguration.readProbeIntervalMs,
cooldownMs: databaseConfiguration.readCircuitCooldownMs,
probe: probeReadEndpoint,
})
// Bootstrap DDL and every following migration run under the primary advisory lock.
await runDatabaseMigrations(db)
export interface WebhookSecretEncryptionHealth {
readonly status: 'ready' | 'degraded'
readonly encryptionAvailable: boolean
readonly legacySecretsRemaining: number | null
}
const getConfiguredWebhookSecretKeyring = (): WebhookSecretEncryptionKeyring =>
parseWebhookSecretEncryptionKeyring({
WEBHOOK_SECRET_ENCRYPTION_KEY: process.env.WEBHOOK_SECRET_ENCRYPTION_KEY,
WEBHOOK_SECRET_ENCRYPTION_PREVIOUS_KEYS: process.env.WEBHOOK_SECRET_ENCRYPTION_PREVIOUS_KEYS,
})
// There is intentionally no development plaintext fallback. Developers who use
// server-managed signing secrets must configure the same encryption key setup.
let webhookSecretEncryptionHealth: WebhookSecretEncryptionHealth = {
status: 'degraded',
encryptionAvailable: false,
legacySecretsRemaining: null,
}
const startupWebhookSecretKeyring = (() => {
try {
return getConfiguredWebhookSecretKeyring()
} catch {
return null
}
})()
try {
const migration = await migrateWebhookSecretEnvelopes(db, startupWebhookSecretKeyring)
webhookSecretEncryptionHealth = {
status: migration.legacyRemaining === 0 ? 'ready' : 'degraded',
encryptionAvailable: migration.encryptionAvailable,
legacySecretsRemaining: migration.legacyRemaining,
}
} catch {
// Do not log the error. Driver errors and bad configuration can include values
// which must not reach process logs. The health endpoint exposes only this state.
webhookSecretEncryptionHealth = {
status: 'degraded',
encryptionAvailable: startupWebhookSecretKeyring !== null,
legacySecretsRemaining: null,
}
}
export const getWebhookSecretEncryptionHealth = (): WebhookSecretEncryptionHealth => ({
...webhookSecretEncryptionHealth,
})
const startupReadHealth = await databaseReadCircuit.probeNow()
if (
databaseConfiguration.requiresProductionSafety &&
hasSeparateDatabaseReadPool &&
(startupReadHealth.mode === 'writable' || startupReadHealth.mode === 'unsafe')
) {
throw new Error(
'DATABASE_READ_URL must use a restricted role with default_transaction_read_only=on outside development and test',
)
}
const primaryPresentationRead = createPresentationReadQueries(db)
const replicaPresentationRead = replicaReadDb ? createPresentationReadQueries(replicaReadDb) : undefined
const withEventualPresentationRead = async <T>(
operation: (queries: typeof primaryPresentationRead) => Promise<T>,
): Promise<T> => {
const replicaQueries = replicaPresentationRead
if (!replicaQueries) return await operation(primaryPresentationRead)
return await databaseReadCircuit.withRead(
async () => await operation(replicaQueries),
async () => await operation(primaryPresentationRead),
)
}
/**
* Explicitly eventual, presentation-only reads. A healthy separate read pool is
* used only after its probe confirms a read-only session and acceptable replay
* lag. The circuit falls back to primary on probe or query failures. Never use
* this facade for authentication, authorization, ownership, tokens, cache waits,
* private data, or mutations.
*/
export const eventualRead = {
getUserStatus: async (did: string) => await withEventualPresentationRead((queries) => queries.getUserStatus(did)),
getSupporterStatus: async (did: string) => await withEventualPresentationRead((queries) => queries.isSupporter(did)),
getSitesForDid: async (did: string) => await withEventualPresentationRead((queries) => queries.getSitesByDid(did)),
getSitesWithDomainsForDid: async (did: string) =>
await withEventualPresentationRead((queries) => queries.getSitesWithDomainsByDid(did)),
getDomainsForDid: async (did: string) =>
await withEventualPresentationRead((queries) => queries.getDomainsForDid(did)),
getDomainsForSite: async (did: string, rkey: string) =>
await withEventualPresentationRead((queries) => queries.getDomainsBySite(did, rkey)),
getDomainStatus: async (domain: string) =>
await withEventualPresentationRead((queries) => queries.getDomainStatus(domain)),
getAdminSites: async (limit: number, offset: number) =>
await withEventualPresentationRead((queries) => queries.getAdminSites(limit, offset)),
getAdminSupporters: async () => await withEventualPresentationRead((queries) => queries.getAllSupporters()),
}
/** Sanitized read-endpoint health; this never includes a connection URL. */
export const getDatabaseReadHealth = async () => await databaseReadCircuit.health()
export const pruneAnalyticsData = async (): Promise<void> => {
await Promise.all([
db`DELETE FROM site_analytics_hourly WHERE bucket_start < NOW() - INTERVAL '90 days'`,
db`DELETE FROM analytics_ingest_batches WHERE received_at < NOW() - INTERVAL '7 days'`,
])
}
// These shared helpers intentionally keep their historic strong/primary semantics.
export const getDomainByDid = primaryPresentationRead.getDomainByDid
// Includes cache/configuration metadata that the replica role cannot access.
export const getAdminDatabaseReport = primaryPresentationRead.getAdminDatabaseReport
// Delivery history can contain operator-controlled URLs; keep it on primary.
export const getWebhookEventHistory = primaryPresentationRead.getWebhookEventHistory
export const getAllWispDomains = primaryPresentationRead.getAllWispDomains
export const countWispDomains = primaryPresentationRead.countWispDomains
export const isDomainAvailable = async (handle: string): Promise<boolean> => {
const h = handle.trim().toLowerCase()
if (!isValidHandle(h)) return false
const domain = toDomain(h)
const rows = await db`SELECT 1 FROM domains WHERE domain = ${domain} LIMIT 1`
return rows.length === 0
}
export const isDomainRegistered = primaryPresentationRead.isDomainRegistered
export const claimDomain = async (did: string, handle: string): Promise<string> => {
const h = handle.trim().toLowerCase()
if (!isValidHandle(h)) throw new Error('invalid_handle')
// Check if user already has 3 domains (unless they're a supporter)
const supporter = await isSupporter(did)
if (!supporter) {
const existingCount = await countWispDomains(did)
if (existingCount >= 3) {
throw new Error('domain_limit_reached')
}
}
const domain = toDomain(h)
try {
await db`
INSERT INTO domains (domain, did)
VALUES (${domain}, ${did})
`
} catch (_err) {
// Unique constraint violations -> already taken
throw new Error('conflict')
}
return domain
}
export const updateDomain = async (did: string, handle: string): Promise<string> => {
const h = handle.trim().toLowerCase()
if (!isValidHandle(h)) throw new Error('invalid_handle')
const domain = toDomain(h)
try {
const rows = await db`
UPDATE domains SET domain = ${domain}
WHERE did = ${did}
RETURNING domain
`
if (rows.length > 0) return rows[0].domain as string
// No existing row, behave like claim
return await claimDomain(did, handle)
} catch (_err) {
// Unique constraint violations -> already taken by someone else
throw new Error('conflict')
}
}
export const updateWispDomainSite = async (domain: string, siteRkey: string | null): Promise<void> => {
await db`
UPDATE domains
SET rkey = ${siteRkey}
WHERE domain = ${domain}
`
}
export const deleteWispDomain = async (domain: string): Promise<void> => {
await db`DELETE FROM domains WHERE domain = ${domain}`
}
export const getCustomDomainsByDid = primaryPresentationRead.getCustomDomainsByDid
export const getCustomDomainInfo = primaryPresentationRead.getCustomDomainInfo
export const getCustomDomainByHash = async (hash: string) => {
const rows = await db`SELECT * FROM custom_domains WHERE id = ${hash}`
return rows[0] ?? null
}
export const getCustomDomainById = async (id: string) => {
const rows = await db`SELECT * FROM custom_domains WHERE id = ${id}`
return rows[0] ?? null
}
export const claimCustomDomain = async (did: string, domain: string, hash: string, rkey: string | null = null) => {
const domainLower = domain.toLowerCase()
try {
const result = await db`
INSERT INTO custom_domains (id, domain, did, rkey, verified, created_at)
VALUES (${hash}, ${domainLower}, ${did}, ${rkey}, false, EXTRACT(EPOCH FROM NOW()))
ON CONFLICT (domain) DO UPDATE SET
id = EXCLUDED.id,
did = EXCLUDED.did,
rkey = EXCLUDED.rkey,
verified = EXCLUDED.verified,
created_at = EXCLUDED.created_at
WHERE custom_domains.verified = false
RETURNING *
`
if (result.length === 0) {
console.log('Failed to claim custom domain - already verified by another user')
throw new Error('conflict')
}
return { success: true, hash }
} catch (err) {
console.error('Failed to claim custom domain', err)
throw new Error('conflict')
}
}
export const updateCustomDomainRkey = async (id: string, rkey: string | null) => {
const rows = await db`
UPDATE custom_domains
SET rkey = ${rkey}
WHERE id = ${id}
RETURNING *
`
return rows[0] ?? null
}
export const updateCustomDomainVerification = async (id: string, verified: boolean) => {
const rows = await db`
UPDATE custom_domains
SET verified = ${verified}, last_verified_at = EXTRACT(EPOCH FROM NOW())
WHERE id = ${id}
RETURNING *
`
return rows[0] ?? null
}
export const deleteCustomDomain = async (id: string) => {
await db`DELETE FROM custom_domains WHERE id = ${id}`
}
export const getSitesByDid = primaryPresentationRead.getSitesByDid
export const waitForSiteCache = async (did: string, rkey: string): Promise<boolean> =>
waitForSiteCacheProjection(async () => {
const rows = await db`
SELECT 1 FROM site_cache
WHERE did = ${did} AND rkey = ${rkey} AND record_cid <> ${DELETED_SITE_RECORD_CID}
LIMIT 1
`
return rows.length > 0
})
// Get all domains (wisp + custom) mapped to a specific site
export const getDomainsBySite = primaryPresentationRead.getDomainsBySite
// Get count of domains mapped to a specific site
export const getDomainCountBySite = primaryPresentationRead.getDomainCountBySite
// Cookie secret management - ensure we have a secret for signing cookies
export const getCookieSecret = async (): Promise<string> => {
// Check if secret already exists
const rows = await db`SELECT secret FROM cookie_secrets WHERE id = 'default' LIMIT 1`
if (rows.length > 0) {
return rows[0].secret as string
}
// Generate new secret if none exists
const secret = crypto.randomUUID() + crypto.randomUUID() // 72 character random string
await db`
INSERT INTO cookie_secrets (id, secret, created_at)
VALUES ('default', ${secret}, EXTRACT(EPOCH FROM NOW()))
`
console.log('[CookieSecret] Generated new cookie signing secret')
return secret
}
export const getServiceIdentityKeypair = async (): Promise<{
publicKeyMultibase: string
privateKeyMultibase: string | null
} | null> => {
const rows = await db`
SELECT public_key_multibase, private_key_multibase
FROM service_identity_keys
WHERE id = 'default'
LIMIT 1
`
if (rows.length === 0) {
return null
}
return {
publicKeyMultibase: rows[0].public_key_multibase as string,
privateKeyMultibase: (rows[0].private_key_multibase as string | undefined) ?? null,
}
}
export const setServiceIdentityKeypair = async (
publicKeyMultibase: string,
privateKeyMultibase: string | null,
): Promise<void> => {
await db`
INSERT INTO service_identity_keys (id, public_key_multibase, private_key_multibase, created_at, updated_at)
VALUES ('default', ${publicKeyMultibase}, ${privateKeyMultibase}, EXTRACT(EPOCH FROM NOW()), EXTRACT(EPOCH FROM NOW()))
ON CONFLICT (id)
DO UPDATE SET
public_key_multibase = EXCLUDED.public_key_multibase,
private_key_multibase = EXCLUDED.private_key_multibase,
updated_at = EXTRACT(EPOCH FROM NOW())
`
}
// Supporter management functions
export const isSupporter = primaryPresentationRead.isSupporter
export const addSupporter = async (did: string): Promise<void> => {
await db`
INSERT INTO supporter (did)
VALUES (${did})
ON CONFLICT (did) DO NOTHING
`
}
export const removeSupporter = async (did: string): Promise<void> => {
await db`DELETE FROM supporter WHERE did = ${did}`
}
export const getAllSupporters = primaryPresentationRead.getAllSupporters
function generateSecretToken(): string {
const bytes = crypto.getRandomValues(new Uint8Array(24))
return `wsk_${Buffer.from(bytes).toString('base64url')}`
}
/**
* Parse on every secret operation so a controlled process restart/env reload
* never leaves a stale active key in memory. This throws only the stable,
* non-secret encryption-unavailable error on missing or malformed config.
*/
const requireWebhookSecretKeyring = (): WebhookSecretEncryptionKeyring => getConfiguredWebhookSecretKeyring()
const WEBHOOK_SECRET_STORAGE_ERROR = 'webhook_secret_storage_unavailable'
const WEBHOOK_SECRET_ID_ERROR = 'invalid_webhook_secret_id'
const assertValidWebhookSecretId = (name: string): void => {
if (!isValidWebhookSecretId(name)) throw new Error(WEBHOOK_SECRET_ID_ERROR)
}
const isUniqueConstraintViolation = (error: unknown): boolean =>
typeof error === 'object' && error !== null && 'code' in error && (error as { code?: unknown }).code === '23505'
export const createWebhookSecret = async (did: string, name: string): Promise<{ token: string; createdAt: string }> => {
assertValidWebhookSecretId(name)
const keyring = requireWebhookSecretKeyring()
const token = generateSecretToken()
const envelope = encryptWebhookSecret(token, keyring)
try {
const rows = await db`
INSERT INTO webhook_secrets (did, name, token, created_at)
VALUES (${did}, ${name}, ${envelope}, NOW())
RETURNING created_at
`
return { token, createdAt: new Date(rows[0].created_at).toISOString() }
} catch (error) {
// Do not attach the driver error: it can contain query parameter values.
if (isUniqueConstraintViolation(error)) throw new Error('already_exists')
throw new Error(WEBHOOK_SECRET_STORAGE_ERROR)
}
}
export const listWebhookSecrets = async (
did: string,
): Promise<Array<{ name: string; createdAt: string; lastRotatedAt?: string }>> => {
const rows = await db<Array<{ name: string; created_at: string; last_rotated_at: string | null }>>`
SELECT name, created_at, last_rotated_at
FROM webhook_secrets
WHERE did = ${did}
ORDER BY created_at ASC
`
return rows.map((r) => ({
name: r.name,
createdAt: new Date(r.created_at).toISOString(),
lastRotatedAt: r.last_rotated_at ? new Date(r.last_rotated_at).toISOString() : undefined,
}))
}
export const deleteWebhookSecret = async (did: string, name: string): Promise<boolean> => {
assertValidWebhookSecretId(name)
const rows = await db`
DELETE FROM webhook_secrets WHERE did = ${did} AND name = ${name} RETURNING name
`
return rows.length > 0
}
export const rotateWebhookSecret = async (
did: string,
name: string,
): Promise<{ token: string; rotatedAt: string } | null> => {
assertValidWebhookSecretId(name)
const keyring = requireWebhookSecretKeyring()
const token = generateSecretToken()
const envelope = encryptWebhookSecret(token, keyring)
const rows = await db`
UPDATE webhook_secrets
SET token = ${envelope}, last_rotated_at = NOW()
WHERE did = ${did} AND name = ${name}
RETURNING last_rotated_at
`
if (rows.length === 0) return null
return { token, rotatedAt: new Date(rows[0].last_rotated_at).toISOString() }
}
/**
* Internal delivery-only lookup. A found value must be a valid encrypted
* envelope. Plaintext legacy values, malformed records, and unavailable keys
* all fail closed with the same retryable generic error.
*/
export const getWebhookSecretToken = async (did: string, name: string): Promise<string | null> => {
assertValidWebhookSecretId(name)
const rows = await db<Array<{ token: string }>>`
SELECT token FROM webhook_secrets WHERE did = ${did} AND name = ${name} LIMIT 1
`
const envelope = rows[0]?.token
if (envelope === undefined) return null
return decryptWebhookSecret(envelope, requireWebhookSecretKeyring())
}
type WebhookMutationAction = 'create' | 'delete'
const WEBHOOK_MUTATION_WINDOW_MS = 60_000
const WEBHOOK_MUTATION_MAX_PER_WINDOW = 10
/**
* Atomically consume one per-DID mutation slot. The backing table is created by
* the normal primary migration; failure to reach it is handled as a generic
* route failure, never a permissive rate-limit bypass.
*/
export const consumeWebhookMutationRateLimit = async (
did: string,
action: WebhookMutationAction,
now = Date.now(),
): Promise<boolean> => {
const windowStart = now - WEBHOOK_MUTATION_WINDOW_MS
const rows = await db<Array<{ mutation_count: number }>>`
INSERT INTO webhook_mutation_rate_limits (did, action, window_started_at, mutation_count)
VALUES (${did}, ${action}, ${now}, 1)
ON CONFLICT (did, action) DO UPDATE
SET
window_started_at = CASE
WHEN webhook_mutation_rate_limits.window_started_at <= ${windowStart} THEN ${now}
ELSE webhook_mutation_rate_limits.window_started_at
END,
mutation_count = CASE
WHEN webhook_mutation_rate_limits.window_started_at <= ${windowStart} THEN 1
ELSE webhook_mutation_rate_limits.mutation_count + 1
END
WHERE webhook_mutation_rate_limits.window_started_at <= ${windowStart}
OR webhook_mutation_rate_limits.mutation_count < ${WEBHOOK_MUTATION_MAX_PER_WINDOW}
RETURNING mutation_count
`
return rows.length === 1
}
/**
* Serialize main-app PDS webhook mutations per owner across regions. The PDS
* has no multi-record transaction, so this is best-effort only for writes
* through this API; firehose intake remains the final enforcement point.
*/
export const withWebhookOwnerMutationLock = async <T>(did: string, operation: () => Promise<T>): Promise<T> => {
let reserved: Awaited<ReturnType<typeof db.reserve>>
try {
reserved = await db.reserve()
} catch {
throw new Error('webhook_mutation_unavailable')
}
const lockName = `wisp-webhook-owner:${did}`
return await withReservedOAuthLock(
{
async acquire(): Promise<void> {
// Do not alter a pooled session setting around a long PDS request.
// A busy owner fails closed and the authenticated caller can retry.
const rows = await reserved<Array<{ locked: boolean }>>`
SELECT pg_try_advisory_lock(hashtextextended(${lockName}, 0)) AS locked
`
if (!rows[0]?.locked) throw new Error('webhook_mutation_unavailable')
},
async unlock(): Promise<void> {
const rows = await reserved<Array<{ unlocked: boolean }>>`
SELECT pg_advisory_unlock(hashtextextended(${lockName}, 0)) AS unlocked
`
if (!rows[0]?.unlocked) throw new Error('webhook_mutation_unavailable')
},
release(): void {
reserved.release()
},
async close(): Promise<void> {
await reserved.close({ timeout: 0 })
},
},
operation,
() => {
// Static only: a cleanup failure must not expose a DID, endpoint, or secret.
console.error('[Webhook] Owner mutation lock cleanup failed')
},
)
}
const closeDatabasePool = async (name: 'primary' | 'read', pool: typeof db): Promise<void> => {
try {
await pool.end()
console.log(`[DB] ${name} database connection closed`)
} catch {
// Do not include driver errors here; they can include a connection URL.
console.error(`[DB] Error closing ${name} database connection`)
}
}
let databaseClosePromise: Promise<void> | undefined
/**
* Close the primary pool and, when configured, the separate eventual-read pool.
* Repeated shutdown signals share this promise so a pool is never ended twice.
*/
export const closeDatabase = (): Promise<void> => {
databaseClosePromise ??= Promise.all([
closeDatabasePool('primary', db),
...(replicaReadDb ? [closeDatabasePool('read', replicaReadDb)] : []),
]).then(() => undefined)
return databaseClosePromise
}