/**
* Org RPC handlers.
*
* Each handler takes (ctx, params) and returns a JSON-serializable result.
* Mutating handlers call requireWriter(ctx) first — UDS bypasses, HTTP
* needs a valid session bound to an IdentityRegistry admin/write user.
*/
import { requireWriter, requireAdmin } from '../auth/check.js'
import { resolveUserId } from '../auth/user-resolver.js'
import { mountOrgGroupedRepos } from '../services/org-grouped-repos.js'
import { assertValidRepoName } from '../repo-name.js'
// Debounce for the org-room re-sweep. Matches attachSshResyncTrigger and
// scheduleMirrorReannounce so every autobase-update-driven path in the daemon
// coalesces on the same interval. See autoMountOrgRepos for why this must be a
// timer (macrotask) and not a bare await.
const ORG_RESWEEP_DEBOUNCE_MS = 1500
export function orgHandlers (svc) {
const orgs = () => svc.modules.orgs
const registry = () => svc.identityRegistry
// ── Org-scoped bidirectional git sync helpers (#16, over the org autobase) ──
// The org tracker mount gives refs/issues; the git OBJECTS ride a punch://
// PACK clone that only mirrors used to trigger. These publish our origin
// coords into the org autobase and boot a scoped reverse RepoPuller toward
// each co-member, so an org join actually clones the org's repos.
// Repos an org reverse-puller may pull = the union of every joined org's
// resources. Add-only (never-delete-storage): dropping a repo from an org
// doesn't retract already-pulled objects. The puller's repoFilter reads this
// live, so repos added to an org post-boot start matching without a restart.
function refreshOrgAllow (org) {
if (!org || !Array.isArray(org.resources)) return
svc._orgRepoAllow = svc._orgRepoAllow || new Set()
for (const r of org.resources) svc._orgRepoAllow.add(r)
}
// Publish our origin coords (serverKey + repoEventLog) into the org autobase.
// WRITERS ONLY — getOrgRoom(name).base.writable gates it (a reader can't
// append and serves nothing). At most once per org (coords are stable); on
// failure the once-guard is released so the next sweep retries.
async function announceOrgCoords (orgName) {
if (!svc.identity || !svc.identity.publicKeyHex) return
if (!svc.bootstrapKeys || !svc.bootstrapKeys.repoEventLog) return
const room = orgs().getOrgRoom(orgName)
if (!room || !room.base || !room.base.writable) return
svc._orgCoordsAnnounced = svc._orgCoordsAnnounced || new Set()
if (svc._orgCoordsAnnounced.has(orgName)) return
svc._orgCoordsAnnounced.add(orgName)
try {
await orgs().setMeta(orgName, 'peer:' + svc.identity.publicKeyHex, {
serverKey: svc.identity.publicKeyHex,
repoEventLog: svc.bootstrapKeys.repoEventLog
})
} catch (err) {
svc._orgCoordsAnnounced.delete(orgName)
throw err
}
}
// Co-members' announced coords, minus our own.
async function listOrgPeerCoords (orgName) {
const own = svc.identity && svc.identity.publicKeyHex
let entries = []
try { entries = await orgs().listMeta(orgName, 'peer:') } catch { entries = [] }
return (entries || [])
.map((e) => e.value)
.filter((v) => v && v.serverKey && v.repoEventLog && v.serverKey !== own)
}
// Boot a reverse RepoPuller toward a co-member, scoped to repos in orgs we
// joined (svc._orgRepoAllow), so a co-member's non-org repos never reach our
// disk (#35). Delegates to the idempotent per-serverKey boot wired by
// setupMirrorWriterCoordination; no-op until that has set svc._bootReversePuller.
function bootOrgReversePuller (serverKey, repoEventLog) {
if (typeof svc._bootReversePuller !== 'function') return
svc._bootReversePuller(serverKey, repoEventLog, {
repoFilter: (entry) => !!(svc._orgRepoAllow && entry && svc._orgRepoAllow.has(entry.name))
})
}
// Phase B: after joining an org, mount its grouped repos (scoped to
// exactly this org's resourceKeys — no global repo stream, so isolation
// holds). Mirrors the join-repo path: bind the tracker bootstrap key,
// open the tracker, rendezvous on its room topic, and let RefReconciler
// materialize the bare repo from the autobase + overlay. Read-only for
// a reader joiner — we never addWriter here, so push stays gated.
//
// The org record + its resourceKeys replicate asynchronously after
// joinOrg returns, and may grow as more repos are added, so we sweep on
// each org-room view update (idempotent — `mounted` dedups) and bound
// the watch so it can't run forever.
async function autoMountOrgRepos (orgName) {
if (!svc.issueRegistry) return
// Shared dedup across the initial sweep and every update-triggered
// re-sweep so each repo is mounted at most once.
const mounted = new Set()
const deps = {
getOrg: async (name) => {
const org = await orgs().get(name)
refreshOrgAllow(org) // keep the reverse-puller scope current
return org
},
issueRegistry: svc.issueRegistry,
activateRoom: svc.activateRoom ? svc.activateRoom.bind(svc) : null,
refReconcilers: svc._refReconcilers,
// #16 loop over the org autobase: announce our coords (writers) + boot a
// scoped reverse RepoPuller toward each co-member, so the org's git
// OBJECTS actually clone (punch:// PACK) — not just the refs/tracker.
announceCoords: announceOrgCoords,
listPeerCoords: listOrgPeerCoords,
bootReversePuller: bootOrgReversePuller,
mounted,
onMount: (repo) => {
// eslint-disable-next-line no-console
console.log('[org-grouped-repos] mounted', repo, 'from org', orgName)
},
onError: (repo, err) => {
if (process.env.PGV3_DEBUG_ORG_REPOS) {
// eslint-disable-next-line no-console
console.log('[org-grouped-repos] mount', repo, 'failed:', err.message)
}
}
}
// The org record + its resourceKeys replicate asynchronously and may grow
// as more repos are added, so re-sweep on each org-room view update
// (idempotent — `mounted` dedups). The watch is PERSISTENT for the lifetime
// of the daemon: the old 120s teardown was correct for a one-off joiner
// (repos replicate seconds after join-org) but wrong for an owner or
// long-running mirror armed at boot — a contributor adding a repo hours
// later must still be auto-mounted. Guard so each org is watched at most
// once even though autoMountOrgRepos runs from boot + create-org + join-org.
svc._orgRepoWatchers = svc._orgRepoWatchers || new Set()
const room = orgs().getOrgRoom(orgName)
if (room && room.base && typeof room.base.on === 'function' && !svc._orgRepoWatchers.has(orgName)) {
svc._orgRepoWatchers.add(orgName)
// DEBOUNCED + SERIALIZED. The previous handler was fire-and-forget, so
// every org-room view advance started ANOTHER full sweep of that org's
// repos while earlier sweeps were still running. `mounted` dedups the
// mount ACTION, but not the sweep: each pass still walks the org's repo
// list calling ensureFor(), which opens a per-repo tracker autobase and
// runs a reconcile.
//
// On an ORIGIN this is harmless — org rooms only advance when someone
// writes. On a MIRROR every org room is a replication target, so during
// catch-up all 11 advance continuously and sweeps stack without bound.
// Measured on stortron: 294 org-mount lines in one boot against a normal
// 117, with the daemon pinned at 100% CPU and boot never reaching its
// banner.
//
// setTimeout is deliberate, not cosmetic: it is a MACROTASK, so the loop
// is handed back between sweeps. Awaiting alone would only drain the
// microtask queue and still starve accept()/SIGCHLD — the same trap
// documented in 0.1.180, 0.1.214 and 0.1.215.
let sweepTimer = null
let sweeping = false
let resweep = false
const runSweep = async () => {
if (sweeping) { resweep = true; return } // coalesce into the running pass
sweeping = true
try {
do {
resweep = false
await mountOrgGroupedRepos(orgName, deps)
} while (resweep)
} catch {
} finally {
sweeping = false
}
}
const onUpdate = () => {
if (sweepTimer) return // a sweep is already scheduled
sweepTimer = setTimeout(() => {
sweepTimer = null
runSweep()
}, ORG_RESWEEP_DEBOUNCE_MS)
if (sweepTimer && typeof sweepTimer.unref === 'function') sweepTimer.unref()
}
room.base.on('update', onUpdate)
}
await mountOrgGroupedRepos(orgName, deps)
}
// Expose for the boot sweep (pear-git.js) so OWNED orgs re-establish the
// grouped-repos watcher on restart — not just orgs we joined. Without this
// an org owner never auto-pulls repos that contributors add: the org record
// + its resourceKeys replicate, but nothing mounts/materializes them.
svc._autoMountOrgRepos = autoMountOrgRepos
return [
{
name: 'list-orgs',
fn: async (ctx, params = {}) => {
const filters = {}
if (params.type) filters.type = params.type
if (params.visibility) filters.visibility = params.visibility
return orgs().list(filters)
}
},
{
name: 'get-org',
fn: async (ctx, params = {}) => {
if (!params.name) throw new Error('get-org requires { name }')
const org = await orgs().get(params.name)
if (!org) throw new Error(`org "${params.name}" not found`)
return org
}
},
{
name: 'create-org',
fn: async (ctx, params = {}) => {
// U36: org creation is privileged — orgs live in a single
// flat namespace that all peers share, so any user spinning
// one up pollutes the global view. Gated on 'admin' (owner).
// UDS bypasses because the local user runs the daemon.
await requireAdmin(ctx, registry())
if (!params.name) throw new Error('create-org requires { name }')
const type = params.type || 'group'
const org = await orgs().create(params.name, type, {
displayName: params.displayName,
description: params.description,
about: params.about,
userId: params.userId,
visibility: params.visibility
})
// Announce this org room on the shared swarm so peers/mirrors can
// rendezvous for it by its discoveryKey (Keet activateRoom).
const createdRoom = orgs().getOrgRoom(params.name)
if (createdRoom && typeof svc.activateRoom === 'function') {
svc.activateRoom(createdRoom.discoveryKey)
}
// Broadcast the org's RECORD autobase key via OrgEventLog so mirrors
// (#37 per-org-autobase model) mountOrg() it and replicate a full
// org. Org issues live IN that record autobase (linked-core #42), so
// there is no separate issue-tracker key to convey — issueBootstrapKey
// stays null on org events.
// Mirrors RepoEventLog's create-emit pattern (pear-git.js 'serve'
// listener). Failures here MUST NOT roll back the org create — the
// record is the canonical thing; propagation is best-effort and
// re-broadcast by the boot backfill walk or `pgh resync-from-origin`.
if (svc._orgEventLog && svc._orgEventLog.writable) {
try {
const orgRoom = orgs().getOrgRoom(params.name)
const orgKeyHex = orgRoom && orgRoom.keyHex
const orgKey = orgKeyHex ? Buffer.from(orgKeyHex, 'hex') : null
const seen = svc._orgEventLogSeen
if (!seen || !seen.has(params.name)) {
await svc._orgEventLog.append({
op: 0, // created
name: params.name,
issueBootstrapKey: null,
orgBootstrapKey: orgKey
})
if (seen) seen.add(params.name)
}
} catch (err) {
// eslint-disable-next-line no-console
console.error('[org-event-log] create-org append failed for', params.name, '-', err && err.message)
}
}
// Watch this OWNED org's room for grouped-repo additions, exactly
// as a joiner does. When a contributor later adds a repo resource,
// the resourceKey replicates into our org autobase, the room fires
// 'update', and the watcher mounts + materializes the repo. Without
// this the owner sees the org record grow but never pulls the git
// content. Best-effort — must not roll back org create.
autoMountOrgRepos(params.name).catch(() => {})
return org
}
},
{
name: 'update-org',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
if (!params.name) throw new Error('update-org requires { name }')
return orgs().update(params.name, {
displayName: params.displayName,
description: params.description,
about: params.about,
visibility: params.visibility
})
}
},
{
name: 'delete-org',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
if (!params.name) throw new Error('delete-org requires { name }')
await orgs().delete(params.name)
return { name: params.name, deleted: true }
}
},
{
// Caller passes any identifier shape (userId, identity-hex, username)
// as `userId`. The handler resolves it to the canonical userId
// before writing, rejects unresolved inputs (so we never persist
// a string that doesn't point at a real User), and rejects
// duplicates (same User already in the org under any identifier).
// This is what guards against the cross-identifier dupe seen in
// 2026-05-11 where one User got added to hyper-fleeky-dev twice —
// once by identity hex, once by username.
name: 'add-org-member',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
if (!params.org || !params.userId) {
throw new Error('add-org-member requires { org, userId, role? }')
}
const resolved = await resolveUserId(registry(), params.userId)
if (!resolved) {
const err = new Error('no user matches "' + params.userId + '" (tried userId, identity, username)')
err.code = 'ENOTFOUND'
err.status = 404
throw err
}
const org = await orgs().get(params.org)
if (!org) {
const err = new Error('org "' + params.org + '" not found')
err.code = 'ENOTFOUND'
err.status = 404
throw err
}
if (Array.isArray(org.members) && org.members.some(m => m.userId === resolved)) {
const err = new Error('user is already a member of "' + params.org + '"')
err.code = 'EEXISTS'
err.status = 409
throw err
}
return orgs().addMember(params.org, resolved, params.role || 'member')
}
},
{
name: 'remove-org-member',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
if (!params.org || !params.userId) {
throw new Error('remove-org-member requires { org, userId }')
}
return orgs().removeMember(params.org, params.userId)
}
},
{
name: 'add-org-resource',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
// The pgh CLI sends the org as `name`; accept either for back-compat.
const orgName = params.org || params.name
if (!orgName || !params.resource) {
throw new Error('add-org-resource requires { org (or name), resource }')
}
// Validate BEFORE the try below: getForRepo MINTS a tracker autobase
// on first reference, and that call is wrapped in a best-effort catch
// — validating inside it would swallow the rejection and still add the
// bad name to the org. Passing an ORG name here is how `7rnx` and
// `hyper-fleeky-dev` became repos (audited 2026-07-28).
assertValidRepoName(params.resource, 'resource')
// Record the repo's tracker bootstrap key on the org record so a
// peer that joins the org can mount this repo (scoped — Phase B).
// Best-effort: if the tracker can't be opened here (e.g. the repo
// doesn't exist yet on a mirror), add the name only; the key gets
// backfilled by a later add-resource once the repo exists.
let opts = {}
try {
const tracker = await svc.issueRegistry.getForRepo(params.resource)
if (tracker && tracker.keyHex) opts = { bootstrapKey: tracker.keyHex }
// Serve this repo's room so a peer that joins the org can actually
// replicate it (the join-side auto-mount rendezvouses on the same
// topic). Without this, a resource minted here isn't advertised
// until the next boot scan. Mirrors join-repo's activateRoom.
if (tracker && tracker.discoveryKey && typeof svc.activateRoom === 'function') {
svc.activateRoom(tracker.discoveryKey)
}
} catch {}
return orgs().addResource(orgName, params.resource, opts)
}
},
{
name: 'remove-org-resource',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
const orgName = params.org || params.name
if (!orgName || !params.resource) {
throw new Error('remove-org-resource requires { org (or name), resource }')
}
return orgs().removeResource(orgName, params.resource)
}
},
// ── Per-org-autobase invites / join / enroll (issue #35) ──
//
// Replaces the old app-layer create-invite/consume-invite (which
// issued grants within a single shared autobase). The bearer invite
// now carries ONE org's autobase bootstrap key; joining mounts only
// that org. Peer mode = these handlers; mirror mode = list-org-keys.
{
// Owner: mint a bearer invite for one org (+ the role it grants).
// Mint a blind-pairing invite for an org (P2P). A joiner pairs over the
// swarm and, for contributor+, is addWritered DURING pairing — no HTTP,
// no separate enroll step.
name: 'create-org-invite',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
if (!params.name) throw new Error('create-org-invite requires { name }')
if (!svc.swarm || !svc.swarm.swarm) throw new Error('swarm not available for pairing')
const { invitation, descriptor } = await orgs().createOrgInvite(params.name, {
role: params.role || 'reader',
swarm: svc.swarm.swarm
})
// Persist so the pairing member is re-registered on daemon restart.
if (svc._pendingPairings && descriptor) svc._pendingPairings.add('orgs', params.name, descriptor)
return { invitation }
}
},
{
// Joiner: pair into ONLY that org over the swarm; become a writer
// (contributor+) DURING pairing. One step, P2P — no enroll-writer.
name: 'join-org',
fn: async (ctx, params = {}) => {
await requireWriter(ctx, registry())
if (!params.invitation) throw new Error('join-org requires { invitation }')
if (!svc.swarm || !svc.swarm.swarm) throw new Error('swarm not available for pairing')
const joined = await orgs().joinOrg(params.invitation, { swarm: svc.swarm.swarm })
const room = orgs().getOrgRoom(joined.name)
if (room && typeof svc.activateRoom === 'function') {
svc.activateRoom(room.discoveryKey)
}
// Phase B: auto-mount this org's grouped repos (scoped). Detached —
// the org record + its repo keys replicate after this returns.
autoMountOrgRepos(joined.name).catch(() => {})
return joined
}
},
{
// Mirror mode: every org's bootstrap key, for a sister daemon to
// mountOrg() each → full replica. Admin-gated (full-forge
// replication capability); peers never call this.
name: 'list-org-keys',
fn: async (ctx, params = {}) => {
await requireAdmin(ctx, registry())
return orgs().listOrgKeys()
}
}
]
}