diff --git a/.github/workflows/cloud-verify.yml b/.github/workflows/cloud-verify.yml index 097a2118827..7a12a4d3184 100644 --- a/.github/workflows/cloud-verify.yml +++ b/.github/workflows/cloud-verify.yml @@ -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 diff --git a/cloud/README.md b/cloud/README.md index f1114655842..67caaaefb41 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -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 diff --git a/cloud/apps/connector/Dockerfile b/cloud/apps/connector/Dockerfile new file mode 100644 index 00000000000..3bb28d5c02e --- /dev/null +++ b/cloud/apps/connector/Dockerfile @@ -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"] diff --git a/cloud/apps/connector/package.json b/cloud/apps/connector/package.json new file mode 100644 index 00000000000..3dbe175e985 --- /dev/null +++ b/cloud/apps/connector/package.json @@ -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" + } +} diff --git a/cloud/apps/connector/src/config.test.ts b/cloud/apps/connector/src/config.test.ts new file mode 100644 index 00000000000..bb351833f02 --- /dev/null +++ b/cloud/apps/connector/src/config.test.ts @@ -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') + }) +}) diff --git a/cloud/apps/connector/src/config.ts b/cloud/apps/connector/src/config.ts new file mode 100644 index 00000000000..c500ba62e65 --- /dev/null +++ b/cloud/apps/connector/src/config.ts @@ -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 + } +} diff --git a/cloud/apps/connector/src/connector-database.ts b/cloud/apps/connector/src/connector-database.ts new file mode 100644 index 00000000000..f2976dfb863 --- /dev/null +++ b/cloud/apps/connector/src/connector-database.ts @@ -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 +): Promise { + return await openCloudDatabase({ + ...input, + service: 'connector', + schemaStatements: CONNECTOR_SCHEMA_STATEMENTS + }) +} + +export async function openInMemoryConnectorDatabase(): Promise { + return await openInMemoryCloudDatabase(CONNECTOR_SCHEMA_STATEMENTS) +} diff --git a/cloud/apps/connector/src/connector-server-harness.test-fixture.ts b/cloud/apps/connector/src/connector-server-harness.test-fixture.ts new file mode 100644 index 00000000000..6e8bd24567d --- /dev/null +++ b/cloud/apps/connector/src/connector-server-harness.test-fixture.ts @@ -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)[] = [] +export async function cleanupConnectorHarnesses(): Promise { + await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) +} + +async function openTestDatabase(): Promise { + 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 = {}) { + const database = await openTestDatabase() + const runtime = createConnectorServer({ ...testConnectorConfig(), ...config }, database) + const request = (path: string, init?: RequestInit): Promise => + Promise.resolve(runtime.app.request(path, init)) + return { runtime, database, request } +} diff --git a/cloud/apps/connector/src/connector-server.test.ts b/cloud/apps/connector/src/connector-server.test.ts new file mode 100644 index 00000000000..dead4f93299 --- /dev/null +++ b/cloud/apps/connector/src/connector-server.test.ts @@ -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) + }) +}) diff --git a/cloud/apps/connector/src/connector-server.ts b/cloud/apps/connector/src/connector-server.ts new file mode 100644 index 00000000000..0ca2238045c --- /dev/null +++ b/cloud/apps/connector/src/connector-server.ts @@ -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) } +} diff --git a/cloud/apps/connector/src/connector-token-verifier.test.ts b/cloud/apps/connector/src/connector-token-verifier.test.ts new file mode 100644 index 00000000000..ebd5abba7b0 --- /dev/null +++ b/cloud/apps/connector/src/connector-token-verifier.test.ts @@ -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>['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 { + 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() + }) +}) diff --git a/cloud/apps/connector/src/connector-token-verifier.ts b/cloud/apps/connector/src/connector-token-verifier.ts new file mode 100644 index 00000000000..6b01603fa97 --- /dev/null +++ b/cloud/apps/connector/src/connector-token-verifier.ts @@ -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 +export type ConnectorTokenVerifier = (token: string) => Promise + +export function connectorHostId(claims: Pick): 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, + 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 +} diff --git a/cloud/apps/connector/src/index.ts b/cloud/apps/connector/src/index.ts new file mode 100644 index 00000000000..12f921fb41f --- /dev/null +++ b/cloud/apps/connector/src/index.ts @@ -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((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) diff --git a/cloud/apps/connector/src/rate-limiter.test.ts b/cloud/apps/connector/src/rate-limiter.test.ts new file mode 100644 index 00000000000..0338638a0a2 --- /dev/null +++ b/cloud/apps/connector/src/rate-limiter.test.ts @@ -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) + }) +}) diff --git a/cloud/apps/connector/src/rate-limiter.ts b/cloud/apps/connector/src/rate-limiter.ts new file mode 100644 index 00000000000..a03c09f3db1 --- /dev/null +++ b/cloud/apps/connector/src/rate-limiter.ts @@ -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() + + 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() + } +} diff --git a/cloud/apps/connector/src/request-drain.ts b/cloud/apps/connector/src/request-drain.ts new file mode 100644 index 00000000000..13ab0f7dab0 --- /dev/null +++ b/cloud/apps/connector/src/request-drain.ts @@ -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 { + this.draining = true + return this.active === 0 ? Promise.resolve() : new Promise((resolve) => (this.idle = resolve)) + } +} diff --git a/cloud/apps/connector/tsconfig.build.json b/cloud/apps/connector/tsconfig.build.json new file mode 100644 index 00000000000..5e71eb0f951 --- /dev/null +++ b/cloud/apps/connector/tsconfig.build.json @@ -0,0 +1,10 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "declaration": true, + "noEmit": false, + "outDir": "dist", + "rootDir": "src" + }, + "exclude": ["src/**/*.test.ts", "src/**/*.test-fixture.ts"] +} diff --git a/cloud/apps/connector/tsconfig.json b/cloud/apps/connector/tsconfig.json new file mode 100644 index 00000000000..a552e34dbe9 --- /dev/null +++ b/cloud/apps/connector/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "noEmit": true }, + "include": ["src/**/*.ts"] +} diff --git a/cloud/apps/connector/vitest.config.ts b/cloud/apps/connector/vitest.config.ts new file mode 100644 index 00000000000..0ddfa1d2604 --- /dev/null +++ b/cloud/apps/connector/vitest.config.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 } +}) diff --git a/cloud/pnpm-lock.yaml b/cloud/pnpm-lock.yaml index 8eeafd28169..cb1b20c4483 100644 --- a/cloud/pnpm-lock.yaml +++ b/cloud/pnpm-lock.yaml @@ -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': diff --git a/config/scripts/check-test-file-runners.mjs b/config/scripts/check-test-file-runners.mjs index f4689ef484f..68f22769568 100644 --- a/config/scripts/check-test-file-runners.mjs +++ b/config/scripts/check-test-file-runners.mjs @@ -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`],