atproto-store / test / sequencer.test.js

// FirehoseSequencer: seq assignment, cursor=0 full backfill, cursor>0 partial, and
// live-tail (a frame appended AFTER subscription arrives) + clean abort.

import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { RepoStore } from '../lib/repo-store.js'
import { FirehoseSequencer } from '../lib/sequencer.js'

async function newSeq (t) {
  const dir = mkdtempSync(join(tmpdir(), 'seq-'))
  const store = await new RepoStore(dir).ready()
  const core = store.corestore.get({ name: 'firehose' })
  await core.ready()
  const seq = await new FirehoseSequencer(core).ready()
  t.teardown(async () => { await seq.close(); await store.close(); rmSync(dir, { recursive: true, force: true }) })
  return seq
}

const F = (n) => new Uint8Array([n]) // a tiny distinguishable "frame"

// 1. seq is 1-based and monotonic
test('append returns 1-based monotonic seq', async (t) => {
  const s = await newSeq(t)
  const a = await s.append(F(10))
  const b = await s.append(F(20))
  const c = await s.append(F(30))
  t.ok(a === 1 && b === 2 && c === 3 && s.length === 3, 'append returns 1-based monotonic seq')
})

// 2. cursor=0 backfills the full history, in order
test('cursor=0 replays all frames in seq order', async (t) => {
  const s = await newSeq(t)
  for (let i = 1; i <= 4; i++) await s.append(F(i))
  const got = []
  const ac = new AbortController()
  for await (const { seq, bytes } of s.subscribe(0, { signal: ac.signal })) {
    got.push([seq, bytes[0]])
    if (got.length === 4) { ac.abort(); break }
  }
  t.ok(JSON.stringify(got) === JSON.stringify([[1, 1], [2, 2], [3, 3], [4, 4]]),
    'cursor=0 replays all frames in seq order')
})

// 3. cursor>0 backfills only frames with seq > cursor
test('cursor=2 replays only seq 3,4', async (t) => {
  const s = await newSeq(t)
  for (let i = 1; i <= 4; i++) await s.append(F(i))
  const got = []
  const ac = new AbortController()
  for await (const { seq } of s.subscribe(2, { signal: ac.signal })) {
    got.push(seq)
    if (got.length === 2) { ac.abort(); break }
  }
  t.ok(JSON.stringify(got) === JSON.stringify([3, 4]), 'cursor=2 replays only seq 3,4')
})

// 4. live-tail: a frame appended AFTER subscribe() started is delivered
test('live frame appended after subscribe is delivered', async (t) => {
  const s = await newSeq(t)
  await s.append(F(1))
  const ac = new AbortController()
  const got = []
  const consumer = (async () => {
    for await (const { seq, bytes } of s.subscribe(0, { signal: ac.signal })) {
      got.push([seq, bytes[0]])
      if (got.length === 2) { ac.abort(); break }
    }
  })()
  // give the consumer a tick to drain backfill and park on 'append', then append live
  await new Promise(r => setTimeout(r, 50))
  await s.append(F(2))
  await consumer
  t.ok(JSON.stringify(got) === JSON.stringify([[1, 1], [2, 2]]),
    'live frame appended after subscribe is delivered')
})