SkateSpots/scripts/community-bot.ts

/**
 * SkateSpots Community — always-on Cordn room bot.
 *
 * The room itself is founder-owned; the bot is a normal MLS member that joins
 * once and then approves new join requests. It never creates a group (that
 * would risk a split if its state were lost): if there is no local group it
 * publishes a key package and asks to join, then waits for an admin's Welcome.
 *
 * Secrets: COMMUNITY_BOT_KEY (hex) comes from a 600 env file. The MLS snapshot
 * on disk also contains group private material, so keep the state file 600 and
 * back it up.
 *
 * Run locally:
 *   COMMUNITY_BOT_KEY=... COMMUNITY_GID=... npx tsx scripts/community-bot.ts
 *
 * Env:
 *   COMMUNITY_BOT_KEY            32-byte hex (required)
 *   COMMUNITY_GID                e.g. 35006:<founder>:community (required)
 *   COMMUNITY_STATE              snapshot path (default /var/lib/skatespots-community/state.json)
 *   COMMUNITY_COORDINATOR        coordinator pubkey (default SkateSpots')
 *   COMMUNITY_RELAY              coordinator relay (default wss://relay.skate-spots.org)
 *   COMMUNITY_POLL_MS            poll interval (default 15000)
 *   COMMUNITY_AUTO_APPROVE=0     leave requests for admins instead of adding
 *   COMMUNITY_MAX_MEMBERS        stop auto-approving at this group size (default 250)
 *   COMMUNITY_MAX_DEVICES_PER_PUBKEY  per-account device cap (default 5)
 *   COMMUNITY_MAX_APPROVALS_PER_HOUR  flood brake for auto-approval (default 30)
 *   COMMUNITY_ONCE=1             one cycle then exit (for a dry run)
 */
import { mkdir, readFile, rename, writeFile } from 'node:fs/promises';
import { dirname } from 'node:path';
import WebSocket from 'ws';
import { CordnSession, type CordnSnapshot, type CordnStore } from '../src/lib/cordn/native/session';

// Node 20 (the relay box) has no global WebSocket. The ContextVM transport
// looks for one at connect time, so install `ws` before opening a session.
(globalThis as unknown as { WebSocket?: unknown }).WebSocket = WebSocket;

const COORDINATOR_PUBKEY =
  process.env.COMMUNITY_COORDINATOR ??
  '02816a05b1d4e70a7df9c712ef6f7272774269d65bd8a2f262ce0330d0885693';
const RELAY = process.env.COMMUNITY_RELAY ?? 'wss://relay.skate-spots.org';
const GID = process.env.COMMUNITY_GID ?? '';
const BOT_KEY = (process.env.COMMUNITY_BOT_KEY ?? '').trim().toLowerCase();
const STATE_PATH =
  process.env.COMMUNITY_STATE ?? '/var/lib/skatespots-community/state.json';
const POLL_MS = Number(process.env.COMMUNITY_POLL_MS ?? 15_000);
const AUTO_APPROVE = process.env.COMMUNITY_AUTO_APPROVE !== '0';
const parsedMaxMembers = Number(process.env.COMMUNITY_MAX_MEMBERS ?? 250);
const MAX_MEMBERS = Number.isFinite(parsedMaxMembers) && parsedMaxMembers >= 2
  ? Math.floor(parsedMaxMembers)
  : 250;
const parsedMaxDevices = Number(process.env.COMMUNITY_MAX_DEVICES_PER_PUBKEY ?? 5);
const MAX_DEVICES_PER_PUBKEY = Number.isFinite(parsedMaxDevices) && parsedMaxDevices >= 1
  ? Math.floor(parsedMaxDevices)
  : 5;
const parsedMaxPerHour = Number(process.env.COMMUNITY_MAX_APPROVALS_PER_HOUR ?? 30);
const MAX_APPROVALS_PER_HOUR = Number.isFinite(parsedMaxPerHour) && parsedMaxPerHour >= 1
  ? Math.floor(parsedMaxPerHour)
  : 30;
const CREATE = process.env.COMMUNITY_CREATE === '1';
const ONCE = process.env.COMMUNITY_ONCE === '1';

if (!/^[0-9a-f]{64}$/.test(BOT_KEY)) {
  console.error('[community] COMMUNITY_BOT_KEY must be 64 lowercase hex chars');
  process.exit(1);
}
if (!GID) {
  console.error('[community] COMMUNITY_GID is required');
  process.exit(1);
}

const started = Date.now();
const log = (...parts: unknown[]) => {
  const uptime = Math.round((Date.now() - started) / 1000);
  console.log(`[community +${uptime}s]`, ...parts);
};
const short = (pk: string) => `${pk.slice(0, 8)}…`;
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

