diff --git a/src/yhub-server/README.md b/src/yhub-server/README.md index 12309345e..5e3727a7d 100644 --- a/src/yhub-server/README.md +++ b/src/yhub-server/README.md @@ -7,13 +7,20 @@ server that synchronizes Yjs documents between editors in real time. It is not a fork of yhub — it is a thin wrapper, written in TypeScript under `src/` and compiled to `dist/` with `yarn build` (see "Container image" below): -- `src/server.ts` — configuration, the auth plugin, and the custom REST endpoints, +- `src/server.ts` — wires the pieces below into one yhub instance: the auth + plugin, the legacy-store seed, the persistence plugins and the worker events, +- `src/config.ts` — every environment variable the server reads, parsed and + validated at startup, +- `src/backend.ts` — the calls this server makes to the Docs Django backend, and + the keys both ends verify each other with, +- `src/api.ts` — the custom REST endpoints mounted under `/collaboration/`, - `src/permissions.ts` — Docs' access policy as yhub permission objects (see "Access control" below), - `src/migration.ts` — everything that reads the legacy Django/S3 document store (both migrations described below), - `src/env.ts` — the `*_FILE` secret indirection shared by the rest, -- `src/yhub.ts` — the couple of `@y/hub` types its package index does not export. +- `src/yhub.ts` — the couple of `@y/hub` types its package index does not export, + plus the `AppAuthInfo` identity shape `server.ts` and `api.ts` share. `src/server.ts`: @@ -393,9 +400,16 @@ from an `mc` container on the stack's network. end with `@aws-sdk/client-s3` and the yhub instance faked and `@y/y` real, so the content maps under assertion are the real ones — plus the module-load validation (`SOFT_MIGRATION`, the `LEGACY_S3_*` checks, addressing style), -- `__tests__/server.spec.ts` covers the boot contract: the environment - `src/server.ts` refuses (`YHUB_ROLE`, the numeric knobs, a half-configured - bucket) and the configuration it hands `createYHub` when it accepts, +- `__tests__/config.spec.ts` loads `src/config.ts` against a stubbed environment + and checks what it parses and what it refuses (`YHUB_ROLE`, the numeric knobs, + the origin allowlist, `PORT`), +- `__tests__/backend.spec.ts` covers `src/backend.ts` with `global.fetch` spied: + the headers `backendFetch` sends, the status it tags on a failure, and the + public-only JWK derived from `YHUB_JWT_PRIVATE_KEY`, +- `__tests__/server.spec.ts` covers the boot contract of the composition root: + the S3 configuration `src/server.ts` refuses (a half-configured bucket) and the + configuration it hands `createYHub` when it accepts (the role split, the + persistence plugins, the endpoint array), - `__tests__/permissions.spec.ts` asks the policy tables the same questions yhub's gates ask, through the real `@y/hub/permissions` pipeline (a subpath export, no redis/postgres pulled in). diff --git a/src/yhub-server/__tests__/backend.spec.ts b/src/yhub-server/__tests__/backend.spec.ts new file mode 100644 index 000000000..fcfed8c75 --- /dev/null +++ b/src/yhub-server/__tests__/backend.spec.ts @@ -0,0 +1,150 @@ +// backend.ts — the calls this server makes to the Docs Django backend, and the +// public half of the RS256 key it signs them with. +// +// `@y/hub` is faked for its logger; `global.fetch` is a spy. `./config.ts` runs +// for real — it reads the same stubbed env. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { exportPKCS8, generateKeyPair } from 'jose'; + +vi.mock('@y/hub', () => { + const logger = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + child: () => logger, + }; + return { logger }; +}); + +const BASE_ENV = { + COLLABORATION_BACKEND_BASE_URL: 'https://backend.example.test', + Y_PROVIDER_API_KEY: 'the-provider-key', +}; + +const load = () => import('../src/backend.js'); + +// A PKCS8 PEM for a fresh RS256 key, for the tests that need a configured signer. +const freshPrivateKeyPem = async (): Promise => { + const { privateKey } = await generateKeyPair('RS256', { extractable: true }); + return exportPKCS8(privateKey); +}; + +const jsonOk = (body: unknown) => ({ + ok: true, + status: 200, + json: async () => body, +}); +const httpError = (status: number) => ({ ok: false, status }); + +let fetchMock: ReturnType; + +beforeEach(() => { + vi.unstubAllEnvs(); + vi.resetModules(); + for (const [key, value] of Object.entries(BASE_ENV)) vi.stubEnv(key, value); + vi.stubEnv('YHUB_JWT_PRIVATE_KEY', ''); + fetchMock = vi.fn(); + vi.stubGlobal('fetch', fetchMock); +}); + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe('backendFetch', () => { + it('sends the provider key and forwards the caller’s cookie and origin', async () => { + fetchMock.mockResolvedValue(jsonOk({ id: 'u1' })); + const { backendFetch } = await load(); + + const body = await backendFetch('/api/v1.0/users/me/', { + cookie: 'sessionid=abc', + origin: 'https://app.example', + }); + + expect(body).toEqual({ id: 'u1' }); + const [url, init] = fetchMock.mock.calls[0]; + expect(url).toBe('https://backend.example.test/api/v1.0/users/me/'); + expect(init.headers).toMatchObject({ + cookie: 'sessionid=abc', + origin: 'https://app.example', + 'X-Y-Provider-Key': 'the-provider-key', + }); + }); + + it('omits cookie and origin entirely when the caller has neither', async () => { + fetchMock.mockResolvedValue(jsonOk({})); + const { backendFetch } = await load(); + + await backendFetch('/api/v1.0/documents/x/', {}); + + const [, init] = fetchMock.mock.calls[0]; + expect(init.headers).not.toHaveProperty('cookie'); + expect(init.headers).not.toHaveProperty('origin'); + expect(init.headers['X-Y-Provider-Key']).toBe('the-provider-key'); + }); + + it('rejects with the HTTP status tagged on the error', async () => { + fetchMock.mockResolvedValue(httpError(403)); + const { backendFetch } = await load(); + + await expect( + backendFetch('/api/v1.0/documents/x/', {}), + ).rejects.toMatchObject({ status: 403 }); + }); +}); + +describe('the backend signing key', () => { + it('publishes no JWK when YHUB_JWT_PRIVATE_KEY is unset', async () => { + const { backendPublicJwk } = await load(); + expect(backendPublicJwk).toBeNull(); + }); + + it('publishes only the public half, with a stable kid, when a key is set', async () => { + vi.stubEnv('YHUB_JWT_PRIVATE_KEY', await freshPrivateKeyPem()); + const { backendPublicJwk } = await load(); + + expect(backendPublicJwk).toMatchObject({ + kty: 'RSA', + alg: 'RS256', + use: 'sig', + }); + expect(backendPublicJwk).toHaveProperty('kid'); + // no private components leak into the published key + expect(backendPublicJwk).not.toHaveProperty('d'); + expect(backendPublicJwk).not.toHaveProperty('p'); + expect(backendPublicJwk).not.toHaveProperty('q'); + }); +}); + +describe('touchDocument', () => { + it('does nothing, and never calls the backend, without a signing key', async () => { + const { touchDocument } = await load(); + await touchDocument('doc-1'); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it('POSTs a bearer-token notification when a key is configured', async () => { + vi.stubEnv('YHUB_JWT_PRIVATE_KEY', await freshPrivateKeyPem()); + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + const { touchDocument } = await load(); + + await touchDocument('doc-1'); + + const [url, init] = fetchMock.mock.calls[0]; + expect(url).toBe( + 'https://backend.example.test/api/v1.0/documents/doc-1/content-updated/', + ); + expect(init.method).toBe('POST'); + expect(init.headers.authorization).toMatch(/^Bearer /); + }); + + it('swallows a backend failure rather than throwing into the worker', async () => { + vi.stubEnv('YHUB_JWT_PRIVATE_KEY', await freshPrivateKeyPem()); + fetchMock.mockRejectedValue(new Error('network down')); + const { touchDocument } = await load(); + + await expect(touchDocument('doc-1')).resolves.toBeUndefined(); + }); +}); diff --git a/src/yhub-server/__tests__/config.spec.ts b/src/yhub-server/__tests__/config.spec.ts new file mode 100644 index 000000000..9717acccc --- /dev/null +++ b/src/yhub-server/__tests__/config.spec.ts @@ -0,0 +1,121 @@ +// config.ts — every environment variable the server reads, parsed and validated +// at import time. Tested directly: no yhub, no stores, no mocks — the env goes +// in, and the exported constants (or the startup throw) come out. + +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +// Only what `config.ts` looks at; REDIS/POSTGRES are passed straight through to +// yhub, which is what validates them, so they are not needed here. +const BASE_ENV = { + COLLABORATION_SERVER_ORIGIN: 'http://localhost:3000', +}; + +const load = () => import('../src/config.js'); + +beforeEach(() => { + vi.unstubAllEnvs(); + vi.resetModules(); + for (const [key, value] of Object.entries(BASE_ENV)) vi.stubEnv(key, value); + // vars a previous test may have set that must not leak in as defaults + for (const key of [ + 'YHUB_ROLE', + 'YHUB_ORG', + 'REDIS_PREFIX', + 'YHUB_TASK_CONCURRENCY', + 'YHUB_TASK_DEBOUNCE_MS', + 'YHUB_MIN_MESSAGE_LIFETIME_MS', + 'PORT', + ]) { + vi.stubEnv(key, ''); + } +}); + +describe('YHUB_ROLE', () => { + it('defaults to running both halves', async () => { + const { ROLE, RUNS_SERVER, RUNS_WORKER } = await load(); + expect(ROLE).toBe('all'); + expect(RUNS_SERVER).toBe(true); + expect(RUNS_WORKER).toBe(true); + }); + + it('a server role runs the server half only', async () => { + vi.stubEnv('YHUB_ROLE', 'server'); + const { RUNS_SERVER, RUNS_WORKER } = await load(); + expect(RUNS_SERVER).toBe(true); + expect(RUNS_WORKER).toBe(false); + }); + + it('a worker role runs the worker half only', async () => { + vi.stubEnv('YHUB_ROLE', 'worker'); + const { RUNS_SERVER, RUNS_WORKER } = await load(); + expect(RUNS_SERVER).toBe(false); + expect(RUNS_WORKER).toBe(true); + }); + + it('refuses an unknown role', async () => { + vi.stubEnv('YHUB_ROLE', 'bogus'); + await expect(load()).rejects.toThrow(/YHUB_ROLE must be one of/); + }); +}); + +describe('the numeric tuning knobs', () => { + it('refuses a non-integer concurrency', async () => { + vi.stubEnv('YHUB_TASK_CONCURRENCY', 'abc'); + await expect(load()).rejects.toThrow( + /YHUB_TASK_CONCURRENCY must be an integer >= 1/, + ); + }); + + it('refuses a concurrency below one', async () => { + vi.stubEnv('YHUB_TASK_CONCURRENCY', '0'); + await expect(load()).rejects.toThrow(/YHUB_TASK_CONCURRENCY/); + }); + + it('accepts a debounce of zero but not of minus one', async () => { + vi.stubEnv('YHUB_TASK_DEBOUNCE_MS', '0'); + await expect(load()).resolves.toBeDefined(); + + vi.resetModules(); + vi.stubEnv('YHUB_TASK_DEBOUNCE_MS', '-1'); + await expect(load()).rejects.toThrow(/YHUB_TASK_DEBOUNCE_MS/); + }); + + it('treats a blank variable as the default', async () => { + vi.stubEnv('YHUB_TASK_CONCURRENCY', ''); + const { TASK_CONCURRENCY } = await load(); + expect(TASK_CONCURRENCY).toBe(5); + }); + + it('carries Docs’ stream-timing defaults', async () => { + const { TASK_DEBOUNCE_MS, MIN_MESSAGE_LIFETIME_MS } = await load(); + expect(TASK_DEBOUNCE_MS).toBe(10000); + expect(MIN_MESSAGE_LIFETIME_MS).toBe(60000); + }); +}); + +describe('the origin allowlist', () => { + it('defaults to localhost:3000', async () => { + const { allowedOrigins } = await load(); + expect(allowedOrigins).toEqual(['http://localhost:3000']); + }); + + it('splits on commas', async () => { + vi.stubEnv( + 'COLLABORATION_SERVER_ORIGIN', + 'https://a.example,https://b.example', + ); + const { allowedOrigins } = await load(); + expect(allowedOrigins).toEqual(['https://a.example', 'https://b.example']); + }); +}); + +describe('PORT', () => { + it('defaults to 3002', async () => { + expect((await load()).PORT).toBe(3002); + }); + + it('is honoured when set', async () => { + vi.stubEnv('PORT', '4000'); + expect((await load()).PORT).toBe(4000); + }); +}); diff --git a/src/yhub-server/__tests__/server.spec.ts b/src/yhub-server/__tests__/server.spec.ts index fa2e66689..8930a1f7f 100644 --- a/src/yhub-server/__tests__/server.spec.ts +++ b/src/yhub-server/__tests__/server.spec.ts @@ -1,10 +1,11 @@ -// server.ts — configuration, the auth plugin, the custom REST endpoints. +// server.ts — the composition root: it builds the persistence plugins and wires +// the auth plugin and the endpoint array into one `await createYHub(...)` call. +// It has no exports, so what a unit test holds onto is that boot contract — the +// S3 configuration it refuses, and the config object it hands yhub. // -// The module has no exports and ends on `await createYHub(...)`, so what a unit -// test can hold onto is its boot contract: the environment it refuses, and the -// configuration it hands yhub when it accepts. `@y/hub`, its S3 plugin and -// `./migration.ts` are faked; the config object passed to `createYHub` and the -// args passed to `S3PersistenceV1` are captured. +// `@y/hub`, its S3 plugin and `./migration.ts` are faked; the config passed to +// `createYHub` and the args passed to `S3PersistenceV1` are captured. Env +// parsing and validation is `config.ts`'s concern — see config.spec.ts. import { beforeEach, describe, expect, it, vi } from 'vitest'; @@ -108,38 +109,16 @@ describe('a valid, minimal environment', () => { expect(config.persistence).toEqual([]); }); - it('passes the stream tuning through with Docs’ defaults', async () => { + it('threads config.ts’ stream tuning and concurrency into the yhub config', async () => { await boot(); - const { redis } = createYHub.lastConfig; + const { redis, worker } = createYHub.lastConfig; expect(redis.taskDebounce).toBe(10000); expect(redis.minMessageLifetime).toBe(60000); - }); - - it('splits the origin allowlist on commas', async () => { - vi.stubEnv( - 'COLLABORATION_SERVER_ORIGIN', - 'https://a.example,https://b.example', - ); - await boot(); - expect(createYHub.lastConfig.server.cors.origin).toEqual([ - 'https://a.example', - 'https://b.example', - ]); - }); - - it('honours PORT', async () => { - vi.stubEnv('PORT', '4000'); - await boot(); - expect(createYHub.lastConfig.server.port).toBe(4000); + expect(worker.taskConcurrency).toBe(5); }); }); -describe('YHUB_ROLE', () => { - it('refuses an unknown role at boot', async () => { - vi.stubEnv('YHUB_ROLE', 'bogus'); - await expect(boot()).rejects.toThrow(/YHUB_ROLE must be one of/); - }); - +describe('the process role', () => { it('a server role binds the port and claims no task', async () => { vi.stubEnv('YHUB_ROLE', 'server'); await boot(); @@ -153,34 +132,14 @@ describe('YHUB_ROLE', () => { expect(createYHub.lastConfig.server).toBeNull(); expect(createYHub.lastConfig.worker).not.toBeNull(); }); -}); -describe('the numeric tuning knobs', () => { - it('refuses a non-integer concurrency', async () => { - vi.stubEnv('YHUB_TASK_CONCURRENCY', 'abc'); - await expect(boot()).rejects.toThrow( - /YHUB_TASK_CONCURRENCY must be an integer >= 1/, - ); - }); - - it('refuses a concurrency below one', async () => { - vi.stubEnv('YHUB_TASK_CONCURRENCY', '0'); - await expect(boot()).rejects.toThrow(/YHUB_TASK_CONCURRENCY/); - }); - - it('accepts a debounce of zero but not of minus one', async () => { - vi.stubEnv('YHUB_TASK_DEBOUNCE_MS', '0'); - await expect(boot()).resolves.toBeDefined(); - - vi.resetModules(); - vi.stubEnv('YHUB_TASK_DEBOUNCE_MS', '-1'); - await expect(boot()).rejects.toThrow(/YHUB_TASK_DEBOUNCE_MS/); - }); - - it('treats a blank variable as the default', async () => { - vi.stubEnv('YHUB_TASK_CONCURRENCY', ''); + it('the endpoint array is handed to the server half', async () => { await boot(); - expect(createYHub.lastConfig.worker.taskConcurrency).toBe(5); + // createApiEndpoint is faked to `{ name, opts }` — assert the routes are wired + const names = createYHub.lastConfig.server.api.map((e: { name: string }) => e.name); + expect(names).toEqual( + expect.arrayContaining(['ping', 'ready', 'jwks', 'create-ydoc']), + ); }); }); diff --git a/src/yhub-server/src/api.ts b/src/yhub-server/src/api.ts new file mode 100644 index 000000000..af968f20f --- /dev/null +++ b/src/yhub-server/src/api.ts @@ -0,0 +1,464 @@ +/** + * The custom REST endpoints, mounted under `/collaboration/` alongside yhub's + * built-ins and the websocket route. Every handler reaches yhub through + * `req.yhub`, so this file holds no reference to the instance `server.ts` + * creates — it only has to be handed to `createYHub`'s `server.api`. + * + * Access to each route is settled by the permission tables in `permissions.ts` + * (yhub runs them before a handler is called); the branch/org/uuid checks here + * are the second fence the admin token needs, since it is not scoped to `main` + * or to this org by `authorize`. + */ +import { + checkPermissions, + createApiEndpoint, + createDocumentPermissions, + logger, +} from '@y/hub'; + +import { backendPublicJwk } from './backend.js'; +import { + EMPTY_UPDATE_MAX_BYTES, + EMPTY_YDOC, + MAX_CREATE_BYTES, + ORG, + READINESS_TIMEOUT_MS, + UUID4, +} from './config.js'; +import { SOFT_MIGRATION, fullMigrate } from './migration.js'; +import type { AppAuthInfo, DocRef, YHub } from './yhub.js'; + +// Mimic the old y-provider REST responses (JSON, not yhub's lib0-any +// encoding) so the Django caller keeps its historical contract. +const jsonResponse = (status: number, body: unknown): Response => + new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json' }, + }); + +// Does the room hold anything? Covers persisted rows and the messages still on +// the stream, which is what makes it an answer about the content rather than +// about the storage. +const hasContent = async (yhubApi: YHub, docRef: DocRef): Promise => { + const { gcDoc } = await yhubApi.getDoc( + docRef, + { gc: true, nongc: false }, + { gcOnMerge: false }, + ); + return gcDoc != null && gcDoc.byteLength > EMPTY_UPDATE_MAX_BYTES; +}; + +// Erase every trace of a room's content and leave it writable again. +// +// The erasure is yhub's hard deletion: it clears the stream, disconnects the +// editors and drops every row and asset, irreversibly. Its tombstone is also +// the barrier that stops a compaction still in flight from writing the content +// back — every `store` is refused while it is there, and the purge runs behind +// it — so the room is only made writable again, by dropping the tombstone, +// once there is nothing left to write back. +// +// Dropping the tombstone is what makes this a reset rather than a deletion: +// yhub has no such operation, a hard deletion is final for the room and even +// `restoreDoc` refuses it. Here the document id belongs to a Django document +// that goes on living, so the room has to be usable again. +const eraseContent = async ( + yhubApi: YHub, + docRef: DocRef, + by: string, +): Promise => { + await yhubApi.deleteDoc(docRef, { hard: true, by }); + await yhubApi.persistence.deleteTombstone(docRef); +}; + +const readyLog = logger.child({ module: 'readiness' }); +const resetLog = logger.child({ module: 'reset-ydoc' }); + +// One readiness check: is that store answering? The error never leaves the +// server — the route is unauthenticated, and a postgres client is happy to put +// its connection string, password included, in the message it raises. +const checkStore = async ( + name: string, + probe: () => PromiseLike, +): Promise<[string, string]> => { + let timer: NodeJS.Timeout | undefined; + try { + await Promise.race([ + probe(), + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`no answer in ${READINESS_TIMEOUT_MS}ms`)), + READINESS_TIMEOUT_MS, + ); + }), + ]); + return [name, 'ok']; + } catch (err) { + readyLog.warn( + { store: name, err: err instanceof Error ? err.message : String(err) }, + 'store is unreachable', + ); + return [name, 'unreachable']; + } finally { + clearTimeout(timer); + } +}; + +export const api = [ + // GET /collaboration/ping/v1 — liveness. It answers, therefore the http + // channel and the event loop are alive, which is all a liveness probe should + // ever conclude: touching redis or postgres here would restart a server that + // holds perfectly good websocket connections every time a store blinks. + createApiEndpoint('ping', { + scope: 'global', + get: { + handler: () => jsonResponse(200, { status: 'pong' }), + }, + }), + // GET /collaboration/ready/v1 — readiness. The two stores this server cannot + // serve a single document without: the postgres holding the persisted state + // and the redis carrying the updates between replicas. Answering 503 takes + // this pod out of the service endpoints and leaves the others serving, which + // is the whole difference with the liveness probe above. + createApiEndpoint('ready', { + scope: 'global', + get: { + handler: async (req) => { + // both at once: a probe is not the place to add the latency of one + // store to the latency of the other + const checks = Object.fromEntries( + await Promise.all([ + checkStore('postgres', () => req.yhub.persistence.sql`SELECT 1`), + checkStore('redis', () => req.yhub.stream.redis.ping()), + ]), + ); + const ready = Object.values(checks).every((state) => state === 'ok'); + return jsonResponse(ready ? 200 : 503, { + status: ready ? 'ready' : 'unready', + checks, + }); + }, + }, + }), + // GET /collaboration/jwks/v1 — the public keys verifying the tokens we sign + // to call the backend, in the JSON Web Key Set format (RFC 7517). Global + // scope: it is about this server, not about a document, so the route carries + // no org and no docid. The counterpart of the backend's own /api/v1.0/jwks, + // which we read above to verify its tokens: neither side stores a copy of + // the other's key, so either can be rolled without the other being changed. + createApiEndpoint('jwks', { + scope: 'global', + get: { + // an empty set when no key is configured: honest, and the backend + // refuses our (equally absent) tokens rather than trusting anything + handler: () => + jsonResponse(200, { + keys: backendPublicJwk == null ? [] : [backendPublicJwk], + }), + }, + }), + // POST /collaboration/reset-connections/v1/{org}/{docid} — replaces + // y-provider's /collaboration/api/reset-connections/?room=. Doc-scoped, so + // the docRef comes from the path; access is the admin token's alone, because + // the browser's grant names no `reset-connections` endpoint and has no `'*'` + // fallback. uws routes are exact: a trailing slash 404s. + createApiEndpoint('reset-connections', { + post: { + handler: async (req) => { + const userId = req.headers['x-user-id'] || null; + if (req.org !== ORG) { + return jsonResponse(400, { error: 'Unknown org' }); + } + if (!UUID4.test(req.docid)) { + return jsonResponse(400, { error: 'Room name is invalid' }); + } + // in-place recheck: every yhub server re-runs `authorize` per + // matching connection and closes 4401 only when the access changed — + // no reconnect churn for unaffected clients + await req.yhub.recheckAuth(req.docRef, { + users: userId ? [userId] : null, + }); + return jsonResponse(200, { message: 'Connections reset' }); + }, + }, + }), + // POST /collaboration/migrate/v1/{org}/{docid} — replay a document's full + // legacy version history from the S3 media bucket into yhub (see README.md). + // Backend-internal, like reset-connections: gated to the admin token via the + // 'migrate' access purpose, since it writes history and reads the legacy + // store. + createApiEndpoint('migrate', { + post: { + handler: async (req) => { + if (req.org !== ORG) { + return jsonResponse(400, { error: 'Unknown org' }); + } + if (!UUID4.test(req.docid)) { + return jsonResponse(400, { error: 'Room name is invalid' }); + } + if (req.branch !== 'main') { + // the legacy store is branchless: `{docid}/file` is the main branch + return jsonResponse(400, { error: 'Unknown branch' }); + } + if (!SOFT_MIGRATION) { + // the flag is what configures the S3 client (migration.ts) + return jsonResponse(503, { error: 'Legacy store is not configured' }); + } + // ?force=true replays a document that is already in the migrated set. + // Only safe while its clock-0 row is still there: once compaction has + // folded that row away, a second replay attributes the same content a + // second time and the activity timestamps become ambiguous. + const { status, ...stats } = await fullMigrate(req.yhub, req.docRef, { + force: req.query.force === 'true', + }); + // `status` is what a backfill driver records per document: 'ok', + // 'already', 'empty' (no legacy object — a brand-new document) or + // 'nothing' (versions exist, none readable). All four are done, hence + // one 2xx; `message` says the same thing to a human, `migrated` + // whether this call is the one that wrote the history. + const messages: Record = { + already: 'Already migrated', + empty: 'No legacy document in s3', + nothing: 'No usable content in the legacy versions', + ok: 'Migration completed', + }; + + return jsonResponse(200, { + status, + message: messages[status], + migrated: status === 'ok', + ...stats, + }); + }, + }, + }), + // POST /collaboration/create-ydoc/v1/{org}/{docid} — create a document's + // initial Yjs state from a RAW binary update (`Y.encodeStateAsUpdate` / + // pycrdt `get_update()` output) posted as application/octet-stream. + // + // The built-in `PATCH ydoc` takes the same update (base64, in a json body + // since 0.5.0) but neither of the two things this endpoint exists for: it is + // a strict create, answering 409 when the room already has content, and it + // attributes the content to the user named in `X-User-Id` rather than to the + // backend making the call. Reads have no such needs and use the built-in + // `GET ydoc`. Default access purpose: guarded like the built-in ydoc routes + // (write access on the doc — the admin JWT, or a user session with update + // ability). + createApiEndpoint('create-ydoc', { + post: { + handler: async (req) => { + // The endpoint facet already admitted this caller — only the admin + // token names this route — but the write itself is a separate facet, and + // yhub's contract is that a handler states the facets it touches. Cheap, + // and it keeps the grant honest if the tables above ever widen. + checkPermissions( + req.permissions, + createDocumentPermissions({ ydoc: '--u-' }), + ); + if (req.org !== ORG) { + return jsonResponse(400, { error: 'Unknown org' }); + } + if (!UUID4.test(req.docid)) { + return jsonResponse(400, { error: 'Room name is invalid' }); + } + if (req.branch !== 'main') { + // cookie users are main-only via `authorize`, but the admin + // token bypasses it — reject explicitly so an admin create can't + // seed an orphan non-main room (and dodge the 409 check, which is + // branch-scoped) + return jsonResponse(400, { error: 'Unknown branch' }); + } + const body = await req.bytes(); + // req.bytes() resolves to a Node Buffer, but the compute-task schema + // requires an exact Uint8Array (lib0 $constructedBy compares the + // constructor) — re-view the same bytes without copying + const update = new Uint8Array( + body.buffer, + body.byteOffset, + body.byteLength, + ); + if (update.byteLength > MAX_CREATE_BYTES) { + // 413 is missing from yhub's status-line map (503 was added in + // 0.5.0, 413 was not), so the reason phrase is empty + // ("HTTP/1.1 413 ") — legal, and callers switch on the code + return jsonResponse(413, { error: 'Update too large' }); + } + // yhub's "no effective content" convention — reject before it reaches + // a worker + if (update.byteLength <= EMPTY_UPDATE_MAX_BYTES) { + return jsonResponse(400, { error: 'Empty update' }); + } + // covers persisted state AND uncompacted stream messages. Not atomic + // with addMessage below (yhub has no atomic create): two concurrent + // creates can both pass the check and their updates merge — with + // independently generated updates (fresh clientIDs) the seeded + // content then appears twice. Accepted: Django creates each doc + // once, and a duplicated seed is user-fixable, unlike corruption. + const { gcDoc } = await req.yhub.getDoc( + req.docRef, + { gc: true, nongc: false }, + { gcOnMerge: false }, + ); + if (gcDoc != null && gcDoc.byteLength > EMPTY_UPDATE_MAX_BYTES) { + return jsonResponse(409, { error: 'Document already exists' }); + } + // Only the backend admin token may attribute the content to another + // user; regular callers always author as themselves — honoring a + // client-supplied header would let any editor forge the attribution + // history (the ws path likewise stamps the server-side identity). + const authInfo = req.authInfo as AppAuthInfo | null; + if (authInfo == null) { + return jsonResponse(401, { error: 'Authentication required' }); + } + const userid = + (authInfo.admin === true && req.headers['x-user-id']) || + authInfo.userid; + let result; + try { + // diffs the posted update against the (empty) current doc and + // stamps the attribution contentmap + result = await req.yhub.computePool.patchYdoc( + { + update, + currentDoc: gcDoc ?? EMPTY_YDOC, + userid, + customAttributions: [], + }, + { docRef: req.docRef }, + ); + } catch { + // a malformed update makes the compute worker throw (yhub logs + // 'worker failed' and replaces the thread). The update is the only + // untrusted input here, so a rejection maps to 400; getDoc / + // addMessage failures stay generic 500s. + return jsonResponse(400, { error: 'Invalid Yjs update' }); + } + if (result == null) { + // structurally valid but no effective content (e.g. delete-set + // only). A "successful" create that leaves the room nonexistent + // would lie to the caller — a later create would not 409. + return jsonResponse(400, { error: 'Empty update' }); + } + // on a fresh room this creates the stream, schedules compaction, and + // fans out to any live subscribers — nothing else to do + await req.yhub.stream.addMessage(req.docRef, { + type: 'ydoc:update:v1', + contentmap: result.contentmap, + update: result.update, + }); + return jsonResponse(201, { message: 'Document created' }); + }, + }, + }), + // POST /collaboration/restore-ydoc/v1/{org}/{docid} — undo the deletion of a + // document, putting back what `DELETE .../ydoc/` took away. + // + // Deleting has a built-in route, restoring does not: yhub 0.6.0 exposes + // `restoreDoc` to the process embedding it and nothing else. Backend-internal + // like reset-connections and migrate, gated to the admin token by the + // 'restore' purpose — a document leaves the trashbin because the backend + // says so, never because an editor asked. + createApiEndpoint('restore-ydoc', { + post: { + handler: async (req) => { + if (req.org !== ORG) { + return jsonResponse(400, { error: 'Unknown org' }); + } + if (!UUID4.test(req.docid)) { + return jsonResponse(400, { error: 'Room name is invalid' }); + } + if (req.branch !== 'main') { + // as in create-ydoc: the admin token is not fenced to main by + // `authorize`, and a deletion is recorded per branch + return jsonResponse(400, { error: 'Unknown branch' }); + } + // read the deletion before undoing it: `restoreDoc` throws a plain + // Error for a document whose content was erased, and that is a + // conflict to report as one — catching around the call would turn + // every failure alike, a database outage included, into the same answer + const tombstone = await req.yhub.persistence.retrieveTombstone( + req.docRef, + ); + if (tombstone == null) { + // not an error: the backend restores a whole subtree, of which only + // the part that was deleted with it has anything to put back + return jsonResponse(200, { + message: 'Document is not deleted', + restored: false, + }); + } + if (tombstone.hard || tombstone.purgedAt != null) { + return jsonResponse(409, { error: 'Document content was erased' }); + } + await req.yhub.restoreDoc(req.docRef); + return jsonResponse(200, { + message: 'Document restored', + restored: true, + }); + }, + }, + }), + // POST /collaboration/reset-ydoc/v1/{org}/{docid} — erase the content of a + // document and leave the room usable, as if it had never been written. + // + // What the backend's `clean_document` command needs to reset the onboarding + // sandbox: the Django document keeps its id and goes on being edited, so + // deleting the room is not an option — a hard deletion is final and even a + // soft one would answer 404 for a document that still exists. Backend-internal + // and admin-only, like the deletions it is built on: this destroys content + // with no way back. + createApiEndpoint('reset-ydoc', { + post: { + handler: async (req) => { + if (req.org !== ORG) { + return jsonResponse(400, { error: 'Unknown org' }); + } + if (!UUID4.test(req.docid)) { + return jsonResponse(400, { error: 'Room name is invalid' }); + } + if (req.branch !== 'main') { + return jsonResponse(400, { error: 'Unknown branch' }); + } + const authInfo = req.authInfo as AppAuthInfo | null; + const by = req.headers['x-user-id'] || authInfo?.userid; + if (!by) { + return jsonResponse(401, { error: 'Authentication required' }); + } + // No `checkPermissions` here, deliberately. The honest facet for what + // follows would be `delete: ['hard']`, and the admin grant withholds + // `'hard'` on purpose so that `DELETE /ydoc?hard=true` stays refused over + // REST. The erasure below reaches `deleteDoc` programmatically, which the + // `delete` facet does not gate; the endpoint facet — this route is named + // by no browser grant — is what admits the caller. + // Nothing compacts this room while the erasure runs: this drops the + // task already waiting for it and refuses to enqueue another, which + // leaves one writer to race with — a task a worker had claimed before + // this call. The tombstone barrier covers it right up to the moment + // the room is made writable again, so it can only land after that, + // and the second pass below is what picks it up. + await req.yhub.stream.disableCompaction(req.docRef); + try { + await eraseContent(req.yhub, req.docRef, by); + if (await hasContent(req.yhub, req.docRef)) { + resetLog.warn( + { docid: req.docid }, + 'content came back while it was being erased, erasing again', + ); + await eraseContent(req.yhub, req.docRef, by); + if (await hasContent(req.yhub, req.docRef)) { + // saying it is erased when it is not is the one answer this + // endpoint must never give + return jsonResponse(500, { + error: 'Document content came back after being erased', + }); + } + } + } finally { + // even on failure: leaving compaction off would freeze the room for + // every later edit, a worse state than the one we came to fix + await req.yhub.stream.enableCompaction(req.docRef); + } + return jsonResponse(200, { message: 'Document content erased' }); + }, + }, + }), +]; diff --git a/src/yhub-server/src/backend.ts b/src/yhub-server/src/backend.ts new file mode 100644 index 000000000..fa8793a65 --- /dev/null +++ b/src/yhub-server/src/backend.ts @@ -0,0 +1,165 @@ +/** + * Everything this server sends to the Docs Django backend, and the keys both + * ends verify each other with. + * + * `backendFetch` — a GET carrying the caller's session (the auth plugin reads + * `/users/me/` and `/documents/{id}/` through it) + * `touchDocument` — the `content-updated` notification a compaction fires + * `JWKS` — Django's public keys, for verifying the admin tokens it signs to + * call us (consumed by `server.ts`) + * `backendPublicJwk` — the public half of *our* signing key, as the + * `/collaboration/jwks/v1` route publishes it + * + * The two token directions are symmetric: neither side stores the other's key, + * so either can be rolled without a change here. + */ +import { createPublicKey } from 'node:crypto'; + +import { logger } from '@y/hub'; +import type { JWK } from 'jose'; +import { + calculateJwkThumbprint, + createRemoteJWKSet, + exportJWK, + importPKCS8, + SignJWT, +} from 'jose'; + +import type { DocumentAbilities } from './permissions.js'; +import { + BACKEND_AUDIENCE, + BACKEND_NOTIFY_TIMEOUT_MS, + BACKEND_TOKEN_LIFETIME_S, + BACKEND_TOKEN_MARGIN_MS, + COLLABORATION_BACKEND_BASE_URL, + Y_PROVIDER_API_KEY, + YHUB_JWT_PRIVATE_KEY, +} from './config.js'; + +// The shape of the backend payload a caller name resolves to. +export interface BackendUser { + id: string | number; +} +// `GET /api/v1.0/documents/{id}/` — only the abilities the policy reads. +export interface BackendDocument { + abilities?: DocumentAbilities; +} + +const touchLog = logger.child({ module: 'updated-at-notifier' }); + +const backendSigningKey = YHUB_JWT_PRIVATE_KEY + ? await importPKCS8(YHUB_JWT_PRIVATE_KEY, 'RS256') + : null; + +if (backendSigningKey == null) { + // not fatal, documents keep being served — only their `updated_at` freezes + touchLog.warn( + 'YHUB_JWT_PRIVATE_KEY is empty, the backend will not be notified of content updates', + ); +} + +// The public half of the signing key, as published on the JWKS endpoint. Its +// "kid" is the RFC 7638 thumbprint of the key: computed from the public +// components only, it is stable across restarts and changes on its own when +// the key is rolled. Every token we sign carries it, which is how the backend +// picks the matching key — and how it knows to fetch the set again when it +// does not know the key yet, so rolling this key needs no change on its side. +export const backendPublicJwk: (JWK & { kid: string }) | null = + backendSigningKey == null + ? null + : await (async () => { + // derived from the PEM rather than exported from `backendSigningKey`: + // exporting a private key as a JWK would carry its private components + const jwk = await exportJWK(createPublicKey(YHUB_JWT_PRIVATE_KEY)); + return { + ...jwk, + alg: 'RS256', + use: 'sig', + kid: await calculateJwkThumbprint(jwk), + }; + })(); + +let backendToken: { token: string; expiresAt: number } | null = null; + +// The token carries no per-document claim, so one is reused until it is about +// to expire rather than signing on every notification. +const getBackendToken = async (): Promise => { + if (backendSigningKey == null || backendPublicJwk == null) { + throw new Error('no backend signing key is configured'); + } + const now = Date.now(); + if ( + backendToken != null && + backendToken.expiresAt - BACKEND_TOKEN_MARGIN_MS > now + ) { + return backendToken.token; + } + const token = await new SignJWT({}) + // the "kid" names the key in our JWKS the backend must verify it with + .setProtectedHeader({ alg: 'RS256', kid: backendPublicJwk.kid }) + .setIssuer('yhub') + .setAudience(BACKEND_AUDIENCE) + .setIssuedAt() + .setExpirationTime(`${BACKEND_TOKEN_LIFETIME_S}s`) + .sign(backendSigningKey); + backendToken = { token, expiresAt: now + BACKEND_TOKEN_LIFETIME_S * 1000 }; + return token; +}; + +// Public keys verifying the RS256 admin tokens Django issues (JWTService). +// Lazily fetched on first use; jose caches the keys and refetches on unknown +// "kid", so Django can rotate the signing key without a yhub restart. +export const JWKS = createRemoteJWKSet( + new URL(`${COLLABORATION_BACKEND_BASE_URL}/api/v1.0/jwks`), +); + +export const backendFetch = async ( + path: string, + { cookie, origin }: { cookie?: string; origin?: string }, +): Promise => { + const res = await fetch(`${COLLABORATION_BACKEND_BASE_URL}${path}`, { + headers: { + // an anonymous caller may have no session at all; `cookie: undefined` would + // reach the backend as the literal string "undefined" + ...(cookie ? { cookie } : {}), + // a same-origin request carries no `Origin` — forwarded when there is one, omitted + // rather than sent empty, which is not a value the header is allowed to take + ...(origin ? { origin } : {}), + 'X-Y-Provider-Key': Y_PROVIDER_API_KEY, + }, + }); + if (!res.ok) { + const err: Error & { status?: number } = new Error( + `Failed to fetch ${path}: ${res.status}`, + ); + err.status = res.status; + throw err; + } + return res.json() as Promise; +}; + +// Django orders the document lists by `updated_at` and no edit goes through it +// anymore, so it is told here that a document moved on. +export const touchDocument = async (docid: string): Promise => { + if (backendSigningKey == null) return; + try { + const res = await fetch( + `${COLLABORATION_BACKEND_BASE_URL}/api/v1.0/documents/${docid}/content-updated/`, + { + method: 'POST', + headers: { authorization: `Bearer ${await getBackendToken()}` }, + signal: AbortSignal.timeout(BACKEND_NOTIFY_TIMEOUT_MS), + }, + ); + if (!res.ok) { + touchLog.warn( + { docid, status: res.status }, + 'backend refused the notification', + ); + } + } catch (err) { + // best effort: a lost notification only leaves `updated_at` behind until + // the document is edited again, it must never fail a compaction + touchLog.warn({ err, docid }, 'could not notify the backend'); + } +}; diff --git a/src/yhub-server/src/config.ts b/src/yhub-server/src/config.ts new file mode 100644 index 000000000..e88d81c37 --- /dev/null +++ b/src/yhub-server/src/config.ts @@ -0,0 +1,157 @@ +/** + * Every environment variable this server reads, in one place: parsed, validated + * and refused at import time when it cannot be trusted, so a misconfiguration is + * a startup error naming the variable rather than a pod that looks healthy and + * is not. + * + * `backend.ts`, `api.ts` and `server.ts` import their settings from here; the + * `*_FILE` secret indirection is `env.ts`. + */ +import { secret } from './env.js'; + +// A numeric setting, read from the environment and refused rather than guessed +// when it is not a whole number at or above `min`: `Number()` reads a typo as +// NaN, which yhub takes as-is and turns into a worker that claims nothing or a +// stream that is never trimmed — a deployment that looks healthy and is not. +// An unset or empty variable is the default, so a kubernetes env var left blank +// behaves as if it had not been set at all. +const intEnv = (name: string, dflt: number, min = 1): number => { + const raw = process.env[name]; + const value = raw == null || raw === '' ? dflt : Number(raw); + if (!Number.isInteger(value) || value < min) { + throw new Error(`${name} must be an integer >= ${min} (got "${raw}")`); + } + return value; +}; + +export const PORT = Number(process.env.PORT || 3002); +export const REDIS = process.env.REDIS; +export const POSTGRES = process.env.POSTGRES; +export const REDIS_PREFIX = process.env.REDIS_PREFIX || 'yhub'; +// How long an update waits on the stream before a worker claims the compaction +// task it belongs to. It is the delay between an edit and its row in postgres, +// and the window over which the edits of a busy document are merged into one +// task: lowering it persists sooner and compacts more often, raising it does +// the reverse. yhub defaults to 120s, which is a long time to lose when a pod +// is killed — Docs asks for 10s. +export const TASK_DEBOUNCE_MS = intEnv('YHUB_TASK_DEBOUNCE_MS', 10000, 0); +// How long messages a worker has already persisted are kept on the stream. The +// trim stops at the older of that age and the point postgres holds, so this is +// not a durability setting — nothing unpersisted is ever trimmed. It is how +// much recent history stays replayable from redis instead of being read back +// out of postgres, paid for in memory on the redis side. +export const MIN_MESSAGE_LIFETIME_MS = intEnv( + 'YHUB_MIN_MESSAGE_LIFETIME_MS', + 60000, + 0, +); +export const COLLABORATION_BACKEND_BASE_URL = + process.env.COLLABORATION_BACKEND_BASE_URL || 'http://app-dev:8000'; +export const allowedOrigins = ( + process.env.COLLABORATION_SERVER_ORIGIN || 'http://localhost:3000' +).split(','); +export const Y_PROVIDER_API_KEY = secret('Y_PROVIDER_API_KEY', 'yprovider-api-key'); +export const ORG = process.env.YHUB_ORG || 'docs'; +// Which halves of yhub this process runs. The server accepts the websocket +// connections and serves the REST routes; the worker drains the redis stream +// into postgres. They share the two stores and nothing else — no in-process +// state, no ordering between them — so one process can run both (the default) +// or a deployment can split them and scale each on its own: the server with +// the connected editors, the worker with the write throughput. +// +// A stream is only drained by the workers that are running: a deployment of +// `server` alone keeps accepting edits and never persists them, so the two +// halves are split together or not at all. +export const ROLE = process.env.YHUB_ROLE || 'all'; +if (!['all', 'server', 'worker'].includes(ROLE)) { + throw new Error( + `YHUB_ROLE must be one of "all", "server" or "worker" (got "${ROLE}")`, + ); +} +export const RUNS_SERVER = ROLE !== 'worker'; +export const RUNS_WORKER = ROLE !== 'server'; +// How many tasks one worker process claims at once. Redis hands each task to a +// single worker, so what a deployment actually runs in parallel is this times +// the number of worker processes — the two knobs are interchangeable up to the +// point where a pod runs out of memory, each task holding the document it +// merges. +export const TASK_CONCURRENCY = intEnv('YHUB_TASK_CONCURRENCY', 5, 1); +// Where the blobs of a compaction go — the garbage-collected document, the one +// that keeps its history, the content map and the content ids. yhub writes the +// four of them into its own postgres; a persistence plugin takes them out of +// it, the row then holding a reference and the bytes living in the plugin's +// store. Off by default, which is postgres alone, the way Docs has been +// running. +// +// This decides where *new* blobs are written, and nothing else. The plugin +// itself is loaded whenever the bucket below is configured, on or off, because +// a row pointing at an object is unreadable without the plugin that wrote it — +// yhub reports such a version as having no content rather than as an error, so +// a deployment that has ever had this on and drops the plugin loses those +// documents silently. Keep the YHUB_S3_* settings in place for as long as the +// bucket holds anything; turning this off then stops the writing and leaves the +// reading alone. See README.md. +export const S3_PERSISTENCE = process.env.YHUB_S3_PERSISTENCE === 'true'; +// Its own bucket, named apart from the backend's `AWS_S3_*` and from the legacy +// document store's `LEGACY_S3_*` (migration.ts): three buckets that may sit on +// three providers with credentials of their own, each read by the process it +// belongs to. +export const YHUB_S3_ENDPOINT_URL = process.env.YHUB_S3_ENDPOINT_URL; +export const YHUB_S3_ACCESS_KEY_ID = secret('YHUB_S3_ACCESS_KEY_ID'); +export const YHUB_S3_SECRET_ACCESS_KEY = secret('YHUB_S3_SECRET_ACCESS_KEY'); +export const YHUB_S3_BUCKET_NAME = process.env.YHUB_S3_BUCKET_NAME; +export const YHUB_S3_REGION_NAME = process.env.YHUB_S3_REGION_NAME; +// Segment every route is mounted under (`server.apiPrefix` below), matching the +// URL scheme Docs already routes to the collaboration server. Hardcoded like +// the audiences: the backend builds its urls with the same prefix. +export const API_PREFIX = 'collaboration'; +// The identity of a caller who is not signed in. It is a userid rather than the +// absence of one on purpose: yhub refuses the websocket upgrade of a caller that +// holds the write but has no identity (401 `unauthenticated`), because +// attributions carry the userid — so without this an anonymous visitor could not +// edit a public document at all. Every anonymous edit is therefore attributed to +// one shared author; yhub's own userids are opaque strings and Docs' are UUIDs, +// so this cannot collide with a real one. +export const ANONYMOUS_USERID = 'anonymous'; +// What the readiness check gives a store before reporting it unreachable. Short +// on purpose: the point of the probe is to answer, and answering "not ready" +// early is more useful than holding the connection until kubelet times out. +export const READINESS_TIMEOUT_MS = 2000; +// Requiring this audience stops a valid admin JWT that Django issued for +// another service (today: the y-converter token in converter_services.py, +// which is handed to the converter process) from being replayed against yhub. +// Hardcoded, like y-provider's Y_CONVERTER_AUDIENCE: both ends of a two-party +// contract, so an env var would only add a way to misconfigure it into a 401. +export const YHUB_AUDIENCE = 'yhub'; +// lowercase only (no /i): Django serializes UUIDs lowercase, while yhub rooms +// and S3 keys are case-sensitive strings — accepting case variants would let a +// client open a parallel room for the same document (and, with soft migration, +// miss its S3 object and fork the document's lineage) +export const UUID4 = + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; +// an empty Yjs update (what `Y.encodeStateAsUpdate(new Y.Doc())` encodes to) — +// hardcoded so we don't import @y/y for two bytes +export const EMPTY_YDOC = new Uint8Array([0, 0]); +// yhub's "no effective content" convention: an empty update encodes to 2 bytes, +// and anything up to 3 is read as an empty document +export const EMPTY_UPDATE_MAX_BYTES = 3; +// uws buffers the whole body before the handler sees it, so this cap does not +// bound upload memory — it bounds what a single create hands to a compute +// worker and writes to the valkey stream as one message. Creates carry one +// freshly-converted snapshot (typically KBs); anything bigger belongs on the +// websocket path. +export const MAX_CREATE_BYTES = 10 * 1024 * 1024; + +export const BACKEND_NOTIFY_TIMEOUT_MS = 5000; +// Audience of the tokens the backend accepts from us. It must match the one +// its CollaborationServerAuthentication requires, a token minted for anything +// else is refused there. +export const BACKEND_AUDIENCE = 'docs-backend'; +export const BACKEND_TOKEN_LIFETIME_S = 60; +// Renew this long before expiry so a token never dies in flight. +export const BACKEND_TOKEN_MARGIN_MS = 10000; + +// We sign the calls we make to the backend, the mirror of the admin JWT it +// signs to call us: no long-lived shared secret, only our private key here and +// its public half published on the JWKS endpoint. +export const YHUB_JWT_PRIVATE_KEY = secret('YHUB_JWT_PRIVATE_KEY', ''); diff --git a/src/yhub-server/src/server.ts b/src/yhub-server/src/server.ts index 299955940..1a35b174d 100644 --- a/src/yhub-server/src/server.ts +++ b/src/yhub-server/src/server.ts @@ -1,64 +1,57 @@ -import { createPublicKey } from 'node:crypto'; - import { apiError, - checkPermissions, - createApiEndpoint, createAuthPlugin, createAuthorize, - createDocumentPermissions, createYHub, logger, } from '@y/hub'; import { S3PersistenceV1 } from '@y/hub/plugins/s3'; -import type { DocRef, YHub } from './yhub.js'; -import type { JWK, JWTPayload } from 'jose'; -import { - calculateJwkThumbprint, - createRemoteJWKSet, - exportJWK, - importPKCS8, - jwtVerify, - SignJWT, -} from 'jose'; +import type { JWTPayload } from 'jose'; +import { jwtVerify } from 'jose'; -import { secret } from './env.js'; +import { api } from './api.js'; +// outbound calls to the Docs Django backend, and the keys both ends verify with +import type { BackendDocument, BackendUser } from './backend.js'; +import { JWKS, backendFetch, touchDocument } from './backend.js'; +import { + ANONYMOUS_USERID, + API_PREFIX, + MIN_MESSAGE_LIFETIME_MS, + ORG, + PORT, + POSTGRES, + REDIS, + REDIS_PREFIX, + ROLE, + RUNS_SERVER, + RUNS_WORKER, + S3_PERSISTENCE, + TASK_CONCURRENCY, + TASK_DEBOUNCE_MS, + UUID4, + YHUB_AUDIENCE, + YHUB_S3_ACCESS_KEY_ID, + YHUB_S3_BUCKET_NAME, + YHUB_S3_ENDPOINT_URL, + YHUB_S3_REGION_NAME, + YHUB_S3_SECRET_ACCESS_KEY, + allowedOrigins, +} from './config.js'; +// legacy Django/S3 document store — see migration.ts and README.md +import { + SOFT_MIGRATION, + isPermanentFailure, + maybeMigrate, + migrationLog, +} from './migration.js'; // Docs' access policy as yhub permission objects — see permissions.spec.ts -import type { DocumentAbilities } from './permissions.js'; import { adminDocumentPermissions, browserDocumentPermissions, publicGlobalPermissions, resolveHistoryFrom, } from './permissions.js'; -// legacy Django/S3 document store — see migration.ts and README.md -import { - SOFT_MIGRATION, - fullMigrate, - isPermanentFailure, - maybeMigrate, - migrationLog, -} from './migration.js'; - -// The caller identity `authenticate` returns and `authorize` is handed. yhub -// only requires `userid`; the rest is Docs' own, read back off `req.authInfo` -// (typed by yhub as the bare `{ userid }`, hence the cast at those sites). -interface AppAuthInfo { - userid: string; - admin?: boolean; - endpoint?: string; - cookie?: string; - origin?: string; -} - -// The shape of the backend payload a caller name resolves to. -interface BackendUser { - id: string | number; -} -// `GET /api/v1.0/documents/{id}/` — only the abilities the policy reads. -interface BackendDocument { - abilities?: DocumentAbilities; -} +import type { AppAuthInfo, DocRef, YHub } from './yhub.js'; // Read the ad-hoc tags the error paths below switch on without trusting the // error's runtime shape. @@ -77,242 +70,6 @@ const errCode = (err: unknown): string | undefined => ? (err as { code: string }).code : undefined; -// A numeric setting, read from the environment and refused rather than guessed -// when it is not a whole number at or above `min`: `Number()` reads a typo as -// NaN, which yhub takes as-is and turns into a worker that claims nothing or a -// stream that is never trimmed — a deployment that looks healthy and is not. -// An unset or empty variable is the default, so a kubernetes env var left blank -// behaves as if it had not been set at all. -const intEnv = (name: string, dflt: number, min = 1): number => { - const raw = process.env[name]; - const value = raw == null || raw === '' ? dflt : Number(raw); - if (!Number.isInteger(value) || value < min) { - throw new Error(`${name} must be an integer >= ${min} (got "${raw}")`); - } - return value; -}; - -const PORT = Number(process.env.PORT || 3002); -const REDIS = process.env.REDIS; -const POSTGRES = process.env.POSTGRES; -const REDIS_PREFIX = process.env.REDIS_PREFIX || 'yhub'; -// How long an update waits on the stream before a worker claims the compaction -// task it belongs to. It is the delay between an edit and its row in postgres, -// and the window over which the edits of a busy document are merged into one -// task: lowering it persists sooner and compacts more often, raising it does -// the reverse. yhub defaults to 120s, which is a long time to lose when a pod -// is killed — Docs asks for 10s. -const TASK_DEBOUNCE_MS = intEnv('YHUB_TASK_DEBOUNCE_MS', 10000, 0); -// How long messages a worker has already persisted are kept on the stream. The -// trim stops at the older of that age and the point postgres holds, so this is -// not a durability setting — nothing unpersisted is ever trimmed. It is how -// much recent history stays replayable from redis instead of being read back -// out of postgres, paid for in memory on the redis side. -const MIN_MESSAGE_LIFETIME_MS = intEnv('YHUB_MIN_MESSAGE_LIFETIME_MS', 60000, 0); -const COLLABORATION_BACKEND_BASE_URL = - process.env.COLLABORATION_BACKEND_BASE_URL || 'http://app-dev:8000'; -const allowedOrigins = ( - process.env.COLLABORATION_SERVER_ORIGIN || 'http://localhost:3000' -).split(','); -const Y_PROVIDER_API_KEY = secret('Y_PROVIDER_API_KEY', 'yprovider-api-key'); -const ORG = process.env.YHUB_ORG || 'docs'; -// Which halves of yhub this process runs. The server accepts the websocket -// connections and serves the REST routes; the worker drains the redis stream -// into postgres. They share the two stores and nothing else — no in-process -// state, no ordering between them — so one process can run both (the default) -// or a deployment can split them and scale each on its own: the server with -// the connected editors, the worker with the write throughput. -// -// A stream is only drained by the workers that are running: a deployment of -// `server` alone keeps accepting edits and never persists them, so the two -// halves are split together or not at all. -const ROLE = process.env.YHUB_ROLE || 'all'; -if (!['all', 'server', 'worker'].includes(ROLE)) { - throw new Error( - `YHUB_ROLE must be one of "all", "server" or "worker" (got "${ROLE}")`, - ); -} -const RUNS_SERVER = ROLE !== 'worker'; -const RUNS_WORKER = ROLE !== 'server'; -// How many tasks one worker process claims at once. Redis hands each task to a -// single worker, so what a deployment actually runs in parallel is this times -// the number of worker processes — the two knobs are interchangeable up to the -// point where a pod runs out of memory, each task holding the document it -// merges. -const TASK_CONCURRENCY = intEnv('YHUB_TASK_CONCURRENCY', 5, 1); -// Where the blobs of a compaction go — the garbage-collected document, the one -// that keeps its history, the content map and the content ids. yhub writes the -// four of them into its own postgres; a persistence plugin takes them out of -// it, the row then holding a reference and the bytes living in the plugin's -// store. Off by default, which is postgres alone, the way Docs has been -// running. -// -// This decides where *new* blobs are written, and nothing else. The plugin -// itself is loaded whenever the bucket below is configured, on or off, because -// a row pointing at an object is unreadable without the plugin that wrote it — -// yhub reports such a version as having no content rather than as an error, so -// a deployment that has ever had this on and drops the plugin loses those -// documents silently. Keep the YHUB_S3_* settings in place for as long as the -// bucket holds anything; turning this off then stops the writing and leaves the -// reading alone. See README.md. -const S3_PERSISTENCE = process.env.YHUB_S3_PERSISTENCE === 'true'; -// Its own bucket, named apart from the backend's `AWS_S3_*` and from the legacy -// document store's `LEGACY_S3_*` (migration.ts): three buckets that may sit on -// three providers with credentials of their own, each read by the process it -// belongs to. -const YHUB_S3_ENDPOINT_URL = process.env.YHUB_S3_ENDPOINT_URL; -const YHUB_S3_ACCESS_KEY_ID = secret('YHUB_S3_ACCESS_KEY_ID'); -const YHUB_S3_SECRET_ACCESS_KEY = secret('YHUB_S3_SECRET_ACCESS_KEY'); -const YHUB_S3_BUCKET_NAME = process.env.YHUB_S3_BUCKET_NAME; -const YHUB_S3_REGION_NAME = process.env.YHUB_S3_REGION_NAME; -// Segment every route is mounted under (`server.apiPrefix` below), matching the -// URL scheme Docs already routes to the collaboration server. Hardcoded like -// the audiences: the backend builds its urls with the same prefix. -const API_PREFIX = 'collaboration'; -// The identity of a caller who is not signed in. It is a userid rather than the -// absence of one on purpose: yhub refuses the websocket upgrade of a caller that -// holds the write but has no identity (401 `unauthenticated`), because -// attributions carry the userid — so without this an anonymous visitor could not -// edit a public document at all. Every anonymous edit is therefore attributed to -// one shared author; yhub's own userids are opaque strings and Docs' are UUIDs, -// so this cannot collide with a real one. -const ANONYMOUS_USERID = 'anonymous'; -// What the readiness check gives a store before reporting it unreachable. Short -// on purpose: the point of the probe is to answer, and answering "not ready" -// early is more useful than holding the connection until kubelet times out. -const READINESS_TIMEOUT_MS = 2000; -// Requiring this audience stops a valid admin JWT that Django issued for -// another service (today: the y-converter token in converter_services.py, -// which is handed to the converter process) from being replayed against yhub. -// Hardcoded, like y-provider's Y_CONVERTER_AUDIENCE: both ends of a two-party -// contract, so an env var would only add a way to misconfigure it into a 401. -const YHUB_AUDIENCE = 'yhub'; -// lowercase only (no /i): Django serializes UUIDs lowercase, while yhub rooms -// and S3 keys are case-sensitive strings — accepting case variants would let a -// client open a parallel room for the same document (and, with soft migration, -// miss its S3 object and fork the document's lineage) -const UUID4 = - /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; -// an empty Yjs update (what `Y.encodeStateAsUpdate(new Y.Doc())` encodes to) — -// hardcoded so we don't import @y/y for two bytes -const EMPTY_YDOC = new Uint8Array([0, 0]); -// yhub's "no effective content" convention: an empty update encodes to 2 bytes, -// and anything up to 3 is read as an empty document -const EMPTY_UPDATE_MAX_BYTES = 3; -// uws buffers the whole body before the handler sees it, so this cap does not -// bound upload memory — it bounds what a single create hands to a compute -// worker and writes to the valkey stream as one message. Creates carry one -// freshly-converted snapshot (typically KBs); anything bigger belongs on the -// websocket path. -const MAX_CREATE_BYTES = 10 * 1024 * 1024; - -const touchLog = logger.child({ module: 'updated-at-notifier' }); -const resetLog = logger.child({ module: 'reset-ydoc' }); - -const BACKEND_NOTIFY_TIMEOUT_MS = 5000; -// Audience of the tokens the backend accepts from us. It must match the one -// its CollaborationServerAuthentication requires, a token minted for anything -// else is refused there. -const BACKEND_AUDIENCE = 'docs-backend'; -const BACKEND_TOKEN_LIFETIME_S = 60; -// Renew this long before expiry so a token never dies in flight. -const BACKEND_TOKEN_MARGIN_MS = 10000; - -// We sign the calls we make to the backend, the mirror of the admin JWT it -// signs to call us: no long-lived shared secret, only our private key here and -// its public half published on the JWKS endpoint below. -const YHUB_JWT_PRIVATE_KEY = secret('YHUB_JWT_PRIVATE_KEY', ''); -const backendSigningKey = YHUB_JWT_PRIVATE_KEY - ? await importPKCS8(YHUB_JWT_PRIVATE_KEY, 'RS256') - : null; - -if (backendSigningKey == null) { - // not fatal, documents keep being served — only their `updated_at` freezes - touchLog.warn( - 'YHUB_JWT_PRIVATE_KEY is empty, the backend will not be notified of content updates', - ); -} - -// The public half of the signing key, as published on the JWKS endpoint. Its -// "kid" is the RFC 7638 thumbprint of the key: computed from the public -// components only, it is stable across restarts and changes on its own when -// the key is rolled. Every token we sign carries it, which is how the backend -// picks the matching key — and how it knows to fetch the set again when it -// does not know the key yet, so rolling this key needs no change on its side. -const backendPublicJwk: (JWK & { kid: string }) | null = - backendSigningKey == null - ? null - : await (async () => { - // derived from the PEM rather than exported from `backendSigningKey`: - // exporting a private key as a JWK would carry its private components - const jwk = await exportJWK(createPublicKey(YHUB_JWT_PRIVATE_KEY)); - return { - ...jwk, - alg: 'RS256', - use: 'sig', - kid: await calculateJwkThumbprint(jwk), - }; - })(); - -let backendToken: { token: string; expiresAt: number } | null = null; - -// The token carries no per-document claim, so one is reused until it is about -// to expire rather than signing on every notification. -const getBackendToken = async (): Promise => { - if (backendSigningKey == null || backendPublicJwk == null) { - throw new Error('no backend signing key is configured'); - } - const now = Date.now(); - if ( - backendToken != null && - backendToken.expiresAt - BACKEND_TOKEN_MARGIN_MS > now - ) { - return backendToken.token; - } - const token = await new SignJWT({}) - // the "kid" names the key in our JWKS the backend must verify it with - .setProtectedHeader({ alg: 'RS256', kid: backendPublicJwk.kid }) - .setIssuer('yhub') - .setAudience(BACKEND_AUDIENCE) - .setIssuedAt() - .setExpirationTime(`${BACKEND_TOKEN_LIFETIME_S}s`) - .sign(backendSigningKey); - backendToken = { token, expiresAt: now + BACKEND_TOKEN_LIFETIME_S * 1000 }; - return token; -}; - -// Public keys verifying the RS256 admin tokens Django issues (JWTService). -// Lazily fetched on first use; jose caches the keys and refetches on unknown -// "kid", so Django can rotate the signing key without a yhub restart. -const JWKS = createRemoteJWKSet( - new URL(`${COLLABORATION_BACKEND_BASE_URL}/api/v1.0/jwks`), -); - -const backendFetch = async ( - path: string, - { cookie, origin }: { cookie?: string; origin?: string }, -): Promise => { - const res = await fetch(`${COLLABORATION_BACKEND_BASE_URL}${path}`, { - headers: { - // an anonymous caller may have no session at all; `cookie: undefined` would - // reach the backend as the literal string "undefined" - ...(cookie ? { cookie } : {}), - // a same-origin request carries no `Origin` — forwarded when there is one, omitted - // rather than sent empty, which is not a value the header is allowed to take - ...(origin ? { origin } : {}), - 'X-Y-Provider-Key': Y_PROVIDER_API_KEY, - }, - }); - if (!res.ok) { - const err: Error & { status?: number } = new Error( - `Failed to fetch ${path}: ${res.status}`, - ); - err.status = res.status; - throw err; - } - return res.json() as Promise; -}; - // First access to a room yhub does not know: seed it from the legacy Django S3 // store before admitting the caller. Awaited inside the upgrade handler, so the // post-upgrade initial sync (which merges postgres and the stream from clock 0) @@ -537,466 +294,6 @@ const auth = createAuthPlugin({ }), }); -// Mimic the old y-provider REST responses (JSON, not yhub's lib0-any -// encoding) so the Django caller keeps its historical contract. -const jsonResponse = (status: number, body: unknown): Response => - new Response(JSON.stringify(body), { - status, - headers: { 'content-type': 'application/json' }, - }); - -// Does the room hold anything? Covers persisted rows and the messages still on -// the stream, which is what makes it an answer about the content rather than -// about the storage. -const hasContent = async (yhubApi: YHub, docRef: DocRef): Promise => { - const { gcDoc } = await yhubApi.getDoc( - docRef, - { gc: true, nongc: false }, - { gcOnMerge: false }, - ); - return gcDoc != null && gcDoc.byteLength > EMPTY_UPDATE_MAX_BYTES; -}; - -// Erase every trace of a room's content and leave it writable again. -// -// The erasure is yhub's hard deletion: it clears the stream, disconnects the -// editors and drops every row and asset, irreversibly. Its tombstone is also -// the barrier that stops a compaction still in flight from writing the content -// back — every `store` is refused while it is there, and the purge runs behind -// it — so the room is only made writable again, by dropping the tombstone, -// once there is nothing left to write back. -// -// Dropping the tombstone is what makes this a reset rather than a deletion: -// yhub has no such operation, a hard deletion is final for the room and even -// `restoreDoc` refuses it. Here the document id belongs to a Django document -// that goes on living, so the room has to be usable again. -const eraseContent = async ( - yhubApi: YHub, - docRef: DocRef, - by: string, -): Promise => { - await yhubApi.deleteDoc(docRef, { hard: true, by }); - await yhubApi.persistence.deleteTombstone(docRef); -}; - -const readyLog = logger.child({ module: 'readiness' }); - -// One readiness check: is that store answering? The error never leaves the -// server — the route is unauthenticated, and a postgres client is happy to put -// its connection string, password included, in the message it raises. -const checkStore = async ( - name: string, - probe: () => PromiseLike, -): Promise<[string, string]> => { - let timer: NodeJS.Timeout | undefined; - try { - await Promise.race([ - probe(), - new Promise((_, reject) => { - timer = setTimeout( - () => reject(new Error(`no answer in ${READINESS_TIMEOUT_MS}ms`)), - READINESS_TIMEOUT_MS, - ); - }), - ]); - return [name, 'ok']; - } catch (err) { - readyLog.warn( - { store: name, err: err instanceof Error ? err.message : String(err) }, - 'store is unreachable', - ); - return [name, 'unreachable']; - } finally { - clearTimeout(timer); - } -}; - -const api = [ - // GET /collaboration/ping/v1 — liveness. It answers, therefore the http - // channel and the event loop are alive, which is all a liveness probe should - // ever conclude: touching redis or postgres here would restart a server that - // holds perfectly good websocket connections every time a store blinks. - createApiEndpoint('ping', { - scope: 'global', - get: { - handler: () => jsonResponse(200, { status: 'pong' }), - }, - }), - // GET /collaboration/ready/v1 — readiness. The two stores this server cannot - // serve a single document without: the postgres holding the persisted state - // and the redis carrying the updates between replicas. Answering 503 takes - // this pod out of the service endpoints and leaves the others serving, which - // is the whole difference with the liveness probe above. - createApiEndpoint('ready', { - scope: 'global', - get: { - handler: async (req) => { - // both at once: a probe is not the place to add the latency of one - // store to the latency of the other - const checks = Object.fromEntries( - await Promise.all([ - checkStore('postgres', () => req.yhub.persistence.sql`SELECT 1`), - checkStore('redis', () => req.yhub.stream.redis.ping()), - ]), - ); - const ready = Object.values(checks).every((state) => state === 'ok'); - return jsonResponse(ready ? 200 : 503, { - status: ready ? 'ready' : 'unready', - checks, - }); - }, - }, - }), - // GET /collaboration/jwks/v1 — the public keys verifying the tokens we sign - // to call the backend, in the JSON Web Key Set format (RFC 7517). Global - // scope: it is about this server, not about a document, so the route carries - // no org and no docid. The counterpart of the backend's own /api/v1.0/jwks, - // which we read above to verify its tokens: neither side stores a copy of - // the other's key, so either can be rolled without the other being changed. - createApiEndpoint('jwks', { - scope: 'global', - get: { - // an empty set when no key is configured: honest, and the backend - // refuses our (equally absent) tokens rather than trusting anything - handler: () => - jsonResponse(200, { - keys: backendPublicJwk == null ? [] : [backendPublicJwk], - }), - }, - }), - // POST /collaboration/reset-connections/v1/{org}/{docid} — replaces - // y-provider's /collaboration/api/reset-connections/?room=. Doc-scoped, so - // the docRef comes from the path; access is the admin token's alone, because - // the browser's grant names no `reset-connections` endpoint and has no `'*'` - // fallback. uws routes are exact: a trailing slash 404s. - createApiEndpoint('reset-connections', { - post: { - handler: async (req) => { - const userId = req.headers['x-user-id'] || null; - if (req.org !== ORG) { - return jsonResponse(400, { error: 'Unknown org' }); - } - if (!UUID4.test(req.docid)) { - return jsonResponse(400, { error: 'Room name is invalid' }); - } - // in-place recheck: every yhub server re-runs `authorize` per - // matching connection and closes 4401 only when the access changed — - // no reconnect churn for unaffected clients - await req.yhub.recheckAuth(req.docRef, { - users: userId ? [userId] : null, - }); - return jsonResponse(200, { message: 'Connections reset' }); - }, - }, - }), - // POST /collaboration/migrate/v1/{org}/{docid} — replay a document's full - // legacy version history from the S3 media bucket into yhub (see README.md). - // Backend-internal, like reset-connections: gated to the admin token via the - // 'migrate' access purpose, since it writes history and reads the legacy - // store. - createApiEndpoint('migrate', { - post: { - handler: async (req) => { - if (req.org !== ORG) { - return jsonResponse(400, { error: 'Unknown org' }); - } - if (!UUID4.test(req.docid)) { - return jsonResponse(400, { error: 'Room name is invalid' }); - } - if (req.branch !== 'main') { - // the legacy store is branchless: `{docid}/file` is the main branch - return jsonResponse(400, { error: 'Unknown branch' }); - } - if (!SOFT_MIGRATION) { - // the flag is what configures the S3 client (migration.ts) - return jsonResponse(503, { error: 'Legacy store is not configured' }); - } - // ?force=true replays a document that is already in the migrated set. - // Only safe while its clock-0 row is still there: once compaction has - // folded that row away, a second replay attributes the same content a - // second time and the activity timestamps become ambiguous. - const { status, ...stats } = await fullMigrate(req.yhub, req.docRef, { - force: req.query.force === 'true', - }); - // `status` is what a backfill driver records per document: 'ok', - // 'already', 'empty' (no legacy object — a brand-new document) or - // 'nothing' (versions exist, none readable). All four are done, hence - // one 2xx; `message` says the same thing to a human, `migrated` - // whether this call is the one that wrote the history. - const messages: Record = { - already: 'Already migrated', - empty: 'No legacy document in s3', - nothing: 'No usable content in the legacy versions', - ok: 'Migration completed', - }; - - return jsonResponse(200, { - status, - message: messages[status], - migrated: status === 'ok', - ...stats, - }); - }, - }, - }), - // POST /collaboration/create-ydoc/v1/{org}/{docid} — create a document's - // initial Yjs state from a RAW binary update (`Y.encodeStateAsUpdate` / - // pycrdt `get_update()` output) posted as application/octet-stream. - // - // The built-in `PATCH ydoc` takes the same update (base64, in a json body - // since 0.5.0) but neither of the two things this endpoint exists for: it is - // a strict create, answering 409 when the room already has content, and it - // attributes the content to the user named in `X-User-Id` rather than to the - // backend making the call. Reads have no such needs and use the built-in - // `GET ydoc`. Default access purpose: guarded like the built-in ydoc routes - // (write access on the doc — the admin JWT, or a user session with update - // ability). - createApiEndpoint('create-ydoc', { - post: { - handler: async (req) => { - // The endpoint facet already admitted this caller — only the admin - // token names this route — but the write itself is a separate facet, and - // yhub's contract is that a handler states the facets it touches. Cheap, - // and it keeps the grant honest if the tables above ever widen. - checkPermissions( - req.permissions, - createDocumentPermissions({ ydoc: '--u-' }), - ); - if (req.org !== ORG) { - return jsonResponse(400, { error: 'Unknown org' }); - } - if (!UUID4.test(req.docid)) { - return jsonResponse(400, { error: 'Room name is invalid' }); - } - if (req.branch !== 'main') { - // cookie users are main-only via `authorize`, but the admin - // token bypasses it — reject explicitly so an admin create can't - // seed an orphan non-main room (and dodge the 409 check, which is - // branch-scoped) - return jsonResponse(400, { error: 'Unknown branch' }); - } - const body = await req.bytes(); - // req.bytes() resolves to a Node Buffer, but the compute-task schema - // requires an exact Uint8Array (lib0 $constructedBy compares the - // constructor) — re-view the same bytes without copying - const update = new Uint8Array( - body.buffer, - body.byteOffset, - body.byteLength, - ); - if (update.byteLength > MAX_CREATE_BYTES) { - // 413 is missing from yhub's status-line map (503 was added in - // 0.5.0, 413 was not), so the reason phrase is empty - // ("HTTP/1.1 413 ") — legal, and callers switch on the code - return jsonResponse(413, { error: 'Update too large' }); - } - // yhub's "no effective content" convention — reject before it reaches - // a worker - if (update.byteLength <= EMPTY_UPDATE_MAX_BYTES) { - return jsonResponse(400, { error: 'Empty update' }); - } - // covers persisted state AND uncompacted stream messages. Not atomic - // with addMessage below (yhub has no atomic create): two concurrent - // creates can both pass the check and their updates merge — with - // independently generated updates (fresh clientIDs) the seeded - // content then appears twice. Accepted: Django creates each doc - // once, and a duplicated seed is user-fixable, unlike corruption. - const { gcDoc } = await req.yhub.getDoc( - req.docRef, - { gc: true, nongc: false }, - { gcOnMerge: false }, - ); - if (gcDoc != null && gcDoc.byteLength > EMPTY_UPDATE_MAX_BYTES) { - return jsonResponse(409, { error: 'Document already exists' }); - } - // Only the backend admin token may attribute the content to another - // user; regular callers always author as themselves — honoring a - // client-supplied header would let any editor forge the attribution - // history (the ws path likewise stamps the server-side identity). - const authInfo = req.authInfo as AppAuthInfo | null; - if (authInfo == null) { - return jsonResponse(401, { error: 'Authentication required' }); - } - const userid = - (authInfo.admin === true && req.headers['x-user-id']) || - authInfo.userid; - let result; - try { - // diffs the posted update against the (empty) current doc and - // stamps the attribution contentmap - result = await req.yhub.computePool.patchYdoc( - { - update, - currentDoc: gcDoc ?? EMPTY_YDOC, - userid, - customAttributions: [], - }, - { docRef: req.docRef }, - ); - } catch { - // a malformed update makes the compute worker throw (yhub logs - // 'worker failed' and replaces the thread). The update is the only - // untrusted input here, so a rejection maps to 400; getDoc / - // addMessage failures stay generic 500s. - return jsonResponse(400, { error: 'Invalid Yjs update' }); - } - if (result == null) { - // structurally valid but no effective content (e.g. delete-set - // only). A "successful" create that leaves the room nonexistent - // would lie to the caller — a later create would not 409. - return jsonResponse(400, { error: 'Empty update' }); - } - // on a fresh room this creates the stream, schedules compaction, and - // fans out to any live subscribers — nothing else to do - await req.yhub.stream.addMessage(req.docRef, { - type: 'ydoc:update:v1', - contentmap: result.contentmap, - update: result.update, - }); - return jsonResponse(201, { message: 'Document created' }); - }, - }, - }), - // POST /collaboration/restore-ydoc/v1/{org}/{docid} — undo the deletion of a - // document, putting back what `DELETE .../ydoc/` took away. - // - // Deleting has a built-in route, restoring does not: yhub 0.6.0 exposes - // `restoreDoc` to the process embedding it and nothing else. Backend-internal - // like reset-connections and migrate, gated to the admin token by the - // 'restore' purpose — a document leaves the trashbin because the backend - // says so, never because an editor asked. - createApiEndpoint('restore-ydoc', { - post: { - handler: async (req) => { - if (req.org !== ORG) { - return jsonResponse(400, { error: 'Unknown org' }); - } - if (!UUID4.test(req.docid)) { - return jsonResponse(400, { error: 'Room name is invalid' }); - } - if (req.branch !== 'main') { - // as in create-ydoc: the admin token is not fenced to main by - // `authorize`, and a deletion is recorded per branch - return jsonResponse(400, { error: 'Unknown branch' }); - } - // read the deletion before undoing it: `restoreDoc` throws a plain - // Error for a document whose content was erased, and that is a - // conflict to report as one — catching around the call would turn - // every failure alike, a database outage included, into the same answer - const tombstone = await req.yhub.persistence.retrieveTombstone( - req.docRef, - ); - if (tombstone == null) { - // not an error: the backend restores a whole subtree, of which only - // the part that was deleted with it has anything to put back - return jsonResponse(200, { - message: 'Document is not deleted', - restored: false, - }); - } - if (tombstone.hard || tombstone.purgedAt != null) { - return jsonResponse(409, { error: 'Document content was erased' }); - } - await req.yhub.restoreDoc(req.docRef); - return jsonResponse(200, { - message: 'Document restored', - restored: true, - }); - }, - }, - }), - // POST /collaboration/reset-ydoc/v1/{org}/{docid} — erase the content of a - // document and leave the room usable, as if it had never been written. - // - // What the backend's `clean_document` command needs to reset the onboarding - // sandbox: the Django document keeps its id and goes on being edited, so - // deleting the room is not an option — a hard deletion is final and even a - // soft one would answer 404 for a document that still exists. Backend-internal - // and admin-only, like the deletions it is built on: this destroys content - // with no way back. - createApiEndpoint('reset-ydoc', { - post: { - handler: async (req) => { - if (req.org !== ORG) { - return jsonResponse(400, { error: 'Unknown org' }); - } - if (!UUID4.test(req.docid)) { - return jsonResponse(400, { error: 'Room name is invalid' }); - } - if (req.branch !== 'main') { - return jsonResponse(400, { error: 'Unknown branch' }); - } - const authInfo = req.authInfo as AppAuthInfo | null; - const by = req.headers['x-user-id'] || authInfo?.userid; - if (!by) { - return jsonResponse(401, { error: 'Authentication required' }); - } - // No `checkPermissions` here, deliberately. The honest facet for what - // follows would be `delete: ['hard']`, and the admin grant withholds - // `'hard'` on purpose so that `DELETE /ydoc?hard=true` stays refused over - // REST. The erasure below reaches `deleteDoc` programmatically, which the - // `delete` facet does not gate; the endpoint facet — this route is named - // by no browser grant — is what admits the caller. - // Nothing compacts this room while the erasure runs: this drops the - // task already waiting for it and refuses to enqueue another, which - // leaves one writer to race with — a task a worker had claimed before - // this call. The tombstone barrier covers it right up to the moment - // the room is made writable again, so it can only land after that, - // and the second pass below is what picks it up. - await req.yhub.stream.disableCompaction(req.docRef); - try { - await eraseContent(req.yhub, req.docRef, by); - if (await hasContent(req.yhub, req.docRef)) { - resetLog.warn( - { docid: req.docid }, - 'content came back while it was being erased, erasing again', - ); - await eraseContent(req.yhub, req.docRef, by); - if (await hasContent(req.yhub, req.docRef)) { - // saying it is erased when it is not is the one answer this - // endpoint must never give - return jsonResponse(500, { - error: 'Document content came back after being erased', - }); - } - } - } finally { - // even on failure: leaving compaction off would freeze the room for - // every later edit, a worse state than the one we came to fix - await req.yhub.stream.enableCompaction(req.docRef); - } - return jsonResponse(200, { message: 'Document content erased' }); - }, - }, - }), -]; - -// Django orders the document lists by `updated_at` and no edit goes through it -// anymore, so it is told here that a document moved on. -const touchDocument = async (docid: string): Promise => { - if (backendSigningKey == null) return; - try { - const res = await fetch( - `${COLLABORATION_BACKEND_BASE_URL}/api/v1.0/documents/${docid}/content-updated/`, - { - method: 'POST', - headers: { authorization: `Bearer ${await getBackendToken()}` }, - signal: AbortSignal.timeout(BACKEND_NOTIFY_TIMEOUT_MS), - }, - ); - if (!res.ok) { - touchLog.warn( - { docid, status: res.status }, - 'backend refused the notification', - ); - } - } catch (err) { - // best effort: a lost notification only leaves `updated_at` behind until - // the document is edited again, it must never fail a compaction - touchLog.warn({ err, docid }, 'could not notify the backend'); - } -}; - // `docUpdate` is the worker event for "this compaction found new content": the // task returns before it when it has nothing to persist, so the awareness-only // traffic of someone merely opening a document never reaches it. Since yhub diff --git a/src/yhub-server/src/yhub.ts b/src/yhub-server/src/yhub.ts index 140573d92..2ce33dea3 100644 --- a/src/yhub-server/src/yhub.ts +++ b/src/yhub-server/src/yhub.ts @@ -9,3 +9,16 @@ export type { YHub }; /** `{ org, docid, branch }` — how yhub addresses one document on one branch. */ export type DocRef = Parameters[0]; + +// The caller identity `authenticate` returns and `authorize` is handed. yhub +// only requires `userid`; the rest is Docs' own, read back off `req.authInfo` +// (typed by yhub as the bare `{ userid }`, hence the cast at those sites). +// Declared here rather than in `server.ts` so `api.ts` can share the shape +// without importing from it. +export interface AppAuthInfo { + userid: string; + admin?: boolean; + endpoint?: string; + cookie?: string; + origin?: string; +}