Hub stage 2: install, update and uninstall apps on several servers

Writes use only <ID>_ADMIN_TOKEN and need the apps.write capability.
Store URLs are resolved from each server's own listing. The config is saved
to history before an uninstall. Zip uploads are fanned out to the servers.
Every target's result is audited. The console gets a row menu on the
matrix, a target picker, a results dialog with Retry failed, the store and
install from file in All servers mode, and single-server actions for writers.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

hub/server.ts

@@ -22,6 +22,8 @@ import { LoginLimiter, lockedMessage } from './ratelimit.ts'
 import { Audit } from './audit.ts'
 import { routeFor } from './routes.ts'
 import { buildMatrix } from './apps-matrix.ts'
+import { AppsWriter, actionForPath, type FileAction, type TargetResult } from './apps-write.ts'
+import { TooLarge, parseMultipart, readLimited, type UploadedFile } from './multipart.ts'
 import {
   mergeQueryLogs,
   mergeSettled,
@@ -80,10 +82,30 @@ async function readForm(req: http.IncomingMessage): Promise<URLSearchParams> {
     chunks.push(chunk as Buffer)
   }
   const type = String(req.headers['content-type'] ?? '')
-  if (size > MAX_BODY || !type.includes('application/x-www-form-urlencoded')) return new URLSearchParams()
-  return new URLSearchParams(Buffer.concat(chunks).toString('utf8'))
+  if (size > MAX_BODY) return new URLSearchParams()
+  const text = Buffer.concat(chunks).toString('utf8')
+  if (type.includes('application/x-www-form-urlencoded')) return new URLSearchParams(text)
+  if (type.includes('application/json')) {
+    // A flat object; an array becomes a comma-separated list, as the form fields are.
+    const out = new URLSearchParams()
+    try {
+      const data = JSON.parse(text) as Record<string, unknown>
+      for (const [k, v] of Object.entries(data)) {
+        if (Array.isArray(v)) out.set(k, v.map(String).join(','))
+        else if (typeof v === 'string' || typeof v === 'number' || typeof v === 'boolean') out.set(k, String(v))
+      }
+    } catch {
+      // Not JSON: no parameters.
+    }
+    return out
+  }
+  return new URLSearchParams()
 }
 