/** JSON snapshot store. Atomic writes, 600, and no duplicated bot private key. */
class FileCordnStore implements CordnStore {
  constructor(private readonly path: string) {}

  async load(): Promise<CordnSnapshot | null> {
    try {
      return JSON.parse(await readFile(this.path, 'utf8')) as CordnSnapshot;
    } catch {
      return null;
    }
  }

  async save(snapshot: CordnSnapshot): Promise<void> {
    await mkdir(dirname(this.path), { recursive: true, mode: 0o700 });
    // The bot key lives in the env file; don't duplicate it in the state.
    const { privateKeyHex, ...safe } = snapshot;
    void privateKeyHex;
    const tmp = `${this.path}.${process.pid}.tmp`;
    await writeFile(tmp, JSON.stringify(safe), { mode: 0o600 });
    await rename(tmp, this.path);
  }

  async clear(): Promise<void> {
    await writeFile(this.path, 'null', { mode: 0o600 });
  }
}

let session!: CordnSession;

const hasRoom = () => session.listGroups().some((group) => group.gid === GID);

async function joinRoom(): Promise<boolean> {
  if (hasRoom()) return true;

  // A Welcome already waiting? Accept it before asking again, so a restart
  // between approval and pickup does not create a duplicate request.
  const welcomes = await session.fetchWelcomes();
  for (const welcome of welcomes) {
    try {
      const joined = await session.acceptWelcome(welcome, 'SkateSpots Community');
      if (joined.gid === GID) {
        log('joined the room as a member');
        return true;
      }
    } catch {
      // Welcome for a key package this bot no longer has — skip.
    }
  }

  // Ask the founder/admin to add us. `requestJoin` publishes a key package if
  // one is not already waiting.
  await session.requestJoin(GID);
  lastJoinRequestAt = Date.now();
  log('join request sent; waiting for an admin to approve this bot');
  return false;
}

// Keyed by the requester's key-package ref, not just their pubkey: one account
// can legitimately join from a second device, and a consumed key package must
// not be retried forever. Entries expire so a re-request after a long gap works.
const APPROVAL_TTL_MS = 7 * 24 * 60 * 60 * 1000;
const approved = new Map<string, number>();
const lastAttempt = new Map<string, number>();
let lastJoinRequestAt = 0;
let lastError = '';
let lastPending = 0;
let capLogged = false;
let hourCapLogged = false;
const approvalTimes: number[] = [];
const deviceCapLogged = new Set<string>();

