orca/cloud/apps/push/src/push-server.ts

import { PushAuthAdmission } from './push-auth-admission.js'
import { createAdaptorServer } from '@hono/node-server'
import {
  PUSH_LIMITS,
  PushDeviceRegistrationRequestSchema,
  PushHostChallengeRequestSchema,
  PushHostSessionRequestSchema,
  PushSendRequestSchema,
  type PushSendResult
} from '@orca-cloud/push-contract'
import { Hono, type MiddlewareHandler } from 'hono'
import { bodyLimit } from 'hono/body-limit'
import { ApnsClient } from './apns-client.js'
import { createApnsHttp2Transport, type ApnsTransport } from './apns-http2-transport.js'
import { clientIpRateLimit, ClientIpRateLimiter, readClientIp } from './client-ip-rate-limit.js'
import { DurablePushStore } from './durable-push-store.js'
import { DurablePushWorker } from './durable-push-worker.js'
import type { PushConfig } from './config.js'
import { PushDeviceRegistryStore } from './device-registry-store.js'
import { createFcmAccessTokenProvider } from './fcm-access-token.js'
import { createFcmFetchTransport, FcmClient, type FcmTransport } from './fcm-client.js'
import { PushHostChallengeStore } from './host-challenge-store.js'
import { PushHostSessionStore } from './host-session-store.js'
import type { PushDatabase } from './push-database.js'
import { PushDispatcher } from './push-dispatcher.js'
import { PushObservability } from './push-observability.js'
import { reserveRequestConnection } from './push-background-database.js'
import { createPushReadiness } from './push-readiness.js'
import { PushRequestDrain } from './push-request-drain.js'

export type PushServerOptions = {
  now?: () => number
  apnsTransport?: ApnsTransport
  fcmTransport?: FcmTransport
  fcmAccessToken?: () => Promise<string>
}

type PushVariables = { hostFingerprint: string }

export function readBearer(header: string | undefined): string | null {
  if (!header) return null
  const [scheme, ...rest] = header.split(' ')
  const token = rest.join(' ').trim()
  return scheme?.toLowerCase() === 'bearer' && token.length > 0 ? token : null
}

// Hono's body limit, not a Content-Length check: a chunked body declares no
// length, and req.json() would buffer all of it before any handler ran.
const limitBody = bodyLimit({
  maxSize: PUSH_LIMITS.maxHttpBodyBytes,
  onError: (context) => context.json({ error: 'request_too_large' }, 413)
})

