mdg-filius/packages/server/test/blobs.test.ts

// The large binary files on F (F5.5): kept by their SHA-256 through POST
// /blobs from the editor's origin with a write grant, given back through GET
// /blobs/<sha> to whoever has a grant, with CORS for the editor's origins;
// refused from another origin, to a reader, in a frozen project, without a
// session, beyond the size. A room gets the blobs its document names before
// its simulation starts: a picture copied in a command line is the same
// blob, and a file it did not touch stays one.

import { randomUUID } from 'node:crypto';
import { readFileSync } from 'node:fs';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import WebSocket from 'ws';
import * as Y from 'yjs';
import { attachBlobs, BLOB_THRESHOLD_BYTES, blobSha, changeFiles, memoryBlobSide, nodeFiles, ROOTS, writeProject } from '@mdg-filius/doc';
import { openProjectArchive } from '@mdg-filius/fls';
import type { HostMessage } from '@mdg-filius/protocol';
import { BLOB_MAX_BYTES, PostgresBlobs, PostgresSessions, PostgresStore, shaOf } from '../src/index.js';
import { BLOB_ASK_AGAIN_MILLISECONDS, BLOB_ASKS } from '../src/simChannel.js';
import { testDatabase, type TestDatabase } from './database.js';
import { EDITOR_ORIGIN, newProject, SERVICE_KEY, signIn, startTestServer, until, type TestServer } from './harness.js';

let database: TestDatabase;
let store: PostgresStore;
let sessions: PostgresSessions;
let blobs: PostgresBlobs;

beforeAll(async () => {
  database = await testDatabase();
  store = PostgresStore.connect(database.url);
  sessions = new PostgresSessions(store.pool);
  blobs = new PostgresBlobs(store.pool);
  await store.migrate();
  await sessions.migrate();
  await blobs.migrate();
}, 60_000);

afterAll(async () => {
  await store?.close();
  await database?.stop();
});

const picture = (length = BLOB_THRESHOLD_BYTES * 2, seed = 7) => Uint8Array.from({ length }, (_, index) => (index * 13 + seed) % 256);

function post(server: TestServer, project: string, body: Uint8Array, headers: Record<string, string>): Promise<Response> {
  return fetch(`${server.base}/blobs?project=${project}`, { method: 'POST', headers: { 'content-type': 'application/octet-stream', ...headers }, body });
}
const get = (server: TestServer, project: string, sha: string, headers: Record<string, string>) => fetch(`${server.base}/blobs/${sha}?project=${project}`, { headers });

/** F's store, slower by so many milliseconds in reading. */
function slowly(inner: PostgresBlobs, { get }: { readonly get: number }): PostgresBlobs {
  return Object.assign(Object.create(inner) as PostgresBlobs, {
    get: async (shas: readonly string[]) => (await new Promise((resolve) => setTimeout(resolve, get)), inner.get(shas)),
  });
}

/** FILIUS' example with a picture as a blob on the notebook (its bytes not in F's store). */
async function projectWithPicture(seed: number) {
  const project = randomUUID();
  const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
  const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
  const doc = new Y.Doc();
  attachBlobs(doc, memoryBlobSide());
  writeProject(doc, imported);
  const bytes = picture(BLOB_THRESHOLD_BYTES * 2, seed);
  changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/spaet.png', bytes, 'image/png'));
  await store.create(project, Y.encodeStateAsUpdate(doc));
  return { project, notebook, bytes };
}

/** In a project's room by WebSocket, signed in with a write grant. */
async function enterRoom(server: TestServer, project: string) {
  const cookie = await signIn(server, project);
  const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
  const got: HostMessage[] = [];
  socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
  socket.on('error', () => {});
  await new Promise((resolve) => socket.on('open', resolve));
  return { got, send: (message: object) => socket.send(JSON.stringify({ v: 1, ...message })), close: () => socket.close() };
}

