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 {
  calculateJwkThumbprint,
  createRemoteJWKSet,
  exportJWK,
  importPKCS8,
  jwtVerify,
  SignJWT,
} from 'jose';

import { secret } from './env.js';
// Docs' access policy as yhub permission objects — see permissions.test.js
import {
  adminDocumentPermissions,
  browserDocumentPermissions,
  publicGlobalPermissions,
} from './permissions.js';
// legacy Django/S3 document store — see migration.js and README.md
import {
  SOFT_MIGRATION,
  fullMigrate,
  isPermanentFailure,
  maybeMigrate,
  migrationLog,
} from './migration.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, dflt, min = 1) => {
  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.js): 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 =
  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),
        };
      })();

/**
 * @type {{ token: string, expiresAt: number } | null}
 */
let backendToken = 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 () => {
  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, { cookie, origin }) => {
  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 = new Error(`Failed to fetch ${path}: ${res.status}`);
    err.status = res.status;
    throw err;
  }
  return res.json();
};

// 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)
// is guaranteed to include the seed.
//
// Seeding never decides whether the caller may read the document — that is the
// backend's answer alone. There are two ways this ends other than a seed:
//
//   the legacy object cannot be migrated (it does not decode) — retrying will
//     not change that, so the room opens as a new document. Refusing instead
//     would lock a document nobody can repair from the outside. Logged per
//     access, because the caller is now editing alongside legacy content that
//     stayed behind in S3.
//   the legacy store could not be reached (timeout, network, backpressure) —
//     the same request later may well succeed, so it answers 503 rather than
//     silently starting an empty document on top of content that exists.
const seedFromLegacyStore = async (docRef) => {
  try {
    // `yhub` is declared at the bottom of this file — safe: auth callbacks only
    // fire once the server is up, i.e. after that assignment
    await maybeMigrate(yhub, docRef);
  } catch (err) {
    if (!isPermanentFailure(err)) {
      throw apiError(503, 'Legacy document store is unavailable');
    }
    // why it failed was logged once, at the attempt, inside maybeMigrate
    migrationLog.warn(
      { event: 'seed.skipped', docid: docRef.docid, err: err?.message },
      'admitting caller to a document that could not be migrated; it opens as new',
    );
  }
};

