chadtmiller.com/pds.js

Switch repository

Loading…

import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
  buildCarFile,
  CID,
  cborEncode,
  cborEncodeDagCbor,
  cidToBytes,
  cidToString,
  createCid,
} from '@pdsjs/core/repo';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { WebSocketServer } from 'ws';
import { watch } from '../src/daemon.js';
import { subscribeRepos } from '../src/firehose.js';
import { createStateStore } from '../src/state.js';

const COLLECTION = 'dev.pdsjs.git.repo';
const DID = 'did:plc:abc123';

/**
 * @param {number} seq
 * @param {string} sha
 */
async function commitFrame(seq, sha) {
  const value = {
    name: 'proj',
    refs: [{ name: 'refs/heads/main', sha }],
    bundles: [],
    createdAt: '2026-08-18T00:00:00.000Z',
  };
  const data = cborEncodeDagCbor(value);
  const cid = cidToString(await createCid(data));
  const header = cborEncode({ op: 1, t: '#commit' });
  const body = cborEncodeDagCbor({
    seq,
    repo: DID,
    rev: '3ktest',
    blocks: buildCarFile(cid, [{ cid, data }]),
    ops: [
      {
        action: 'update',
        path: `${COLLECTION}/proj`,
        cid: new CID(cidToBytes(cid)),
      },
    ],
    time: '2026-08-18T00:00:00.000Z',
  });
  const frame = new Uint8Array(header.length + body.length);
  frame.set(header, 0);
  frame.set(body, header.length);
  return frame;
}

/** @type {WebSocketServer} */
let server;
/** @type {string} */
let service;
/** @type {string[]} */
let requestedUrls;

beforeEach(async () => {
  requestedUrls = [];
  server = new WebSocketServer({ port: 0 });
  await new Promise((resolve) => server.once('listening', resolve));
  const address = /** @type {{port: number}} */ (server.address());
  service = `http://127.0.0.1:${address.port}`;
});

afterEach(async () => {
  await new Promise((resolve) => server.close(resolve));
});

/**
 * @param {(connection: number) => Promise<Uint8Array[]>} framesFor
 * @param {{closeAfter?: boolean}} [options]
 */
function serve(framesFor, options = {}) {
  let connections = 0;
  server.on('connection', async (socket, request) => {
    requestedUrls.push(request.url ?? '');
    const frames = await framesFor(connections++);
    for (const frame of frames) socket.send(frame);
    if (options.closeAfter) socket.close();
  });
}

describe('subscribeRepos', () => {
  it('delivers every record write on the socket', async () => {
    serve(async () => [await commitFrame(1, 'a'.repeat(40))]);

    /** @type {import('../src/firehose.js').RecordWrite[]} */
    const writes = [];
    const controller = new AbortController();
    const done = subscribeRepos({
      service,
      collections: [COLLECTION],
      signal: controller.signal,
      onWrite: (write) => {
        writes.push(write);
        controller.abort();
      },
    });
    await done;

    expect(writes).toHaveLength(1);
    expect(writes[0].rkey).toBe('proj');
    expect(writes[0].value.refs[0].sha).toBe('a'.repeat(40));
  });

  it('reconnects past the last sequence it saw', async () => {
    serve(
      async (connection) =>
        connection === 0
          ? [await commitFrame(7, 'a'.repeat(40))]
          : [await commitFrame(8, 'b'.repeat(40))],
      { closeAfter: true },
    );

    /** @type {number[]} */
    const seqs = [];
    const controller = new AbortController();
    await subscribeRepos({
      service,
      collections: [COLLECTION],
      signal: controller.signal,
      onWrite: () => {},
      onSeq: (seq) => {
        seqs.push(seq);
        if (seqs.length === 2) controller.abort();
      },
    });

    expect(seqs).toEqual([7, 8]);
    expect(requestedUrls[0]).not.toContain('cursor');
    expect(requestedUrls[1]).toContain('cursor=7');
  });

  it('resumes from a cursor it was given', async () => {
    serve(async () => [await commitFrame(1, 'a'.repeat(40))]);

    const controller = new AbortController();
    await subscribeRepos({
      service,
      collections: [COLLECTION],
      cursor: 99,
      signal: controller.signal,
      onWrite: () => controller.abort(),
    });

    expect(requestedUrls[0]).toContain('cursor=99');
  });

  it('reports an error frame without dropping the connection', async () => {
    serve(async () => {
      const header = cborEncode({ op: -1 });
      const body = cborEncode({ error: 'FutureCursor', message: 'too far' });
      const frame = new Uint8Array(header.length + body.length);
      frame.set(header, 0);
      frame.set(body, header.length);
      return [frame, await commitFrame(1, 'a'.repeat(40))];
    });

    /** @type {string[]} */
    const notices = [];
    const controller = new AbortController();
    await subscribeRepos({
      service,
      collections: [COLLECTION],
      signal: controller.signal,
      onNotice: (message) => notices.push(message),
      onWrite: () => controller.abort(),
    });

    expect(notices.join('\n')).toContain('FutureCursor');
  });
});

describe('watch', () => {
  /** @type {string} */
  let dir;

  beforeEach(async () => {
    dir = await mkdtemp(join(tmpdir(), 'git-ci-watch-'));
  });

  afterEach(async () => {
    await rm(dir, { recursive: true, force: true });
  });

  it('runs a check for a push and persists the cursor', async () => {
    serve(async () => [await commitFrame(12, 'a'.repeat(40))]);

    const state = createStateStore(join(dir, 'state.json'));
    /** @type {Array<Record<string, *>>} */
    const runs = [];
    const controller = new AbortController();
    await watch({
      repos: [{ did: DID, repoName: 'proj', service }],
      state,
      signal: controller.signal,
      runCheck: async (/** @type {Record<string, *>} */ request) => {
        runs.push(request);
        controller.abort();
        return 'success';
      },
    });

    expect(runs).toHaveLength(1);
    expect(runs[0].ref).toBe('refs/heads/main');
    expect(state.cursor(service)).toBe(12);
    expect(state.refs(`${DID}/proj`)).toEqual({
      'refs/heads/main': 'a'.repeat(40),
    });
  });
});