/** A computer's files as F's store holds them. */
async function storedFiles(project: string, node: string): Promise<Y.Map<Y.Map<unknown>>> {
  const stored = await store.load(project);
  const doc = new Y.Doc();
  Y.applyUpdate(doc, stored!.state);
  for (const update of stored!.updates) Y.applyUpdate(doc, update);
  return doc.getMap<Y.Map<unknown>>(ROOTS.nodes).get(node)!.get('files') as Y.Map<Y.Map<unknown>>;
}

describe('the blobs on F', () => {
  it('keeps bytes under the name F gives them, and gives them back to whoever has a grant', async () => {
    const server = await startTestServer(store, sessions);
    const project = await newProject(store);
    const cookie = await signIn(server, project, { login: 'ab12' });
    const bytes = picture();
    const kept = await post(server, project, bytes, { cookie, origin: EDITOR_ORIGIN });
    expect([kept.status, await kept.json()]).toEqual([201, { sha: blobSha(bytes) }]);
    expect(kept.headers.get('access-control-allow-origin')).toBe(EDITOR_ORIGIN);
    expect(kept.headers.get('access-control-allow-credentials')).toBe('true');
    expect(shaOf(bytes)).toBe(blobSha(bytes));
    // Again: the same name, kept once.
    expect(await (await post(server, project, bytes, { cookie, origin: EDITOR_ORIGIN })).json()).toEqual({ sha: blobSha(bytes) });
    expect((await store.pool.query('SELECT count(*)::int AS n FROM filius_blob WHERE sha = $1', [blobSha(bytes)])).rows[0].n).toBe(1);
    const reader = await signIn(server, project, { level: 'read' });
    const back = await get(server, project, blobSha(bytes), { cookie: reader, origin: EDITOR_ORIGIN });
    expect(back.status).toBe(200);
    expect(new Uint8Array(await back.arrayBuffer())).toEqual(bytes);
    expect([back.headers.get('content-type'), back.headers.get('cache-control'), back.headers.get('access-control-allow-origin')]).toEqual([
      'application/octet-stream',
      'private, max-age=31536000, immutable',
      EDITOR_ORIGIN,
    ]);
    // Without an origin (the same origin), no CORS.
    expect((await get(server, project, blobSha(bytes), { cookie: reader })).headers.get('access-control-allow-origin')).toBeNull();
    expect(server.log.lines.find((line) => line.event === 'blob.kept')).toMatchObject({ project, user: 'ab12', sha: blobSha(bytes), size: bytes.length });
    await server.close();
  });

  it('answers the preflight of the editor’s origin, and no other', async () => {
    const server = await startTestServer(store, sessions);
    const ask = (origin: string) => fetch(`${server.base}/blobs?project=x`, { method: 'OPTIONS', headers: { origin, 'access-control-request-method': 'POST' } });
    const ours = await ask(EDITOR_ORIGIN);
    expect([ours.status, ours.headers.get('access-control-allow-methods'), ours.headers.get('access-control-allow-headers')]).toEqual([204, 'GET, POST', 'content-type']);
    expect((await ask('https://elsewhere.test')).status).toBe(403);
    await server.close();
  });

  it('refuses another origin, a reader, a frozen project, no session, no project and too many bytes', async () => {
    const server = await startTestServer(store, sessions);
    const project = await newProject(store);
    const cookie = await signIn(server, project);
    const reader = await signIn(server, project, { level: 'read' });
    const bytes = picture(100);
    const answer = async (response: Response) => [response.status, await response.json()];
    expect(await answer(await post(server, project, bytes, { cookie }))).toEqual([403, { error: 'foreignOrigin' }]);
    expect(await answer(await post(server, project, bytes, { cookie, origin: 'https://elsewhere.test' }))).toEqual([403, { error: 'foreignOrigin' }]);
    expect(await answer(await post(server, project, bytes, { cookie: reader, origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'readOnly' }]);
    expect(await answer(await post(server, project, bytes, { origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'notSignedIn' }]);
    expect(await answer(await post(server, 'nothing', bytes, { cookie, origin: EDITOR_ORIGIN }))).toEqual([400, { error: 'notAProject' }]);
    expect(await answer(await post(server, randomUUID(), bytes, { cookie, origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'unknownProject' }]);
    expect(await answer(await post(server, project, new Uint8Array(BLOB_MAX_BYTES + 1), { cookie, origin: EDITOR_ORIGIN }))).toEqual([413, { error: 'tooLarge' }]);
    expect(server.log.lines.filter((line) => line.event === 'blob.foreignOrigin')).toHaveLength(2);
    await fetch(`${server.base}/svc/projects/${project}/freeze`, { method: 'POST', headers: { 'x-filius-service-key': SERVICE_KEY }, body: JSON.stringify({ frozen: true }) });
    expect(await answer(await post(server, project, bytes, { cookie, origin: EDITOR_ORIGIN }))).toEqual([403, { error: 'frozen' }]);
    expect(await answer(await get(server, project, 'f'.repeat(64), { cookie }))).toEqual([404, { error: 'unknownBlob' }]);
    expect(await answer(await get(server, project, 'not-a-sha', { cookie }))).toEqual([404, { error: 'unknownBlob' }]);
    expect(await answer(await get(server, randomUUID(), blobSha(bytes), { cookie }))).toEqual([403, { error: 'unknownProject' }]);
    expect(await answer(await fetch(`${server.base}/blobs/x/y?project=${project}`, { headers: { cookie } }))).toEqual([404, { error: 'notFound' }]);
    expect(BLOB_MAX_BYTES).toBe(16 * 1024 * 1024);
    await server.close();
  });

  it('gives a room the blobs of its document before the simulation starts', async () => {
    const server = await startTestServer(store, sessions);
    // FILIUS' example, with a picture as a blob on the notebook.
    const project = randomUUID();
    const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
    const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
    const doc = new Y.Doc();
    const side = memoryBlobSide();
    attachBlobs(doc, side);
    writeProject(doc, imported);
    const bytes = picture();
    changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/bild.png', bytes, 'image/png'));
    await blobs.put(bytes);
    await store.create(project, Y.encodeStateAsUpdate(doc));
    const cookie = await signIn(server, project);
    const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
    const got: HostMessage[] = [];
    socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
    await new Promise((resolve) => socket.on('open', resolve));
    const send = (message: object) => socket.send(JSON.stringify({ v: 1, ...message }));
    send({ kind: 'start' });
    await until(() => got.some((message) => message.kind === 'state' && message.mode === 'simulation'));
    send({ kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'cp bild.png kopie.png' }, reqId: 'r' });
    await until(() => got.some((message) => message.kind === 'ack'));
    send({ kind: 'stop' });
    await until(() => got.filter((message) => message.kind === 'state' && message.mode === 'design').length === 2);
    socket.close();
    await server.close();
    const stored = await store.load(project);
    const after = new Y.Doc();
    Y.applyUpdate(after, stored!.state);
    for (const update of stored!.updates) Y.applyUpdate(after, update);
    const files = after.getMap<Y.Map<unknown>>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map<Y.Map<unknown>>;
    // The copy is the same blob: the room had the bytes; the original is untouched.
    expect([files.get('/kopie.png')?.get('blob'), files.get('/bild.png')?.get('blob')]).toEqual([blobSha(bytes), blobSha(bytes)]);
    expect(nodeFiles(after, notebook)!.entry('/kopie.png')!.content).toMatchObject({ kind: 'image', mediaType: 'image/png' });
  }, 30_000);

  it('asks again for a blob F gets only after the start, and the running simulation reads it', async () => {
    const server = await startTestServer(store, sessions);
    const project = randomUUID();
    const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
    const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
    const doc = new Y.Doc();
    attachBlobs(doc, memoryBlobSide());
    writeProject(doc, imported);
    const bytes = picture(BLOB_THRESHOLD_BYTES * 2, 42);
    changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/spaet.png', bytes, 'image/png'));
    await store.create(project, Y.encodeStateAsUpdate(doc));
    const cookie = await signIn(server, project);
    const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
    const got: HostMessage[] = [];
    socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
    await new Promise((resolve) => socket.on('open', resolve));
    socket.send(JSON.stringify({ v: 1, kind: 'start' }));
    await until(() => got.some((message) => message.kind === 'state' && message.mode === 'simulation'));
    // Taken to F only now (an editor still uploading it); the thread asks again a second later.
    await blobs.put(bytes);
    await new Promise((resolve) => setTimeout(resolve, BLOB_ASK_AGAIN_MILLISECONDS + 500));
    socket.send(JSON.stringify({ v: 1, kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'cp spaet.png kopie.png' }, reqId: 'r' }));
    await until(() => got.some((message) => message.kind === 'ack'));
    socket.close();
    await server.close();
    const stored = await store.load(project);
    const after = new Y.Doc();
    Y.applyUpdate(after, stored!.state);
    for (const update of stored!.updates) Y.applyUpdate(after, update);
    const files = after.getMap<Y.Map<unknown>>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map<Y.Map<unknown>>;
    expect(files.get('/kopie.png')?.get('blob')).toBe(blobSha(bytes));
    expect(BLOB_ASKS).toBe(5);
  }, 30_000);

  it("starts a simulation only once the blobs asked for have come, however slow F's store is", async () => {
    const { project, notebook, bytes } = await projectWithPicture(51);
    await blobs.put(bytes);
    const server = await startTestServer(store, sessions, { blobs: slowly(blobs, { get: 800 }) });
    const room = await enterRoom(server, project);
    room.send({ kind: 'start' });
    await until(() => room.got.some((message) => message.kind === 'state' && message.mode === 'simulation'), 5_000);
    room.send({ kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'cp spaet.png kopie.png' }, reqId: 'r' });
    await until(() => room.got.some((message) => message.kind === 'ack'));
    room.close();
    await server.close();
    expect((await storedFiles(project, notebook)).get('/kopie.png')?.get('blob')).toBe(blobSha(bytes));
  }, 30_000);

  it('starts a room whose blob F has not got, and empties no file by it', async () => {
    const server = await startTestServer(store, sessions);
    const project = randomUUID();
    const imported = openProjectArchive(new Uint8Array(readFileSync(new URL('../../../corpus/java-examples/internetwork_komplett.fls', import.meta.url))), () => randomUUID());
    const notebook = imported.graph.nodes.find((each) => each.displayName === 'Notebook 1.10')!.id;
    const doc = new Y.Doc();
    attachBlobs(doc, memoryBlobSide());
    writeProject(doc, imported);
    const bytes = picture(BLOB_THRESHOLD_BYTES * 3, 99);
    changeFiles(doc, notebook, 'person', (files) => files.writeImageFile('/fehlt.png', bytes, 'image/png'));
    await store.create(project, Y.encodeStateAsUpdate(doc));
    const cookie = await signIn(server, project);
    const socket = new WebSocket(`ws://127.0.0.1:${server.port}/sim?project=${project}`, { headers: { cookie } });
    const got: HostMessage[] = [];
    socket.on('message', (data) => got.push(JSON.parse(data.toString()) as HostMessage));
    await new Promise((resolve) => socket.on('open', resolve));
    socket.send(JSON.stringify({ v: 1, kind: 'start' }));
    await until(() => got.some((message) => message.kind === 'state' && message.mode === 'simulation'));
    socket.send(JSON.stringify({ v: 1, kind: 'input', node: notebook, app: 'terminal', action: 'enter', args: { line: 'touch neu.txt' }, reqId: 'r' }));
    await until(() => got.some((message) => message.kind === 'ack'));
    socket.close();
    await server.close();
    const stored = await store.load(project);
    const after = new Y.Doc();
    Y.applyUpdate(after, stored!.state);
    for (const update of stored!.updates) Y.applyUpdate(after, update);
    const files = after.getMap<Y.Map<unknown>>(ROOTS.nodes).get(notebook)!.get('files') as Y.Map<Y.Map<unknown>>;
    expect([files.get('/fehlt.png')?.get('blob'), files.has('/neu.txt')]).toEqual([blobSha(bytes), true]);
  }, 30_000);
});