const auth = createAuthPlugin({
  /**
   * Who is asking. Returning `null` here would mean "an anonymous caller" — it is
   * not a refusal, and `authorize` is still asked — so every rejection below is a
   * thrown `apiError(401)` and every unauthenticated caller gets an identity.
   * Under 0.7 this callback denied by returning `null`, which is the one change
   * in this file that would fail open rather than closed if it were missed.
   */
  async authenticate(req) {
    // uws req is only valid synchronously — read headers AND the url before the
    // first await.
    const authorization = req.getHeader('authorization');
    const cookie = req.getHeader('cookie');
    const origin = req.getHeader('origin');
    // 0.8 dropped `accessPurpose`, and `authorize` is told the scope and the
    // resource but not the route. The seed below needs to know one thing about
    // the route — whether this is the `migrate` call — so the endpoint name is
    // read off the path here, where the request still exists, and carried on the
    // identity. `/collaboration/{name}/{version}/...` → index 2.
    const endpoint = req.getUrl().split('/')[2] ?? '';

    if (authorization !== '') {
      // backend-to-server call: RS256 JWT signed by Django, verified against
      // its JWKS. A browser cannot attach an Authorization header to a ws
      // upgrade or a credentialed cross-origin fetch, so this never shadows a
      // real user session. present-but-invalid is rejected here (401) instead of
      // falling through to the cookie flow, which would mask a misconfiguration
      // as an anonymous browse.
      const token = authorization.startsWith('Bearer ')
        ? authorization.slice('Bearer '.length)
        : authorization;
      let payload;
      try {
        // clockTolerance absorbs Django's cache-at-exp race (the admin token
        // is cached for exactly its lifetime, so it can arrive here moments
        // after exp) plus small clock skew — without it a kick would be
        // silently dropped as a 401.
        ({ payload } = await jwtVerify(token, JWKS, {
          algorithms: ['RS256'],
          audience: YHUB_AUDIENCE,
          clockTolerance: 5,
        }));
      } catch (err) {
        // jose tags every token-validation failure with an `ERR_J…` code (bad
        // signature, expired, wrong or missing audience) — those are permanent,
        // fail closed. A JWKS fetch that times out or never connects has no
        // such code (or ERR_JWKS_TIMEOUT): the token may be perfectly valid and
        // we simply cannot check it, so report it as retryable instead of
        // accusing the caller of forging it.
        if (
          err?.code === 'ERR_JWKS_TIMEOUT' ||
          typeof err?.code !== 'string' ||
          !err.code.startsWith('ERR_J')
        ) {
          throw apiError(503, 'Token verification keys are unavailable');
        }
        throw apiError(401, 'Invalid token');
      }
      if (payload.admin !== true) {
        // a valid token that is not an admin one is a credential we refuse, not
        // an anonymous caller: returning null here — which is what 0.7 did —
        // would now silently downgrade a service call into a public browse
        throw apiError(401, 'Not an admin token');
      }
      // admin tokens act as the "system" user (no per-user admin identities yet)
      return { userid: 'system', admin: true, endpoint };
    }
    // No origin check here: `server.cors` below is the allowlist, and yhub applies it to the
    // websocket upgrade and to every REST request before this runs. Checking it a second time
    // would also refuse the http fallback's polls — a same-origin `fetch` GET carries no
    // `Origin` header at all, so on a deployment where the page and /collaboration/ share a
    // host every round would 401 while the PATCH beside it succeeded.
    //
    // A caller with no cookie at all is anonymous too: the probes arrive that way, and so does
    // a public document opened in a fresh browser. What they may do is `authorize`'s answer.
    if (!cookie) {
      return { userid: ANONYMOUS_USERID, origin, endpoint };
    }
    try {
      const user = await backendFetch('/api/v1.0/users/me/', {
        cookie,
        origin,
      });
      return { userid: String(user.id), cookie, origin, endpoint };
    } catch (err) {
      // Only a genuine "not signed in" falls back to the anonymous identity.
      // On backend failure (5xx/network) still refuse to admit the connection —
      // a signed-in editor authorized under an anon userid would be invisible
      // to the targeted reset-connections recheck (users: [<uuid>]) for the
      // connection's whole lifetime — but report it as retryable rather than as
      // an authentication failure the client should give up on.
      if (err?.status !== 401 && err?.status !== 403) {
        throw apiError(503, 'Authentication backend is unavailable');
      }
      // the cookie is kept: it is a session the backend still answers document
      // questions about, which is how a public document resolves below
      return { userid: ANONYMOUS_USERID, cookie, origin, endpoint };
    }
  },
  /**
   * What that caller may do. One handler per scope; a scope without a handler
   * denies, which is what closes `org` and `branch` here. Denial is a value —
   * returning `null`, never throwing: an error thrown during a websocket recheck
   * disconnects with the transient close code 1013 instead of the revoke code
   * 4401.
   */
  authorize: createAuthorize({
    async document({ org, docid, branch }, user) {
      if (user?.admin === true) {
        // Django's admin token: full access. It still goes through the legacy
        // seed, on the same terms as a user, but not for `migrate` — that call
        // replays the whole version history itself, and seeding first would put
        // the newest version in the room underneath it. Without the seed a
        // backend read of an unmigrated document would answer with an *empty*
        // doc, and a create-ydoc against one would write a second lineage next
        // to the legacy content the first user access is about to seed in.
        // Access itself is never in question here — the token already granted it.
        // The same org/branch fence the user path applies below. The admin token
        // is the only identity that can name an arbitrary org or branch, and the
        // legacy store is branchless — `{docid}/file` *is* main — so seeding any
        // other room would write main's content into an orphan room, and the
        // per-docid verdict cache would then report that docid as done and leave
        // the real room empty.
        if (
          SOFT_MIGRATION &&
          user.endpoint !== 'migrate' &&
          org === ORG &&
          branch === 'main' &&
          UUID4.test(docid)
        ) {
          await seedFromLegacyStore({ org, docid, branch });
        }
        return adminDocumentPermissions;
      }
      if (org !== ORG || branch !== 'main' || !UUID4.test(docid)) {
        return null;
      }
      let doc;
      try {
        doc = await backendFetch(`/api/v1.0/documents/${docid}/`, user);
      } catch (err) {
        // the backend answered "no": a real, permanent denial (403 Forbidden)
        if (err?.status === 401 || err?.status === 403 || err?.status === 404) {
          return null;
        }
        // it did not answer at all — say so, so the caller retries instead of
        // reading a 5xx or a network blip as a permission decision
        throw apiError(503, 'Document authorization backend is unavailable');
      }
      if (!doc.abilities?.retrieve) {
        return null;
      }
      // the backend has already decided the caller may read this document; the
      // seed only decides what is in it
      if (SOFT_MIGRATION) {
        await seedFromLegacyStore({ org, docid, branch });
      }
      // When this caller was given access, which is where the history they may
      // read starts. The backend sends ISO-8601 (null for a link-reach reader,
      // who holds no access and so has no date); `history.from` is unix ms.
      //
      // Anything unparseable is *no* history rather than full history, and zero
      // is refused with it: `from: 0` is the one value that also unlocks a
      // `gc=false` websocket, and no real access date is ever zero, so a zero
      // here could only ever be a bug upstream.
      const accessSince = Date.parse(doc.user_access_since ?? '');
      return browserDocumentPermissions(
        doc.abilities.update === true,
        Number.isFinite(accessSince) && accessSince > 0 ? accessSince : null,
      );
    },
    async global() {
      return publicGlobalPermissions;
    },
  }),
});

