// Co-owned DID regression (the wj62 prod topology: stortron notary + tabertron co-owner). BOTH co-owners
// serve the SAME did with THEIR OWN membrane cores (distinct firehose keys, identical deterministic content,
// differing only in replication lag). The canonical DidPeerRouter must front the SET of co-owners and serve
// reads from the FRESHEST by commit REV — never regressing onto a lagging co-owner (the prod symptom: an
// 83-frame stale membrane out-"lengthed" a 6-frame fresh one, so freshness is REV not frame count).
//
// Ported from atproto-bridge/pds/test/co-owned-front-thrash.test.js onto atproto-did-peer's own DidPeer
// holders (the shared module's announce path), per EXTRACTION-PLAN Phase 1.
// Run: node test/co-owned.test.mjs
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import createTestnet from '@hyperswarm/testnet'
import Hyperswarm from 'hyperswarm'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { RepoManager } from 'atproto-store/lib/repo-manager.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { createAtprotoIdentityFromKey } from 'atproto-identity/lib/atproto-identity.js'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { DidPeer, DidPeerRouter } from '../index.js'
const assert = (c, m) => { if (!c) { console.error('FAIL:', m); cleanupSync(); process.exit(1) } console.log(' ok -', m) }
const iso = '2026-07-21T00:00:00.000Z'
const dirs = []
const toClose = []
function newDir (p) { const d = mkdtempSync(join(tmpdir(), p)); dirs.push(d); return d }
function cleanupSync () { for (const d of dirs) { try { rmSync(d, { recursive: true, force: true }) } catch {} } }
const post = (rk, text) => ({ action: 'create', collection: 'app.bsky.feed.post', rkey: rk, record: { $type: 'app.bsky.feed.post', text, createdAt: iso, langs: ['en'] } })
// The shared DID identity (both co-owners hold the SAME did + signing key — that is what "co-owned" means).
const id = createAtprotoIdentityFromKey(new Uint8Array(32).fill(7), { handle: 'coowned.test', pdsUrl: 'https://pds.local' })
const DID = id.did
const testnet = await createTestnet(3)
const bootstrap = testnet.bootstrap
// A co-owner: its own corestore/cores, nCommits signed over the SHARED DID key, served as a DidPeer holder.
// Injected clock is shared-base, so a co-owner with MORE commits sits at a strictly higher head rev.
async function coOwner (nCommits) {
const store = openCorestore(newDir('coowner-'))
const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready()
let micros = 1_700_000_000_000_000n
const mgr = await new RepoManager(repo, { did: DID, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready()
for (let i = 0; i < nCommits; i++) { const w = await mgr.applyWrite(post('3post' + i + 'aaaaaa2', 'c' + i)); await seq.sequenceCommit(w, { time: iso }) }
const peer = new DidPeer(store, { did: DID, verifyKey: id.publicKey })
await peer.provide({ repoKey: repo.core.key, firehoseKey: seq.core.key })
const swarm = new Hyperswarm({ bootstrap })
swarm.on('connection', (c) => peer.attach(c))
await swarm.join(discoveryTopic(DID), { server: true, client: false }).flushed()
toClose.push(async () => { try { await swarm.destroy() } catch {}; try { await peer.close() } catch {}; try { await store.close() } catch {} })
return { peer, fhKey: seq.core.key.toString('hex') }
}
const behind = await coOwner(1) // lagging co-owner: 1 commit, LOWER head rev
const ahead = await coOwner(2) // notary: 2 commits, HIGHER head rev + more records
console.log('co-owners serving DID', DID.slice(0, 24), '| behind fh', behind.fhKey.slice(0, 8), '| ahead fh', ahead.fhKey.slice(0, 8))
// Reader: caller-owned swarm (favela posture), the CANONICAL router fronting the co-owned DID.
const store = openCorestore(newDir('reader-'))
const swarm = new Hyperswarm({ bootstrap })
const router = new DidPeerRouter(store)
swarm.on('connection', (c) => router.onconnection(c))
router.add(DID, { verifyKey: id.publicKey })
await swarm.join(discoveryTopic(DID), { server: false, client: true }).flushed()
// Poll front() FRESH each time (as a real getRepo request does) so any re-fronting/regression is visible.
const counts = []
for (let i = 0; i < 40; i++) {
const f = await router.front(DID).catch(() => null)
// listRecords returns { records, cursor } per the lexicon (0.2.10) — NOT a bare array.
const recs = f ? ((await f.listRecords('app.bsky.feed.post').catch(() => null)) || {}).records || [] : []
counts.push(recs.length)
await new Promise((r) => setTimeout(r, 300))
}
console.log(' served record counts over time:', JSON.stringify(counts))
const firstTwo = counts.indexOf(2)
assert(firstTwo >= 0, 'the AHEAD co-owner (2 records) was fronted at some point')
assert(counts[counts.length - 1] === 2, 'reads END on the FRESHEST (ahead) co-owner')
assert(counts.slice(firstTwo).every((x) => x >= 2), 'once fronting the freshest, it NEVER regresses to the behind co-owner')
// The router tracks BOTH co-owners as a SET (multi-front), and picks freshest by rev.
const diag = router.diagnose()
assert(diag.cachedFronts === 2, `both co-owners fronted (SET), got ${diag.cachedFronts}`)
console.log('\nCO-OWNED OK: freshest-by-rev across a co-owner SET; never regresses to the lagging membrane')
await swarm.destroy(); await router.close(); await store.close()
for (const close of toClose) await close()
await testnet.destroy()
cleanupSync()