chadtmiller.com/pds.js

Switch repository

Loading…

import { sign } from '@pdsjs/core/crypto';
import { beforeEach, describe, expect, it } from 'vitest';
import { ATPROTO_PDS_ID } from '../src/authority.js';
import { createMemorySpaceStorage } from '../src/memory-storage.js';
import {
  createNotifyOutbox,
  DELIVERY_DEADLINE_MS,
  retryDelay,
} from '../src/outbox.js';
import { makeSpaceRow } from '../src/space-row.js';

const WRITER = 'did:plc:writer';
const AUTHORITY = 'did:plc:authority';
const SPACE = `at://${AUTHORITY}/space/com.example.forum/self`;
const DATA = new Uint8Array(32).fill(5);
const T0 = Date.parse('2026-10-01T00:00:00.000Z');

/** @type {any} */ let store;
/** @type {any[]} */ let sent;
/** @type {number[]} */ let wakeups;
/** @type {Array<number|Error>} */ let answers;
/** @type {number} */ let clock;
/** @type {ReturnType<typeof createNotifyOutbox>} */ let outbox;

beforeEach(async () => {
  const keys = await crypto.subtle.generateKey(
    { name: 'ECDSA', namedCurve: 'P-256' },
    false,
    ['sign', 'verify'],
  );

  store = createMemorySpaceStorage();
  sent = [];
  wakeups = [];
  answers = [];
  clock = T0;
  outbox = createNotifyOutbox({
    spaceStorage: store,
    getDid: async () => WRITER,
    getSigner: async () => ({ sign: (bytes) => sign(keys.privateKey, bytes) }),
    resolveDid: async (did) =>
      did === AUTHORITY
        ? {
            id: AUTHORITY,
            service: [
              {
                id: ATPROTO_PDS_ID,
                type: 'AtprotoPersonalDataServer',
                serviceEndpoint: 'https://authority.example',
              },
            ],
          }
        : null,
    fetch: async (url, init) => {
      sent.push({
        url: String(url),
        signal: init?.signal,
        body: JSON.parse(String(init?.body)),
        token: String(
          /** @type {any} */ (init?.headers)?.authorization,
        ).replace(/^Bearer /, ''),
      });
      const answer = answers.shift() ?? 200;
      if (answer instanceof Error) throw answer;
      return new Response('{}', { status: answer });
    },
    scheduleWakeup: async (at) => {
      wakeups.push(at);
    },
    now: () => clock,
  });
});

/** @param {string} jwt */
const payloadOf = (jwt) =>
  JSON.parse(atob(jwt.split('.')[1].replace(/-/g, '+').replace(/_/g, '/')));

describe('notify outbox', () => {
  it('delivers at once and owes nothing afterwards', async () => {
    await outbox.deliver(SPACE, { rev: 'r1', data: DATA });
    expect(sent).toHaveLength(1);
    expect(sent[0].url).toBe(
      'https://authority.example/xrpc/com.atproto.space.notifyWrite',
    );
    expect(sent[0].body).toEqual({
      space: SPACE,
      repo: WRITER,
      repoRev: 'r1',
      data: { $bytes: btoa(String.fromCharCode(...DATA)).replace(/=+$/, '') },
    });
    expect(payloadOf(sent[0].token)).toMatchObject({
      iss: WRITER,
      aud: `${AUTHORITY}#atproto_pds`,
      lxm: 'com.atproto.space.notifyWrite',
    });
    expect(await store.nextSpaceNotifyAt()).toBeNull();
    expect(wakeups).toEqual([]);
  });

  it('waits no longer than its timeout for an answer', async () => {
    await outbox.deliver(SPACE, { rev: 'r1', data: DATA });
    expect(sent[0].signal).toBeInstanceOf(AbortSignal);
  });

  it('retries a transient failure with backoff until it lands', async () => {
    answers = [503, new Error('network down'), 200];
    await outbox.deliver(SPACE, { rev: 'r1', data: DATA });
    expect(wakeups).toEqual([T0 + retryDelay(1)]);

    // Not yet due.
    clock = T0 + retryDelay(1) - 1;
    await outbox.tick();
    expect(sent).toHaveLength(1);

    clock = T0 + retryDelay(1);
    expect(await outbox.tick()).toBe(clock + retryDelay(2));
    expect(sent).toHaveLength(2);

    clock += retryDelay(2);
    expect(await outbox.tick()).toBeNull();
    expect(sent).toHaveLength(3);
    expect(sent.every((c) => c.body.repoRev === 'r1')).toBe(true);
  });

  it('stops on a permanent failure', async () => {
    answers = [400];
    await outbox.deliver(SPACE, { rev: 'r1', data: DATA });
    expect(await store.nextSpaceNotifyAt()).toBeNull();
    expect(wakeups).toEqual([]);
  });

  it('retries a rate limit', async () => {
    answers = [429];
    await outbox.deliver(SPACE, { rev: 'r1', data: DATA });
    expect(await store.nextSpaceNotifyAt()).not.toBeNull();
  });

  it('owes only the newest write, with its retries started afresh', async () => {
    answers = [503, 503, 503, 200];
    await outbox.deliver(SPACE, { rev: 'r1', data: DATA });
    clock += retryDelay(1);
    await outbox.tick();
    await outbox.deliver(SPACE, { rev: 'r2', data: DATA });
    const [row] = await store.listDueSpaceNotify(
      new Date(clock + retryDelay(1)).toISOString(),
      10,
    );
    expect(row).toMatchObject({ repoRev: 'r2', attempts: 1 });
    clock += retryDelay(1);
    await outbox.tick();
    expect(sent.at(-1).body.repoRev).toBe('r2');
    expect(await store.nextSpaceNotifyAt()).toBeNull();
  });

  it('gives up once the next retry would pass the deadline', async () => {
    answers = Array(100).fill(503);
    await outbox.deliver(SPACE, { rev: 'r1', data: DATA });
    for (;;) {
      const next = await store.nextSpaceNotifyAt();
      if (next === null) break;
      clock = Date.parse(next);
      await outbox.tick();
    }
    expect(clock).toBeLessThanOrEqual(T0 + DELIVERY_DEADLINE_MS);
    expect(sent.length).toBeGreaterThan(5);
  });
});

