// RecordLog — the convergent half of a feed or a document.
//
// Names and POINTERS live here; bytes live in the RecordStore. The KV is injected — an autobase's
// linearized view in production, a Map in the tests — so this layer is provable without a swarm.
// The KV surface is the minimum an autobase view presents: set / get / del / list(prefix).
//
// ── The three states, kept apart on purpose ────────────────────────────────────────────────────
//
// ABSENT no entry. In a replicated log this ALSO means "not replicated to me yet", so it
// can never be read as a deletion.
// DELETED a tombstone. Positive evidence. Absence is ambiguous; a tombstone is not.
// UNRESOLVED a pointer whose bytes we cannot obtain. The record exists and we cannot read it.
//
// Collapsing UNRESOLVED into ABSENT is how a lagging node concludes "deleted" and publishes a
// delete for a record that is alive. Collapsing ABSENT into DELETED is the same bug from the other
// side. `resolve()` is the only honest read, and `get()` is the convenience that flattens all three
// to null for callers that genuinely do not care.
import b4a from 'b4a'
const SEP = '/'
const byteOrder = (a, b) => b4a.compare(b4a.from(a), b4a.from(b))
export const OK = 'ok'
export const ABSENT = 'absent'
export const DELETED = 'deleted'
export const UNRESOLVED = 'unresolved'
const isTombstone = (v) => !!(v && typeof v === 'object' && v.$deleted === true)
// ⛔ A pointer is EXACTLY {$oid, size} and nothing else. Measured: a value carrying the record
// itself has no $oid, so a peer resolving it gets null — it reads only on the node that wrote it
// ("UNFETCHABLE BY EVERY OTHER NODE… that DID cannot be co-owned at all"). That shape is refused
// here rather than left as a rule to remember, because nothing fails visibly when it is broken.
const isPointer = (v) => !!(v && typeof v === 'object' && '$oid' in v && typeof v.size === 'number' &&
Object.keys(v).length === 2)
// Split a KV key into { collection, name } on the FIRST separator. Reserved keys are never records:
// a single-segment key (a lease, a head) and any collection beginning with '__' (control data).
function splitKey (key) {
const i = key.indexOf(SEP)
if (i < 0) return null
const collection = key.slice(0, i)
if (collection.startsWith('__')) return null
return { collection, name: key.slice(i + 1) }
}
function recordKey (collection, name) {
if (typeof collection !== 'string' || !collection.length || collection.includes(SEP)) {
throw new Error('RecordLog: collection must be a non-empty string without "' + SEP + '"')
}
if (typeof name !== 'string' || !name.length || name.includes(SEP)) {
throw new Error('RecordLog: name must be a non-empty string without "' + SEP + '"')
}
return collection + SEP + name
}
export class RecordLog {
constructor (kv, store) {
if (!kv || typeof kv.set !== 'function' || typeof kv.list !== 'function') {
throw new Error('RecordLog: a KV with async set/get/del/list is required')
}
if (!store || typeof store.put !== 'function' || typeof store.ensureLocal !== 'function') {
throw new Error('RecordLog: a RecordStore is required for the bytes')
}
this.kv = kv
this.store = store
}
// Bytes first, then the pointer. The order is a safety property: a pointer naming bytes that were
// never written is unresolvable for everyone, while bytes with no pointer yet are merely not
// announced. Fail towards the recoverable state.
async put (collection, name, record) {
const key = recordKey(collection, name)
const { oid, size } = await this.store.put(collection, name, record)
await this.kv.set(key, { $oid: oid, size })
return { oid, size }
}
// Write a raw pointer — for a caller replicating someone else's entry. Refuses anything that is
// not exactly {$oid,size}; see isPointer.
async setPointer (collection, name, pointer) {
const key = recordKey(collection, name)
if (!isPointer(pointer)) {
throw new Error('RecordLog: a log value must be exactly { $oid, size } — an inline record ' +
'resolves only for its author and is invisible to every peer')
}
await this.kv.set(key, pointer)
}
// The honest read. { status, record, pointer }.
async resolve (collection, name) {
const key = recordKey(collection, name)
const ptr = await this.kv.get(key)
if (ptr == null) return { status: ABSENT, record: null, pointer: null }
if (isTombstone(ptr)) return { status: DELETED, record: null, pointer: ptr }
const record = await this.store.ensureLocal(collection, name, ptr.$oid)
if (record == null) return { status: UNRESOLVED, record: null, pointer: ptr }
return { status: OK, record, pointer: ptr }
}
// Flattens all three not-readable states to null. Use resolve() wherever the difference matters —
// which is anywhere a decision is made, as opposed to anywhere something is rendered.
async get (collection, name) {
const r = await this.resolve(collection, name)
return r.status === OK ? r.record : null
}
// Deletion is a TOMBSTONE, then the bytes. A tombstone converges symmetrically like any other
// value, so a peer can tell "deleted" from "not yet replicated to me".
async del (collection, name) {
const key = recordKey(collection, name)
await this.kv.set(key, { $deleted: true })
try { await this.store.del(collection, name) } catch {}
}
// Live, readable records as { key, collection, name, record }, in byte order.
async list (collection = null) {
const rows = await this._scan(collection)
return rows.filter((r) => r.status === OK).map(({ key, collection, name, record }) =>
({ key, collection, name, record }))
}
/**
* Every live pointer, WITHOUT resolving any of them.
*
* ⛔ The one enumeration that does not fetch. `list()` and `unresolved()` both go through
* `_scan`, which calls `resolve()` on every record — so both of them attempt a network fetch for
* everything missing, all at once. That is right for a reader who wants the records, and wrong
* for anything that needs to decide WHICH bytes to go and get: you cannot bound a fetch you have
* already made by asking the question.
*
* Tombstones are omitted: a deleted record has no bytes to hold, and offering it here would send
* a mirror looking for something that is positively known not to exist.
*/
async pointers (collection = null) {
const prefix = collection ? collection + SEP : ''
const map = await this.kv.list(prefix)
const out = []
for (const key of Object.keys(map).filter((k) => splitKey(k) !== null).sort(byteOrder)) {
const pointer = map[key]
if (pointer == null || isTombstone(pointer)) continue
const { collection: c, name } = splitKey(key)
out.push({ key, collection: c, name, pointer })
}
return out
}
/**
* Names in `collection` that are POSITIVELY deleted, in byte order.
*
* ⛔ The evidence a bridge retracts on. A name missing from list() could be "deleted" or "not
* replicated to me yet"; a tombstone can only be the first, so this is the one listing that may
* send a Delete to bsky or the fediverse.
*/
async tombstones (collection) {
const prefix = collection + SEP
const map = await this.kv.list(prefix)
return Object.keys(map)
.filter((k) => splitKey(k) !== null && isTombstone(map[k]))
.sort(byteOrder)
.map((k) => splitKey(k).name)
}
// Pointers we hold whose bytes we cannot obtain. Never merge this into list() — a record that is
// silently missing from a listing is indistinguishable from one that was never written, and that
// is the ambiguity this whole layer exists to refuse.
async unresolved (collection = null) {
const rows = await this._scan(collection)
return rows.filter((r) => r.status === UNRESOLVED).map(({ key, collection, name, pointer }) =>
({ key, collection, name, pointer }))
}
async _scan (collection) {
const prefix = collection ? collection + SEP : ''
const map = await this.kv.list(prefix)
const keys = Object.keys(map).filter((k) => splitKey(k) !== null).sort(byteOrder)
// ⛔ Concurrent, and the ORDER is preserved by the map rather than by the await. Resolving in
// series meant a feed whose bytes are unavailable paid the fetch timeout once PER RECORD — a
// freshly peeked feed of fifty posts would have taken fifty timeouts to list. They are
// independent reads; nothing about one depends on another.
//
// ⚠ Unbounded fan-out. Records already on disk cost nothing, so the width is the number of
// MISSING ones — which for a fresh peek is all of them. Fine at the sizes this is built for and
// it will want a limiter before it wants anything else.
return Promise.all(keys.map(async (key) => {
const { collection: c, name } = splitKey(key)
return { key, collection: c, name, ...(await this.resolve(c, name)) }
}))
}
}