feat(connector): connector service skeleton that checks auth-service tokens (#33723)

2026-10-30 16:47:34 +00:00 · 2026-10-26 01:20:53 -04:00
parent 4843f85f2a
commit 8cc8024504
21 changed files with 650 additions and 2 deletions
@@ -115,7 +115,7 @@ jobs:
    strategy:
      fail-fast: false
      matrix:
-        app: [relay, push]
+        app: [relay, push, connector]
    env:
      APP: ${{ matrix.app }}
      TRIVY_IMAGE: aquasec/trivy:0.76.1@sha256:790fd66d1697395d8693d1acaa1ea702406a20f109b6fc7bac8db623eca6c06c
@@ -138,6 +138,7 @@ jobs:
            ! command -v pnpm
            module=app
            if test "$1" = push; then module=push-server; fi
+            if test "$1" = connector; then module=connector-server; fi
            node --input-type=module -e "await import(\"./apps/$1/dist/$module.js\")"
          ' sh "$APP"

@@ -187,6 +188,7 @@ jobs:
    env:
      ORCA_BACKGROUND_LAUNCH: '1'
      ORCA_PUSH_TEST_DATABASE_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test
+      ORCA_CONNECTOR_TEST_DATABASE_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test
      ORCA_RELAY_TEST_POSTGRES_URL: postgres://relay_test:relay_test@127.0.0.1:5432/orca_relay_test
    steps:
      - uses: actions/checkout@v4
@@ -28,6 +28,10 @@ the validation required before rolling out the updated images and host pins.
 - `apps/push` and `packages/push-contract`: the mobile push gateway that holds
  the APNs key and sends to phones through APNs and FCM, and its wire contract.
  Deployment and operations are owned by the private `stablyai/orca-cloud` repository.
+- `apps/connector` and `packages/connector-contract`: the Slack connector service
+  (experimental). A signed-in Orca calls it with short-lived connector tokens
+  from the auth service, checked the way the relay checks relay tokens.
+  Deployed like push from the private repository.

 ## Mobile push gateway

@@ -0,0 +1,36 @@
+FROM node:24.22.1-alpine@sha256:feb76650b7195319268c59cb2d419c732d6c499b13bf5296f16ab59eb1f0a4f3 AS build
+WORKDIR /app
+ENV PATH="/opt/pnpm/node_modules/.bin:$PATH"
+COPY package.json pnpm-lock.yaml pnpm-workspace.yaml tsconfig.base.json ./
+RUN npm install --prefix /opt/pnpm --ignore-scripts --no-audit --no-fund "$(node -p 'require("./package.json").packageManager.split("+")[0]')" \
+  && npm audit signatures --prefix /opt/pnpm
+COPY packages/postgres-schema/package.json packages/postgres-schema/package.json
+COPY packages/cloud-database/package.json packages/cloud-database/package.json
+COPY apps/connector/package.json apps/connector/package.json
+RUN pnpm install --frozen-lockfile
+COPY packages/postgres-schema packages/postgres-schema
+COPY packages/cloud-database packages/cloud-database
+COPY apps/connector apps/connector
+RUN pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/cloud-database build \
+  && pnpm --filter @orca-cloud/connector build
+
+FROM node:24.22.1-alpine@sha256:feb76650b7195319268c59cb2d419c732d6c499b13bf5296f16ab59eb1f0a4f3 AS runtime
+ENV NODE_ENV=production
+ENV PORT=8080
+WORKDIR /app
+COPY --from=build /opt/pnpm /opt/pnpm
+ENV PATH="/opt/pnpm/node_modules/.bin:$PATH"
+COPY package.json pnpm-lock.yaml pnpm-workspace.yaml ./
+COPY packages/postgres-schema/package.json packages/postgres-schema/package.json
+COPY packages/cloud-database/package.json packages/cloud-database/package.json
+COPY apps/connector/package.json apps/connector/package.json
+COPY --from=build /app/packages/postgres-schema/dist packages/postgres-schema/dist
+COPY --from=build /app/packages/cloud-database/dist packages/cloud-database/dist
+COPY --from=build /app/apps/connector/dist apps/connector/dist
+RUN pnpm install --prod --frozen-lockfile --filter @orca-cloud/connector... \
+  && rm -rf /opt/pnpm /usr/local/lib/node_modules /opt/yarn-* /root/.cache /root/.local \
+  && rm -f /usr/local/bin/npm /usr/local/bin/npx /usr/local/bin/corepack \
+    /usr/local/bin/pnpm /usr/local/bin/pnpx /usr/local/bin/yarn /usr/local/bin/yarnpkg
+USER node
+EXPOSE 8080
+CMD ["node", "apps/connector/dist/index.js"]
@@ -0,0 +1,32 @@
+{
+  "name": "@orca-cloud/connector",
+  "private": true,
+  "version": "0.0.0",
+  "type": "module",
+  "main": "dist/index.js",
+  "scripts": {
+    "build": "pnpm clean && tsc -p tsconfig.build.json",
+    "clean": "node -e \"require('fs').rmSync('dist', { recursive: true, force: true })\"",
+    "dev": "tsx watch src/index.ts",
+    "lint": "tsc -p tsconfig.json --noEmit",
+    "pretest": "pnpm --filter @orca-cloud/postgres-schema build && pnpm --filter @orca-cloud/cloud-database build",
+    "start": "node dist/index.js",
+    "test": "vitest run",
+    "typecheck": "tsc -p tsconfig.json --noEmit"
+  },
+  "dependencies": {
+    "@hono/node-server": "^2.1.5",
+    "@orca-cloud/cloud-database": "workspace:*",
+    "hono": "^4.13.13",
+    "jose": "^6.2.13",
+    "zod": "^4.6.7"
+  },
+  "devDependencies": {
+    "@types/node": "^24.19.2",
+    "@types/pg": "^8.23.1",
+    "pg": "^8.23.1",
+    "tsx": "^4.23.15",
+    "typescript": "^7.0.2",
+    "vitest": "^5.0.5"
+  }
+}
@@ -0,0 +1,44 @@
+import { describe, expect, it } from 'vitest'
+import { loadConnectorConfig } from './config.js'
+
+const ENV = {
+  ORCA_CONNECTOR_PUBLIC_URL: 'https://connector.onorca.dev',
+  ORCA_CONNECTOR_AUTH_ISSUER: 'https://login.onorca.dev',
+  ORCA_CONNECTOR_JWKS_URL: 'https://login.onorca.dev/.well-known/jwks.json'
+}
+
+describe('connector config', () => {
+  it('fills defaults around the required settings', () => {
+    expect(loadConnectorConfig(ENV)).toEqual({
+      mode: 'active',
+      port: 8080,
+      publicUrl: 'https://connector.onorca.dev',
+      dataDir: './data/connector',
+      databasePoolMax: 10,
+      authIssuer: 'https://login.onorca.dev',
+      jwksUrl: 'https://login.onorca.dev/.well-known/jwks.json',
+      trustedProxyHops: 0
+    })
+  })
+
+  it('treats a blank database URL as absent', () => {
+    expect(loadConnectorConfig({ ...ENV, ORCA_CONNECTOR_DATABASE_URL: ' ' }).databaseUrl).toBe(
+      undefined
+    )
+  })
+
+  it.each([
+    ['a path on the public URL', { ORCA_CONNECTOR_PUBLIC_URL: 'https://connector.onorca.dev/x' }],
+    ['plain HTTP off loopback', { ORCA_CONNECTOR_AUTH_ISSUER: 'http://login.onorca.dev' }],
+    ['a missing key set URL', { ORCA_CONNECTOR_JWKS_URL: undefined }],
+    ['a plain HTTP key set URL', { ORCA_CONNECTOR_JWKS_URL: 'http://login.onorca.dev/jwks' }]
+  ])('rejects %s', (_name, override) => {
+    expect(() => loadConnectorConfig({ ...ENV, ...override })).toThrow()
+  })
+
+  it('allows plain HTTP on loopback for local development', () => {
+    expect(
+      loadConnectorConfig({ ...ENV, ORCA_CONNECTOR_PUBLIC_URL: 'http://127.0.0.1:8080' }).publicUrl
+    ).toBe('http://127.0.0.1:8080')
+  })
+})
@@ -0,0 +1,66 @@
+import { z } from 'zod'
+
+const OptionalTextSchema = z.string().min(1).optional()
+
+const EnvSchema = z.object({
+  ORCA_CONNECTOR_MODE: z.enum(['active', 'validation']).default('active'),
+  PORT: z.coerce.number().int().positive().default(8080),
+  ORCA_CONNECTOR_PUBLIC_URL: z.string().url(),
+  ORCA_CONNECTOR_DATABASE_URL: OptionalTextSchema,
+  ORCA_CONNECTOR_DATA_DIR: z.string().min(1).default('./data/connector'),
+  ORCA_CONNECTOR_DATABASE_POOL_MAX: z.coerce.number().int().positive().max(100).default(10),
+  ORCA_CONNECTOR_AUTH_ISSUER: z.string().url(),
+  ORCA_CONNECTOR_JWKS_URL: z.string().url(),
+  // As push: proxies appending to x-forwarded-for after the client (0 is Cloud Run alone).
+  ORCA_CONNECTOR_TRUSTED_PROXY_HOPS: z.coerce.number().int().nonnegative().max(8).default(0)
+})
+
+export type ConnectorConfig = {
+  mode: 'active' | 'validation'
+  port: number
+  publicUrl: string
+  databaseUrl?: string
+  dataDir: string
+  databasePoolMax: number
+  authIssuer: string
+  jwksUrl: string
+  trustedProxyHops: number
+}
+
+function secureUrl(value: string, name: string): URL {
+  const url = new URL(value)
+  const loopback = ['127.0.0.1', 'localhost', '[::1]'].includes(url.hostname)
+  if (url.protocol !== 'https:' && !(loopback && url.protocol === 'http:')) {
+    throw new Error(`${name} must use HTTPS outside loopback development`)
+  }
+  return url
+}
+
+function canonicalOrigin(value: string, name: string): string {
+  if (secureUrl(value, name).origin !== value) {
+    throw new Error(`${name} must be an origin`)
+  }
+  return value
+}
+
+export function loadConnectorConfig(env: NodeJS.ProcessEnv = process.env): ConnectorConfig {
+  const parsed = EnvSchema.parse(
+    Object.fromEntries(
+      Object.entries(env).map(([key, value]) => [key, value?.trim() === '' ? undefined : value])
+    )
+  )
+  return {
+    mode: parsed.ORCA_CONNECTOR_MODE,
+    port: parsed.PORT,
+    publicUrl: canonicalOrigin(parsed.ORCA_CONNECTOR_PUBLIC_URL, 'ORCA_CONNECTOR_PUBLIC_URL'),
+    ...(parsed.ORCA_CONNECTOR_DATABASE_URL === undefined
+      ? {}
+      : { databaseUrl: parsed.ORCA_CONNECTOR_DATABASE_URL }),
+    dataDir: parsed.ORCA_CONNECTOR_DATA_DIR,
+    databasePoolMax: parsed.ORCA_CONNECTOR_DATABASE_POOL_MAX,
+    authIssuer: canonicalOrigin(parsed.ORCA_CONNECTOR_AUTH_ISSUER, 'ORCA_CONNECTOR_AUTH_ISSUER'),
+    // The signing keys come from here, so plain HTTP would let anyone on the path mint tokens.
+    jwksUrl: secureUrl(parsed.ORCA_CONNECTOR_JWKS_URL, 'ORCA_CONNECTOR_JWKS_URL').href,
+    trustedProxyHops: parsed.ORCA_CONNECTOR_TRUSTED_PROXY_HOPS
+  }
+}
@@ -0,0 +1,22 @@
+import {
+  openCloudDatabase,
+  openInMemoryCloudDatabase,
+  type CloudDatabase,
+  type OpenCloudDatabaseInput
+} from '@orca-cloud/cloud-database'
+
+const CONNECTOR_SCHEMA_STATEMENTS: string[] = []
+
+export async function openConnectorDatabase(
+  input: Omit<OpenCloudDatabaseInput, 'service' | 'schemaStatements'>
+): Promise<CloudDatabase> {
+  return await openCloudDatabase({
+    ...input,
+    service: 'connector',
+    schemaStatements: CONNECTOR_SCHEMA_STATEMENTS
+  })
+}
+
+export async function openInMemoryConnectorDatabase(): Promise<CloudDatabase> {
+  return await openInMemoryCloudDatabase(CONNECTOR_SCHEMA_STATEMENTS)
+}
@@ -0,0 +1,59 @@
+import { randomUUID } from 'node:crypto'
+import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import pg from 'pg'
+import type { ConnectorConfig } from './config.js'
+import { openConnectorDatabase, openInMemoryConnectorDatabase } from './connector-database.js'
+import { createConnectorServer } from './connector-server.js'
+
+export const PUBLIC_URL = 'https://connector.onorca.dev'
+export const AUTH_ISSUER = 'https://login.onorca.dev'
+
+// Real Postgres when the CI service is present; each harness gets its own schema.
+const testDatabaseUrl = process.env.ORCA_CONNECTOR_TEST_DATABASE_URL
+
+export function testConnectorConfig(): ConnectorConfig {
+  return {
+    mode: 'active',
+    port: 0,
+    publicUrl: PUBLIC_URL,
+    dataDir: './data/connector-test',
+    databasePoolMax: 4,
+    authIssuer: AUTH_ISSUER,
+    jwksUrl: `${AUTH_ISSUER}/.well-known/jwks.json`,
+    trustedProxyHops: 0
+  }
+}
+
+const cleanups: (() => Promise<void>)[] = []
+export async function cleanupConnectorHarnesses(): Promise<void> {
+  await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()))
+}
+
+async function openTestDatabase(): Promise<CloudDatabase> {
+  if (!testDatabaseUrl) {
+    const database = await openInMemoryConnectorDatabase()
+    cleanups.push(() => database.close())
+    return database
+  }
+  const admin = new pg.Client({ connectionString: testDatabaseUrl })
+  await admin.connect()
+  const schema = `connector_${randomUUID().replaceAll('-', '')}`
+  await admin.query(`CREATE SCHEMA ${schema}`)
+  const url = new URL(testDatabaseUrl)
+  url.searchParams.set('options', `-c search_path=${schema}`)
+  const database = await openConnectorDatabase({ databaseUrl: url.toString(), dataDir: '', poolMax: 4 })
+  cleanups.push(async () => {
+    await database.close()
+    await admin.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`)
+    await admin.end()
+  })
+  return database
+}
+
+export async function createConnectorHarness(config: Partial<ConnectorConfig> = {}) {
+  const database = await openTestDatabase()
+  const runtime = createConnectorServer({ ...testConnectorConfig(), ...config }, database)
+  const request = (path: string, init?: RequestInit): Promise<Response> =>
+    Promise.resolve(runtime.app.request(path, init))
+  return { runtime, database, request }
+}
@@ -0,0 +1,21 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import {
+  cleanupConnectorHarnesses,
+  createConnectorHarness
+} from './connector-server-harness.test-fixture.js'
+
+afterEach(cleanupConnectorHarnesses)
+
+describe('connector server', () => {
+  it('reports health and database readiness', async () => {
+    const { request } = await createConnectorHarness()
+    expect(await (await request('/health')).json()).toEqual({ ok: true, mode: 'active' })
+    expect((await request('/ready')).status).toBe(200)
+  })
+
+  it('answers only health and readiness in validation mode', async () => {
+    const { request } = await createConnectorHarness({ mode: 'validation' })
+    expect((await request('/health')).status).toBe(200)
+    expect((await request('/v1/connector/poll', { method: 'POST' })).status).toBe(503)
+  })
+})
@@ -0,0 +1,54 @@
+import { createAdaptorServer } from '@hono/node-server'
+import type { CloudDatabase } from '@orca-cloud/cloud-database'
+import { Hono } from 'hono'
+import type { ConnectorConfig } from './config.js'
+import { limitByClientIp, RateLimiter } from './rate-limiter.js'
+import { RequestDrain } from './request-drain.js'
+
+const UNAUTHENTICATED_PER_MINUTE_PER_IP = 30
+
+export type ConnectorServerOptions = {
+  now?: () => number
+}
+
+export function createConnectorServer(
+  config: ConnectorConfig,
+  database: CloudDatabase,
+  options: ConnectorServerOptions = {}
+) {
+  const now = options.now ?? Date.now
+  const limitUnauthenticated = limitByClientIp(
+    new RateLimiter(UNAUTHENTICATED_PER_MINUTE_PER_IP, now),
+    config.trustedProxyHops
+  )
+
+  const app = new Hono()
+  const requestDrain = new RequestDrain()
+  app.use('*', requestDrain.middleware)
+  // Default logging can carry row detail; log only the error's name.
+  app.onError((error, context) => {
+    console.warn(
+      JSON.stringify({
+        event: 'orca_connector_request_failed',
+        error: error instanceof Error ? error.name : 'unknown'
+      })
+    )
+    return context.json({ error: 'internal' }, 500)
+  })
+
+  app.get('/health', (context) => context.json({ ok: true, mode: config.mode }))
+  app.get('/ready', limitUnauthenticated, async (context) =>
+    (await database.query('SELECT 1 AS ready').then(
+      () => true,
+      () => false
+    ))
+      ? context.json({ ok: true })
+      : context.json({ error: 'dependency_unavailable' }, 503)
+  )
+
+  if (config.mode === 'validation') {
+    app.use('*', async (context) => context.json({ error: 'validation_only' }, 503))
+  }
+
+  return { app, requestDrain, server: createAdaptorServer(app) }
+}
@@ -0,0 +1,68 @@
+import { createLocalJWKSet, exportJWK, generateKeyPair, SignJWT, type JWTPayload } from 'jose'
+import { beforeAll, describe, expect, it } from 'vitest'
+import { AUTH_ISSUER, testConnectorConfig } from './connector-server-harness.test-fixture.js'
+import {
+  connectorHostId,
+  createConnectorTokenVerifier,
+  readBearer,
+  type ConnectorTokenVerifier
+} from './connector-token-verifier.js'
+
+const MACHINE_ID = 'm4chine-1d-0123456789'
+const CLAIMS = { prof: 'profile-1', sid: 'family-1', mid: MACHINE_ID, purpose: 'connector-host' }
+
+let privateKey: Awaited<ReturnType<typeof generateKeyPair>>['privateKey']
+let verify: ConnectorTokenVerifier
+
+beforeAll(async () => {
+  const pair = await generateKeyPair('ES256')
+  privateKey = pair.privateKey
+  const jwk = { ...(await exportJWK(pair.publicKey)), kid: 'k1', alg: 'ES256' }
+  verify = createConnectorTokenVerifier(testConnectorConfig(), createLocalJWKSet({ keys: [jwk] }))
+})
+
+async function sign(
+  payload: JWTPayload = CLAIMS,
+  { audience = 'orca-connector', issuer = AUTH_ISSUER, expiresIn = '5m', key = privateKey } = {}
+): Promise<string> {
+  return await new SignJWT(payload)
+    .setProtectedHeader({ alg: 'ES256', kid: 'k1' })
+    .setIssuer(issuer)
+    .setAudience(audience)
+    .setSubject('user-1')
+    .setIssuedAt()
+    .setExpirationTime(expiresIn)
+    .sign(key)
+}
+
+describe('connector token verifier', () => {
+  it('accepts a connector token and names the machine by account and machine id', async () => {
+    const claims = await verify(await sign())
+    expect(claims).toMatchObject({ sub: 'user-1', mid: MACHINE_ID, sid: 'family-1' })
+    expect(claims && connectorHostId(claims)).toBe(`user-1:${MACHINE_ID}`)
+  })
+
+  it('refuses tokens meant for something else, expired, or signed by another key', async () => {
+    const otherKey = (await generateKeyPair('ES256')).privateKey
+    const refused = [
+      await sign(CLAIMS, { audience: 'orca-relay' }),
+      await sign({ ...CLAIMS, purpose: 'host-control' }),
+      await sign(CLAIMS, { issuer: 'https://elsewhere.example' }),
+      await sign(CLAIMS, { expiresIn: '-1s' }),
+      await sign(CLAIMS, { key: otherKey }),
+      await sign({ ...CLAIMS, mid: 'short' }),
+      await sign({ ...CLAIMS, mid: 'has:colon-0123456789' }),
+      'not-a-jwt'
+    ]
+    for (const token of refused) {
+      expect(await verify(token)).toBeNull()
+    }
+  })
+
+  it('reads only a single bearer token', () => {
+    expect(readBearer('Bearer abc')).toBe('abc')
+    expect(readBearer('Bearer a b')).toBeNull()
+    expect(readBearer('Basic abc')).toBeNull()
+    expect(readBearer(undefined)).toBeNull()
+  })
+})
@@ -0,0 +1,49 @@
+import { createRemoteJWKSet, jwtVerify, type JWTVerifyGetKey } from 'jose'
+import { z } from 'zod'
+import type { ConnectorConfig } from './config.js'
+
+export const CONNECTOR_TOKEN_AUDIENCE = 'orca-connector'
+
+// Minted by the auth service's connector token route, as relay tokens are by its relay route.
+const ClaimsSchema = z.object({
+  sub: z.string().min(1),
+  prof: z.string().min(1),
+  org: z.string().min(1).optional(),
+  sid: z.string().min(1),
+  // Random per-install id; its alphabet has no ':', so `${sub}:${mid}` names one machine of one account.
+  mid: z.string().regex(/^[A-Za-z0-9_-]{16,64}$/),
+  purpose: z.literal('connector-host'),
+  exp: z.number().int().positive()
+})
+
+export type ConnectorTokenClaims = z.infer<typeof ClaimsSchema>
+export type ConnectorTokenVerifier = (token: string) => Promise<ConnectorTokenClaims | null>
+
+export function connectorHostId(claims: Pick<ConnectorTokenClaims, 'sub' | 'mid'>): string {
+  return `${claims.sub}:${claims.mid}`
+}
+
+// jose's plain key cache, without the relay's stale-if-error fallback: machines already need the
+// auth service for a fresh token every 5 min.
+export function createConnectorTokenVerifier(
+  config: Pick<ConnectorConfig, 'authIssuer' | 'jwksUrl'>,
+  jwks: JWTVerifyGetKey = createRemoteJWKSet(new URL(config.jwksUrl))
+): ConnectorTokenVerifier {
+  return async (token) => {
+    try {
+      const verified = await jwtVerify(token, jwks, {
+        issuer: config.authIssuer,
+        audience: CONNECTOR_TOKEN_AUDIENCE,
+        algorithms: ['ES256']
+      })
+      return ClaimsSchema.parse(verified.payload)
+    } catch {
+      return null
+    }
+  }
+}
+
+export function readBearer(value: string | undefined): string | null {
+  const match = /^Bearer ([^\s]+)$/.exec(value ?? '')
+  return match?.[1] ?? null
+}
@@ -0,0 +1,41 @@
+import { loadConnectorConfig } from './config.js'
+import { openConnectorDatabase } from './connector-database.js'
+import { createConnectorServer } from './connector-server.js'
+
+const config = loadConnectorConfig()
+const database = await openConnectorDatabase({
+  ...(config.databaseUrl === undefined ? {} : { databaseUrl: config.databaseUrl }),
+  dataDir: config.dataDir,
+  poolMax: config.databasePoolMax,
+  applicationName: 'orca-connector',
+  readOnly: config.mode === 'validation'
+})
+const runtime = createConnectorServer(config, database)
+
+runtime.server.listen(config.port, () => {
+  console.log(`[orca-connector] listening on ${config.publicUrl} (port ${config.port})`)
+})
+
+let stopping = false
+const shutdown = (): void => {
+  if (stopping) {
+    return
+  }
+  stopping = true
+  // Cloud Run sends SIGKILL after ten seconds; leave time for explicit cleanup.
+  const deadline = setTimeout(() => process.exit(1), 9_000)
+  deadline.unref()
+  const requests = runtime.requestDrain.begin()
+  const connections = new Promise<void>((resolve) => runtime.server.close(() => resolve()))
+  void Promise.all([requests, connections])
+    .then(async () => {
+      await database.close()
+      clearTimeout(deadline)
+    })
+    .catch(() => {
+      console.warn(JSON.stringify({ event: 'orca_connector_shutdown_failed' }))
+      process.exitCode = 1
+    })
+}
+process.once('SIGTERM', shutdown)
+process.once('SIGINT', shutdown)
@@ -0,0 +1,13 @@
+import { describe, expect, it } from 'vitest'
+import { RateLimiter } from './rate-limiter.js'
+
+describe('rate limiter', () => {
+  it('keeps unused allowance when the clock steps back', () => {
+    let now = 120_000
+    const limiter = new RateLimiter(2, () => now)
+    expect(limiter.allow('a')).toBe(true)
+    now = 60_000
+    expect(limiter.allow('a')).toBe(true)
+    expect(limiter.allow('a')).toBe(false)
+  })
+})
@@ -0,0 +1,53 @@
+import type { Context, MiddlewareHandler } from 'hono'
+
+type Bucket = { tokens: number; updatedAt: number }
+
+// Per-instance token buckets keyed by client address or account; capacity scales with instances.
+export class RateLimiter {
+  private readonly buckets = new Map<string, Bucket>()
+
+  constructor(
+    private readonly capacityPerMinute: number,
+    private readonly now: () => number,
+    private readonly maxKeys = 10_000
+  ) {}
+
+  allow(key: string): boolean {
+    const now = this.now()
+    const bucket = this.buckets.get(key)
+    const refilled = bucket
+      ? Math.min(
+          this.capacityPerMinute,
+          bucket.tokens + (Math.max(0, now - bucket.updatedAt) * this.capacityPerMinute) / 60_000
+        )
+      : this.capacityPerMinute
+    // Re-inserting keeps the map in recency order, so the oldest key is evicted first.
+    this.buckets.delete(key)
+    this.buckets.set(key, { tokens: refilled >= 1 ? refilled - 1 : refilled, updatedAt: now })
+    if (this.buckets.size > this.maxKeys) {
+      const oldest = this.buckets.keys().next().value
+      if (oldest !== undefined) {
+        this.buckets.delete(oldest)
+      }
+    }
+    return refilled >= 1
+  }
+}
+
+// Read from the right, as push does: Cloud Run appends the peer, earlier entries can be forged.
+export function readClientIp(context: Context, trustedProxyHops: number): string {
+  const hops = (context.req.header('x-forwarded-for') ?? '')
+    .split(',')
+    .map((hop) => hop.trim())
+    .filter((hop) => hop.length > 0)
+  return hops[hops.length - 1 - trustedProxyHops] ?? 'unknown'
+}
+
+export function limitByClientIp(limiter: RateLimiter, trustedProxyHops: number): MiddlewareHandler {
+  return async (context, next) => {
+    if (!limiter.allow(readClientIp(context, trustedProxyHops))) {
+      return context.json({ error: 'rate_limited' }, 429)
+    }
+    await next()
+  }
+}
@@ -0,0 +1,27 @@
+import type { MiddlewareHandler } from 'hono'
+
+// Refuses new requests once shutdown begins and resolves when the in-flight ones finish.
+export class RequestDrain {
+  private draining = false
+  private active = 0
+  private idle: (() => void) | null = null
+
+  readonly middleware: MiddlewareHandler = async (context, next) => {
+    if (this.draining) {
+      return context.json({ error: 'shutting_down' }, 503)
+    }
+    this.active++
+    try {
+      await next()
+    } finally {
+      if (--this.active === 0) {
+        this.idle?.()
+      }
+    }
+  }
+
+  begin(): Promise<void> {
+    this.draining = true
+    return this.active === 0 ? Promise.resolve() : new Promise((resolve) => (this.idle = resolve))
+  }
+}
@@ -0,0 +1,10 @@
+{
+  "extends": "./tsconfig.json",
+  "compilerOptions": {
+    "declaration": true,
+    "noEmit": false,
+    "outDir": "dist",
+    "rootDir": "src"
+  },
+  "exclude": ["src/**/*.test.ts", "src/**/*.test-fixture.ts"]
+}
@@ -0,0 +1,5 @@
+{
+  "extends": "../../tsconfig.base.json",
+  "compilerOptions": { "noEmit": true },
+  "include": ["src/**/*.ts"]
+}
@@ -0,0 +1,5 @@
+import { defineConfig } from 'vitest/config'
+
+export default defineConfig({
+  test: { name: 'connector', include: ['src/**/*.test.ts'], testTimeout: 15_000, hookTimeout: 15_000 }
+})
@@ -185,6 +185,43 @@ importers:
        specifier: ^5.0.5
        version: 5.0.5(@types/node@24.19.2)(vite@8.3.4(@types/node@24.19.2)(esbuild@0.28.2)(tsx@4.23.15))

+  apps/connector:
+    dependencies:
+      '@hono/node-server':
+        specifier: ^2.1.5
+        version: 2.1.5(hono@4.13.13)
+      '@orca-cloud/cloud-database':
+        specifier: workspace:*
+        version: link:../../packages/cloud-database
+      hono:
+        specifier: ^4.13.13
+        version: 4.13.13
+      jose:
+        specifier: ^6.2.13
+        version: 6.2.13
+      zod:
+        specifier: ^4.6.7
+        version: 4.6.7
+    devDependencies:
+      '@types/node':
+        specifier: ^24.19.2
+        version: 24.19.2
+      '@types/pg':
+        specifier: ^8.23.1
+        version: 8.23.1
+      pg:
+        specifier: ^8.23.1
+        version: 8.23.1
+      tsx:
+        specifier: ^4.23.15
+        version: 4.23.15
+      typescript:
+        specifier: ^7.0.2
+        version: 7.0.2
+      vitest:
+        specifier: ^5.0.5
+        version: 5.0.5(@types/node@24.19.2)(vite@8.3.4(@types/node@24.19.2)(esbuild@0.28.2)(tsx@4.23.15))
+
  apps/push:
    dependencies:
      '@hono/node-server':
@@ -58,7 +58,7 @@ const NON_UNIT_TEST_RUNNERS = [
      "'src/**/*.test.ts', 'src/**/*.test.tsx', 'scripts/**/*.test.ts'"
    ]
  },
-  ...['push', 'relay'].map((app) => ({
+  ...['connector', 'push', 'relay'].map((app) => ({
    name: `cloud ${app}`,
    include: [`cloud/apps/${app}/src/**/*.test.ts`],
    sources: [`cloud/apps/${app}/vitest.config.ts`],