Something went wrong. Try again.
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
Something went wrong. Try again.
16 kB · 504 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505import { afterEach, expect, test } from 'bun:test'import type { ServerWebSocket } from 'bun'import { Cause, Effect, Exit } from 'effect'import { type HydrantSourceOptions, runHydrantSource } from './hydrant-source'
const cid = 'bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'const record = (id = 1, overrides: Record<string, unknown> = {}) => JSON.stringify({ id, type: 'record', record: { live: false, did: 'did:plc:aaaaaaaaaaaaaaaaaaaaaaaa', rev: '3mabcdef23456', collection: 'place.wisp.fs', rkey: 'site', action: 'create', cid, record: { $type: 'place.wisp.fs', site: 'hello' }, ...overrides, }, })const wait = (ms = 2) => new Promise<void>((resolve) => setTimeout(resolve, ms))async function until(predicate: () => boolean) { for (let n = 0; n < 500 && !predicate(); n++) await wait() expect(predicate()).toBe(true)}const cleanups: (() => Promise<unknown> | unknown)[] = []afterEach(async () => { for (const cleanup of cleanups.splice(0).reverse()) await cleanup()})
function fixture(head: string = '{"id":100}') { let socket: ServerWebSocket<undefined> | undefined let path: string | undefined let closed = false let pings = 0 const server = Bun.serve<undefined>({ hostname: '127.0.0.1', port: 0, fetch(request, server) { const url = new URL(request.url) if (url.pathname === '/stream/head') return new Response(head) path = url.pathname + url.search if (server.upgrade(request)) return return new Response('no', { status: 400 }) }, websocket: { open(ws) { socket = ws }, message() {}, ping() { pings++ }, close() { closed = true }, }, }) cleanups.push(() => server.stop(true)) return { url: `http://127.0.0.1:${server.port}`, send: (text: string) => socket?.send(text), get socket() { return socket }, get path() { return path }, get closed() { return closed }, get pings() { return pings }, }}function start<E = never>(url: string, overrides: Partial<HydrantSourceOptions<E, never>> = {}) { const seen: number[] = [] const heads: (number | null)[] = [] let connected = 0 let disconnected = 0 let done = false const controller = new AbortController() const result = Effect.runPromiseExit( runHydrantSource({ url, getCursor: () => Effect.succeed(undefined), handleEvent: (event) => Effect.sync(() => { seen.push(event.id) }), onReplayHead: (head) => Effect.sync(() => { heads.push(head) }), onConnect: () => Effect.sync(() => { connected++ }), onDisconnect: () => Effect.sync(() => { disconnected++ }), heartbeatIntervalMs: 10, heartbeatTimeoutMs: 100, ...overrides, }), { signal: controller.signal }, ).then((exit) => { done = true return exit }) cleanups.push(async () => { controller.abort() await result }) return { result, controller, seen, heads, get connected() { return connected }, get disconnected() { return disconnected }, get done() { return done }, }}function error(exit: Exit.Exit<void, unknown>): unknown { if (Exit.isSuccess(exit)) throw new Error('expected failure') const found = Cause.findError(exit.cause) return found._tag === 'Success' ? found.success : undefined}
test('real quiet Bun socket uses matching pong without inventing progress', async () => { const peer = fixture() const source = start(peer.url) await until(() => source.connected === 1) await until(() => peer.pings >= 4) expect(source.done).toBe(false) expect(source.seen).toEqual([]) expect(source.heads).toEqual([100]) expect(peer.path).toBe('/stream?cursor=0') source.controller.abort() await source.result await until(() => peer.closed) expect(source.disconnected).toBe(1)})
test('inclusive cursor, historical records and missing history body retain order; ephemeral IDs do not checkpoint', async () => { const peer = fixture() const events: unknown[] = [] const source = start(peer.url, { getCursor: () => Effect.succeed(7), handleEvent: (event) => Effect.sync(() => { events.push(event) }), }) await until(() => source.connected === 1) expect(peer.path).toBe('/stream?cursor=7') peer.send(record(7)) peer.send(JSON.stringify({ id: 8, type: 'identity', identity: { did: 'did:plc:aaaaaaaaaaaaaaaaaaaaaaaa' } })) peer.send(record(9, { record: undefined })) peer.send( JSON.stringify({ id: 10, type: 'account', account: { did: 'did:plc:aaaaaaaaaaaaaaaaaaaaaaaa', active: false } }), ) peer.send(record(11, { action: 'delete', cid: undefined, record: undefined })) await until(() => events.length === 3) expect(events).toMatchObject([ { id: 7, live: false }, { id: 9, cid }, { id: 11, action: 'delete' }, ]) expect((events[1] as { record?: unknown }).record).toBeUndefined() expect(source.heads).toEqual([100])})
for (const [frame, code] of [ ['not json', 'MalformedFrame'], [record().replace('"id":1', '"id":9007199254740993'), 'MalformedFrame'], [record().replace('"id":1', '"id":1.00000000000000001'), 'MalformedFrame'], [record().replace('"id":1', '"id":1,"id":2'), 'MalformedFrame'], [record(1, { cid: 'not-a-cid' }), 'InvalidCid'], [record(1, { cid: undefined }), 'InvalidCid'], ['{"id":1,"type":"identity","identity":{}}', 'MalformedFrame'], ['{"id":1,"type":"account","account":{"did":"a","active":"false"}}', 'MalformedFrame'], ['{"type":"error","error":"secret body"}', 'MalformedFrame'],] as const) { test(`fail closed ${code}: ${frame.slice(0, 45)}`, async () => { const peer = fixture() const source = start(peer.url) await until(() => source.connected === 1) peer.send(frame) const failure = error(await source.result) expect(failure).toMatchObject({ code, recoveryRequired: true }) expect(String(failure)).not.toContain('secret body') expect(source.seen).toEqual([]) expect(source.disconnected).toBe(1) })}
test('repeated or decreasing frame ID fails this generation without reconnect', async () => { const peer = fixture() const source = start(peer.url) await until(() => source.connected === 1) peer.send(record(2)) await until(() => source.seen.length === 1) peer.send(record(1)) expect(error(await source.result)).toMatchObject({ code: 'MalformedFrame' }) expect(source.seen).toEqual([2]) expect(source.connected).toBe(1)})
for (const head of ['{"id":null}', '{"id":2}']) { test(`cursor ahead of ${head} is rejected before websocket`, async () => { const peer = fixture(head) const source = start(peer.url, { getCursor: () => Effect.succeed(3) }) expect(error(await source.result)).toMatchObject({ code: 'CursorAhead', recoveryRequired: true }) expect(peer.path).toBeUndefined() })}for (const head of ['{"id":9007199254740993}', '{"id":1.00000000000000001}', 'x'.repeat(4097)]) { test(`malformed/bounded head ${head.slice(0, 25)} cannot connect`, async () => { const peer = fixture(head) const source = start(peer.url) expect(error(await source.result)).toMatchObject({ code: 'MalformedHead', recoveryRequired: true }) expect(peer.path).toBeUndefined() })}
test('caller errors keep their identity and never reconnect', async () => { const peer = fixture() const caller = { _tag: 'CallerFailure' } const source = start(peer.url, { handleEvent: () => Effect.fail(caller) }) await until(() => source.connected === 1) peer.send(record()) expect(error(await source.result)).toBe(caller) expect(source.connected).toBe(1)})
for (const cap of ['count', 'bytes', 'single'] as const) { test(`${cap} overload includes blocked callback retention and joins callback settlement`, async () => { const peer = fixture() let release!: () => void let calls = 0 const gate = new Promise<void>((resolve) => { release = resolve }) cleanups.push(() => release()) const source = start(peer.url, { handleEvent: () => Effect.promise(async () => { calls++ await gate }), }) await until(() => source.connected === 1) peer.send(record(1, cap === 'bytes' ? { record: { text: 'a'.repeat(600_000) } } : {})) await until(() => calls === 1) if (cap === 'count') for (let id = 2; id <= 65; id++) peer.send(record(id)) else peer.send(record(2, { record: { text: 'a'.repeat(cap === 'single' ? 8_400_000 : 600_000) } })) await until(() => peer.closed) expect(source.done).toBe(false) release() expect(error(await source.result)).toMatchObject({ code: 'Overload', recoveryRequired: true }) expect(calls).toBe(1) })}
test('interruption joins an in-flight foreign callback before disconnect hook', async () => { const peer = fixture() let release!: () => void let entered = false const gate = new Promise<void>((resolve) => { release = resolve }) cleanups.push(() => release()) const source = start(peer.url, { handleEvent: () => Effect.promise(async () => { entered = true await gate }), }) await until(() => source.connected === 1) peer.send(record()) await until(() => entered) source.controller.abort() await wait(20) expect(source.done).toBe(false) expect(source.disconnected).toBe(0) release() await source.result expect(source.disconnected).toBe(1)})
// Run the raw peer under Node: Bun's node:http upgrade compatibility is not a raw TCP socket.for (const pong of ['none', 'mismatch'] as const) { test(`real Node ${pong} pong fails liveness`, async () => { const script = ` import { createServer } from 'node:http'; import { createHash } from 'node:crypto'; const server = createServer((request, response) => response.end('{"id":null}')); server.on('upgrade', (request, socket) => { const accept = createHash('sha1') .update(request.headers['sec-websocket-key'] + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11') .digest('base64'); socket.write([ 'HTTP/1.1 101 Switching Protocols', 'Upgrade: websocket', 'Connection: Upgrade', 'Sec-WebSocket-Accept: ' + accept, '', '' ].join('\\r\\n')); socket.on('error', () => {}); socket.on('data', chunk => { const opcode = chunk[0] & 15; if (opcode === 9 && ${JSON.stringify(pong)} === 'mismatch') { socket.write(Buffer.from([138, 5, ...Buffer.from('wrong')])); } if (opcode === 8) socket.end(Buffer.from([136, 0])); }); }); server.listen(0, '127.0.0.1', () => console.log(server.address().port)); ` const peer = Bun.spawn(['node', '--input-type=module', '-e', script], { stdout: 'pipe', stderr: 'pipe' }) cleanups.push(async () => { peer.kill() await peer.exited }) const reader = peer.stdout.getReader() const first = await reader.read() reader.releaseLock() const port = Number(new TextDecoder().decode(first.value).trim()) expect(Number.isSafeInteger(port) && port > 0).toBe(true) const source = start(`http://127.0.0.1:${port}`, { heartbeatTimeoutMs: 30 }) expect(error(await source.result)).toMatchObject({ code: 'PongTimeout' }) expect(source.connected).toBe(1) expect(source.seen).toEqual([]) })}
class DelayedSocket extends EventTarget { static instances: DelayedSocket[] = [] closeRequested = false constructor(readonly url: string | URL) { super() DelayedSocket.instances.push(this) } open() { this.dispatchEvent(new Event('open')) } ping(data: string) { this.dispatchEvent(new MessageEvent('pong', { data })) } close() { this.closeRequested = true } closed() { this.dispatchEvent(new Event('close')) } frame(data: unknown) { this.dispatchEvent(new MessageEvent('message', { data })) }}function fakeSocket() { const original = globalThis.WebSocket DelayedSocket.instances = [] globalThis.WebSocket = DelayedSocket as unknown as typeof WebSocket cleanups.push(() => { globalThis.WebSocket = original }) return DelayedSocket.instances}
test('socket finalizer joins physical close and old generation callbacks cannot reach new source', async () => { const peer = fixture() const sockets = fakeSocket() const source = start(peer.url) cleanups.push(() => { for (const socket of sockets) socket.closed() }) await until(() => sockets.length === 1) const old = sockets[0]! old.open() await until(() => source.connected === 1) source.controller.abort() await until(() => old.closeRequested) expect(source.done).toBe(false) expect(source.disconnected).toBe(0) old.closed() await source.result const replacement = start(peer.url) cleanups.push(() => { for (const socket of sockets) socket.closed() }) await until(() => sockets.length === 2) sockets[1]!.open() await until(() => replacement.connected === 1) old.open() old.frame(record(90)) sockets[1]!.frame(record(3)) await until(() => replacement.seen.length === 1) expect(source.seen).toEqual([]) expect(replacement.seen).toEqual([3]) replacement.controller.abort() await until(() => sockets[1]!.closeRequested) sockets[1]!.closed() await replacement.result})
test('opening timeout joins delayed physical close', async () => { const peer = fixture() const sockets = fakeSocket() const source = start(peer.url, { openTimeoutMs: 10 }) cleanups.push(() => { for (const socket of sockets) socket.closed() }) await until(() => sockets[0]?.closeRequested === true) expect(source.done).toBe(false) sockets[0]!.closed() expect(error(await source.result)).toMatchObject({ code: 'OpenTimeout' }) expect(source.connected).toBe(0)})
test('unsupported control-ping socket fails explicitly; binary frames fail closed', async () => { const peer = fixture() const sockets = fakeSocket() const source = start(peer.url) cleanups.push(() => { for (const socket of sockets) socket.closed() }) await until(() => sockets.length === 1) Object.defineProperty(sockets[0], 'ping', { value: undefined }) sockets[0]!.open() await until(() => sockets[0]!.closeRequested) sockets[0]!.closed() expect(error(await source.result)).toMatchObject({ code: 'UnsupportedRuntime' }) const binary = start(peer.url) cleanups.push(() => { for (const socket of sockets) socket.closed() }) await until(() => sockets.length === 2) sockets[1]!.open() await until(() => binary.connected === 1) sockets[1]!.frame(new ArrayBuffer(1)) await until(() => sockets[1]!.closeRequested) sockets[1]!.closed() expect(error(await binary.result)).toMatchObject({ code: 'MalformedFrame', recoveryRequired: true })})
test('head interruption aborts and joins noncooperative fetch settlement', async () => { const original = globalThis.fetch let entered = false let signal: AbortSignal | null | undefined let release!: () => void const gate = new Promise<void>((resolve) => { release = resolve }) globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { entered = true signal = init?.signal await gate throw new Error('foreign fetch settled') }) as unknown as typeof fetch cleanups.push(() => { globalThis.fetch = original }) const source = start('http://operator.invalid') cleanups.push(() => release()) await until(() => entered) source.controller.abort() await until(() => signal?.aborted === true) expect(source.done).toBe(false) release() await source.result expect(source.connected).toBe(0)})
test('escaped root id is checked without mistaking nested/string IDs for envelope ID', async () => { const peer = fixture() const source = start(peer.url) await until(() => source.connected === 1) peer.send( record(4, { record: { id: 9007199254740992, text: '"id":9999999999999999999' } }).replace('"id":4', '"i\\u0064":4'), ) await until(() => source.seen.length === 1) expect(source.seen).toEqual([4])})
test('wire ID zero is valid; undefined cursor subscribes inclusively from zero', async () => { const peer = fixture('{"id":0}') const source = start(peer.url) await until(() => source.connected === 1) peer.send(record(0)) await until(() => source.seen.length === 1) expect(peer.path).toBe('/stream?cursor=0') expect(source.heads).toEqual([0]) expect(source.seen).toEqual([0])})