+const isMultipart = (req: http.IncomingMessage): boolean => String(req.headers['content-type'] ?? '').toLowerCase().startsWith('multipart/form-data')
+
+const csv = (value: string | undefined): string[] => (value ?? '').split(',').map((v) => v.trim()).filter((v) => v !== '')
+
 function bearer(req: http.IncomingMessage): string | null {
   const header = String(req.headers.authorization ?? '')
   return header.startsWith('Bearer ') ? header.slice(7).trim() : null
@@ -108,6 +130,7 @@ export function createHub(config: HubConfig, users: Users = Users.load(config.da
   const sessions = new Sessions(config, users)
   const limiter = new LoginLimiter()
   const audit = new Audit(config.dataDir)
+  const writer = new AppsWriter(config, upstream)
   const byId = new Map(config.servers.map((s) => [s.id, s]))
 
   /* Ask every server; the ones that fail become warnings (see mergeSettled). */
@@ -162,6 +185,16 @@ export function createHub(config: HubConfig, users: Users = Users.load(config.da
           const pages = data.map((page, i) => ({ node: answered[i], page }))
           return ok(mergeQueryLogs(pages, { pageNumber, entriesPerPage, descending: params.descendingOrder !== 'false' }), warnings)
         }
+        case 'apps/listStoreApps': {
+          // The first answering server's listing, each app tagged with the servers it is installed on.
+          const { data, warnings } = mergeSettled(await everywhere<{ storeApps?: { name: string }[] }>(apiPath, params))
+          const installedOn = new Map<string, string[]>()
+          for (const r of await everywhere<{ apps?: { name: string }[] }>('apps/list', {})) {
+            if (r.ok) for (const a of r.data.apps ?? []) installedOn.set(a.name, [...(installedOn.get(a.name) ?? []), r.node])
+          }
+          const storeApps = (data[0].storeApps ?? []).map((a) => ({ ...a, installedOn: installedOn.get(a.name) ?? [] }))
+          return ok({ ...data[0], storeApps }, warnings)
+        }
         case 'apps/list': {
           const { data, warnings } = mergeSettled(await everywhere<{ apps?: { name: string }[] }>(apiPath, params))
           const seen = new Map<string, { name: string }>()
@@ -178,8 +211,131 @@ export function createHub(config: HubConfig, users: Users = Users.load(config.da
     }
   }
 
+  type Who = { user: string; role: string }
+
+  /* One audit line per target: who, which server, which app, what happened. */
+  function auditResults(who: Who, ip: string, action: string, name: string, results: TargetResult[]): void {
+    for (const r of results) {
+      audit.log({
+        user: who.user, role: who.role, ip, action: `apps.${action}`, server: r.server, target: name,
+        outcome: r.ok ? 'ok' : 'failed', detail: r.message,
+      })
+    }
+  }
+
+  async function uploadOf(req: http.IncomingMessage, res: http.ServerResponse, params: Params): Promise<UploadedFile | null> {
+    let parsed
+    try {
+      parsed = parseMultipart(await readLimited(req), String(req.headers['content-type'] ?? ''))
+    } catch (e) {
+      if (e instanceof TooLarge) {
+        res.setHeader('connection', 'close')
+        send(res, 413, fail(e.message))
+      } else send(res, 400, fail('The upload could not be read.'))
+      return null
+    }
+    Object.assign(params, parsed.fields)
+    const file = parsed.files[0]
+    if (file === undefined || file.data.length === 0) {
+      send(res, 400, fail('The zip file is missing.'))
+      return null
+    }
+    return file
+  }
+
+  /* The write endpoints. The role has already been checked against the route's capability. */
+  async function write(req: http.IncomingMessage, res: http.ServerResponse, apiPath: string, kind: string, params: Params, who: Who, ip: string): Promise<void> {
+    if (kind === 'write-all') {
+      if (req.method !== 'POST') {
+        send(res, 405, fail('Use POST.'))
+        return
+      }
+      if (apiPath === 'hub/apps/run') {
+        const action = params.action
+        if (action !== 'install-store' && action !== 'update-store' && action !== 'uninstall') {
+          send(res, 400, fail('action must be install-store, update-store or uninstall.'))
+          return
+        }
+        const outcome = await writer.runMany(action, params.name ?? '', csv(params.targets), { suppliedUrl: params.url })
+        if ('error' in outcome) {
+          audit.log({ user: who.user, role: who.role, ip, action: `apps.${action}`, target: params.name, outcome: 'refused', detail: outcome.error })
+          send(res, 400, fail(outcome.error))
+          return
+        }
+        auditResults(who, ip, action, params.name ?? '', outcome.results)
+        send(res, 200, ok({ results: outcome.results }))
+        return
+      }
+      if (apiPath === 'hub/apps/upload') {
+        if (!isMultipart(req)) {
+          send(res, 400, fail('Send multipart/form-data.'))
+          return
+        }
+        const file = await uploadOf(req, res, params)
+        if (file === null) return
+        if (params.action !== 'install' && params.action !== 'update') {
+          send(res, 400, fail('action must be install or update.'))
+          return
+        }
+        const action: FileAction = params.action === 'install' ? 'install-file' : 'update-file'
+        const outcome = await writer.runMany(action, params.name ?? '', csv(params.targets), { file })
+        if ('error' in outcome) {
+          audit.log({ user: who.user, role: who.role, ip, action: `apps.${action}`, target: params.name, outcome: 'refused', detail: outcome.error })
+          send(res, 400, fail(outcome.error))
+          return
+        }
+        auditResults(who, ip, action, params.name ?? '', outcome.results)
+        send(res, 200, ok({ results: outcome.results }))
+        return
+      }
+    }
+
+    // write-one: an upstream endpoint aimed at ONE server (`node=<id>`).
+    const action = actionForPath(apiPath)
+    const node = params.node ?? ''
+    const server = byId.get(node)
+    if (action === null) {
+      send(res, 403, fail('The hub is read-only.'))
+      return
+    }
+    if (server === undefined) {
+      send(res, 400, fail(node === '' || node === 'cluster' ? 'Choose one server: this change cannot be made on every server at once here.' : `Unknown node: ${node}`))
+      return
+    }
+    const name = params.name ?? ''
+    let file: UploadedFile | undefined
+    if (action === 'install-file' || action === 'update-file') {
+      if (!isMultipart(req)) {
+        send(res, 400, fail('Send multipart/form-data.'))
+        return
+      }
+      const uploaded = await uploadOf(req, res, params)
+      if (uploaded === null) return
+      file = uploaded
+    }
+    let result: TargetResult
+    let response: unknown = {}
+    try {
+      let url: string | undefined
+      if (action === 'install-store' || action === 'update-store') {
+        const fromStore = server.adminToken === undefined ? null : await writer.storeUrl(server, name).catch(() => null)
+        if (server.adminToken !== undefined && (fromStore === null || params.url !== fromStore)) {
+          throw new Error("the url is not from this server's store listing")
+        }
+        url = fromStore ?? undefined
+      }
+      response = await writer.perform(server, action, name, url, file)
+      result = { server: server.id, ok: true }
+    } catch (e) {
+      result = { server: server.id, ok: false, message: e instanceof Error ? e.message : 'failed' }
+    }
+    auditResults(who, ip, action, name, [result])
+    send(res, 200, result.ok ? ok(response) : fail(`${server.id}: ${result.message}`))
+  }
+
   async function api(req: http.IncomingMessage, res: http.ServerResponse, apiPath: string, url: URL): Promise<void> {
-    const form = await readForm(req)
+    // A multipart body (an app zip) is read later, once the caller is known to be allowed to send it.
+    const form = isMultipart(req) ? new URLSearchParams() : await readForm(req)
     const params: Params = {}
     for (const [k, v] of url.searchParams) params[k] = v
     for (const [k, v] of form) params[k] = v
@@ -249,6 +405,10 @@ export function createHub(config: HubConfig, users: Users = Users.load(config.da
       send(res, 200, ok({ entries: audit.list(Number(params.limit ?? 200), params.before) }))
       return
     }
+    if (route.kind !== 'read') {
+      await write(req, res, apiPath, route.kind, params, who, ip)
+      return
+    }
     if (apiPath === 'hub/apps/matrix') {
       try {
         const { matrix, warnings } = await buildMatrix(config, upstream)