// Observations — held until the peer collects them, and never mixed between peers.
//
// ⛔ THE ISOLATION IS THE SAFETY PROPERTY, exactly as it is for the inbound store. One bridge serves
// every registered identity, and an observation belongs to precisely one of them. A read that could
// return another peer's rows would put the door's authorisation between one person's social graph and
// another's, which is too much weight for a transport check to carry.
import test from 'brittle'
import fs from 'bare-fs'
import os from 'bare-os'
import { join } from 'bare-path'
import { ObservationStore, KIND } from '../lib/observations.js'
const tmp = (t) => {
const d = fs.mkdtempSync(join(os.tmpdir(), 'obs-'))
t.teardown(() => { try { fs.rmSync(d, { recursive: true, force: true }) } catch {} })
return join(d, 'observations.jsonl')
}
const ALICE = 'https://mastodon.social/users/alice'
const BOB = 'did:plc:wj62ebbusln6gxo5azo6ogzo'
test('an observation is recorded and read back by its recipient', (t) => {
const s = new ObservationStore({ path: tmp(t) })
s.note({ recipient: 'fleeky', subject: 'https://ap.7rnx.net/ap/objects/1', actor: ALICE, kind: KIND.LIKE, at: 1000 })
const r = s.list({ recipient: 'fleeky' })
t.is(r.rows.length, 1)
t.is(r.rows[0].actor, ALICE)
t.is(r.rows[0].kind, KIND.LIKE)
t.is(r.cursor, 1, 'the cursor is the seq of the last row returned')
})
test('★★★ one peer never sees another peer\'s observations', (t) => {
const s = new ObservationStore({ path: tmp(t) })
s.note({ recipient: 'fleeky', subject: 'o1', actor: ALICE, at: 1 })
s.note({ recipient: 'ashatron', subject: 'o2', actor: BOB, at: 2 })
t.is(s.list({ recipient: 'fleeky' }).rows.length, 1)
t.is(s.list({ recipient: 'ashatron' }).rows[0].actor, BOB)
// ⛔ And there is no call that returns both.
t.is(s.list({}).rows.length, 0, 'a read with no recipient returns nothing rather than everything')
})
test('★★ the cursor advances only over rows actually returned', (t) => {
const s = new ObservationStore({ path: tmp(t) })
for (let i = 0; i < 5; i++) s.note({ recipient: 'fleeky', subject: 'o', actor: ALICE + i, at: i })
const first = s.list({ recipient: 'fleeky', limit: 2 })
t.is(first.rows.length, 2)
t.ok(first.more, 'it says there is more')
const second = s.list({ recipient: 'fleeky', since: first.cursor, limit: 2 })
t.is(second.rows[0].actor, ALICE + '2', 'resumes exactly where the last page ended — no gap, no repeat')
const rest = s.list({ recipient: 'fleeky', since: second.cursor })
t.is(rest.rows.length, 1)
t.absent(rest.more)
})
test('⛔ a retry is not a second like', (t) => {
const s = new ObservationStore({ path: tmp(t) })
const a = s.note({ recipient: 'fleeky', subject: 'o', actor: ALICE, id: 'act-1', at: 1 })
const b = s.note({ recipient: 'fleeky', subject: 'o', actor: ALICE, id: 'act-1', at: 1 })
t.ok(a)
t.is(b, null, 'the duplicate is refused')
t.is(s.count({ recipient: 'fleeky' }), 1)
})
test('⛔ a row nobody can be given, or attributed to, is refused', (t) => {
const s = new ObservationStore({ path: tmp(t) })
// Noise that will be read later as data is worse than a missing row.
t.is(s.note({ subject: 'o', actor: ALICE }), null, 'no recipient')
t.is(s.note({ recipient: 'fleeky', subject: 'o' }), null, 'no actor')
t.is(s.count({}), 0)
})
test('an undo is its own kind, not an absence', (t) => {
const s = new ObservationStore({ path: tmp(t) })
s.note({ recipient: 'fleeky', subject: 'o', actor: ALICE, kind: KIND.LIKE, at: 1 })
s.note({ recipient: 'fleeky', subject: 'o', actor: ALICE, kind: KIND.UNDO, at: 2 })
const rows = s.list({ recipient: 'fleeky' }).rows
// ⛔ Absence is ambiguous in a replicated log, so a retraction is positive evidence
// (interaction.md §4.3) — the reader decides, the store does not silently forget.
t.alike(rows.map((r) => r.kind), [KIND.LIKE, KIND.UNDO])
})
test('⚠ bounded, oldest out — a full buffer must not refuse what we just learned', (t) => {
const s = new ObservationStore({ path: tmp(t), cap: 3 })
for (let i = 0; i < 5; i++) s.note({ recipient: 'fleeky', subject: 'o', actor: ALICE + i, at: i })
const rows = s.list({ recipient: 'fleeky' }).rows
t.is(rows.length, 3)
t.is(rows[0].actor, ALICE + '2', 'the oldest were dropped, not the newest refused')
})
test('it survives a restart, and a torn tail does not cost the file', (t) => {
const path = tmp(t)
const s = new ObservationStore({ path })
s.note({ recipient: 'fleeky', subject: 'o', actor: ALICE, at: 1 })
s.note({ recipient: 'fleeky', subject: 'o', actor: BOB, at: 2 })
// A process killed mid-append leaves half a line. An append-only log will eventually have one.
fs.appendFileSync(path, '{"seq":99,"recipient":"fle')
const reopened = new ObservationStore({ path })
const rows = reopened.list({ recipient: 'fleeky' }).rows
t.is(rows.length, 2, 'the intact rows survive')
t.is(rows[1].actor, BOB)
// And the next write continues the sequence rather than colliding with what was read.
const next = reopened.note({ recipient: 'fleeky', subject: 'o', actor: 'x', at: 3 })
t.is(next.seq, 3)
})