describe('deletion notices', () => {
  const SYNCER = 'did:web:syncer.example#atproto_space_syncer';
  const recipient = (expiresAt = new Date(T0 + 600_000).toISOString()) => ({
    serviceDid: SYNCER,
    serviceEndpoint: 'https://syncer.example',
    expiresAt,
  });

  /** As the authority: the space exists, then is deleted. */
  async function deleteOwnSpace() {
    await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
    await store.deleteSpace(SPACE, new Date(T0).toISOString());
  }

  it('tells each registered syncer at once', async () => {
    await deleteOwnSpace();
    await outbox.deliverSpaceDeleted(SPACE, [recipient()]);
    expect(sent).toHaveLength(1);
    expect(sent[0].url).toBe(
      'https://syncer.example/xrpc/com.atproto.space.notifySpaceDeleted',
    );
    expect(sent[0].body).toEqual({ space: SPACE });
    expect(payloadOf(sent[0].token)).toMatchObject({
      iss: WRITER,
      aud: SYNCER,
      lxm: 'com.atproto.space.notifySpaceDeleted',
    });
    expect(await store.nextSpaceNotifyAt()).toBeNull();
  });

  it('retries until it lands', async () => {
    await deleteOwnSpace();
    answers = [503, 200];
    await outbox.deliverSpaceDeleted(SPACE, [recipient()]);
    expect(wakeups).toEqual([T0 + retryDelay(1)]);
    clock = T0 + retryDelay(1);
    expect(await outbox.tick()).toBeNull();
    expect(sent).toHaveLength(2);
  });

  it('stops when the registration expires, trying once on its last moment', async () => {
    await deleteOwnSpace();
    const expiresAt = T0 + retryDelay(1) / 2;
    answers = [503, 503];
    await outbox.deliverSpaceDeleted(SPACE, [
      recipient(new Date(expiresAt).toISOString()),
    ]);
    expect(wakeups).toEqual([expiresAt]);
    clock = expiresAt;
    expect(await outbox.tick()).toBeNull();
    expect(sent).toHaveLength(2);
  });

  it('drops a notice once the space exists again', async () => {
    await deleteOwnSpace();
    answers = [503];
    await outbox.deliverSpaceDeleted(SPACE, [recipient()]);
    await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
    clock = T0 + retryDelay(1);
    expect(await outbox.tick()).toBeNull();
    expect(sent).toHaveLength(1);
  });

  it('stops on a permanent failure', async () => {
    await deleteOwnSpace();
    answers = [404];
    await outbox.deliverSpaceDeleted(SPACE, [recipient()]);
    expect(await store.nextSpaceNotifyAt()).toBeNull();
  });
});