// 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, body) =>
  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 (yhub, docRef) => {
  const { gcDoc } = await yhub.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 (yhub, docRef, by) => {
  await yhub.deleteDoc(docRef, { hard: true, by });
  await yhub.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, probe) => {
  let timer;
  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?.message }, '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.js)
          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 = {
          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 userid =
          (req.authInfo.admin === true && req.headers['x-user-id']) ||
          req.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 by = req.headers['x-user-id'] || req.authInfo.userid;
        // 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) => {
  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
// 0.5.0 it is handed the room of the task alongside the merged document.
const workerEvents = {
  docUpdate: ({ docRef }) => {
    // Django knows the documents of this org, on the main branch, by their uuid
    if (
      docRef.org !== ORG ||
      docRef.branch !== 'main' ||
      !UUID4.test(docRef.docid)
    ) {
      return;
    }
    // deliberately not awaited: a slow backend must not hold the worker
    touchDocument(docRef.docid);
  },
};

// The persistence plugins yhub consults, in order, before writing a blob to
// postgres and before reading one back. An empty list keeps everything in the
// database, which is what a deployment that names no bucket gets.
//
// Naming a bucket is enough to get the plugin, whether or not the toggle asks
// for the writing: reading is the half that must never be taken away, since the
// objects an earlier run wrote are the only copy of those versions. `branches`
// is what the toggle actually moves — every branch, or none, which is a plugin
// that retrieves and deletes what is in the bucket and adds nothing to it.
//
// Read here rather than in the call below so that an incomplete configuration
// is a startup error naming what is missing: the client would otherwise be
// built anonymous or against the wrong host and only say so on the first
// compaction, which is a background task — the failure would show up as
// documents quietly not being persisted.
const persistencePlugins = () => {
  const settings = [
    ['YHUB_S3_ENDPOINT_URL', YHUB_S3_ENDPOINT_URL],
    ['YHUB_S3_ACCESS_KEY_ID', YHUB_S3_ACCESS_KEY_ID],
    ['YHUB_S3_SECRET_ACCESS_KEY', YHUB_S3_SECRET_ACCESS_KEY],
    ['YHUB_S3_BUCKET_NAME', YHUB_S3_BUCKET_NAME],
  ];
  const missing = settings.filter(([, value]) => !value).map(([name]) => name);
  // no bucket named at all, and the toggle does not ask for one: postgres
  // alone, and no object anywhere that would need reading back
  if (missing.length === settings.length && !S3_PERSISTENCE) return [];
  if (missing.length > 0) {
    // half a configuration is always a mistake, and the half that is set says
    // which mistake: a bucket was meant to be reachable and is not
    const named = missing.join(', ');
    throw new Error(
      S3_PERSISTENCE
        ? `YHUB_S3_PERSISTENCE=true requires ${named}`
        : `The YHUB_S3_* bucket is partly configured, missing ${named}`,
    );
  }

  const url = new URL(YHUB_S3_ENDPOINT_URL);
  if (url.pathname !== '/' && url.pathname !== '') {
    // the client is given a host and a port, so a base path would be dropped
    // without a word and the objects written next to where they belong
    throw new Error('YHUB_S3_ENDPOINT_URL must not contain a path');
  }
  if (url.protocol !== 'http:' && url.protocol !== 'https:') {
    // the client is told "SSL or not", so any other scheme would read as "not"
    // and send the credentials in clear
    throw new Error('YHUB_S3_ENDPOINT_URL must be http:// or https://');
  }
  const useSSL = url.protocol === 'https:';

  return [
    new S3PersistenceV1({
      bucket: YHUB_S3_BUCKET_NAME,
      endPoint: url.hostname,
      // an implicit port parses as "", which the client reads as 0 — its way
      // of saying "whatever the scheme defaults to"
      port: Number(url.port),
      useSSL,
      accessKey: YHUB_S3_ACCESS_KEY_ID,
      secretKey: YHUB_S3_SECRET_ACCESS_KEY,
      // left out rather than passed empty: unset, the client discovers the
      // region of the bucket instead of validating an empty string
      ...(YHUB_S3_REGION_NAME ? { region: YHUB_S3_REGION_NAME } : {}),
      // The branches whose blobs are written here: all of them, or none, which
      // is how the plugin is kept for reading while the writing goes back to
      // postgres. `store` declines a branch it is not given and yhub falls
      // through to the database, while `retrieve` and `delete` go on answering
      // for every object already in the bucket.
      branches: S3_PERSISTENCE ? true : [],
      // On a versioned bucket a plain delete deletes nothing: it writes a
      // delete marker over the object and keeps every version underneath it.
      // Each compaction supersedes the blobs of the one before, so that would
      // be every version ever written kept forever — and a document a user
      // asked to erase still readable by anyone who can list versions. The
      // plugin records the version id it wrote at store time so that the
      // delete can name it, which is what removes the bytes. Asked for here
      // rather than left to the plugin's default: the delete is only as
      // thorough as this line.
      deleteVersions: true,
    }),
  ];
};

// the instance is referenced by the soft-migration helpers above — safe: auth
// callbacks only fire once the server is up, i.e. after this assignment
const yhub = await createYHub({
  redis: {
    url: REDIS,
    prefix: REDIS_PREFIX,
    taskDebounce: TASK_DEBOUNCE_MS,
    minMessageLifetime: MIN_MESSAGE_LIFETIME_MS,
  },
  postgres: POSTGRES,
  // where the blobs live: nothing here keeps them in yhub's postgres
  persistence: persistencePlugins(),
  // Both halves are declared, and YHUB_ROLE decides which are built: a null
  // server binds no port at all (a `worker` pod has no http surface, hence no
  // probes and no service in front of it), a null worker claims no task.
  //
  // apiPrefix mounts every route — built-ins, our custom endpoints, and the
  // websocket (/collaboration/ws/v1/{org}/{docid}) — under /collaboration/.
  server: RUNS_SERVER
    ? {
        port: PORT,
        auth,
        api,
        apiPrefix: API_PREFIX,
        // What a browser may reach this server from, applied by yhub to the websocket upgrade
        // and to every REST route — the only origin check there is, `authenticate` no longer
        // does its own. `credentials` is what lets the http fallback send the session cookie
        // on a cross-origin `fetch`; it is also why the list has to be concrete, browsers
        // refusing "*" together with Access-Control-Allow-Credentials.
        cors: { origin: allowedOrigins, credentials: true },
      }
    : null,
  worker: RUNS_WORKER
    ? { taskConcurrency: TASK_CONCURRENCY, events: workerEvents }
    : null,
});

// What this process was configured to be, in one line: yhub's own startup log
// reports neither the role nor the stream settings, and every one of them is an
// environment variable a deployment can get wrong. The two timings are read
// back off the instance rather than from the constants above, so the line says
// what yhub is using and not merely what it was asked for.
logger.info(
  {
    role: ROLE,
    server: RUNS_SERVER,
    worker: RUNS_WORKER,
    taskConcurrency: RUNS_WORKER ? TASK_CONCURRENCY : null,
    // the bucket the plugin is attached to, null when there is no plugin at
    // all, and whether the compaction blobs are written to it or to yhub's own
    // postgres — a bucket with `s3Writes` false is one that is only read
    s3Bucket: YHUB_S3_BUCKET_NAME ?? null,
    s3Writes: S3_PERSISTENCE,
    taskDebounceMs: yhub.stream.taskDebounce,
    minMessageLifetimeMs: yhub.stream.minMessageLifetime,
  },
  'yhub configuration',
);