From d4b61891d993a2c45de63c484def14662ba3f097 Mon Sep 17 00:00:00 2001 From: Anthony LC Date: Mon, 7 Sep 2026 18:06:50 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=8E=A8(yhub)=20split=20server.ts=20into?= =?UTF-8?q?=20focused=20modules?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Divide server.ts into 3 focused modules: - config.ts: every environment variable the server reads, parsed and validated at import time, - backend.ts: the calls made to the Docs Django backend, the token signing and the JWKS both ends verify each other with, - api.ts: the custom REST endpoints mounted under /collaboration/. server.ts keeps only what wires these together: the auth plugin and its legacy-store seed (which closes over the yhub instance), the persistence plugins, the worker events and the createYHub call. The AppAuthInfo identity shape moves to yhub.ts so server.ts and api.ts can share it without importing from each other. Tests follow the split: config.spec.ts and backend.spec.ts are new, server.spec.ts narrows to the composition root's boot contract. No behaviour change. --- src/yhub-server/README.md | 24 +- src/yhub-server/__tests__/backend.spec.ts | 150 +++++ src/yhub-server/__tests__/config.spec.ts | 121 ++++ src/yhub-server/__tests__/server.spec.ts | 75 +-- src/yhub-server/src/api.ts | 464 +++++++++++++ src/yhub-server/src/backend.ts | 165 +++++ src/yhub-server/src/config.ts | 157 +++++ src/yhub-server/src/server.ts | 779 ++-------------------- src/yhub-server/src/yhub.ts | 13 + 9 files changed, 1144 insertions(+), 804 deletions(-) create mode 100644 src/yhub-server/__tests__/backend.spec.ts create mode 100644 src/yhub-server/__tests__/config.spec.ts create mode 100644 src/yhub-server/src/api.ts create mode 100644 src/yhub-server/src/backend.ts create mode 100644 src/yhub-server/src/config.ts 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; +}