pithos/hosted/supervisor.js

// One s3d per hosted account, each wrapped in the same createPithos() the
// desktop runs, so the server speaks exactly the /api the Pithos UI already
// knows. Accounts live under <root>/users/<id>/:
//
//   account.json   who (public key), how they got in (gate), port slot, state
//   s3d.yml        loopback-only addresses from the slot, random admin password
//   data/          the s3d data folder moved here from the user's PC
//   rules.json     public folders, links and shared keys
//   usage.json     bytes served this month
//
// An s3d starts on the account's first request and is stopped after
// `idleMinutes` without one, flushing leftovers to Sia first. At most
// `maxRunning` run at once; the least recently used idle one makes room.

import fs from 'node:fs';
import path from 'node:path';
import { createPithos } from '../core/server.js';
import { makeCli } from '../core/cli.js';
import { readConfig, writeConfig, randomPassword } from '../core/config.js';
import { writePrefs } from '../core/prefs.js';
import { S3Client } from '../core/s3.js';
import * as admin from '../core/admin.js';
import { readJson, writeJson } from './store.js';

// reservedKey(akid): true for key ids the gateway itself hands out (grants),
// which no account's s3d may also use.
export function createSupervisor(cfg, log = console.log, { reservedKey = () => false } = {}) {
  const usersDir = path.join(cfg.root, 'users');
  const slotsFile = path.join(cfg.root, 'slots.json');
  const keysFile = path.join(cfg.root, 'keys.json');
  const running = new Map(); // id -> instance
  const loading = new Map(); // id -> Promise<instance>
  const closing = new Map(); // id -> Promise, while an s3d is being stopped
  const keyIndex = readJson(keysFile, {}); // accessKeyId -> account id (owners' s3d keys)

  const dir = (id) => path.join(usersDir, id);
  const accountFile = (id) => path.join(dir(id), 'account.json');
  const configFile = (id) => path.join(dir(id), 's3d.yml');
  const dataDir = (id) => path.join(dir(id), 'data');

  function account(id) {
    if (!/^[0-9a-f]{32}$/.test(id)) return null;
    return readJson(accountFile(id), null);
  }

  function saveAccount(a) { writeJson(accountFile(a.id), a); nameIndex = null; }

  // Accounts by the BNS name they signed up with (one slot per name), for
  // public addresses like /p/alice.x/<drive>/… instead of the 32-hex id.
  let nameIndex = null;
  function accountByName(name) {
    if (!nameIndex) {
      nameIndex = new Map();
      let ids = [];
      try { ids = fs.readdirSync(usersDir); } catch {}
      for (const id of ids) {
        const a = account(id);
        if (a?.gate?.type === 'bns' && a.gate.name) nameIndex.set(String(a.gate.name).toLowerCase(), id);
      }
    }
    return nameIndex.get(String(name).toLowerCase()) || null;
  }

  function createAccount({ id, pubkey, gate }) {
    const slots = readJson(slotsFile, { next: 0 });
    const slot = slots.next++;
    writeJson(slotsFile, slots);
    const a = { id, pubkey, gate, slot, state: 'empty', createdAt: Date.now() };
    fs.mkdirSync(dataDir(id), { recursive: true });
    const port = cfg.portBase + slot * 2;
    writeConfig(configFile(id), {
      apiAddress: `127.0.0.1:${port}`,
      adminAddress: `127.0.0.1:${port + 1}`,
      adminPassword: randomPassword(),
      directory: dataDir(id),
      'sia.diskUsageLimit': cfg.bufferLimitBytes,
    });
    // Leftovers always go to Sia in hosted mode: an idle s3d gets stopped.
    writePrefs(configFile(id), { autostart: false, autoFlush: true, autoFlushMinutes: 10 });
    saveAccount(a);
    return a;
  }

  function setState(id, state) {
    const a = account(id);
    a.state = state;
    saveAccount(a);
    writePrefs(configFile(id), { autostart: state === 'active' });
    return a;
  }

  async function load(id) {
    // Never start a second s3d on a folder whose first one is still stopping.
    while (closing.has(id)) await closing.get(id);
    if (running.has(id)) return running.get(id);
    if (loading.has(id)) return loading.get(id);
    const p = (async () => {
      await makeRoom();
      const pithos = await createPithos({
        host: '127.0.0.1', port: 0, hostName: 'hosted',
        configFile: configFile(id), binary: cfg.s3dBinary,
      });
      const inst = { id, pithos, cli: makeCli(pithos.daemon), cookie: pithos.internalSession(), lastUsed: Date.now(), busy: 0, buckets: new Map() };
      running.set(id, inst);
      pithos.daemon.on('log', (l) => { if (l.stream !== 'out') log(`[${id.slice(0, 8)}] ${l.line}`); });
      return inst;
    })().finally(() => loading.delete(id));
    loading.set(id, p);
    return p;
  }

  // The instance with s3d up, for anything that talks S3.
  async function ensure(id) {
    const a = account(id);
    if (!a) throw httpError(404, 'no such account');
    if (a.state === 'empty') throw httpError(409, 'nothing is hosted yet: move your drives here from Pithos first');
    if (a.state === 'moving-out') throw httpError(409, 'this account is being moved to its owner\'s computer');
    const inst = await load(id);
    // The account may have started moving out while it loaded.
    if (account(id).state !== 'active') throw httpError(409, 'this account is being moved');
    inst.lastUsed = Date.now();
    const d = inst.pithos.daemon;
    if (d.state !== 'running') {
      if (d.state !== 'starting') d.start();
      const until = Date.now() + 30_000;
      while (d.state === 'starting' && Date.now() < until) await sleep(200);
      if (d.state !== 'running') throw httpError(503, 'your s3d did not start; try again in a minute');
      await indexKeys(inst);
    }
    return inst;
  }

  // Routes S3 requests: access key id -> account. A key id another account
  // (or a grant) already uses is never taken over; that account's traffic
  // would otherwise land in this one's s3d. Such keys are left unrouted and
  // reported in inst.conflicts.
  async function indexKeys(inst) {
    const keys = await inst.cli.listKeys().catch(() => []);
    let changed = false;
    for (const [k, v] of Object.entries(keyIndex)) if (v === inst.id && !keys.some((x) => x.accessKeyId === k)) { delete keyIndex[k]; changed = true; }
    const conflicts = [];
    for (const k of keys) {
      const owner = Object.hasOwn(keyIndex, k.accessKeyId) ? keyIndex[k.accessKeyId] : null;
      if (owner === inst.id) continue;
      if (owner || reservedKey(k.accessKeyId)) { conflicts.push(k.accessKeyId); continue; }
      keyIndex[k.accessKeyId] = inst.id;
      changed = true;
    }
    if (changed) writeJson(keysFile, keyIndex);
    if (conflicts.length) log(`[${inst.id.slice(0, 8)}] access keys already used elsewhere, not routed: ${conflicts.join(', ')}`);
    inst.keys = keys.filter((k) => !conflicts.includes(k.accessKeyId));
    inst.conflicts = conflicts;
    inst.buckets.clear();
    return inst.keys;
  }

  const keyTaken = (akid, id) => (Object.hasOwn(keyIndex, akid) && keyIndex[akid] !== id) || reservedKey(akid);

  // Every drive on the account, mapped to a key of the s3d user owning it.
  async function mapBuckets(inst) {
    inst.buckets.clear();
    for (const k of inst.keys || await indexKeys(inst)) {
      const list = await clientFor(inst, k).listBuckets().catch(() => []);
      for (const b of list) if (!inst.buckets.has(b.name)) inst.buckets.set(b.name, k);
    }
    return [...inst.buckets.keys()];
  }

  // An owner key that can act on `bucket`.
  async function bucketClient(inst, bucket) {
    if (!inst.buckets.has(bucket)) await mapBuckets(inst);
    const k = inst.buckets.get(bucket);
    return k ? clientFor(inst, k) : null;
  }

  function clientFor(inst, k) {
    return new S3Client({ endpoint: `http://${readConfig(configFile(inst.id)).apiAddress}`, accessKeyId: k.accessKeyId, secretKey: k.secretKey });
  }

  async function close(id, reason) {
    if (closing.has(id)) return closing.get(id);
    const inst = running.get(id);
    if (!inst) return;
    const p = (async () => {
      log(`[${id.slice(0, 8)}] stopping s3d (${reason})`);
      await inst.pithos.close().catch((e) => log(`[${id.slice(0, 8)}] close failed: ${e.message}`));
      running.delete(id);
    })().finally(() => closing.delete(id));
    closing.set(id, p);
    return p;
  }

  async function flushQuietly(inst, timeoutMs = 120_000) {
    const c = readConfig(configFile(inst.id));
    try {
      const st = await admin.uploadStats(c);
      if (!st.pendingObjects) return true;
      await Promise.race([admin.flush(c), sleep(timeoutMs).then(() => { throw new Error('timed out'); })]);
      return true;
    } catch (e) {
      log(`[${inst.id.slice(0, 8)}] flush before stop failed: ${e.message}`);
      return false;
    }
  }

  async function makeRoom() {
    // Loads in flight count too, or parallel first requests overshoot the cap.
    if (running.size + loading.size < cfg.maxRunning) return;
    const idle = [...running.values()].filter((i) => !i.busy).sort((a, b) => a.lastUsed - b.lastUsed);
    if (!idle.length) throw httpError(503, 'the server is busy; try again in a minute');
    await flushQuietly(idle[0]);
    await close(idle[0].id, 'making room');
  }

  async function sweep() {
    const cutoff = Date.now() - cfg.idleMinutes * 60_000;
    for (const inst of [...running.values()]) {
      if (inst.busy || inst.lastUsed > cutoff || closing.has(inst.id)) continue;
      await flushQuietly(inst);
      if (!inst.busy && inst.lastUsed <= cutoff) await close(inst.id, 'idle');
    }
  }
  const sweeper = setInterval(() => { sweep().catch((e) => log('sweep failed:', e.message)); }, 60_000);
  sweeper.unref?.();

  // Holds an instance "in use" for the duration of a streamed request, so the
  // sweeper does not stop s3d under a long download.
  function hold(inst) {
    inst.busy++;
    let done = false;
    return () => { if (!done) { done = true; inst.busy--; inst.lastUsed = Date.now(); } };
  }

  return {
    dir, dataDir, configFile, account, accountByName, createAccount, saveAccount, setState,
    load, ensure, close, flushQuietly, bucketClient, mapBuckets, indexKeys, hold, keyTaken,
    ownerOfKey: (akid) => (Object.hasOwn(keyIndex, akid) ? keyIndex[akid] : null),
    async closeAll() {
      clearInterval(sweeper);
      for (const id of [...running.keys()]) {
        await flushQuietly(running.get(id), 30_000);
        await close(id, 'shutdown');
      }
    },
    runningCount: () => running.size,
  };
}

export function httpError(status, message) { return Object.assign(new Error(message), { status }); }
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));