orca/cloud/apps/relay/src/index.ts

import {
  formatAssignmentInventorySnapshot,
  readAssignmentInventorySnapshot
} from './assignment-inventory-snapshot.js'
import { readRegionCorrectionOutcomes } from './region-correction-outcomes.js'
import { RelayAssignmentStore } from './assignment-store.js'
import { loadRelayConfig } from './config.js'
import { startCellHeartbeat } from './cell-heartbeat-client.js'
import {
  CELL_FLAG_DEFAULTS,
  CELL_FLAGS_APPLIED_EVENT,
  type CellFlags,
  cellFlagObjectName,
  cellFlagParser
} from './cell-flags.js'
import { startControlFlagChannel } from './relay-control-flag-channel.js'
import {
  reconcileCellAdmissionAtStartup,
  roleOwnsAssignmentMaintenance
} from './cell-admission-startup.js'
import { openRelayDatabaseAtBoot } from './boot-database-open.js'
import {
  consumeRelayCellInventoryHold,
  consumeRelayDatabasePoolPressure,
  readRelayDatabasePoolPressure
} from './database.js'
import { runAssignmentCleanup } from './assignment-cleanup-steps.js'
import { runRelayBackgroundOperation } from './relay-background-operation.js'
import { readPostgresLockWaitSample } from './postgres-lock-wait-sample.js'
import { jitteredSweepIntervalMs } from './relay-sweep-schedule.js'
import { observedRelayRequests } from './relay-observability.js'
import { handleRelayUnhandledRejection } from './relay-database-rejection-fence.js'
import { startRegionalRehomeWorker } from './regional-rehome-worker.js'
import { createRelayServer } from './relay-server.js'
import {
  formatRegisteredMigrationInventory,
  readRegisteredMigrationInventory
} from './registered-migration-inventory.js'

// Both read the cell's switch file once it exists; until then, the boot values.
let cellFlagsApplied: (() => CellFlags) | null = null
// Before anything can start a database promise.
process.on('unhandledRejection', (reason) =>
  handleRelayUnhandledRejection(reason, cellFlagsApplied?.().rejectionFence ?? true)
)
const config = loadRelayConfig()
const database = await openRelayDatabaseAtBoot({
  databaseUrl: config.databaseUrl,
  dataDir: config.dataDir,
  poolMax: config.databasePoolMax,
  applicationName: `orca-relay/${config.role}/${config.cellId}`,
  appliesPostgresSchema: config.role !== 'cell',
  readTimeoutMarginOverrideMs: () => cellFlagsApplied?.().readTimeoutMarginMs
})
await reconcileCellAdmissionAtStartup(config, new RelayAssignmentStore(database))
// Never awaited: the cell listens on defaults (all off) until the first read lands.
const cellFlagChannel =
  config.role === 'cell'
    ? startControlFlagChannel({
        objectName: cellFlagObjectName(config.cellId),
        defaults: CELL_FLAG_DEFAULTS,
        parse: cellFlagParser(config.cellId),
        appliedEvent: CELL_FLAGS_APPLIED_EVENT
      })
    : null
cellFlagsApplied = cellFlagChannel ? () => cellFlagChannel.applied().flags : null
const {
  server,
  sessions,
  store,
  assignments,
  observability,
  runtimeCounts,
  connectionSnapshot,
  ready,
  cellIncarnation,
  shadowSeatPoller
} = createRelayServer(config, database, { cellFlags: cellFlagChannel?.applied })
// Same owner as the assignment sweep: the cleanup only expires credentials that every reader
// already re-checks at read time, so running it in all 23 cells multiplied one table scan by 23
// without changing any answer.
const cleanupTimer = roleOwnsAssignmentMaintenance(config.role)
  ? setInterval(
      () =>
        void runRelayBackgroundOperation(
          () => store.cleanup(),
          '[orca-relay] credential cleanup failed'
        ),
      jitteredSweepIntervalMs(30_000)
    )
  : null
const assignmentCleanupTimer = roleOwnsAssignmentMaintenance(config.role)
  ? setInterval(() => {
      void runAssignmentCleanup(assignments)
    }, jitteredSweepIntervalMs(30_000))
  : null