async function cycle(): Promise<void> {
  try {
    if (!hasRoom()) {
      if (CREATE) {
        await session.createGroup({ groupId: GID, name: 'SkateSpots Community' });
        log('created the room (bootstrap); set COMMUNITY_CREATE=0 and restart');
      }
      const joined = await joinRoom();
      if (!joined) {
        // Retry the request every 5 minutes; a fresh key package may be needed
        // if the original one was consumed or expired.
        if (Date.now() - lastJoinRequestAt > 5 * 60_000) {
          await session.requestJoin(GID);
          lastJoinRequestAt = Date.now();
          log('join request refreshed');
        }
        return;
      }
    }

    // Keep the bot's MLS state current with commits from other members.
    await session.fetch(GID);

    if (!AUTO_APPROVE) return;

    const room = session.listGroups().find((group) => group.gid === GID);
    const memberCount = room?.members.length ?? 0;
    const requests = (await session.fetchJoinRequests(GID))
      .filter((request) => request.pk !== session.pubkey); // our own boot request
    if (requests.length !== lastPending && requests.length > 0) {
      log('pending join requests', requests.length);
    }
    lastPending = requests.length;

    // A public room can be flooded by sybil key packages; MLS also gets more
    // expensive for every member. Stop auto-adding at a sane cap and leave the
    // queue for a human admin to review (raise the cap via env if needed).
    if (memberCount >= MAX_MEMBERS) {
      if (!capLogged) {
        capLogged = true;
        log('member cap reached', memberCount, '/', MAX_MEMBERS, '— leaving requests for admins');
      }
      return;
    }
    if (capLogged) {
      capLogged = false;
      log('member cap cleared; auto-approval resumed');
    }

    const hourAgo = Date.now() - 60 * 60_000;
    while (approvalTimes.length > 0 && (approvalTimes[0] ?? 0) < hourAgo) approvalTimes.shift();
    if (approvalTimes.length >= MAX_APPROVALS_PER_HOUR) {
      if (!hourCapLogged) {
        hourCapLogged = true;
        log('hourly approval cap reached', approvalTimes.length, '/', MAX_APPROVALS_PER_HOUR, '— leaving requests for admins');
      }
      return;
    }
    if (hourCapLogged) {
      hourCapLogged = false;
      log('hourly approval cap cleared; auto-approval resumed');
    }

    const handled: Array<{ pk: string; at: number }> = [];
    for (const request of requests.slice(0, 10)) {
      if (!/^[0-9a-f]{64}$/i.test(request.pk)) {
        log('skipping malformed join request');
        continue;
      }
      const currentRoom = session.listGroups().find((group) => group.gid === GID);
      if ((currentRoom?.members.length ?? 0) >= MAX_MEMBERS) {
        if (!capLogged) {
          capLogged = true;
          log('member cap reached', currentRoom?.members.length ?? 0, '/', MAX_MEMBERS, '— stopping this cycle');
        }
        break;
      }
      const devices = currentRoom?.members.filter((member) => member.pubkey === request.pk).length ?? 0;
      if (devices >= MAX_DEVICES_PER_PUBKEY) {
        if (!deviceCapLogged.has(request.pk)) {
          deviceCapLogged.add(request.pk);
          log('device cap reached for', short(request.pk), devices, '/', MAX_DEVICES_PER_PUBKEY);
        }
        continue;
      }
      const approvalKey = request.kp_ref || request.pk;
      const approvedAt = approved.get(approvalKey) ?? 0;
      if (Date.now() - approvedAt < APPROVAL_TTL_MS) continue;
      const last = lastAttempt.get(approvalKey) ?? 0;
      if (Date.now() - last < 5 * 60_000) continue;
      try {
        // Prefer the exact key package the requester published. Falling back to
        // the pubkey keeps older coordinator responses working.
        try {
          await session.addMemberByIdentifier(GID, approvalKey);
        } catch (error) {
          if (request.kp_ref && request.kp_ref !== request.pk) {
            await session.addMemberByIdentifier(GID, request.pk);
          } else {
            throw error;
          }
        }
        approved.set(approvalKey, Date.now());
        approvalTimes.push(Date.now());
        handled.push({ pk: request.pk, at: request.at });
        log('approved member', short(request.pk));
        await sleep(1500); // be gentle with the coordinator
      } catch (error) {
        const message = error instanceof Error ? error.message : String(error);
        // A stale request for an account already in the group cannot be added
        // again. Retire it instead of retrying forever (and let the user use
        // the Link-device flow for a second device).
        if (/already in the group/i.test(message)) {
          approved.set(approvalKey, Date.now());
          handled.push({ pk: request.pk, at: request.at });
          log('request already satisfied', short(request.pk));
          continue;
        }
        lastAttempt.set(approvalKey, Date.now());
        log('could not approve', short(request.pk), message);
        await sleep(1500);
      }
    }

    if (handled.length > 0) {
      try {
        await session.fetchJoinRequests(GID, handled);
        log('retired handled requests', handled.length);
      } catch (error) {
        log('could not retire handled requests', error instanceof Error ? error.message : String(error));
      }
    }

    // Keep the in-memory approval memory bounded.
    const cutoff = Date.now() - APPROVAL_TTL_MS;
    for (const [key, at] of approved) {
      if (at < cutoff) approved.delete(key);
    }
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error);
    if (message !== lastError) {
      lastError = message;
      log('cycle error', message);
    }
  }
}

let stopping = false;
async function stop(signal: string): Promise<void> {
  if (stopping) return;
  stopping = true;
  log(`stopping (${signal})`);
  try {
    if (session) await session.disconnect();
  } finally {
    process.exit(0);
  }
}
process.on('SIGINT', () => void stop('SIGINT'));
process.on('SIGTERM', () => void stop('SIGTERM'));

async function main(): Promise<void> {
  session = await CordnSession.open({
    coordinator: { serverPubkey: COORDINATOR_PUBKEY, relays: [RELAY] },
    privateKeyHex: BOT_KEY,
    store: new FileCordnStore(STATE_PATH),
  });
  log(
    'bot pubkey', session.pubkey, 'room', GID,
    'auto-approve', AUTO_APPROVE ? 'on' : 'off',
    'max', MAX_MEMBERS,
    'devices/account', MAX_DEVICES_PER_PUBKEY,
    'approvals/hour', MAX_APPROVALS_PER_HOUR,
  );

  do {
    await cycle();
    if (ONCE) break;
    await sleep(POLL_MS);
  } while (!stopping);

  // `ONCE` is used for setup/dry runs and tests; close the ContextVM client so
  // the process can exit instead of hanging on the WebSocket.
  if (ONCE) await session.disconnect();
}

void main().catch((error) => {
  console.error('[community] fatal', error instanceof Error ? error.message : String(error));
  process.exit(1);
});