export function createPushServer(
  config: PushConfig,
  database: PushDatabase,
  options: PushServerOptions = {}
) {
  const now = options.now ?? Date.now
  const observability = new PushObservability()
  const challenges = new PushHostChallengeStore(database, config.publicUrl, now)
  const sessions = new PushHostSessionStore(database, now)
  const devices = new PushDeviceRegistryStore(database, now)
  const deliveryStore = new DurablePushStore(
    database,
    now,
    reserveRequestConnection(database, config.databasePoolMax)
  )
  const apnsTransport = options.apnsTransport ?? (config.apns ? createApnsHttp2Transport() : null)
  const dispatcher = new PushDispatcher({
    devices,
    ...(config.apns && apnsTransport
      ? {
          apns: new ApnsClient({
            topic: config.apnsTopic,
            credentials: config.apns,
            transport: apnsTransport,
            now
          })
        }
      : {}),
    fcm: new FcmClient({
      now,
      projectId: config.fcmProjectId,
      accessToken: options.fcmAccessToken ?? createFcmAccessTokenProvider(),
      transport: options.fcmTransport ?? createFcmFetchTransport()
    }),
    onOutcome: (status) =>
      observability.record(
        status === 'sent' ? 'delivery_sent' : status === 'dead' ? 'delivery_dead' : 'delivery_error'
      )
  })
  const worker = new DurablePushWorker(deliveryStore, dispatcher, {
    now,
    onRetry: () => observability.record('delivery_retry')
  })
  const ready = createPushReadiness(database, { now })
  const unauthenticatedIps = new ClientIpRateLimiter({ now })
  const limitUnauthenticatedIp = clientIpRateLimit(unauthenticatedIps, {
    trustedProxyHops: config.trustedProxyHops,
    onLimited: () => observability.record('ip_rate_limited')
  })
  const limitAuthenticatedIp = clientIpRateLimit(
    new ClientIpRateLimiter({
      now,
      capacity: PUSH_LIMITS.authenticatedRequestsPerMinutePerIp
    }),
    {
      trustedProxyHops: config.trustedProxyHops,
      onLimited: () => observability.record('ip_rate_limited')
    }
  )
  const authAdmission = new PushAuthAdmission()
  const invalidBearerIps = new ClientIpRateLimiter({ now })
  const authenticatedHosts = new ClientIpRateLimiter({
    now,
    capacity: PUSH_LIMITS.authenticatedRequestsPerMinutePerHost
  })
  const app = new Hono<{ Variables: PushVariables }>()
  const requestDrain = new PushRequestDrain()
  app.use('*', requestDrain.middleware)
  // Hono's default handler prints the whole error, and a pg error carries the
  // offending row in `detail`. Only the error's name may reach the logs.
  app.onError((error, context) => {
    observability.record('request_error')
    console.warn(
      JSON.stringify({
        event: 'orca_push_request_failed',
        error: error instanceof Error ? error.name : 'unknown'
      })
    )
    return context.json({ error: 'internal' }, 500)
  })

  app.get('/health', (context) =>
    context.json({ ok: true, pushProtocol: 1, deliveryProtocol: 2, mode: config.mode })
  )
  app.get('/ready', limitUnauthenticatedIp, async (context) =>
    (await ready())
      ? context.json({ ok: true })
      : context.json({ error: 'dependency_unavailable' }, 503)
  )

  if (config.mode === 'validation') {
    app.use('*', async (context) => context.json({ error: 'validation_only' }, 503))
  }

  const bearerSession: MiddlewareHandler<{ Variables: PushVariables }> = async (context, next) => {
    const ip = readClientIp(context, config.trustedProxyHops)
    if (!invalidBearerIps.available(ip)) return context.json({ error: 'rate_limited' }, 429)
    const bearer = readBearer(context.req.header('authorization'))
    if (!bearer) {
      invalidBearerIps.allow(ip)
      return context.json({ error: 'invalid_token' }, 401)
    }
    const session = await authAdmission.run(async () => {
      if (!invalidBearerIps.available(ip)) return null
      const result = await sessions.resolve(bearer)
      if (!result.ok) invalidBearerIps.allow(ip)
      return result
    })
    if (!session) {
      context.header('Retry-After', '1')
      return context.json({ error: 'busy' }, 503)
    }
    if (!session.ok) {
      return context.json(
        { error: session.reason === 'session_expired' ? 'session_expired' : 'invalid_token' },
        401
      )
    }
    if (!authenticatedHosts.allow(session.hostFingerprint))
      return context.json({ error: 'rate_limited' }, 429)
    context.set('hostFingerprint', session.hostFingerprint)
    await next()
    return
  }
  // `/v1/devices/*` matches `/v1/devices` itself; a second registration for the
  // bare path would run both middlewares twice on it.
  app.use('/v1/devices/*', limitAuthenticatedIp, bearerSession)
  app.use('/v1/send', limitAuthenticatedIp, bearerSession)

  app.post('/v1/host/challenge', limitUnauthenticatedIp, limitBody, async (context) => {
    const body = PushHostChallengeRequestSchema.safeParse(
      await context.req.json().catch(() => null)
    )
    if (!body.success) return context.json({ error: 'invalid_request' }, 400)
    const issued = await challenges.issue(body.data.hostPublicKeyB64)
    if (!issued) {
      observability.record('challenge_rejected')
      return context.json({ error: 'invalid_request' }, 400)
    }
    observability.record('challenge_issued')
    const { hostFingerprint: _bound, ...response } = issued
    return context.json(response)
  })

  app.post('/v1/host/session', limitUnauthenticatedIp, limitBody, async (context) => {
    const body = PushHostSessionRequestSchema.safeParse(await context.req.json().catch(() => null))
    if (!body.success) return context.json({ error: 'invalid_request' }, 400)
    const verification = await challenges.verify(body.data.challengeId, body.data.proofB64)
    if (!verification.ok) {
      observability.record('session_rejected')
      return context.json(
        {
          error: verification.reason === 'unknown_challenge' ? 'invalid_challenge' : 'invalid_proof'
        },
        401
      )
    }
    observability.record('session_issued')
    return context.json(await sessions.create(verification.hostFingerprint))
  })

  app.post('/v1/devices', limitBody, async (context) => {
    const body = PushDeviceRegistrationRequestSchema.safeParse(
      await context.req.json().catch(() => null)
    )
    if (!body.success) return context.json({ error: 'invalid_request' }, 400)
    const registered = await devices.upsert({
      hostFingerprint: context.get('hostFingerprint'),
      deviceId: body.data.deviceId,
      platform: body.data.platform,
      token: body.data.token,
      ...(body.data.apnsEnvironment === undefined
        ? {}
        : { apnsEnvironment: body.data.apnsEnvironment })
    })
    if (!registered.ok) {
      observability.record('device_rejected')
      return context.json({ error: 'too_many_devices' }, 409)
    }
    observability.record('device_registered')
    return context.json({ registrationId: registered.registrationId })
  })

  app.delete('/v1/devices/:registrationId', async (context) => {
    const deleted = await devices.deleteOwned(
      context.get('hostFingerprint'),
      context.req.param('registrationId')
    )
    if (!deleted) return context.json({ error: 'not_found' }, 404)
    observability.record('device_deleted')
    return context.body(null, 204)
  })

  app.get('/v1/devices', async (context) =>
    context.json({ devices: await devices.list(context.get('hostFingerprint')) })
  )

  app.post('/v1/send', limitBody, async (context) => {
    const body = PushSendRequestSchema.safeParse(await context.req.json().catch(() => null))
    if (!body.success) return context.json({ error: 'invalid_request' }, 400)
    const hostFingerprint = context.get('hostFingerprint')
    const owned = await devices.findOwned(hostFingerprint, body.data.registrationIds)
    const results: PushSendResult[] = []
    for (const registrationId of body.data.registrationIds) {
      const device = owned.get(registrationId)
      if (!device) {
        observability.record('send_error')
        results.push({ registrationId, status: 'error' })
        continue
      }
      if (device.dead) {
        observability.record('send_dead')
        results.push({ registrationId, status: 'dead' })
        continue
      }
      const reservation = await deliveryStore.accept(
        hostFingerprint,
        registrationId,
        body.data.notification
      )
      if (reservation !== 'queued') {
        observability.record(reservation === 'rate_limited' ? 'send_rate_limited' : 'send_error')
        results.push({ registrationId, status: reservation })
        continue
      }
      observability.record('send_queued')
      results.push({ registrationId, status: 'queued' })
    }
    return context.json({ results })
  })

  return {
    app,
    requestDrain,
    server: createAdaptorServer(app),
    challenges,
    sessions,
    devices,
    deliveryStore,
    unauthenticatedIps,
    worker,
    observability,
    ready,
    closeTransports: (): void => {
      if (apnsTransport && 'close' in apnsTransport) {
        ;(apnsTransport as { close: () => void }).close()
      }
    }
  }
}