const inventorySnapshotTimer = roleOwnsAssignmentMaintenance(config.role)
  ? setInterval(() => {
      void runRelayBackgroundOperation(async () => {
        const snapshot = await readAssignmentInventorySnapshot(database, Date.now())
        for (const line of formatAssignmentInventorySnapshot(snapshot)) console.warn(line)
      }, '[orca-relay] inventory snapshot failed')
    }, 60_000)
  : null
const migrationInventoryTimer = roleOwnsAssignmentMaintenance(config.role)
  ? setInterval(() => {
      void runRelayBackgroundOperation(async () => {
        const inventory = await readRegisteredMigrationInventory(database, Date.now())
        for (const line of formatRegisteredMigrationInventory(inventory)) console.warn(line)
        console.log(
          JSON.stringify({
            event: 'orca_relay_region_correction_outcomes',
            observedAt: Date.now(),
            outcomes: await readRegionCorrectionOutcomes(database, Date.now())
          })
        )
      }, '[orca-relay] migration inventory failed')
    }, 5 * 60_000)
  : null
// Directors only: one role's view covers every backend, and cells roll separately.
// Single-flight, so a slow database never stacks samples on the 3-slot pool.
let lockWaitSampling = false
const lockWaitSampleTimer = roleOwnsAssignmentMaintenance(config.role)
  ? setInterval(() => {
      if (database.dialect !== 'postgres' || lockWaitSampling) return
      lockWaitSampling = true
      void runRelayBackgroundOperation(async () => {
        observability.recordDatabaseLockWaitSample(await readPostgresLockWaitSample(database))
      }, '[orca-relay] lock wait sample failed').finally(() => {
        lockWaitSampling = false
      })
    }, 5_000)
  : null
cleanupTimer?.unref()
assignmentCleanupTimer?.unref()
lockWaitSampleTimer?.unref()
inventorySnapshotTimer?.unref()
migrationInventoryTimer?.unref()
observability.start(() => ({
  ...runtimeCounts(),
  ...consumeRelayDatabasePoolPressure(database),
  ...consumeRelayCellInventoryHold(database)
}))
const regionalRehomeWorker = startRegionalRehomeWorker(config, assignments, {
  safetySnapshot: () => ({
    ...observability.regionalRehomeRuntimeSafety(),
    ...readRelayDatabasePoolPressure(database)
  })
})
const heartbeat = startCellHeartbeat(config, {
  ready,
  incarnation: cellIncarnation,
  observedRequests: () => observedRelayRequests(runtimeCounts()),
  connectionCounts: () => {
    const snapshot = connectionSnapshot()
    const counts = runtimeCounts()
    return {
      totalConnections: snapshot?.physicalConnections ?? counts.totalConnections,
      inFlightConnections:
        snapshot?.inFlightConnections ?? counts.inFlightConnections ?? 0,
      reservedConnectionUnits:
        snapshot?.reservedConnectionUnits ?? counts.reservedConnectionUnits ?? 0,
      enforcedConnectionUnits:
        snapshot?.enforcedConnectionUnits ??
        counts.enforcedConnectionUnits ??
        counts.totalConnections,
      inclusionWatermark: snapshot?.inclusionWatermark
    }
  },
  regionalRehomeSafety: () => ({
    ...observability.regionalRehomeRuntimeSafety(),
    ...readRelayDatabasePoolPressure(database)
  })
})

server.listen(config.port, () => {
  console.log(`[orca-relay] listening on ${config.publicUrl} (port ${config.port})`)
})

const shutdown = (): void => {
  if (cleanupTimer) clearInterval(cleanupTimer)
  if (assignmentCleanupTimer) clearInterval(assignmentCleanupTimer)
  if (inventorySnapshotTimer) clearInterval(inventorySnapshotTimer)
  if (migrationInventoryTimer) clearInterval(migrationInventoryTimer)
  if (lockWaitSampleTimer) clearInterval(lockWaitSampleTimer)
  observability.stop()
  heartbeat?.stop()
  regionalRehomeWorker?.stop()
  shadowSeatPoller?.stop()
  cellFlagChannel?.stop()
  sessions.drain(0)
  server.close(() => void database.close().catch(() => undefined))
}
process.once('SIGTERM', shutdown)
process.once('SIGINT', shutdown)