chadtmiller.com/pds.js

Switch repository

Loading…

import Database from 'better-sqlite3';
import { afterEach, describe, expect, it } from 'vitest';
import { WebSocketServer } from 'ws';
import { createAppViewStore } from '../src/store.js';
import { watch } from '../src/watch.js';

const COLLECTION = 'dev.pdsjs.git.pull';
const DID = 'did:plc:stranger';

/**
 * @param {Partial<Record<string, unknown>>} commit
 * @param {Record<string, unknown>} [extra]
 */
const frame = (commit, extra = {}) => ({
  $type: 'message',
  payload: {
    $type: 'network.bsky.jetstream.subscribeEvents#commit',
    did: DID,
    seq: 10,
    time: '2026-08-13T06:47:43.959305Z',
    operation: 'create',
    collection: COLLECTION,
    rkey: '3kpull',
    rev: '3krev',
    record: { subject: { uri: 'at://did:plc:owner/dev.pdsjs.git.repo/proj' } },
    cid: 'bafypull',
    ...commit,
  },
  ...extra,
});

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

  afterEach(() => {
    server?.close();
  });

  it('folds writes, deletes and identities into the store', async () => {
    const store = createAppViewStore(new Database(':memory:'));
    server = new WebSocketServer({ port: 0 });
    const port = /** @type {import('node:net').AddressInfo} */ (
      server.address()
    ).port;
    server.on('connection', (socket) => {
      socket.send(JSON.stringify(frame({})));
      socket.send(
        JSON.stringify({
          $type: 'message',
          payload: {
            $type: 'network.bsky.jetstream.subscribeEvents#identity',
            did: DID,
            seq: 11,
            identity: { did: DID, handle: 'stranger.example.com' },
          },
        }),
      );
      socket.send(JSON.stringify(frame({ operation: 'delete', seq: 12 })));
      socket.send(JSON.stringify(frame({ operation: 'delete', seq: 13 })));
      // Repeat the delete: at-least-once delivery changes nothing.
      socket.send(
        JSON.stringify({
          $type: 'message',
          payload: {
            $type: 'network.bsky.jetstream.subscribeEvents#account',
            did: 'did:plc:gone',
            seq: 14,
            active: false,
            status: 'deleted',
          },
        }),
      );
    });

    const controller = new AbortController();
    await watch({
      store,
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      onSeq: (seq) => {
        if (seq >= 13) controller.abort();
      },
    });

    expect(store.getRecord(`at://${DID}/${COLLECTION}/3kpull`)).toBe(null);
    expect(store.handleOf(DID)).toBe('stranger.example.com');
    // The checkpoint is behind or at the last handled event, never ahead.
    expect(store.cursor('jetstream')).toBeGreaterThanOrEqual(13);
  });

  it('indexes the subject so discovery answers', async () => {
    const store = createAppViewStore(new Database(':memory:'));
    server = new WebSocketServer({ port: 0 });
    const port = /** @type {import('node:net').AddressInfo} */ (
      server.address()
    ).port;
    server.on('connection', (socket) => {
      socket.send(JSON.stringify(frame({})));
    });

    const controller = new AbortController();
    await watch({
      store,
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      onSeq: () => controller.abort(),
    });

    const found = store.about(
      'at://did:plc:owner/dev.pdsjs.git.repo/proj',
      COLLECTION,
    );
    expect(found).toHaveLength(1);
    expect(found[0].rkey).toBe('3kpull');
  });
});