import { decodeOp, OpTypes, validateOp } from './operations.js'
import b4a from 'b4a'
/**
* Contract validation — checks that the writer who appended an operation
* has permission to execute it. This prevents rogue Autobase writers from
* injecting admin-level operations.
*
* Permission model:
* - CREATE_USER: any writer (but first user gets admin via autoRegister)
* - UPDATE_USER: writer must be the user themselves or an admin
* - DELETE_USER: admin only
* - ADD_IDENTITY / REMOVE_IDENTITY / LINK_IDENTITY: admin or self
* - UPGRADE_PERMISSIONS: admin only
* - MERGE_ACCOUNTS: admin only
* - REVOKE_IDENTITY / RESTORE_IDENTITY: admin only
* - SET_USERNAME: writer must be the user themselves or an admin
*
* Note: The first writer (bootstrapper) is always trusted for initial setup.
* Contract validation is skipped if opts.contractValidation is false (default for
* backwards compatibility during migration).
*/
// Operations that require admin permissions
const ADMIN_OPS = new Set([
OpTypes.DELETE_USER,
OpTypes.UPGRADE_PERMISSIONS,
OpTypes.MERGE_ACCOUNTS,
OpTypes.REVOKE_IDENTITY,
OpTypes.RESTORE_IDENTITY,
OpTypes.DEACTIVATE_USER,
OpTypes.REACTIVATE_USER
])
// Operations that require admin OR self (userId in op matches writer's user)
const SELF_OR_ADMIN_OPS = new Set([
OpTypes.UPDATE_USER,
OpTypes.ADD_IDENTITY,
OpTypes.REMOVE_IDENTITY,
OpTypes.LINK_IDENTITY,
OpTypes.SET_USERNAME
])
/**
* Create the Autobase apply function for identity registry
*
* @param {Object} opts - Options
* @param {string} opts.prefix - Key prefix for namespacing (default: '')
* @param {boolean} opts.contractValidation - Enable contract validation (default: false for migration compat)
* @returns {Function} Autobase apply function
*/
export function createApplyFunction (opts = {}) {
const prefix = opts.prefix || ''
const enableContract = opts.contractValidation === true
// How to reach the LOG's writer-set controls from whatever object the engine hands apply.
//
// ⛔ This exists because the two engines disagree about one word, silently. autobase takes
// `addWriter(key, { indexer })`; autobee takes `addWriter(key, { isIndexer = true, weight =
// isIndexer ? 2 : 1 })` (autobee 2.1.0 lib/apply-calls.js:33). An unknown `indexer` key is
// IGNORED and `isIndexer` DEFAULTS TO TRUE — so passing autobase's spelling to autobee grants
// every writer indexer standing, with no error, which is the thing hyper-identity's own
// constructor comment warns deadlocks a manifest migration at 2→3 indexers.
//
// So the CORE speaks one vocabulary — `{ isIndexer }` — and each engine package translates on the
// way out. The translation is one line, in the package that knows which engine it is, instead of
// a convention every future caller has to remember.
const adaptHost = opts.adaptHost || ((host) => host)
// Which writer authored a node — the second thing the two engines spell differently.
//
// ⛔ autobase attributes on `node.from.key`; autobee on `node.key` (favela-peer/lib/feed-base.js
// marks this as a trap it hit). The default below is autobase's shape, so under autobee
// `node.from` is UNDEFINED and the contract check below is skipped entirely — a permission gate
// that silently stops gating. Latent today only because `contractValidation` is opt-in.
//
// Attribution must never come from the op body; it comes from whatever the engine vouches for,
// which is exactly why each binding names its own accessor.
const writerKeyOf = opts.writerKeyOf || ((node) => (node && node.from ? node.from.key : null))
return async function apply (batch, view, rawHost) {
const base = adaptHost(rawHost)
const b = view.batch({ update: false })
try {
for (const node of batch) {
try {
const op = decodeOp(node.value)
const validation = validateOp(op)
if (!validation.valid) {
console.error(`[IdentityApply] Invalid op: ${validation.error}`)
continue
}
// Contract validation: check writer permissions
const authorKey = writerKeyOf(node)
if (enableContract && authorKey) {
const writerKey = b4a.toString(authorKey, 'hex')
const violation = await checkContract(b, prefix, writerKey, op)
if (violation) {
console.error(`[IdentityApply] Contract violation: ${violation} (writer: ${writerKey.slice(0, 16)}..., op: ${op.type})`)
continue // Drop the operation
}
}
// The op's own clock, stamped by encodeOp on the authoring peer. Every timestamp and
// every time COMPARISON below uses it, so this apply is a pure function of the log.
//
// ⚠ The `at` fallback is for ops written BEFORE this field existed, already
// durable in deployed logs. Those replay non-deterministically exactly as they do today —
// no worse, and unfixable without rewriting history. Every NEW op converges.
const at = op.timestamp || Date.now()
switch (op.type) {
case OpTypes.CREATE_USER: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (existing) continue // Don't overwrite
// Deterministic admin bootstrap: the FIRST user created gets
// admin permissions. All subsequent users get read-only.
// This decision happens in the apply function (deterministic,
// linearized by Autobase) — NOT in the caller (which would be
// racy in multi-writer scenarios).
const isFirstUser = !(await hasAnyUser(b, prefix))
const permissions = isFirstUser
? ['read', 'write', 'admin']
: ['read']
const resources = isFirstUser
? [{ name: '*', permissions: ['read', 'write', 'admin'] }]
: [{ name: '*', permissions: ['read'] }]
const user = {
userId: op.userId,
username: op.username || null,
name: op.name || 'Anonymous',
email: op.email || null,
profilePicture: op.profilePicture || null,
globalPermissions: permissions,
resources,
identities: op.identities || [],
createdAt: op.createdAt || at,
updatedAt: op.updatedAt || at
}
await b.put(key, user)
// Index identities
for (const id of user.identities) {
await b.put(`${prefix}user-identity:${id}`, { userId: op.userId })
}
break
}
case OpTypes.UPDATE_USER: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
if (op.username !== undefined) user.username = op.username
if (op.name !== undefined) user.name = op.name
if (op.email !== undefined) user.email = op.email
if (op.profilePicture !== undefined) user.profilePicture = op.profilePicture
if (op.theme !== undefined) user.theme = op.theme
if (op.globalPermissions !== undefined) user.globalPermissions = op.globalPermissions
user.updatedAt = at
await b.put(key, user)
break
}
case OpTypes.DELETE_USER: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
// Remove identity index entries AND the identity records + their
// reverse indexes FROM THE VIEW. Since the view-migration these
// records live in the view; deleteUserCascade only cleaned the
// legacy db, so the view's pubkey/sshkey/noisekey indexes
// survived a delete. Mirror REGISTER_IDENTITY's index writes in
// reverse so a cascade is complete and replicates to peers.
for (const id of (user.identities || [])) {
await b.del(`${prefix}user-identity:${id}`)
const idEntry = await b.get(`${prefix}identity:${id}`)
if (idEntry) {
const idRec = idEntry.value
if (idRec.publicKey) await b.del(`${prefix}pubkey:${idRec.publicKey}`)
if (idRec.method === 'ssh-key' && idRec.credentials && idRec.credentials.fingerprint) {
await b.del(`${prefix}sshkey:${idRec.credentials.fingerprint}`)
}
if (idRec.method === 'noise-key' && idRec.credentials && idRec.credentials.publicKey) {
await b.del(`${prefix}noisekey:${idRec.credentials.publicKey}`)
}
if (idRec.method === 'atproto' && idRec.credentials) {
if (idRec.credentials.did) await b.del(`${prefix}atproto-did:${idRec.credentials.did}`)
if (idRec.credentials.handle) await b.del(`${prefix}atproto-handle:${idRec.credentials.handle}`)
}
await b.del(`${prefix}identity:${id}`)
}
}
if (user.username) await b.del(`${prefix}username:${user.username}`)
await b.del(key)
break
}
case OpTypes.ADD_IDENTITY: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
if (!user.identities.includes(op.identityId)) {
user.identities.push(op.identityId)
user.updatedAt = at
await b.put(key, user)
await b.put(`${prefix}user-identity:${op.identityId}`, { userId: op.userId })
}
break
}
case OpTypes.REMOVE_IDENTITY: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
user.identities = user.identities.filter(id => id !== op.identityId)
user.updatedAt = at
await b.put(key, user)
await b.del(`${prefix}user-identity:${op.identityId}`)
break
}
case OpTypes.LINK_IDENTITY: {
await b.put(`${prefix}user-identity:${op.identityId}`, { userId: op.userId })
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (existing) {
const user = existing.value
if (!user.identities.includes(op.identityId)) {
user.identities.push(op.identityId)
user.updatedAt = at
await b.put(key, user)
}
}
// Stamp the owning userId onto the identity record IN THE VIEW.
// Since the view-migration, identity records live in the view and
// getIdentity reads it first; the registry method only wrote
// userId to the legacy db, leaving the view record without it
// (cross-daemon link ownership invisible). Idempotent.
const idKey = `${prefix}identity:${op.identityId}`
const idEntry = await b.get(idKey)
if (idEntry && idEntry.value.userId !== op.userId) {
const idRec = idEntry.value
idRec.userId = op.userId
idRec.updatedAt = at
await b.put(idKey, idRec)
}
break
}
case OpTypes.UPGRADE_PERMISSIONS: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
const resourceName = op.resourceName
if (resourceName === '*') {
// Global permissions
user.globalPermissions = op.permissions
} else {
// Resource-specific permissions
if (!user.resources) user.resources = []
const idx = user.resources.findIndex(r => r.name === resourceName)
if (idx >= 0) {
user.resources[idx].permissions = op.permissions
} else {
user.resources.push({ name: resourceName, permissions: op.permissions })
}
}
user.updatedAt = at
await b.put(key, user)
break
}
case OpTypes.MERGE_ACCOUNTS: {
const primaryKey = `${prefix}user:${op.primaryUserId}`
const secondaryKey = `${prefix}user:${op.secondaryUserId}`
const primary = await b.get(primaryKey)
const secondary = await b.get(secondaryKey)
if (!primary || !secondary) continue
const pUser = primary.value
const sUser = secondary.value
// Merge identities
for (const id of sUser.identities) {
if (!pUser.identities.includes(id)) {
pUser.identities.push(id)
}
await b.put(`${prefix}user-identity:${id}`, { userId: op.primaryUserId })
}
// Merge resources
for (const res of (sUser.resources || [])) {
const idx = (pUser.resources || []).findIndex(r => r.name === res.name)
if (idx < 0) {
if (!pUser.resources) pUser.resources = []
pUser.resources.push(res)
}
}
// Merge global permissions
const allPerms = new Set([...pUser.globalPermissions, ...sUser.globalPermissions])
pUser.globalPermissions = [...allPerms]
pUser.updatedAt = at
await b.put(primaryKey, pUser)
// Mark secondary as merged
sUser.mergedInto = op.primaryUserId
sUser.updatedAt = at
await b.put(secondaryKey, sUser)
break
}
case OpTypes.REVOKE_IDENTITY: {
// Write revocation record to view (consensus-visible)
// Direct-write db is updated separately in registry method
await b.put(`${prefix}revoked:${op.identityId}`, {
identityId: op.identityId,
revokedAt: op.revokedAt || at,
reason: op.reason || null
})
break
}
case OpTypes.RESTORE_IDENTITY: {
await b.del(`${prefix}revoked:${op.identityId}`)
break
}
case OpTypes.DEACTIVATE_USER: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
// Guard: can't deactivate the last admin
if (user.globalPermissions?.includes('admin')) {
let adminCount = 0
for await (const e of b.createReadStream({ gt: `${prefix}user:`, lt: `${prefix}user:\xff` })) {
if (e.value.globalPermissions?.includes('admin') && !e.value.deactivatedAt && !e.value.mergedInto) adminCount++
}
if (adminCount <= 1) continue // Silently skip — can't deactivate last admin
}
user.deactivatedAt = at
user.deactivatedReason = op.reason || null
user.updatedAt = at
await b.put(key, user)
break
}
case OpTypes.REACTIVATE_USER: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
delete user.deactivatedAt
delete user.deactivatedReason
user.updatedAt = at
await b.put(key, user)
break
}
case OpTypes.CREATE_LINK_TOKEN: {
// Store the token row keyed by its hash. Idempotent — a
// duplicate op for the same hash is silently dropped.
const key = `${prefix}link-token:${op.tokenHash}`
const existing = await b.get(key)
if (existing) break
await b.put(key, {
tokenHash: op.tokenHash,
userId: op.userId,
expiresAt: op.expiresAt,
createdAt: op.createdAt || at,
consumed: false
})
break
}
case OpTypes.CONSUME_LINK_TOKEN: {
const tokenKey = `${prefix}link-token:${op.tokenHash}`
const tokenEntry = await b.get(tokenKey)
if (!tokenEntry) break // unknown token
const token = tokenEntry.value
if (token.consumed) break // double-consume
if (token.expiresAt && token.expiresAt < at) break // expired
// 1. Mark token consumed (kept for audit, never deleted)
await b.put(tokenKey, {
...token,
consumed: true,
consumedAt: at,
consumedBy: op.newIdentityId
})
// 2. Attach the new identity to the token's target user
const targetKey = `${prefix}user:${token.userId}`
const targetEntry = await b.get(targetKey)
if (targetEntry) {
const target = targetEntry.value
if (!target.identities.includes(op.newIdentityId)) {
target.identities.push(op.newIdentityId)
target.updatedAt = at
await b.put(targetKey, target)
}
// Always rewrite the user-identity index — if the identity
// was previously linked to a throwaway, this rebinds it.
await b.put(`${prefix}user-identity:${op.newIdentityId}`, { userId: token.userId })
}
// 3. Optionally deactivate the throwaway user the new
// identity used to belong to. Strip the identity from
// that user first so it isn't double-counted.
if (op.deactivateUserId && op.deactivateUserId !== token.userId) {
const dKey = `${prefix}user:${op.deactivateUserId}`
const dEntry = await b.get(dKey)
if (dEntry) {
const u = dEntry.value
u.identities = (u.identities || []).filter(id => id !== op.newIdentityId)
u.deactivatedAt = at
u.deactivatedReason = 'merged-via-link-token'
u.updatedAt = at
await b.put(dKey, u)
}
}
break
}
case OpTypes.SET_USERNAME: {
const key = `${prefix}user:${op.userId}`
const existing = await b.get(key)
if (!existing) continue
const user = existing.value
// Remove old username index
if (user.username) {
await b.del(`${prefix}username:${user.username}`)
}
// Set new username
user.username = op.username
user.updatedAt = at
await b.put(key, user)
await b.put(`${prefix}username:${op.username}`, { userId: op.userId })
break
}
case OpTypes.ADD_WRITER: {
// Audit trail in the view
await b.put(`${prefix}writer:${op.writerKey}`, {
addedAt: at,
addedBy: op.addedBy || null,
role: op.role || 'writer'
})
// Propagate to the Autobase host so the new writer's core
// participates in linearization.
if (base && typeof base.addWriter === 'function') {
try {
await base.addWriter(b4a.from(op.writerKey, 'hex'), {
isIndexer: op.role === 'indexer'
})
} catch (err) {
console.error('[IdentityApply] base.addWriter failed:', err.message)
}
}
break
}
case OpTypes.REMOVE_WRITER: {
await b.del(`${prefix}writer:${op.writerKey}`)
if (base && typeof base.removeWriter === 'function') {
try {
await base.removeWriter(b4a.from(op.writerKey, 'hex'))
} catch (err) {
console.error('[IdentityApply] base.removeWriter failed:', err.message)
}
}
break
}
case OpTypes.REGISTER_MIRROR_WRITERS: {
// View-only record. The enclosing app watches this key
// range and issues in-band addWriter ops on the matching
// OTHER autobases it hosts. Apply must stay deterministic:
// no side effects on other autobases from here.
await b.put(`${prefix}mirror-writers:${op.devicePubkey}`, {
keys: op.keys,
recordedAt: at
})
break
}
case OpTypes.REGISTER_IDENTITY: {
// Full identity record into the view (replicated to mirrors).
// Replaces the previous direct-write to a per-daemon
// Hyperbee (`db`), which was local-only. Mirror peers now
// see the full record + reverse indexes via standard
// autobase replication. See pear-git-v3/test/
// identity-record-replication.test.js for the proof.
const identity = op.identity
if (!identity || !identity.identityId) break
await b.put(`${prefix}identity:${identity.identityId}`, identity)
if (identity.publicKey) {
await b.put(`${prefix}pubkey:${identity.publicKey}`, { identityId: identity.identityId })
}
if (identity.method === 'ssh-key' && identity.credentials && identity.credentials.fingerprint) {
await b.put(`${prefix}sshkey:${identity.credentials.fingerprint}`, { identityId: identity.identityId })
}
if (identity.method === 'noise-key' && identity.credentials && identity.credentials.publicKey) {
await b.put(`${prefix}noisekey:${identity.credentials.publicKey}`, { identityId: identity.identityId })
}
if (identity.method === 'atproto' && identity.credentials) {
if (identity.credentials.did) {
await b.put(`${prefix}atproto-did:${identity.credentials.did}`, { identityId: identity.identityId })
}
if (identity.credentials.handle) {
await b.put(`${prefix}atproto-handle:${identity.credentials.handle}`, { identityId: identity.identityId })
}
}
break
}
}
} catch (err) {
console.error('[IdentityApply] Skipping malformed op:', err.message)
continue
}
}
await b.flush()
} catch (err) {
console.error('[IdentityApply] Batch error:', err.message)
await b.close()
throw err
}
}
}
/**
* Check if a writer has permission to execute an operation.
*
* Resolves the writer's key to a user by scanning all users' identities,
* then checks if that user has the required permission level.
*
* @param {Object} b - Hyperbee batch
* @param {string} prefix - Key prefix
* @param {string} writerKey - Writer's Hypercore public key (hex)
* @param {Object} op - The operation being executed
* @returns {string|null} Violation reason, or null if authorized
*/
async function checkContract (b, prefix, writerKey, op) {
// CREATE_USER is always allowed (autoRegister handles first-user-admin)
if (op.type === OpTypes.CREATE_USER) return null
// Find the user associated with this writer key
const writerUser = await resolveWriterUser(b, prefix, writerKey)
// If no user found for this writer, check if ANY users exist yet.
// During bootstrap (empty registry), allow all ops from the first writer.
if (!writerUser) {
const hasUsers = await hasAnyUser(b, prefix)
if (!hasUsers) return null // Bootstrap: empty registry, first writer is trusted
return `Writer ${writerKey.slice(0, 16)} has no associated user`
}
const isAdmin = writerUser.globalPermissions && writerUser.globalPermissions.includes('admin')
// Admin-only operations
if (ADMIN_OPS.has(op.type)) {
if (!isAdmin) return `${op.type} requires admin permissions`
return null
}
// Self-or-admin operations
if (SELF_OR_ADMIN_OPS.has(op.type)) {
if (isAdmin) return null
if (op.userId === writerUser.userId) return null
return `${op.type} requires admin or self (writer is ${writerUser.userId}, op targets ${op.userId})`
}
return null // Unknown op types pass through (validated by validateOp)
}
/**
* Find the user associated with a writer's Hypercore key.
* Looks up the writer key as a pubkey identity, then resolves to user.
*/
async function resolveWriterUser (b, prefix, writerKey) {
// Check pubkey reverse lookup (writer key may be registered as an identity)
const pubkeyEntry = await b.get(`${prefix}pubkey:${writerKey}`)
if (pubkeyEntry) {
const identityLink = await b.get(`${prefix}user-identity:${pubkeyEntry.value.identityId}`)
if (identityLink) {
const userEntry = await b.get(`${prefix}user:${identityLink.value.userId}`)
if (userEntry) return userEntry.value
}
}
// Scan all users to find one with this writer key in their identities
// (fallback for when the key isn't indexed as a pubkey)
for await (const entry of b.createReadStream({ gt: `${prefix}user:`, lt: `${prefix}user:\xff` })) {
const user = entry.value
if (user.identities && user.identities.includes(writerKey)) {
return user
}
}
return null
}
async function hasAnyUser (b, prefix) {
for await (const _ of b.createReadStream({ gt: `${prefix}user:`, lt: `${prefix}user:\xff`, limit: 1 })) {
return true
}
return false
}