/**
* RPC audit log — an append-only JSONL record of every method the daemon
* dispatches, on every transport.
*
* WHY THIS EXISTS
*
* The daemon's control plane was entirely unlogged. `/rpc/<method>` and
* `/api/rpc` accept mutating calls (join-repo, create-access-token,
* consume-link-token, ref-update-observed…) and left no trace anywhere: no
* dispatch log, and the request body — which carries the method name on
* /api/rpc — is not in the upstream Apache log either. Reconstructing "who
* called what" on 2026-07-29 required correlating Apache access lines against
* daemon-side side-effects (identity records, tracker writer entries, pack
* mtimes), and several calls could not be attributed at all.
*
* hyper-svc funnels EVERY transport through Service.dispatch (see
* lib/service.js:78 — uds, http and swarm all call it), so wrapping that one
* method captures the whole surface with no per-transport plumbing.
*
* DESIGN CONSTRAINTS
*
* - Never breaks dispatch. Every code path here is wrapped; a logging failure
* must never fail, delay, or alter an RPC. Fails open, always.
* - Never blocks. Lines buffer in memory and flush on an interval in one
* batched write, so a busy daemon does not pay a syscall per call.
* - Durable. Buffered lines are bounded-loss (one flush interval); the whole
* point is that this survives the restart that erased the daemon log.
* - Cheap attribution. ctx.identity.publicKey is already resolved by the time
* dispatch runs; mapping it to a userId costs a registry lookup, so results
* are memoised in a bounded cache rather than looked up per call.
*/
import fs from '#node:fs'
import { dirname } from '#node:path'
// Methods excluded by default: high-frequency, read-only, and dominated by
// the SSR pages themselves (every page render issues a couple). Logging them
// buries the mutations that matter. Anything NOT listed here is recorded.
export const DEFAULT_SKIP = [
'status',
'whoami',
'get-profile',
'list-repos',
'list-repos-detailed',
'repo-activity',
'issue-counts'
]
const MAX_USERID_CACHE = 500
export class RpcAuditLog {
constructor ({ path, registry = null, skip = DEFAULT_SKIP,
flushMs = 1000, maxBytes = 32 * 1024 * 1024, now = null } = {}) {
if (!path) throw new Error('RpcAuditLog requires { path }')
this._path = path
this._registry = registry
this._skip = new Set(skip || [])
this._flushMs = flushMs
this._maxBytes = maxBytes
this._now = typeof now === 'function' ? now : () => Date.now()
this._buf = []
this._userIds = new Map() // identity pubkey -> userId (bounded)
this._timer = null
this._closed = false
}
start () {
if (this._timer || this._closed) return
try { fs.mkdirSync(dirname(this._path), { recursive: true }) } catch {}
this._timer = setInterval(() => this.flush(), this._flushMs)
if (this._timer && this._timer.unref) this._timer.unref()
}
/**
* Resolve ctx.identity.publicKey -> userId, memoised. Best-effort: an
* unknown or unresolvable identity logs as null rather than failing.
*/
async _userIdFor (pubkey) {
if (!pubkey || !this._registry) return null
if (this._userIds.has(pubkey)) return this._userIds.get(pubkey)
// `registry` may be a getter — the IdentityRegistry is built AFTER the
// Service this wraps, so callers pass () => svc.identityRegistry.
const registry = typeof this._registry === 'function' ? this._registry() : this._registry
if (!registry || typeof registry.getUserByIdentity !== 'function') return null
let userId = null
try {
const user = await registry.getUserByIdentity(pubkey)
userId = (user && user.userId) || null
} catch {
return null // do not cache a lookup failure — it may be transient
}
// Bounded cache: drop the oldest entry once full (insertion-ordered Map).
if (this._userIds.size >= MAX_USERID_CACHE) {
const oldest = this._userIds.keys().next()
if (!oldest.done) this._userIds.delete(oldest.value)
}
this._userIds.set(pubkey, userId)
return userId
}
/**
* Record one dispatch. `outcome` is 'ok' or an error message.
*
* SYNCHRONOUS on purpose. The dispatch wrapper does not await this — if
* buffering were async, an entry could still be in flight when close()
* flushes, and the tail of the log would be lost exactly when it matters
* most (shutdown, crash, restart). The identity pubkey is carried raw and
* mapped to a userId later, in flush(), off the request path.
*/
record (method, ctx, outcome, ms) {
if (this._closed || this._skip.has(method)) return
try {
const identity = (ctx && ctx.identity) || null
// ctx.identity is NOT one shape. http/uds carry { publicKey } from
// httpCtxFactory (a real authenticated user). swarm-rpc instead builds
// { type:'swarm', remotePublicKey, trust:'network' } — non-null, with no
// publicKey at all (hyper-svc/lib/transports/swarm-rpc.js:139-147).
// Reading only .publicKey would log every swarm peer as anonymous; and
// note for anyone writing a GATE here rather than a log: "is ctx.identity
// truthy?" admits every swarm peer on the network.
const pubkey = (identity && identity.publicKey) || null
const swarmKey = (identity && identity.remotePublicKey) || null
const origin = (ctx && ctx.origin) || {}
const entry = {
ts: this._now(),
method,
transport: (ctx && ctx.transport) || null,
identity: (pubkey || swarmKey) ? (pubkey || swarmKey).slice(0, 16) : null,
trust: (identity && identity.trust) || (pubkey ? 'authenticated' : null),
_pubkey: pubkey, // resolved to userId at flush, then stripped
remote: origin.remoteAddress || null,
ua: origin.userAgent || null,
ok: outcome === 'ok',
ms
}
if (outcome !== 'ok') entry.error = String(outcome).slice(0, 200)
this._buf.push(entry)
} catch {
// Logging must never affect the call it is describing.
}
}
/**
* Resolve pending userIds, then write the batch in one append. Serialized:
* a flush arriving while one is in progress chains behind it, so lines can
* never interleave mid-write.
*/
async flush () {
this._flushing = (this._flushing || Promise.resolve()).then(() => this._flushOnce())
return this._flushing
}
async _flushOnce () {
if (!this._buf.length) return
const batch = this._buf
this._buf = []
const lines = []
for (const entry of batch) {
try {
const pubkey = entry._pubkey
delete entry._pubkey
entry.userId = await this._userIdFor(pubkey)
lines.push(JSON.stringify(entry))
} catch {
// Skip a single malformed entry rather than lose the whole batch.
}
}
if (!lines.length) return
try {
try {
const st = fs.statSync(this._path)
if (st.size > this._maxBytes) fs.renameSync(this._path, this._path + '.1')
} catch {} // no file yet, or rotation failed — append anyway
fs.appendFileSync(this._path, lines.join('\n') + '\n')
} catch {
// Disk full / permissions — drop the batch rather than crash the daemon.
}
}
async close () {
this._closed = true
if (this._timer) { clearInterval(this._timer); this._timer = null }
await this.flush()
}
}
/**
* Wrap svc.dispatch so every call is recorded. Returns the RpcAuditLog (also
* set on svc._rpcAuditLog) so callers can flush/close it on shutdown.
*
* The wrapper preserves dispatch's contract exactly: same arguments, same
* return value, same thrown errors. It only observes.
*/
export function installRpcAuditLog (svc, opts = {}) {
if (!svc || typeof svc.dispatch !== 'function') return null
if (svc._rpcAuditLog) return svc._rpcAuditLog
const audit = new RpcAuditLog(opts)
audit.start()
const inner = svc.dispatch.bind(svc)
svc.dispatch = async function (method, params, ctx) {
const started = Date.now()
try {
const result = await inner(method, params, ctx)
audit.record(method, ctx, 'ok', Date.now() - started)
return result
} catch (err) {
audit.record(method, ctx, (err && err.message) || 'error', Date.now() - started)
throw err
}
}
svc._rpcAuditLog = audit
return audit
}