atproto-bridge / pds / test / fleet-relay-seqbase.test.js

// FleetRelay seqBase — the emitted seq is lifted above a downstream relay's stored cursor (the
// indigo recovery path: relays keep a strictly-forward cursor and never reset; we must push our
// emitted seq above it). Verifies: emitted = seqBase + internal, cursor translated in emitted
// space, and setSeqBase is monotonic + persisted.

import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { accountFrame } from 'atproto-store/lib/firehose.js'
import { FleetRelay } from '../lib/fleet-relay.js'

const T = '2026-07-04T00:00:00.000Z'

async function collect (relay, cursor, n) {
  const ac = new AbortController()
  const it = relay.subscribe(cursor, { signal: ac.signal })
  const out = []
  try { for (let k = 0; k < n; k++) { const { value, done } = await it.next(); if (done || !value) break; out.push(value.seq) } } finally { ac.abort() }
  return out
}

test('FleetRelay seqBase: emitted = seqBase + internal, cursor translation, monotonic', async (t) => {
  const dir = mkdtempSync(join(tmpdir(), 'relay-seqbase-'))
  const store = openCorestore(dir)
  let persisted = 0
  const relay = await new FleetRelay(store.get({ name: 'agg', valueEncoding: 'json' }), { seqBase: 0, onSeqBase: (n) => { persisted = n } }).ready()
  t.teardown(async () => { await relay.close(); await store.close(); rmSync(dir, { recursive: true, force: true }) })

  for (let i = 1; i <= 3; i++) await relay.emit(`f:${i}`, accountFrame({ seq: 0, did: 'did:plc:x', active: true, time: T }))

  t.alike(await collect(relay, 0, 3), [1, 2, 3], 'no base: emitted seq = internal (1,2,3)')

  const base = await relay.setSeqBase(100)
  t.ok(base === 100 && persisted === 100 && relay.length === 103, 'setSeqBase(100): base+persisted=100, length=seqBase+internal=103')

  t.alike(await collect(relay, 0, 3), [101, 102, 103], 'cursor 0 (< base): all frames at seqBase+internal')
  t.alike(await collect(relay, 102, 1), [103], 'cursor 102 (emitted space): yields 103 only')

  const same = await relay.setSeqBase(50)
  t.ok(same === 100 && persisted === 100, 'setSeqBase(50) is a monotonic no-op — base stays 100')
})