openvibers/openvibe.chat Commits

Party desktops on OpenVibe's own computer pool (#479)

OP OpenVibers committed 3 days ago · ec293c66cf · 13 files, +1105 −23

* Party desktops on OpenVibe's own computer pool

* Pool docs: current machines page; schema test names the new tables

* Renumber the party pool migration after main

* Merge origin/main into feat/party-pool-1109

* Renumber the party pool migration after the Global insight

* Merge origin/main into feat/party-pool-1109

Changes

ddiff --git a/.env.example b/.env.example
iindex 0eb2e33..e438146 100644
---- a/.env.example
++++ b/.env.example
@@@ -49,6 +49,15 @@ OV_OAUTH_CLIENT_SECRET=written-by-service-principal-js
#CHAT_PARTY_OAUTH_REDIRECT_URI=https://openvibe.party/auth/callback
# The same random value must be set as CONNECT_PARTY_SHARED_SECRET in Connect.
#CHAT_PARTY_CONNECT_SECRET=
++# OpenVibe's own computers for party desktops (server/rooms/party-pool.js). The switch is the owner's, in the
++# party_pool_settings row (off until the owner turns it on at openvibe.party/pool). CHAT_PARTY_POOL=1 is an
++# operator override for a development box. The windows are when a hold nobody renewed goes idle, and how long
++# a place in line lasts.
++#CHAT_PARTY_POOL=0
++#CHAT_PARTY_POOL_IDLE_MS=600000
++#CHAT_PARTY_POOL_QUEUE_MS=1800000
++#CHAT_PARTY_POOL_MAX_PER_ACTOR=2
++#CHAT_PARTY_POOL_MAX_QUEUE=50
# OpenVibe.Live — every Live-owned read and side effect (server/live-context.js)
OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
ddiff --git a/docs/party-pool.md b/docs/party-pool.md
nnew file mode 100644
iindex 0000000..e778de0
---- /dev/null
++++ b/docs/party-pool.md
@@@ -0,0 +1,113 @@
++# OpenVibe's own computers for parties
++
++`openvibe.party` can run a party's shared computer on the person's own Link machine, or on OpenVibe's own Link
++machines (message 11, 2026-11-08). This document covers the second path: the pool, its owner switch, placement, the
++holds it keeps, and the machine-facing assignment contract.
++
++A pool machine is one of the owner's Link machines. Nothing here rents capacity or spends money: there is no quote,
++price, budget or provider call in `server/rooms/party-pool.js`. It chooses only among machines the owner registered.
++
++## The owner turns it on and registers machines
++
++The switch is the one row in `party_pool_settings`, written from `openvibe.party/pool` by a session with the owner
++rank. It starts **off**. `CHAT_PARTY_POOL=1` is an operator override for a development box; the panel's value wins
++otherwise.
++
++The owner adds a machine with its Link principal (`lnk_…`, shown on the machines page at openvibe.link/machines), a name, what it can
++run (`desktop`, and optionally `browser`), and how many desktops it may host at once (`1–50`). The row is the
++machine's advertisement:
++
++```
++principal_id the node principal in the machine's node token
++name what the party page calls it
++can_run ["desktop"] (a party needs desktop)
++desktops capacity: how many parties it may hold at once
++last_seen_at when its Link last asked for an assignment
++```
++
++Removing a machine releases its holds and stops those rooms' desktops.
++
++## A party asks the pool
++
++`POST /api/chat/parties` with `desktop_hosting: "openvibe"` asks the pool before the room is made:
++
++1. The switch is off → 503 `party.pool_off`.
++2. No machine is registered → 503 `party.pool_empty`.
++3. A machine has a free desktop → a `held` hold, the room is created, and the room view's `desktop.where` names it.
++4. Every machine is full → a `queued` hold with `desktop.where.queue_position` (1 is next) and a plain line. A
++ released machine seats the oldest queued room on the next sweep.
++
++Placement is serialized on the settings row. It picks the machine with the most free desktops, then the least loaded
++(used / capacity), never a disabled one. Load is the number of `held`, `hosting` and `ready` holds on the machine.
++
++The person has at most `CHAT_PARTY_POOL_MAX_PER_ACTOR` (default 2) waiting or running OpenVibe parties at once; the
++start route already carries the per-actor rate limit (`chat.party.create`), and the pool's management routes carry
++`chat.party.pool`.
++
++## The hold and the room
++
++`party_pool_holds` is one room's reservation:
++
++| state | meaning |
++| --- | --- |
++| `queued` | no free desktop; the room waits in line |
++| `held` | a machine is reserved; its Link has not claimed the party yet |
++| `hosting` | the machine's Link claimed it (its claim moves the hold) |
++| `ready` | the machine's Connect share is up |
++| `released` | the room ended, a machine was removed, the hold went idle, or the place in line expired |
++
++The partial unique index keeps one open hold per room. Released rows stay as the record.
++
++`GET /api/chat/parties/:slug` adds `party.desktop.where` for every desktop party:
++
++```json
++{ "state": "ready", "machine": "Dell tower 1", "queue_position": null,
++ "line": "Running on Dell tower 1, an OpenVibe computer." }
++```
++
++A Link-hosted party gets `{ "state": "link", …, "line": "Runs on the host's computer through OpenVibe Link." }`.
++
++## Releasing
++
++- The host ends the party → the hold is released and the room's desktop is stopped.
++- The machine stops or its Link says `stopped` → released.
++- A `held`/`hosting`/`ready` hold with no renewal for `CHAT_PARTY_POOL_IDLE_MS` (default 10 minutes) is released as
++ idle and the room says so. Link renews by asking for the party's `access` every three seconds while it hosts.
++- A `queued` hold older than `CHAT_PARTY_POOL_QUEUE_MS` (default 30 minutes) is released as `queue_expired`.
++
++A minute sweep on the server (`start()` in the module) does the idle and queue expiry and seats the line; tests call
++`sweep()` directly.
++
++## The machine-facing assignment
++
++A pool machine's Link presents its own node token (`sub: link:lnk_…`) and asks what to host:
++
++```
++GET /api/chat/parties/desktop/assignment
++→ { "assignment": { "party": "party-0123abcd", "hold": "ph_…", "title": "Friday film" } }
++→ { "assignment": null }
++```
++
++The poll also updates the machine's `last_seen_at`. The answer names the oldest `held` party for that principal whose
++room is still waiting; once its claim moves the hold to `hosting`, the next poll says null.
++
++On the machine, `party_pool: { "enabled": true }` in the Link's `config.json` turns the poll on (off by default). Link
++asks every five seconds and starts the same host path as `openvibe-link party host <party>`: it claims the party,
++creates the Connect share, reports ready and polls `access` while it hosts. The owner still registers the machine's
++principal on the panel; the machine itself is not trusted to join the pool.
++
++## Routes
++
++| Route | Who | Meaning |
++| --- | --- | --- |
++| `GET /api/chat/parties/pool` | owner | the switch, machines with load/free, open holds |
++| `PUT /api/chat/parties/pool/enabled` | owner | `{ enabled }` → the panel |
++| `POST /api/chat/parties/pool/machines` | owner | `{ principal_id, name, desktops, can_run? }` → 201 |
++| `PATCH /api/chat/parties/pool/machines/:principal` | owner | change name, desktops, can_run or enabled |
++| `DELETE /api/chat/parties/pool/machines/:principal` | owner | remove it and release its holds |
++| `GET /api/chat/parties/desktop/assignment` | the machine | what to host next |
++| `GET /pool` | owner page | the plain panel (design pass owns its look) |
++
++Tests: `test/party-pool.test.js` drives the placement engine with a fake registry (capacity, load, queue, promotion,
++idle release), then the HTTP flow around it (switch, owner-only registration, party start, claim, ready, end, queue,
++per-person cap and the panel page).
ddiff --git a/migrations/0064_party_pool.sql b/migrations/0064_party_pool.sql
nnew file mode 100644
iindex 0000000..bda07fe
---- /dev/null
++++ b/migrations/0064_party_pool.sql
@@@ -0,0 +1,53 @@
++-- phase: expand
++-- OpenVibe's own computers for openvibe.party (message 11: a party's desktop runs on a Link machine or on
++-- OpenVibe's compute). A party with no machine of its own asks this pool. The owner (rank owner) registers a
++-- Link machine and what it can run; placement holds it for one room and releases it when the room ends or the
++-- hold goes idle. Pool machines are the owner's own: nothing here rents capacity or spends money.
++--
++-- party_pool_machines what one Link machine offers: its node principal (lnk_…, the identity in its node
++-- token), a display name, what it can run, and how many desktops it may host at once
++-- party_pool_holds one room's reservation: queued (no free desktop), held (a machine is reserved),
++-- hosting (Link claimed it), ready (its Connect share is up) or released. The partial
++-- unique index keeps one open hold per room; released rows stay as the record.
++-- party_pool_settings the owner's switch, one row, off by default. CHAT_PARTY_POOL=1 is the operator override.
++--
++-- Expand only: new tables, and an older release ignores all of it.
++CREATE TABLE IF NOT EXISTS party_pool_settings (
++ id smallint PRIMARY KEY CHECK (id = 1),
++ enabled boolean NOT NULL DEFAULT false,
++ updated_at timestamptz NOT NULL DEFAULT now(),
++ updated_by text COLLATE "C"
++);
++INSERT INTO party_pool_settings (id, enabled) VALUES (1, false) ON CONFLICT (id) DO NOTHING;
++
++CREATE TABLE IF NOT EXISTS party_pool_machines (
++ principal_id text COLLATE "C" PRIMARY KEY,
++ owner_subject text COLLATE "C" NOT NULL,
++ name text COLLATE "C" NOT NULL,
++ can_run jsonb NOT NULL DEFAULT '["desktop"]'::jsonb,
++ desktops integer NOT NULL DEFAULT 1 CHECK (desktops BETWEEN 1 AND 50),
++ enabled boolean NOT NULL DEFAULT true,
++ last_seen_at timestamptz,
++ created_at timestamptz NOT NULL DEFAULT now(),
++ updated_at timestamptz NOT NULL DEFAULT now(),
++ CHECK (principal_id ~ '^lnk_[0-9A-HJKMNP-TV-Z]{26}$')
++);
++
++CREATE TABLE IF NOT EXISTS party_pool_holds (
++ id text COLLATE "C" PRIMARY KEY,
++ slug text COLLATE "C" NOT NULL,
++ room_id bigint REFERENCES rooms(id) ON DELETE CASCADE,
++ actor_subject text COLLATE "C" NOT NULL,
++ principal_id text COLLATE "C",
++ state text COLLATE "C" NOT NULL DEFAULT 'queued',
++ held_at timestamptz,
++ renewed_at timestamptz NOT NULL DEFAULT now(),
++ released_at timestamptz,
++ release_reason text COLLATE "C",
++ created_at timestamptz NOT NULL DEFAULT now(),
++ CHECK (state IN ('queued', 'held', 'hosting', 'ready', 'released'))
++);
++CREATE UNIQUE INDEX IF NOT EXISTS party_pool_holds_open_room ON party_pool_holds (room_id) WHERE state <> 'released' AND room_id IS NOT NULL;
++CREATE INDEX IF NOT EXISTS party_pool_holds_open ON party_pool_holds (state, created_at) WHERE state <> 'released';
++CREATE INDEX IF NOT EXISTS party_pool_holds_machine ON party_pool_holds (principal_id, state) WHERE state <> 'released';
++CREATE INDEX IF NOT EXISTS party_pool_holds_actor ON party_pool_holds (actor_subject, state) WHERE state <> 'released';
ddiff --git a/server/config.js b/server/config.js
iindex 068ec4e..024bccc 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -97,6 +97,16 @@ module.exports = {
baseUrl: strip(process.env.CHAT_PARTY_URL || (isProduction ? 'https://openvibe.party' : `http://localhost:${port}`)),
redirectUri: process.env.CHAT_PARTY_OAUTH_REDIRECT_URI || `${strip(process.env.CHAT_PARTY_URL || (isProduction ? 'https://openvibe.party' : `http://localhost:${port}`))}/auth/callback`,
scope: 'profile theme',
++ // OpenVibe's own computers for party desktops (server/rooms/party-pool.js). The switch is the owner's, in
++ // party_pool_settings (off until the owner turns it on at openvibe.party/pool); CHAT_PARTY_POOL=1 is the
++ // operator override for a development box. The windows are the idle release and the line's lifetime.
++ pool: {
++ enabled: bool(process.env.CHAT_PARTY_POOL, false),
++ idleMs: int(process.env.CHAT_PARTY_POOL_IDLE_MS, 10 * 60_000),
++ queueMs: int(process.env.CHAT_PARTY_POOL_QUEUE_MS, 30 * 60_000),
++ maxPerActor: int(process.env.CHAT_PARTY_POOL_MAX_PER_ACTOR, 2),
++ maxQueue: int(process.env.CHAT_PARTY_POOL_MAX_QUEUE, 50),
++ },
},
// Pictures in rooms and channels (server/rooms/pictures.js): uploaded to OpenVibe.Pics as the person (their own
// Network token) through MediaHub on this host, and shown inline. Off without CHAT_PICS_INTERNAL_URL.
ddiff --git a/server/index.js b/server/index.js
iindex 090101e..cb4ac25 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -90,6 +90,9 @@ async function start() {
chatAi.start();
// Workspace rooms' queues (server/rooms/workspace.js): the OpenVibe.Actor tasks they started, followed every 10 s.
require('./rooms/workspace').start();
++ // OpenVibe's own computers for parties (server/rooms/party-pool.js): release holds that went idle and seat the
++ // line once a minute. The owner switch is read at each placement, so this is safe while it is off.
++ require('./rooms/party-pool').start({ intervalMs: 60_000 });
// Zero-downtime restarts (as OpenVibe.Live): when systemd owns the listener (deploy/systemd/openvibe-chat.socket) it
// hands it over as file descriptor 3, and connections that arrive during a restart wait in the kernel's accept queue
@@@ -127,6 +130,7 @@ async function start() {
() => events.stop(),
() => chatAi.stop(),
() => require('./rooms/workspace').stop(),
++ () => require('./rooms/party-pool').stop(),
() => subs.stop(),
() => { if (valkey) return valkey.close(); },
() => app.locals.chatApps.stop(),
ddiff --git a/server/rooms/parties.js b/server/rooms/parties.js
iindex 8ba9653..89a0bc9 100644
---- a/server/rooms/parties.js
++++ b/server/rooms/parties.js
@@@ -21,6 +21,7 @@ const db = require('../db/database');
const rooms = require('./rooms');
const syncClock = require('./sync-clock');
const controlLease = require('./control-lease');
++const pool = require('./party-pool');
const AGENT_STREAM_RE = /^ags_[0-9A-HJKMNP-TV-Z]{26}$/;
const VOD_RE = /^\d{1,16}$/;
@@@ -98,7 +99,13 @@ function joinUrl(row, kind, rawSecret = null) {
return rawSecret ? `${base}?k=${encodeURIComponent(rawSecret)}` : base;
}
--function view(row, party, { rawSecret = null, now = Date.now() } = {}) {
++/** The plain line a Link-hosted party's page shows for where it runs. */
++const LINK_WHERE = Object.freeze({
++ state: 'link', machine: null, queue_position: null,
++ line: "Runs on the host's computer through OpenVibe Link.",
++});
++
++function view(row, party, { rawSecret = null, now = Date.now(), where } = {}) {
const kind = kindOfParty(party);
return {
id: row.slug,
@@@ -110,6 +117,9 @@ function view(row, party, { rawSecret = null, now = Date.now() } = {}) {
...(kind === 'desktop' ? { desktop: {
hosting: party.desktop_hosting, state: party.desktop_state, layout: party.desktop_layout || 'watch',
url: party.desktop_state === 'ready' ? party.desktop_share_url : null,
++ // Where this party's computer runs: the pool's hold (named machine, or its place in line), or the
++ // host's own Link machine. The page draws one plain line from it; the data is here for the API/SDK.
++ where: where || (party.desktop_hosting === 'link' ? LINK_WHERE : null),
} } : {}),
host: party.host_subject,
title: row.name,
@@@ -134,7 +144,7 @@ async function detail(row, party, { user = null, key = null, now = Date.now() }
if (!member && !keyMatches(party, key)) return null;
const host = await db.get('SELECT username, display_name FROM ctx_users WHERE subject_id = ? LIMIT 1', [party.host_subject]);
const out = {
-- ...view(row, party, { now }),
++ ...view(row, party, { now, ...(kindOfParty(party) === 'desktop' ? { where: await pool.where(party.room_id) } : {}) }),
host_name: host ? (host.display_name || host.username) : null,
member,
};
@@@ -167,17 +177,31 @@ async function createDesktop(user, { title = 'Shared computer', desktop_hosting:
if (!user) fail(401, 'party.sign_in', 'Sign in to start a shared computer');
if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(user.subject_id || ''))) fail(409, 'party.no_subject', 'This account has no OpenVibe subject yet; sign in again');
if (hosting !== 'link' && hosting !== 'openvibe') fail(422, 'party.hosting', 'Choose your computer or an OpenVibe computer');
-- if (hosting === 'openvibe') fail(503, 'party.no_openvibe_machine', 'No OpenVibe machine is free right now');
const name = titleOf(title);
const raw = secret();
const slug = `party-${crypto.randomBytes(4).toString('hex')}`;
-- const room = await rooms.create(user, { name: name.slice(0, 40), slug, kind: 'party', visibility: 'private', topic: name, party: true });
-- await db.run(`INSERT INTO room_parties (room_id, agent_stream_id, vod_id, host_user_id, host_subject, mode, playing, listed, join_secret_hash, party_kind, desktop_hosting, desktop_state)
-- VALUES (?, NULL, NULL, ?, ?, 'live', false, false, ?, 'desktop', ?, 'waiting')`,
-- [room.id, user.id, user.subject_id, sha256(raw), hosting]);
-- await controlLease.take({ id: room.id, slug }, user.subject_id, { watchers: [] });
++ // An OpenVibe party asks the pool before the room exists: a machine is held for this slug, or the party is
++ // given a place in line. A failure here (the switch is off, no machine is registered, the person's own limit)
++ // starts no room at all. A Link party never touches the pool.
++ const placed = hosting === 'openvibe' ? await pool.place({ actor: user.subject_id, slug }) : null;
++ let room;
++ try {
++ room = await rooms.create(user, { name: name.slice(0, 40), slug, kind: 'party', visibility: 'private', topic: name, party: true });
++ await db.run(`INSERT INTO room_parties (room_id, agent_stream_id, vod_id, host_user_id, host_subject, mode, playing, listed, join_secret_hash, party_kind, desktop_hosting, desktop_state)
++ VALUES (?, NULL, NULL, ?, ?, 'live', false, false, ?, 'desktop', ?, 'waiting')`,
++ [room.id, user.id, user.subject_id, sha256(raw), hosting]);
++ if (placed) await pool.bind(placed.hold.id, room.id);
++ await controlLease.take({ id: room.id, slug }, user.subject_id, { watchers: [] });
++ } catch (err) {
++ if (placed) await pool.cancel(placed.hold.id, 'room_failed').catch(() => {});
++ throw err;
++ }
const row = await bySlug(slug);
-- return { party: view(row, partyOf(row), { rawSecret: raw }), roomId: room.id, join_secret: raw };
++ return {
++ party: view(row, partyOf(row), { rawSecret: raw, ...(placed ? { where: await pool.where(room.id) } : {}) }),
++ roomId: room.id,
++ join_secret: raw,
++ };
}
async function createAgentStream(user, { agent_stream_id: agentStreamId, vod_id: vodId = null, title, listed = false } = {}) {
@@@ -318,8 +342,9 @@ async function count(agentStreamId) {
async function end(row, party, user) {
if (!user || Number(user.id) !== Number(party.host_user_id)) fail(403, 'party.not_host', 'Only the host ends the party');
await db.run("UPDATE room_parties SET ended_at = now(), playing = false, desktop_state = CASE WHEN party_kind = 'desktop' THEN 'stopped' ELSE desktop_state END WHERE room_id = ? AND ended_at IS NULL", [party.room_id]);
++ if (party.party_kind === 'desktop') await pool.releaseRoom(party.room_id, 'ended');
const fresh = await bySlug(row.slug);
-- return view(fresh, partyOf(fresh));
++ return view(fresh, partyOf(fresh), { where: await pool.where(party.room_id) });
}
async function setDesktopLayout(row, party, user, layout) {
@@@ -328,7 +353,7 @@ async function setDesktopLayout(row, party, user, layout) {
if (layout !== 'watch' && layout !== 'listen') fail(422, 'party.layout', 'Choose Watch or Listen');
await db.run('UPDATE room_parties SET desktop_layout = ? WHERE room_id = ? AND ended_at IS NULL', [layout, party.room_id]);
const fresh = await bySlug(row.slug);
-- return view(fresh, partyOf(fresh));
++ return view(fresh, partyOf(fresh), { where: await pool.where(party.room_id) });
}
module.exports = {
ddiff --git a/server/rooms/party-desktop.js b/server/rooms/party-desktop.js
iindex 6df7d30..7122403 100644
---- a/server/rooms/party-desktop.js
++++ b/server/rooms/party-desktop.js
@@@ -11,6 +11,7 @@ const { iceServers } = require('../net/turn');
const controlLease = require('./control-lease');
const rooms = require('./rooms');
const parties = require('./parties');
++const pool = require('./party-pool');
const NODE = /^link:(lnk_[0-9A-HJKMNP-TV-Z]{26})$/;
const SHARE = /^shr_[0-9A-HJKMNP-TV-Z]{26}$/;
@@@ -38,7 +39,17 @@ async function authorized(req) {
const node = await nodeOf(req);
const row = await parties.bySlug(req.params.slug);
const party = parties.partyOf(row);
-- if (!node || !row || !party || party.party_kind !== 'desktop' || party.host_subject !== node.owner) return null;
++ if (!node || !row || !party || party.party_kind !== 'desktop') return null;
++ if (party.desktop_hosting === 'link') {
++ // The host's own machine, acting for the host.
++ if (party.host_subject !== node.owner) return null;
++ } else if (party.desktop_hosting === 'openvibe') {
++ // An OpenVibe party belongs to the machine the pool held for it, not to the person who started it.
++ const hold = await pool.holdForRoom(party.room_id);
++ if (!hold || hold.principal_id !== node.id || hold.state === 'queued' || hold.state === 'released') return null;
++ } else {
++ return null;
++ }
if (party.desktop_node_id && party.desktop_node_id !== node.id) return null;
return { node, row, party };
}
@@@ -46,7 +57,6 @@ async function authorized(req) {
async function claim({ node, row, party }, port) {
if (party.ended_at || party.desktop_state === 'stopped') throw new parties.PartyError(410, 'party.ended', 'This party has ended');
if (!Number.isInteger(port) || port < 1 || port > 65535) throw new parties.PartyError(422, 'party.port', 'Name a local port');
-- if (party.desktop_hosting === 'openvibe') throw new parties.PartyError(503, 'party.no_openvibe_machine', 'No OpenVibe machine is free right now');
// neko reads ICE credentials at container start. Match the Connect share's
// 24-hour expiry; ordinary call rooms retain their one-hour credentials.
const turn = iceServers(`p${row.id}`, undefined, Date.now(), 24 * 3600);
@@@ -60,6 +70,8 @@ async function claim({ node, row, party }, port) {
await db.run('UPDATE room_parties SET desktop_node_id = ? WHERE room_id = ? AND desktop_node_id IS NULL AND ended_at IS NULL', [node.id, party.room_id]);
const fresh = parties.partyOf(await parties.bySlug(row.slug));
if (fresh.desktop_node_id !== node.id) throw new parties.PartyError(409, 'party.host_taken', 'Another machine is hosting this party');
++ // The pool's hold moves held → hosting on the machine the pool chose (a Link party has no hold).
++ if (party.desktop_hosting === 'openvibe') await pool.claiming(party.room_id, node.id);
return {
ice_servers: ice,
share_grant: grant({ type: 'create', party: row.slug, owner: party.host_subject, username: host.username, node: node.id, port, exp: Math.floor(Date.now() / 1000) + 90 }),
@@@ -67,6 +79,7 @@ async function claim({ node, row, party }, port) {
}
async function access({ node, row, party }, subject) {
++ if (party.desktop_hosting === 'openvibe') await pool.renew(party.room_id, node.id);
const members = (await rooms.members(row, { subjects: true }))
.filter((m) => m.subject_id && m.role !== 'blocked')
.map((m) => ({ subject: m.subject_id, username: m.username }));
@@@ -89,8 +102,9 @@ async function ready({ node, row, party }, { share_id: shareId, url }) {
throw new parties.PartyError(422, 'party.share_url', 'Invalid Connect share address');
}
await db.run("UPDATE room_parties SET desktop_state = 'ready', desktop_share_id = ?, desktop_share_url = ? WHERE room_id = ? AND desktop_node_id = ? AND ended_at IS NULL", [shareId, u.toString().replace(/\/$/, ''), party.room_id, node.id]);
++ if (party.desktop_hosting === 'openvibe') await pool.ready(party.room_id, node.id);
const fresh = await parties.bySlug(row.slug);
-- return parties.view(fresh, parties.partyOf(fresh));
++ return parties.view(fresh, parties.partyOf(fresh), { where: await pool.where(party.room_id) });
}
async function ticket(row, party, user) {
@@@ -106,7 +120,22 @@ async function ticket(row, party, user) {
}
async function stopped({ node, party }) {
++ if (party.desktop_hosting === 'openvibe') await pool.releaseRoom(party.room_id, 'stopped');
await db.run("UPDATE room_parties SET desktop_state = 'stopped', desktop_share_url = NULL WHERE room_id = ? AND desktop_node_id = ?", [party.room_id, node.id]);
}
--module.exports = { nodeOf, authorized, grant, claim, access, ready, stopped, ticket };
++/**
++ * The pool machine's Link asks what to host next. It presents its own node token; the answer is the oldest party
++ * the pool is holding for it, or none. The poll is also how a registered machine says it is alive.
++ */
++async function assignment(req) {
++ const node = await nodeOf(req);
++ if (!node) return null;
++ await pool.seen(node.id);
++ const hold = await pool.assignmentFor(node.id);
++ return {
++ assignment: hold ? { party: hold.party_slug, hold: hold.id, title: hold.party_title || null } : null,
++ };
++}
++
++module.exports = { nodeOf, authorized, grant, claim, access, ready, stopped, ticket, assignment };
ddiff --git a/server/rooms/party-pool.js b/server/rooms/party-pool.js
nnew file mode 100644
iindex 0000000..1e8fd16
---- /dev/null
++++ b/server/rooms/party-pool.js
@@@ -0,0 +1,433 @@
++'use strict';
++/**
++ * OpenVibe's own computer pool for openvibe.party (message 11, 2026-11-08: a party's desktop runs on someone's Link
++ * machine or on OpenVibe's compute).
++ *
++ * A Link machine registered by the owner advertises what it can run (`desktop`, and `browser` for later) and how many
++ * desktops it may host at once. A party started with `desktop_hosting: 'openvibe'` asks this pool:
++ *
++ * - placement picks a machine with a free desktop, most free first and then the least loaded, never a disabled one;
++ * - the pool holds it for the room (`held`), the machine's Link claims it (`hosting`) and reports its Connect share
++ * (`ready`); the room view says where it runs, plainly;
++ * - when every machine is busy the room waits in line (`queued`) and the view says its position. A released hold
++ * promotes the oldest queued room;
++ * - the room ending releases the hold, and a hold the machine never claimed (or stopped renewing) is released after
++ * the idle window. Link renews by asking for the party's access every three seconds while it hosts;
++ * - pool machines are the owner's own. There is no rental, quote, price or spending path in this module: it only
++ * chooses among machines the owner registered.
++ *
++ * The owner switch is `party_pool_settings.enabled` (one row, off by default, changed on openvibe.party/pool by an
++ * owner session); CHAT_PARTY_POOL=1 is an operator override for a development box. An ordinary person never sees the
++ * panel or changes the pool.
++ *
++ * The registry is an interface, so placement can be tested with a fake:
++ *
++ * list() → [{ principal_id, owner_subject, name, can_run, desktops, enabled, last_seen_at }]
++ * get(principal_id) → a machine or null
++ * upsert(fields) / update(principal_id, fields) / remove(principal_id) / touch(principal_id)
++ */
++const crypto = require('crypto');
++const config = require('../config');
++const db = require('../db/database');
++
++const PRINCIPAL = /^lnk_[0-9A-HJKMNP-TV-Z]{26}$/;
++const SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const CAN_RUN = new Set(['desktop', 'browser']);
++const OPEN_STATES = ['queued', 'held', 'hosting', 'ready'];
++
++class PoolError extends Error {
++ constructor(status, code, message) { super(message); this.status = status; this.code = code; }
++}
++const fail = (status, code, message) => { throw new PoolError(status, code, message); };
++
++const settingsOf = () => ({
++ enabled: !!config.party.pool.enabled,
++ idleMs: Math.max(60_000, Number(config.party.pool.idleMs) || 600_000),
++ queueMs: Math.max(60_000, Number(config.party.pool.queueMs) || 1_800_000),
++ maxPerActor: Math.max(1, Number(config.party.pool.maxPerActor) || 2),
++ maxQueue: Math.max(1, Number(config.party.pool.maxQueue) || 50),
++});
++
++const newId = () => `ph_${crypto.randomBytes(12).toString('hex')}`;
++const iso = (ms) => new Date(ms).toISOString();
++
++/** A stored machine as the module and its callers use it (jsonb can come back parsed or as text). */
++function machineOf(row) {
++ if (!row) return null;
++ let can = row.can_run;
++ if (typeof can === 'string') { try { can = JSON.parse(can); } catch { can = []; } }
++ return {
++ principal_id: String(row.principal_id || ''),
++ owner_subject: String(row.owner_subject || ''),
++ name: String(row.name || ''),
++ can_run: (Array.isArray(can) ? can : []).filter((x) => CAN_RUN.has(String(x))),
++ desktops: Math.max(1, Number(row.desktops) || 1),
++ enabled: row.enabled !== false,
++ last_seen_at: row.last_seen_at || null,
++ created_at: row.created_at || null,
++ updated_at: row.updated_at || null,
++ };
++}
++
++/** The owner registration fields, checked. → the fields a registry write takes. */
++function registrationOf(input = {}) {
++ if (!input || typeof input !== 'object' || Array.isArray(input)) fail(422, 'party.pool_input', 'Send one machine');
++ const principal = String(input.principal_id || '');
++ if (!PRINCIPAL.test(principal)) fail(422, 'party.pool_principal', 'A pool machine is a Link machine principal (lnk_…)');
++ const name = typeof input.name === 'string' ? input.name.replace(/\s+/g, ' ').trim().slice(0, 80) : '';
++ if (!name) fail(422, 'party.pool_name', 'Give the machine a name of 1–80 characters');
++ const desktops = Number(input.desktops == null ? 1 : input.desktops);
++ if (!Number.isInteger(desktops) || desktops < 1 || desktops > 50) fail(422, 'party.pool_desktops', 'Desktops is a whole number of 1–50');
++ const wanted = input.can_run == null ? ['desktop'] : input.can_run;
++ const list = Array.isArray(wanted) ? wanted : [wanted];
++ const can = [];
++ for (const raw of list) {
++ const id = String(raw || '');
++ if (!CAN_RUN.has(id)) fail(422, 'party.pool_can_run', 'A pool machine can run desktop or browser');
++ if (!can.includes(id)) can.push(id);
++ }
++ if (!can.length) fail(422, 'party.pool_can_run', 'Say what the machine can run');
++ if (!can.includes('desktop')) fail(422, 'party.pool_can_run', 'A pool machine for parties runs desktop');
++ return { principal_id: principal, name, desktops, can_run: can, enabled: input.enabled !== false };
++}
++
++/** The settings row's owner switch, if the owner has used it. → true, false or null (never set). */
++function createRegistry({ store = db, now = Date.now } = {}) {
++ return {
++ async list() {
++ return (await store.all('SELECT * FROM party_pool_machines ORDER BY created_at, principal_id')).map(machineOf);
++ },
++ async get(id) {
++ if (!PRINCIPAL.test(String(id || ''))) return null;
++ return machineOf(await store.get('SELECT * FROM party_pool_machines WHERE principal_id = ?', [String(id)]));
++ },
++ async upsert(fields) {
++ const at = new Date(now()).toISOString();
++ await store.run(`INSERT INTO party_pool_machines (principal_id, owner_subject, name, can_run, desktops, enabled, updated_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT (principal_id) DO UPDATE SET owner_subject = excluded.owner_subject, name = excluded.name,
++ can_run = excluded.can_run, desktops = excluded.desktops, enabled = excluded.enabled, updated_at = excluded.updated_at`,
++ [fields.principal_id, fields.owner_subject, fields.name, JSON.stringify(fields.can_run), fields.desktops, fields.enabled !== false, at]);
++ return await this.get(fields.principal_id);
++ },
++ async update(id, fields) {
++ const cur = await this.get(id);
++ if (!cur) return null;
++ const next = { ...cur, ...fields, principal_id: cur.principal_id };
++ return await this.upsert(next);
++ },
++ async remove(id) {
++ return (await store.run('DELETE FROM party_pool_machines WHERE principal_id = ?', [String(id)])) > 0;
++ },
++ async touch(id) {
++ if (!PRINCIPAL.test(String(id || ''))) return;
++ await store.run('UPDATE party_pool_machines SET last_seen_at = now() WHERE principal_id = ?', [String(id)]);
++ },
++ };
++}
++
++function createPool({ registry = createRegistry(), store = db, now = Date.now, log = console } = {}) {
++ const cfg = () => settingsOf();
++
++ /** The effective switch: the owner's row, unless the operator set CHAT_PARTY_POOL=1. */
++ async function enabled() {
++ if (cfg().enabled) return true;
++ const row = await store.get('SELECT enabled FROM party_pool_settings WHERE id = 1');
++ return !!(row && row.enabled);
++ }
++
++ /** The owner flips it; the row is the switch, so every process reads the same answer on its next call. */
++ async function setEnabled(on, by = null) {
++ await store.run(`INSERT INTO party_pool_settings (id, enabled, updated_at, updated_by) VALUES (1, ?, now(), ?)
++ ON CONFLICT (id) DO UPDATE SET enabled = excluded.enabled, updated_at = now(), updated_by = excluded.updated_by`,
++ [!!on, by]);
++ return { enabled: await enabled() };
++ }
++
++ async function activeHolds() {
++ return await store.all(`SELECT * FROM party_pool_holds WHERE state <> 'released'`);
++ }
++
++ async function loadByMachine() {
++ const rows = await store.all(`SELECT principal_id, COUNT(*) AS n FROM party_pool_holds
++ WHERE state IN ('held', 'hosting', 'ready') AND principal_id IS NOT NULL GROUP BY principal_id`);
++ const out = new Map();
++ for (const row of rows) out.set(String(row.principal_id), Number(row.n) || 0);
++ return out;
++ }
++
++ /** The machine with the most free desktops, then the least loaded, among enabled machines that can run desktop. */
++ function pick(machines, load) {
++ const candidates = machines
++ .filter((m) => m.enabled && m.can_run.includes('desktop'))
++ .map((m) => {
++ const used = load.get(m.principal_id) || 0;
++ return { machine: m, used, free: m.desktops - used };
++ })
++ .filter((x) => x.free > 0);
++ if (!candidates.length) return null;
++ candidates.sort((a, b) =>
++ (b.free - a.free) ||
++ (a.used / a.machine.desktops - b.used / b.machine.desktops) ||
++ a.machine.name.localeCompare(b.machine.name) ||
++ a.machine.principal_id.localeCompare(b.machine.principal_id));
++ return candidates[0].machine;
++ }
++
++ async function queueLength() {
++ const row = await store.get("SELECT COUNT(*) AS n FROM party_pool_holds WHERE state = 'queued'");
++ return Number(row && row.n) || 0;
++ }
++
++ /** A hold's current position in line: 1 is next. */
++ async function queuePosition(hold) {
++ const row = await store.get(`SELECT COUNT(*) AS n FROM party_pool_holds WHERE state = 'queued'
++ AND (created_at < ? OR (created_at = ? AND id < ?))`, [hold.created_at, hold.created_at, hold.id]);
++ return (Number(row && row.n) || 0) + 1;
++ }
++
++ /**
++ * Reserve a machine for a party, or put the party in line. Serialized on the settings row so two people starting
++ * at the same moment cannot take the same free desktop.
++ * → { queued: false, hold, machine } or { queued: true, hold, position }
++ */
++ async function place({ actor, slug }) {
++ if (!SUBJECT.test(String(actor || ''))) fail(401, 'party.pool_sign_in', 'Sign in to ask for an OpenVibe computer');
++ if (!(await enabled())) fail(503, 'party.pool_off', 'OpenVibe computers are not available yet');
++ const want = cfg();
++ return await store.tx(async () => {
++ await store.get('SELECT id FROM party_pool_settings WHERE id = 1 FOR UPDATE');
++ const mine = await store.get(`SELECT COUNT(*) AS n FROM party_pool_holds
++ WHERE actor_subject = ? AND state <> 'released'`, [actor]);
++ if ((Number(mine && mine.n) || 0) >= want.maxPerActor) {
++ fail(429, 'party.pool_limit', `You already have ${want.maxPerActor} OpenVibe computer${want.maxPerActor === 1 ? '' : 's'} waiting or running; end one first.`);
++ }
++ const machines = await registry.list();
++ const available = machines.filter((m) => m.enabled && m.can_run.includes('desktop'));
++ if (!available.length) fail(503, 'party.pool_empty', 'No OpenVibe computer is registered yet. The owner can add one on the pool panel.');
++ const chosen = pick(available, await loadByMachine());
++ const id = newId();
++ if (chosen) {
++ await store.run(`INSERT INTO party_pool_holds (id, slug, room_id, actor_subject, principal_id, state, held_at, renewed_at)
++ VALUES (?, ?, NULL, ?, ?, 'held', ?, ?)`, [id, slug, actor, chosen.principal_id, iso(now()), iso(now())]);
++ return { queued: false, hold: await store.get('SELECT * FROM party_pool_holds WHERE id = ?', [id]), machine: chosen };
++ }
++ if ((await queueLength()) >= want.maxQueue) {
++ fail(503, 'party.pool_full', 'Every OpenVibe computer is busy and the line is full right now. Try again in a little while.');
++ }
++ await store.run(`INSERT INTO party_pool_holds (id, slug, room_id, actor_subject, principal_id, state, renewed_at, created_at)
++ VALUES (?, ?, NULL, ?, NULL, 'queued', ?, ?)`, [id, slug, actor, iso(now()), iso(now())]);
++ const hold = await store.get('SELECT * FROM party_pool_holds WHERE id = ?', [id]);
++ return { queued: true, hold, position: await queuePosition(hold) };
++ });
++ }
++
++ /** Once the room exists, the hold names it. */
++ async function bind(holdId, roomId) {
++ await store.run('UPDATE party_pool_holds SET room_id = ? WHERE id = ? AND room_id IS NULL', [roomId, holdId]);
++ }
++
++ /** The newest open hold of a room, or null. */
++ async function holdForRoom(roomId) {
++ if (roomId == null) return null;
++ return await store.get(`SELECT * FROM party_pool_holds WHERE room_id = ? AND state <> 'released'
++ ORDER BY created_at DESC LIMIT 1`, [roomId]);
++ }
++
++ /** The machine's Link is alive: its assignment poll is the pool's presence. */
++ async function seen(principal) {
++ await registry.touch(principal);
++ }
++
++ /** The oldest party waiting for this machine's Link. */
++ async function assignmentFor(principal) {
++ if (!PRINCIPAL.test(String(principal || ''))) return null;
++ return await store.get(`SELECT h.*, r.slug AS party_slug, r.name AS party_title
++ FROM party_pool_holds h JOIN rooms r ON r.id = h.room_id
++ JOIN room_parties p ON p.room_id = h.room_id
++ WHERE h.principal_id = ? AND h.state = 'held' AND p.ended_at IS NULL
++ AND p.desktop_state = 'waiting' AND p.desktop_node_id IS NULL
++ ORDER BY h.created_at, h.id LIMIT 1`, [principal]);
++ }
++
++ /** Link claimed the room's reserved machine. */
++ async function claiming(roomId, principal) {
++ const row = await store.get(`UPDATE party_pool_holds SET state = 'hosting', renewed_at = ?
++ WHERE room_id = ? AND principal_id = ? AND state = 'held' RETURNING *`, [iso(now()), roomId, principal]);
++ if (!row) fail(409, 'party.pool_hold', 'This OpenVibe computer is not holding that party');
++ return row;
++ }
++
++ /** A live hosting Link renewed (its access poll); the idle clock starts over. */
++ async function renew(roomId, principal) {
++ await store.run(`UPDATE party_pool_holds SET renewed_at = ?
++ WHERE room_id = ? AND principal_id = ? AND state IN ('held', 'hosting', 'ready')`, [iso(now()), roomId, principal]);
++ }
++
++ /** Link's Connect share is up. */
++ async function ready(roomId, principal) {
++ await store.run(`UPDATE party_pool_holds SET state = 'ready', renewed_at = ?
++ WHERE room_id = ? AND principal_id = ? AND state IN ('held', 'hosting')`, [iso(now()), roomId, principal]);
++ }
++
++ /** Release one hold and stop its room's desktop. Idempotent. → the released row or null. */
++ async function releaseHold(hold, reason) {
++ const row = await store.get(`UPDATE party_pool_holds SET state = 'released', released_at = ?, release_reason = ?
++ WHERE id = ? AND state <> 'released' RETURNING *`, [iso(now()), String(reason || 'released').slice(0, 40), hold.id]);
++ if (!row) return null;
++ if (row.room_id != null) {
++ await store.run(`UPDATE room_parties SET desktop_state = 'stopped', desktop_share_url = NULL
++ WHERE room_id = ? AND party_kind = 'desktop' AND ended_at IS NULL`, [row.room_id]);
++ }
++ return row;
++ }
++
++ /** A room ended (or its desktop stopped): free its machine. */
++ async function releaseRoom(roomId, reason = 'ended') {
++ const hold = await holdForRoom(roomId);
++ return hold ? releaseHold(hold, reason) : null;
++ }
++
++ /** A placement that never became a room (the room insert failed): give the machine or the place in line back. */
++ async function cancel(holdId, reason = 'cancelled') {
++ const hold = await store.get("SELECT * FROM party_pool_holds WHERE id = ? AND state <> 'released'", [String(holdId || '')]);
++ return hold ? releaseHold(hold, reason) : null;
++ }
++
++ /** The owner takes a machine out of the pool: its rooms are released and it stops being chosen. */
++ async function removeMachine(principal) {
++ const holds = await store.all(`SELECT * FROM party_pool_holds
++ WHERE principal_id = ? AND state <> 'released'`, [String(principal)]);
++ for (const hold of holds) await releaseHold(hold, 'machine_removed');
++ return { removed: await registry.remove(String(principal)), released: holds.length };
++ }
++
++ /** Where the room runs, or where it waits. → null when the room has no pool hold. */
++ async function where(roomId) {
++ if (roomId == null) return null;
++ // Open holds first; a released one is still the room's "where" line ("released because it went idle").
++ const hold = await holdForRoom(roomId)
++ || await store.get(`SELECT * FROM party_pool_holds WHERE room_id = ? ORDER BY created_at DESC, id DESC LIMIT 1`, [roomId]);
++ if (!hold) return null;
++ const machine = hold.principal_id ? await registry.get(hold.principal_id) : null;
++ const name = machine ? machine.name : 'an OpenVibe computer';
++ if (hold.state === 'queued') {
++ const position = await queuePosition(hold);
++ return { state: 'queued', machine: null, queue_position: position,
++ line: `Waiting for an OpenVibe computer: you are ${position === 1 ? 'next' : `${position} in line`}.` };
++ }
++ if (hold.state === 'held') {
++ return { state: 'held', machine: machine ? machine.name : null, queue_position: null,
++ line: `${name} is reserved for this party and will start your computer.` };
++ }
++ if (hold.state === 'hosting') {
++ return { state: 'hosting', machine: machine ? machine.name : null, queue_position: null,
++ line: `Starting on ${name}, an OpenVibe computer.` };
++ }
++ if (hold.state === 'ready') {
++ return { state: 'ready', machine: machine ? machine.name : null, queue_position: null,
++ line: `Running on ${name}, an OpenVibe computer.` };
++ }
++ return { state: 'released', machine: machine ? machine.name : null, queue_position: null,
++ line: hold.release_reason === 'idle'
++ ? 'The OpenVibe computer was released because the party went idle.'
++ : 'The OpenVibe computer was released.' };
++ }
++
++ /** What the start screen needs: whether the pool is on, its size, its free desktops and its line. */
++ async function availability() {
++ const on = await enabled();
++ const machines = on ? (await registry.list()).filter((m) => m.enabled && m.can_run.includes('desktop')) : [];
++ let free = 0;
++ if (on) {
++ const load = await loadByMachine();
++ for (const m of machines) free += Math.max(0, m.desktops - (load.get(m.principal_id) || 0));
++ }
++ return { enabled: on, machines: machines.length, desktops: machines.reduce((n, m) => n + m.desktops, 0),
++ free, queued: on ? await queueLength() : 0 };
++ }
++
++ /**
++ * Release holds that went idle and promote the line. A hold whose machine never claimed it (or stopped renewing)
++ * is released after the idle window; a queued room that nobody could seat is released after the queue window.
++ * Called every minute by the server and directly by tests.
++ */
++ async function sweep() {
++ const at = now();
++ const out = { released: 0, promoted: 0 };
++ const want = cfg();
++ const idle = await store.all(`SELECT * FROM party_pool_holds
++ WHERE state IN ('held', 'hosting', 'ready') AND renewed_at < ?`, [iso(at - want.idleMs)]);
++ for (const hold of idle) if (await releaseHold(hold, 'idle')) out.released++;
++ const stale = await store.all(`SELECT * FROM party_pool_holds WHERE state = 'queued' AND created_at < ?`,
++ [iso(at - want.queueMs)]);
++ for (const hold of stale) if (await releaseHold(hold, 'queue_expired')) out.released++;
++ // With the owner's switch off nothing new is seated; stale holds above are still released.
++ if (!(await enabled())) return out;
++ // Oldest first, and only as far as there is room: a later room never jumps ahead in line.
++ const waiting = await store.all(`SELECT * FROM party_pool_holds WHERE state = 'queued' ORDER BY created_at, id`);
++ if (waiting.length) {
++ const machines = (await registry.list()).filter((m) => m.enabled && m.can_run.includes('desktop'));
++ const load = await loadByMachine();
++ for (const hold of waiting) {
++ const chosen = pick(machines, load);
++ if (!chosen) break;
++ const row = await store.get(`UPDATE party_pool_holds SET state = 'held', principal_id = ?, held_at = ?, renewed_at = ?
++ WHERE id = ? AND state = 'queued' RETURNING *`, [chosen.principal_id, iso(at), iso(at), hold.id]);
++ if (!row) continue;
++ load.set(chosen.principal_id, (load.get(chosen.principal_id) || 0) + 1);
++ out.promoted++;
++ }
++ }
++ if (out.released || out.promoted) log.info?.(`[Party pool] released ${out.released}, seated ${out.promoted}`);
++ return out;
++ }
++
++ /** The owner panel: the switch, the machines with their load, and every open hold. */
++ async function panel() {
++ const on = await enabled();
++ const machines = await registry.list();
++ const load = await loadByMachine();
++ const holds = await store.all(`SELECT h.*, r.slug AS party_slug, r.name AS party_title
++ FROM party_pool_holds h LEFT JOIN rooms r ON r.id = h.room_id
++ WHERE h.state <> 'released' ORDER BY h.created_at, h.id`);
++ return {
++ enabled: on,
++ machines: machines.map((m) => ({
++ ...m,
++ load: load.get(m.principal_id) || 0,
++ free: Math.max(0, m.desktops - (load.get(m.principal_id) || 0)),
++ })),
++ holds: holds.map((h) => ({
++ id: h.id, state: h.state, party: h.party_slug || h.slug, title: h.party_title || null,
++ actor: h.actor_subject, machine: h.principal_id || null, queue_position: null,
++ created_at: h.created_at, renewed_at: h.renewed_at, release_reason: h.release_reason || null,
++ })),
++ queued: holds.filter((h) => h.state === 'queued').length,
++ };
++ }
++
++ // The registry writes the owner panel and its API use (the Link machines live in the registry interface).
++ const upsert = (fields) => registry.upsert(fields);
++ const update = (principal, fields) => registry.update(principal, fields);
++
++ return { enabled, setEnabled, place, bind, holdForRoom, assignmentFor, seen, claiming, renew, ready, releaseRoom, cancel,
++ removeMachine, where, availability, sweep, panel, pick, upsert, update };
++}
++
++const registry = createRegistry();
++const pool = createPool({ registry });
++let timer = null;
++
++/** The server's minute sweep (server/index.js). Off in tests, which call sweep() themselves. */
++function start({ intervalMs = 60_000, log = console } = {}) {
++ if (timer || config.nodeEnv === 'test') return;
++ timer = setInterval(() => { pool.sweep().catch((err) => log.warn?.('[Party pool] sweep failed:', err && err.message)); }, Math.max(15_000, intervalMs));
++ if (timer.unref) timer.unref();
++}
++function stop() { if (timer) clearInterval(timer); timer = null; }
++
++module.exports = {
++ PoolError, PRINCIPAL, CAN_RUN, machineOf, registrationOf, createRegistry, createPool,
++ start, stop, ...pool,
++};
ddiff --git a/server/rooms/party-routes.js b/server/rooms/party-routes.js
iindex 0c3f328..07d8a23 100644
---- a/server/rooms/party-routes.js
++++ b/server/rooms/party-routes.js
@@@ -14,18 +14,31 @@
const express = require('express');
const cache = require('openvibe-shared/cache-policy');
const { requireAuth, optionalAuth } = require('../auth/auth');
++const { isOwner } = require('../auth/permissions');
const { limits } = require('../net/actor-limits');
const parties = require('../rooms/parties');
const radio = require('../rooms/radio');
const desktop = require('../rooms/party-desktop');
++const partyPool = require('../rooms/party-pool');
const router = express.Router();
// Starting a party: the same budget as making a room (owners keep at most MAX_ROOMS_PER_OWNER).
const createLimit = limits('chat.party.create', { minute: 10, hour: 60 });
++// The owner's pool panel is a write surface; the person's own pool budget is in party-pool.js (maxPerActor).
++const poolLimit = limits('chat.party.pool', { minute: 30, hour: 120 });
++
++/** Only the owner rank runs OpenVibe's own computers. */
++function ownerOnly(req, res, next) {
++ if (!req.user || !isOwner(req.user)) return res.status(403).json({ error: 'Only the owner runs OpenVibe’s computer pool.', code: 'party.pool_owner' });
++ const subject = String(req.user.subject_id || '');
++ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(subject)) return res.status(409).json({ error: 'This account has no OpenVibe subject yet; sign in again.', code: 'party.no_subject' });
++ req.poolSubject = subject;
++ next();
++}
function send(res, err) {
-- if (err instanceof parties.PartyError) return res.status(err.status).json({ error: err.message, code: err.code });
++ if (err instanceof parties.PartyError || err instanceof partyPool.PoolError) return res.status(err.status).json({ error: err.message, code: err.code });
console.error('[Parties]', err && err.message);
return res.status(500).json({ error: 'Something went wrong' });
}
@@@ -41,6 +54,59 @@ router.get('/radio/tracks', async (_req, res) => {
res.set('Cache-Control', 'public, max-age=30').json({ tracks: await radio.catalogue() });
});
++// ── OpenVibe's own computers (the owner's pool) ───────────────────────────────────────────────────────────────
++// The switch and the machines are the owner rank's alone; the panel is openvibe.party/pool. When the switch is
++// off (the default), a party cannot start on OpenVibe and the start screen says so.
++router.get('/pool', requireAuth, ownerOnly, poolLimit, async (req, res) => {
++ try {
++ res.set('Cache-Control', 'private, no-store').json(await partyPool.panel());
++ } catch (err) { send(res, err); }
++});
++
++router.put('/pool/enabled', requireAuth, ownerOnly, poolLimit, express.json({ limit: '2kb' }), async (req, res) => {
++ try {
++ const on = !!(req.body && req.body.enabled);
++ await partyPool.setEnabled(on, req.poolSubject);
++ res.set('Cache-Control', 'private, no-store').json(await partyPool.panel());
++ } catch (err) { send(res, err); }
++});
++
++router.post('/pool/machines', requireAuth, ownerOnly, poolLimit, express.json({ limit: '4kb' }), async (req, res) => {
++ try {
++ const fields = partyPool.registrationOf(req.body || {});
++ const machine = await partyPool.upsert({ ...fields, owner_subject: req.poolSubject });
++ res.status(201).json({ machine });
++ } catch (err) { send(res, err); }
++});
++
++router.patch('/pool/machines/:principal', requireAuth, ownerOnly, poolLimit, express.json({ limit: '4kb' }), async (req, res) => {
++ try {
++ const id = String(req.params.principal || '');
++ const current = (await partyPool.panel()).machines.find((m) => m.principal_id === id);
++ if (!current) return res.status(404).json({ error: 'No such pool machine', code: 'party.pool_missing' });
++ const fields = partyPool.registrationOf({ ...current, ...(req.body || {}), principal_id: id });
++ const machine = await partyPool.update(id, { ...fields, owner_subject: current.owner_subject });
++ res.set('Cache-Control', 'private, no-store').json({ machine });
++ } catch (err) { send(res, err); }
++});
++
++router.delete('/pool/machines/:principal', requireAuth, ownerOnly, poolLimit, async (req, res) => {
++ try {
++ const out = await partyPool.removeMachine(String(req.params.principal || ''));
++ if (!out.removed) return res.status(404).json({ error: 'No such pool machine', code: 'party.pool_missing' });
++ res.set('Cache-Control', 'private, no-store').json(out);
++ } catch (err) { send(res, err); }
++});
++
++// A registered pool machine's Link asks what to host next; its node token is the authority.
++router.get('/desktop/assignment', async (req, res) => {
++ try {
++ const out = await desktop.assignment(req);
++ if (!out) return res.status(404).json({ code: 'rooms.not_found' });
++ res.set('Cache-Control', 'no-store').json(out);
++ } catch (err) { send(res, err); }
++});
++
// Link presents its Network node token for Chat, acting for the party owner.
// These routes never accept the invite secret or a browser's session as machine authority.
router.post('/:slug/desktop/claim', express.json({ limit: '2kb' }), async (req, res) => {
ddiff --git a/server/web/party-pages.js b/server/web/party-pages.js
iindex 3cd721d..8a6e436 100644
---- a/server/web/party-pages.js
++++ b/server/web/party-pages.js
@@@ -27,6 +27,8 @@ const { createSessionRoutes } = require('./session');
const { viewerOf } = require('./pages');
const parties = require('../rooms/parties');
const radio = require('../rooms/radio');
++const partyPool = require('../rooms/party-pool');
++const { isOwner } = require('../auth/permissions');
const radioSource = (v) => ({ creator: 'Creator upload', creative_commons: 'Creative Commons', public_domain: 'Public domain', ai: 'AI-made music' })[v] || v;
const radioLicence = (v) => v === 'creator_grant' ? 'Creator’s permission to stream' : v === 'public_domain' ? 'Public domain mark' : v;
@@@ -152,12 +154,22 @@ function createPartyRoutes({ config }) {
<ul class="pty-mine-list">${mine.map((x) => `<li><a href="/p/${encodeURIComponent(x.id)}"><strong>${esc(x.title)}</strong><span>${x.hosting ? 'You started it' : 'You joined'} · ${esc(since(x.started_at))}</span></a></li>`).join('')}</ul>
</section>` : '';
const startHref = (x) => `/new?media=${encodeURIComponent(x.page_url)}&title=${encodeURIComponent(x.title)}`;
-- // Two plain choices, not a dropdown (Opus's design pass): where the shared computer runs.
++ // Two plain choices, not a dropdown (Opus's design pass): where the shared computer runs. The OpenVibe
++ // choice is live only while the owner's pool has a machine; the line below it says the plain truth.
++ const available = await partyPool.availability().catch(() => ({ enabled: false, machines: 0, desktops: 0, free: 0, queued: 0 }));
++ const poolCanStart = available.enabled && available.machines > 0;
++ const poolLine = !available.enabled
++ ? 'OpenVibe computers are not on yet.'
++ : available.machines === 0
++ ? 'No OpenVibe computer is registered yet.'
++ : available.free > 0
++ ? `${available.free} of ${available.desktops} OpenVibe computer${available.desktops === 1 ? '' : 's'} free.`
++ : `All ${available.desktops} OpenVibe computer${available.desktops === 1 ? '' : 's'} busy${available.queued ? ` · ${available.queued} in line` : ''}.`;
const desktopStart = signedIn ? `<form class="pty-desktop-start" method="get" action="/new">
<h2 class="pty-start-h">Start a shared computer</h2>
<div class="pty-host-choice">
<button type="submit" name="desktop" value="link" class="pty-host"><strong>On my computer</strong><span>Free. Runs in the Link app on your computer.</span></button>
--<button type="submit" name="desktop" value="openvibe" class="pty-host" disabled aria-describedby="pty-openvibe-availability"><strong>On OpenVibe</strong><span id="pty-openvibe-availability">No OpenVibe machine is free right now.</span></button>
++<button type="submit" name="desktop" value="openvibe" class="pty-host"${poolCanStart ? '' : ' disabled'} aria-describedby="pty-openvibe-availability"><strong>On OpenVibe</strong><span id="pty-openvibe-availability">${esc(poolLine)}</span></button>
</div>
</form>` : `<p class="pty-cta"><a class="oc-button" href="/auth/login?next=%2F">Sign in to start a shared computer</a></p>`;
const start = signedIn
@@@ -211,8 +223,16 @@ ${grid}
if (req.query.desktop) {
const hosting = String(req.query.desktop);
if (hosting !== 'link' && hosting !== 'openvibe') return res.redirect(303, '/');
-- if (hosting === 'openvibe') return page(req, res, 503, { actor, title: 'No OpenVibe computer is free', path: '/new', robots: 'noindex, nofollow', footer: 'compact', partyHome: true,
-- body: '<section class="oc-party-confirm pty-confirm"><h1>No OpenVibe machine is free right now</h1><p>You can host this party on your own computer with Link.</p><p><a class="oc-button" href="/new?desktop=link">Host it on my computer</a></p></section>' });
++ if (hosting === 'openvibe') {
++ const available = await partyPool.availability().catch(() => ({ enabled: false, machines: 0, desktops: 0, free: 0, queued: 0 }));
++ if (!available.enabled || available.machines === 0) return page(req, res, 503, { actor, title: 'No OpenVibe computer is available', path: '/new', robots: 'noindex, nofollow', footer: 'compact', partyHome: true,
++ body: `<section class="oc-party-confirm pty-confirm"><h1>${available.enabled ? 'No OpenVibe computer is registered yet' : 'OpenVibe computers are not on yet'}</h1><p>${available.enabled ? 'The owner can add one on the pool panel; until then you can host this party on your own computer with Link.' : 'You can host this party on your own computer with Link.'}</p><p><a class="oc-button" href="/new?desktop=link">Host it on my computer</a> <a href="/" class="oc-muted">Not now</a></p></section>` });
++ const line = available.free > 0
++ ? `An OpenVibe computer is free. It starts when the OpenVibe Link on that machine picks the party up.`
++ : `Every OpenVibe computer is busy right now. You will wait in line (${available.queued + 1}${available.queued ? ' behind the parties already waiting' : 'st'}) and it starts as soon as one is free.`;
++ return page(req, res, 200, { actor, title: 'Start a shared computer on OpenVibe', path: '/new', robots: 'noindex, nofollow', footer: 'compact', partyHome: true,
++ body: `<section class="oc-party-confirm pty-confirm"><h1>Start a shared computer on OpenVibe?</h1><p>OpenVibe runs a temporary browser for this party on one of its own computers. People you invite can watch, and you can pass the remote.</p><p class="oc-muted">${esc(line)}</p><form method="post" action="/new"><input type="hidden" name="desktop_hosting" value="openvibe"><button class="oc-button" type="submit">Start the party</button> <a href="/" class="oc-muted">Not now</a></form></section>` });
++ }
return page(req, res, 200, { actor, title: 'Start a shared computer', path: '/new', robots: 'noindex, nofollow', footer: 'compact', partyHome: true,
body: `<section class="oc-party-confirm pty-confirm"><h1>Start a shared computer?</h1><p>Link runs a temporary browser in a container on your computer. People you invite can watch, and you can pass the remote.</p><p class="oc-muted">It needs the OpenVibe Link app on this computer. <a href="https://openvibe.link">Get Link</a></p><form method="post" action="/new"><input type="hidden" name="desktop_hosting" value="link"><button class="oc-button" type="submit">Start the party</button> <a href="/" class="oc-muted">Not now</a></form></section>` });
}
@@@ -261,7 +281,7 @@ ${source ? `<input type="hidden" name="source_url" value="${esc(source)}">` : ''
const out = await parties.createDesktop(actor.user, { desktop_hosting: String(req.body.desktop_hosting), title: 'Shared computer' });
return res.redirect(303, `/p/${encodeURIComponent(out.party.id)}?k=${encodeURIComponent(out.join_secret)}`);
} catch (err) {
-- return res.redirect(303, `/?error=${encodeURIComponent(err instanceof parties.PartyError ? err.message : 'The shared computer could not be started')}`);
++ return res.redirect(303, `/?error=${encodeURIComponent(err && err.status && err.message ? err.message : 'The shared computer could not be started')}`);
}
}
if (req.body && req.body.radio_track_id) {
@@@ -288,6 +308,88 @@ ${source ? `<input type="hidden" name="source_url" value="${esc(source)}">` : ''
}
});
++ // ── The owner's pool panel (openvibe.party/pool): the switch and the Link machines OpenVibe may use ──
++ // Plain markup on purpose: layout, styling and copy are the design pass's.
++ const sameOrigin = (req) => {
++ const origin = String(req.get('origin') || '');
++ const site = String(req.get('sec-fetch-site') || '');
++ return (!origin || origin === base) && (!site || site === 'same-origin');
++ };
++ const poolPanel = async (req, res, { error = '' } = {}) => {
++ const data = await partyPool.panel();
++ const machineRows = data.machines.length ? `<table><thead><tr><th>Machine</th><th>Principal</th><th>Runs</th><th>Desktops</th><th>In use</th><th>Offered</th><th></th></tr></thead><tbody>${data.machines.map((m) => `<tr>
++<td>${esc(m.name)}</td><td><code>${esc(m.principal_id)}</code></td><td>${esc(m.can_run.join(', '))}</td><td>${esc(String(m.desktops))}</td><td>${esc(String(m.load))}</td><td>${m.enabled ? 'yes' : 'no'}</td>
++<td><form method="post" action="/pool/machines/${encodeURIComponent(m.principal_id)}/delete"><button type="submit" class="oc-link">Remove</button></form></td>
++</tr>`).join('')}</tbody></table>` : '<p class="oc-muted">No OpenVibe computers are registered yet.</p>';
++ const holdRows = data.holds.length ? `<table><thead><tr><th>Party</th><th>State</th><th>Machine</th><th>Person</th><th>Held</th></tr></thead><tbody>${data.holds.map((h) => `<tr><td><a href="/p/${encodeURIComponent(h.party)}">${esc(h.title || h.party)}</a></td><td>${esc(h.state)}</td><td>${h.machine ? `<code>${esc(h.machine)}</code>` : '—'}</td><td><code>${esc(h.actor)}</code></td><td>${esc(String(h.created_at || ''))}</td></tr>`).join('')}</tbody></table>` : '<p class="oc-muted">No party is waiting for or running on an OpenVibe computer.</p>';
++ return await page(req, res, 200, {
++ actor: await viewerOf(req), title: 'OpenVibe computers', path: '/pool', robots: 'noindex, nofollow', footer: 'compact', partyHome: true,
++ body: `<section class="oc-party-confirm pty-confirm">
++<h1>OpenVibe computers</h1>
++<p class="oc-muted">These are OpenVibe’s own Link machines. When a party asks for an OpenVibe computer, placement picks one with a free desktop and holds it for that party. There is no rental and no charge: the machines are the owner’s own.</p>
++${error ? notice('error', error) : ''}
++<p>OpenVibe computers are <strong>${data.enabled ? 'on' : 'off'}</strong>.</p>
++<form method="post" action="/pool/enabled"><input type="hidden" name="enabled" value="${data.enabled ? '0' : '1'}"><button type="submit" class="oc-button">${data.enabled ? 'Turn the pool off' : 'Turn the pool on'}</button></form>
++</section>
++<section class="oc-party-confirm pty-confirm"><h2>Machines</h2>
++${machineRows}
++<h3>Add a machine</h3>
++<p class="oc-muted">Pair the machine in OpenVibe Link first; its principal is on your machines page on openvibe.network. Enter what it can run and how many desktops it may host at once.</p>
++<form method="post" action="/pool/machines">
++<label>Principal <input name="principal_id" required pattern="lnk_[0-9A-HJKMNP-TV-Z]{26}" placeholder="lnk_…"></label>
++<label>Name <input name="name" required maxlength="80" placeholder="Dell tower 1"></label>
++<label>Desktops <input name="desktops" type="number" min="1" max="50" value="1" required></label>
++<label><input type="checkbox" name="can_run" value="desktop" checked> Desktop parties</label>
++<label><input type="checkbox" name="can_run" value="browser"> Browsers</label>
++<button type="submit" class="oc-button">Add the machine</button>
++</form></section>
++<section class="oc-party-confirm pty-confirm"><h2>Parties on OpenVibe computers</h2>
++${holdRows}
++<p class="oc-muted">${data.queued} waiting in line.</p>
++</section>`,
++ });
++ };
++
++ router.get('/pool', async (req, res) => {
++ const actor = await viewerOf(req);
++ if (actor.kind !== 'user' || !isOwner(actor.user)) return page(req, res, 403, {
++ actor, title: 'Not the owner', path: '/pool', robots: 'noindex, nofollow', footer: 'compact',
++ body: '<h1>Only the owner runs OpenVibe’s computers</h1><p><a href="/">Start a party</a></p>',
++ });
++ return await poolPanel(req, res);
++ });
++
++ router.post('/pool/enabled', express.urlencoded({ extended: false, limit: '2kb' }), async (req, res) => {
++ const actor = await viewerOf(req);
++ if (actor.kind !== 'user' || !isOwner(actor.user)) return res.status(403).type('text/plain').send('Only the owner changes the pool.');
++ if (!sameOrigin(req)) return res.status(403).type('text/plain').send('Change the pool from openvibe.party itself.');
++ await partyPool.setEnabled(String(req.body && req.body.enabled) === '1', actor.user.subject_id || null);
++ return res.redirect(303, '/pool');
++ });
++
++ router.post('/pool/machines', express.urlencoded({ extended: false, limit: '4kb' }), async (req, res) => {
++ const actor = await viewerOf(req);
++ if (actor.kind !== 'user' || !isOwner(actor.user)) return res.status(403).type('text/plain').send('Only the owner changes the pool.');
++ if (!sameOrigin(req)) return res.status(403).type('text/plain').send('Change the pool from openvibe.party itself.');
++ try {
++ const raw = req.body || {};
++ const fields = partyPool.registrationOf({ principal_id: raw.principal_id, name: raw.name, desktops: raw.desktops,
++ can_run: [].concat(raw.can_run == null ? [] : raw.can_run), enabled: true });
++ await partyPool.upsert({ ...fields, owner_subject: actor.user.subject_id });
++ return res.redirect(303, '/pool');
++ } catch (err) {
++ return await poolPanel(req, res, { error: err && err.message ? err.message : 'That machine was not added' });
++ }
++ });
++
++ router.post('/pool/machines/:principal/delete', async (req, res) => {
++ const actor = await viewerOf(req);
++ if (actor.kind !== 'user' || !isOwner(actor.user)) return res.status(403).type('text/plain').send('Only the owner changes the pool.');
++ if (!sameOrigin(req)) return res.status(403).type('text/plain').send('Change the pool from openvibe.party itself.');
++ await partyPool.removeMachine(String(req.params.principal || ''));
++ return res.redirect(303, '/pool');
++ });
++
const notFound = async (req, res) => page(req, res, 404, {
title: 'Page not found', path: req.path, robots: 'noindex, nofollow', footer: 'compact',
body: `<h1>This party is not here</h1><p class="oc-muted">The link may be wrong, or the party may have ended.</p><p><a class="oc-button" href="/">Start a party</a></p>`,
@@@ -315,7 +417,7 @@ ${source ? `<input type="hidden" name="source_url" value="${esc(source)}">` : ''
description: 'Watch together on OpenVibe.',
mainAttrs: `data-party="${esc(out.id)}"`, party: true, wide: true,
body: `<article class="oc-party${out.kind === 'radio' ? ' is-radio' : ''}">
--<header class="oc-party-head"><h1>${esc(out.title)}</h1><p class="oc-muted">Started by ${esc(out.host_name || 'someone')}</p></header>
++<header class="oc-party-head"><h1>${esc(out.title)}</h1><p class="oc-muted">Started by ${esc(out.host_name || 'someone')}</p>${out.kind === 'desktop' && out.desktop && out.desktop.where ? `<p class="oc-party-where" data-party-where>${esc(out.desktop.where.line)}</p>` : ''}</header>
<div class="oc-party-stage" data-party-stage aria-label="${out.kind === 'radio' ? 'The music stage' : 'The video'}"></div>
<div class="oc-party-people" data-party-people aria-label="People here"></div>
<div class="oc-party-chat" data-party-chat aria-label="Chat"></div>
ddiff --git a/test/party-media.test.js b/test/party-media.test.js
iindex a03e7ba..11e7876 100644
---- a/test/party-media.test.js
++++ b/test/party-media.test.js
@@@ -314,7 +314,7 @@ t('the home offers OpenVibe.Video\'s latest, and a pasted watch page plays its f
assert.ok(!home.text.includes('evil.example'), 'a page on another site is never offered');
assert.match(home.text, /\/web\/party-home\.css/, 'the home\'s styles');
assert.match(home.text, /How it works/);
-- assert.match(home.text, /No OpenVibe machine is free right now/);
++ assert.match(home.text, /OpenVibe computers are not on yet/, 'the plain truth while the owner pool is off');
assert.match(home.text, /name="desktop" value="link" class="pty-host"><strong>On my computer<\/strong>/, 'hosting on my computer is one of two plain choices');
assert.match(home.text, /value="openvibe" class="pty-host" disabled aria-describedby="pty-openvibe-availability"/, 'On OpenVibe cannot be chosen while no OpenVibe machine is free');
ddiff --git a/test/party-pool.test.js b/test/party-pool.test.js
nnew file mode 100644
iindex 0000000..2830a1e
---- /dev/null
++++ b/test/party-pool.test.js
@@@ -0,0 +1,237 @@
++'use strict';
++// OpenVibe's own computers for openvibe.party (message 11): the owner registers a Link machine and what it can run,
++// a party with no machine of its own asks the pool, placement holds a machine by capacity and load (or gives the
++// party a place in line), the machine claims it, and the hold is released when the room ends or goes idle.
++//
++// The placement engine is also driven directly with a fake registry (no owner routes, no Link): that is the unit
++// under test, and the HTTP flow around it proves the party, panel and machine sides meet.
++const assert = require('assert');
++const http = require('http');
++const { ids, serviceAuth } = require('openvibe-contracts');
++const { boot, suite } = require('./helpers');
++
++const t = suite('party pool');
++let h, owner, ann, bob, cat, dave, db, poolModule;
++const LINK_SECRET = 'party-pool-test-shared-secret-01234567890123456789';
++const api = (method, path, u, body) => h.http(method, `/api/chat${path}`, { token: u && u.token, body });
++/** A page on the party host: Node's fetch keeps the URL's Host, so a Host header needs its own request. */
++const page = (path, u) => new Promise((resolve, reject) => {
++ const req = http.request({ host: '127.0.0.1', port: h.port, path, method: 'GET', headers: { host: 'openvibe.party', cookie: `ov_token=${u.token}` } }, (res) => {
++ let text = '';
++ res.on('data', (c) => { text += c; });
++ res.on('end', () => resolve({ status: res.statusCode, headers: res.headers, text }));
++ });
++ req.on('error', reject);
++ req.end();
++});
++const newLink = () => `lnk_${ids.ulid()}`;
++const nodeToken = (principal, subject) => {
++ const at = Math.floor(Date.now() / 1000);
++ return serviceAuth.signServiceToken({
++ iss: h.ISS, sub: `link:${principal}`, actor_type: 'link', on_behalf_of: subject,
++ aud: ['openvibe.chat'], cap: [], iat: at, exp: at + 300, jti: `pool_${ids.ulid()}`,
++ }, h.keys.privateKey);
++};
++
++/** A registry the placement engine can use with no database and no Link: the fake the brief asks for. */
++function fakeRegistry(machines) {
++ const rows = machines.map((m) => ({ can_run: ['desktop'], enabled: true, ...m }));
++ return {
++ rows,
++ async list() { return rows.map((m) => ({ ...m, can_run: [...m.can_run] })); },
++ async get(id) { const m = rows.find((x) => x.principal_id === id); return m ? { ...m, can_run: [...m.can_run] } : null; },
++ async upsert(f) {
++ const i = rows.findIndex((x) => x.principal_id === f.principal_id);
++ if (i >= 0) rows[i] = { ...rows[i], ...f }; else rows.push({ ...f });
++ return await this.get(f.principal_id);
++ },
++ async update(id, f) { const i = rows.findIndex((x) => x.principal_id === id); if (i < 0) return null; rows[i] = { ...rows[i], ...f }; return await this.get(id); },
++ async remove(id) { const i = rows.findIndex((x) => x.principal_id === id); if (i >= 0) rows.splice(i, 1); return i >= 0; },
++ async touch(id) { const m = rows.find((x) => x.principal_id === id); if (m) m.last_seen_at = new Date().toISOString(); },
++ };
++}
++
++t('boot', async () => {
++ h = await boot({ env: {
++ CHAT_PARTY_URL: 'https://openvibe.party', CHAT_PARTY_OAUTH_REDIRECT_URI: 'https://openvibe.party/auth/callback',
++ CHAT_PARTY_CONNECT_SECRET: LINK_SECRET, TURN_URL: 'turns:turn.openvibe.live:443', TURN_AUTH_SECRET: 'party-test-turn-secret',
++ } });
++ owner = h.addUser('owner', { role: 'admin', is_owner: 1, subject: ids.newId('user') });
++ ann = h.addUser('ann', { subject: ids.newId('user') });
++ bob = h.addUser('bob', { subject: ids.newId('user') });
++ cat = h.addUser('cat', { subject: ids.newId('user') });
++ dave = h.addUser('dave', { subject: ids.newId('user') });
++ for (const u of [owner, ann, bob, cat, dave]) await h.ctx.upsertUser(h.live.users.get(u.id));
++ db = h.db;
++ poolModule = require('../server/rooms/party-pool');
++});
++
++t('the pool is off by default, only the owner runs it, and off means no OpenVibe party', async () => {
++ const off = await api('GET', '/parties/pool', owner);
++ assert.strictEqual(off.status, 200, off.text);
++ assert.strictEqual(off.body.enabled, false, 'the owner switch starts off');
++ assert.strictEqual((await api('GET', '/parties/pool', ann)).status, 403, 'a person who is not the owner cannot read the pool');
++ assert.strictEqual((await api('PUT', '/parties/pool/enabled', ann, { enabled: true })).status, 403, 'nor turn it on');
++ const no = await api('POST', '/parties', ann, { desktop_hosting: 'openvibe', title: 'Off party' });
++ assert.strictEqual(no.status, 503, no.text);
++ assert.strictEqual(no.body.code, 'party.pool_off');
++ const on = await api('PUT', '/parties/pool/enabled', owner, { enabled: true });
++ assert.strictEqual(on.status, 200, on.text);
++ assert.strictEqual(on.body.enabled, true);
++});
++
++const M1 = newLink();
++const M2 = newLink();
++
++t('the owner registers machines, and a party asks the pool and is held a machine by capacity', async () => {
++ assert.strictEqual((await api('POST', '/parties/pool/machines', ann, { principal_id: M1, name: 'Dell tower 1', desktops: 2 })).status, 403);
++ const bad = await api('POST', '/parties/pool/machines', owner, { principal_id: 'dev_nope', name: 'X', desktops: 1 });
++ assert.strictEqual(bad.status, 422, bad.text);
++ const one = await api('POST', '/parties/pool/machines', owner, { principal_id: M1, name: 'Dell tower 1', desktops: 2, can_run: ['desktop'] });
++ assert.strictEqual(one.status, 201, one.text);
++ assert.strictEqual(one.body.machine.desktops, 2);
++ const two = await api('POST', '/parties/pool/machines', owner, { principal_id: M2, name: 'Dell tower 2', desktops: 1, can_run: ['desktop'] });
++ assert.strictEqual(two.status, 201, two.text);
++ const panel = await api('GET', '/parties/pool', owner);
++ assert.strictEqual(panel.body.machines.length, 2);
++ assert.strictEqual(panel.body.machines.reduce((n, m) => n + m.free, 0), 3);
++
++ const made = await api('POST', '/parties', ann, { desktop_hosting: 'openvibe', title: 'Film on the pool' });
++ assert.strictEqual(made.status, 201, made.text);
++ assert.strictEqual(made.body.party.desktop.hosting, 'openvibe');
++ assert.strictEqual(made.body.party.desktop.where.state, 'held');
++ assert.strictEqual(made.body.party.desktop.where.machine, 'Dell tower 1');
++ assert.match(made.body.party.desktop.where.line, /Dell tower 1/);
++ ann.party = made.body.party;
++ ann.key = made.body.join_secret;
++
++ const seen = await api('GET', `/parties/${ann.party.id}`, ann);
++ assert.strictEqual(seen.body.party.desktop.where.machine, 'Dell tower 1', 'the room view says where it runs');
++ assert.strictEqual(seen.body.party.desktop.where.state, 'held');
++
++ const assignment = await h.http('GET', '/api/chat/parties/desktop/assignment', { token: nodeToken(M1, owner.subject_id) });
++ assert.strictEqual(assignment.status, 200, assignment.text);
++ assert.strictEqual(assignment.body.assignment.party, ann.party.id, 'the held machine is told which party to host');
++ assert.ok(assignment.body.assignment.hold);
++});
++
++t('the held machine claims the party, and the room says when it runs there', async () => {
++ const wrong = await h.http('POST', `/api/chat/parties/${ann.party.id}/desktop/claim`, { token: nodeToken(M2, owner.subject_id), body: { port: 18180 } });
++ assert.strictEqual(wrong.status, 404, 'a machine the pool did not hold cannot claim it');
++ const claim = await h.http('POST', `/api/chat/parties/${ann.party.id}/desktop/claim`, { token: nodeToken(M1, owner.subject_id), body: { port: 18180 } });
++ assert.strictEqual(claim.status, 200, claim.text);
++ assert.ok(claim.body.share_grant);
++ assert.ok(claim.body.ice_servers.some((s) => String(s.urls).startsWith('turns:')));
++ const share = `shr_${ids.ulid()}`;
++ const ready = await h.http('POST', `/api/chat/parties/${ann.party.id}/desktop/ready`, {
++ token: nodeToken(M1, owner.subject_id),
++ body: { share_id: share, url: `https://${ann.party.id}--x.openvibe.link/` },
++ });
++ assert.strictEqual(ready.status, 200, ready.text);
++ const seen = await api('GET', `/parties/${ann.party.id}`, ann);
++ assert.strictEqual(seen.body.party.desktop.state, 'ready');
++ assert.strictEqual(seen.body.party.desktop.where.state, 'ready');
++ assert.match(seen.body.party.desktop.where.line, /Running on Dell tower 1/);
++ const after = await h.http('GET', '/api/chat/parties/desktop/assignment', { token: nodeToken(M1, owner.subject_id) });
++ assert.strictEqual(after.body.assignment, null, 'a claimed party is no longer an assignment');
++});
++
++t('the pool seats three parties, queues the fourth with its place in line, and promotes it when one ends', async () => {
++ bob.party = (await api('POST', '/parties', bob, { desktop_hosting: 'openvibe', title: 'Bob pool party' })).body.party;
++ cat.party = (await api('POST', '/parties', cat, { desktop_hosting: 'openvibe', title: 'Cat pool party' })).body.party;
++ assert.ok(bob.party.desktop.where.state === 'held' && cat.party.desktop.where.state === 'held');
++ const waiting = await api('POST', '/parties', dave, { desktop_hosting: 'openvibe', title: 'Dave pool party' });
++ assert.strictEqual(waiting.status, 201, waiting.text);
++ assert.strictEqual(waiting.body.party.desktop.where.state, 'queued');
++ assert.strictEqual(waiting.body.party.desktop.where.queue_position, 1);
++ assert.match(waiting.body.party.desktop.where.line, /next/);
++ dave.party = waiting.body.party;
++
++ const panel = await api('GET', '/parties/pool', owner);
++ assert.strictEqual(panel.body.queued, 1);
++ assert.strictEqual(panel.body.machines.reduce((n, m) => n + m.free, 0), 0);
++
++ const ended = await api('POST', `/rooms/${bob.party.id}/party/end`, bob);
++ assert.strictEqual(ended.status, 200, ended.text);
++ const swept = await poolModule.sweep();
++ assert.strictEqual(swept.promoted, 1);
++ const seated = await api('GET', `/parties/${dave.party.id}`, dave);
++ assert.strictEqual(seated.body.party.desktop.where.state, 'held');
++ assert.strictEqual(seated.body.party.desktop.where.machine, bob.party.desktop.where.machine, 'the freed machine goes to the oldest room in line');
++});
++
++t('ending a room releases its machine, and a hold nobody renews goes idle and is released', async () => {
++ const ended = await api('POST', `/rooms/${cat.party.id}/party/end`, cat);
++ assert.strictEqual(ended.status, 200, ended.text);
++ let panel = await api('GET', '/parties/pool', owner);
++ assert.strictEqual(panel.body.holds.filter((x) => x.state !== 'released').length, 2, 'the ended room no longer holds a machine');
++
++ // Ann's machine is Ready; make its heartbeat old and sweep: the room loses the desktop and says why.
++ await db.run("UPDATE party_pool_holds SET renewed_at = ? WHERE room_id = (SELECT room_id FROM room_parties WHERE host_subject = ? AND ended_at IS NULL ORDER BY created_at LIMIT 1)", ['2020-01-01T00:00:00.000Z', ann.subject_id]);
++ const released = await poolModule.sweep();
++ assert.ok(released.released >= 1);
++ const seen = await api('GET', `/parties/${ann.party.id}`, ann);
++ assert.strictEqual(seen.body.party.desktop.state, 'stopped');
++ assert.strictEqual(seen.body.party.desktop.where.state, 'released');
++ assert.match(seen.body.party.desktop.where.line, /idle/);
++ panel = await api('GET', '/parties/pool', owner);
++ assert.strictEqual(panel.body.holds.filter((x) => x.state !== 'released').length, 1, 'only Dave keeps a machine');
++});
++
++t('a person has a cap on waiting and running OpenVibe parties', async () => {
++ const first = await api('POST', '/parties', bob, { desktop_hosting: 'openvibe', title: 'Bob one' });
++ const second = await api('POST', '/parties', bob, { desktop_hosting: 'openvibe', title: 'Bob two' });
++ assert.strictEqual(first.status, 201, first.text);
++ assert.strictEqual(second.status, 201, second.text);
++ const third = await api('POST', '/parties', bob, { desktop_hosting: 'openvibe', title: 'Bob three' });
++ assert.strictEqual(third.status, 429, third.text);
++ assert.strictEqual(third.body.code, 'party.pool_limit');
++});
++
++t('placement is tested with a fake registry: capacity, load, queue, promotion and idle release', async () => {
++ assert.ok(poolModule, 'the module is loaded');
++ const A = newLink(), B = newLink();
++ const registry = fakeRegistry([
++ { principal_id: A, owner_subject: owner.subject_id, name: 'Fake one', desktops: 2, can_run: ['desktop'] },
++ { principal_id: B, owner_subject: owner.subject_id, name: 'Fake two', desktops: 1, can_run: ['desktop'] },
++ ]);
++ let clock = Date.now();
++ const pool = poolModule.createPool({ registry, now: () => clock });
++ const actor = () => ids.newId('user');
++ const p1 = await pool.place({ actor: actor(), slug: 'party-fake1' });
++ const p2 = await pool.place({ actor: actor(), slug: 'party-fake2' });
++ const p3 = await pool.place({ actor: actor(), slug: 'party-fake3' });
++ const p4 = await pool.place({ actor: actor(), slug: 'party-fake4' });
++ assert.deepStrictEqual([p1.machine.principal_id, p2.machine.principal_id, p3.machine.principal_id], [A, B, A], 'most free first, then the least loaded');
++ assert.strictEqual(p4.queued, true);
++ assert.strictEqual(p4.position, 1);
++ await pool.cancel(p2.hold.id, 'test');
++ const swept = await pool.sweep();
++ assert.strictEqual(swept.promoted, 1);
++ assert.strictEqual(await pool.where(null), null, 'a fake hold has no room, so no room line');
++ const after = await db.get('SELECT * FROM party_pool_holds WHERE id = ?', [p4.hold.id]);
++ assert.strictEqual(after.state, 'held');
++ assert.strictEqual(after.principal_id, B);
++ // Idle: the machine is held but nobody renewed it for longer than the window.
++ clock += 11 * 60_000;
++ const idle = await pool.sweep();
++ assert.ok(idle.released >= 2, 'the stale holds are released');
++ const rows = await db.all("SELECT * FROM party_pool_holds WHERE slug LIKE 'party-fake%'");
++ assert.ok(rows.every((r) => r.state === 'released'));
++ await db.run("DELETE FROM party_pool_holds WHERE slug LIKE 'party-fake%'");
++});
++
++t('the owner panel is openvibe.party/pool, shows the machines and holds, and is the owner only', async () => {
++ const panel = await page('/pool', owner);
++ assert.strictEqual(panel.status, 200, panel.text.slice(0, 200));
++ assert.match(panel.text, /OpenVibe computers/);
++ assert.match(panel.text, /Dell tower 1/);
++ const nobody = await page('/pool', ann);
++ assert.strictEqual(nobody.status, 403);
++});
++
++t('close', async () => {
++ await h.close();
++});
++
++t.run();
ddiff --git a/test/pg-schema.test.js b/test/pg-schema.test.js
iindex 4e48626..336954f 100644
---- a/test/pg-schema.test.js
++++ b/test/pg-schema.test.js
@@@ -36,6 +36,7 @@ const TABLES = [
'room_sync_clocks', 'room_control_leases',
'room_message_reactions', 'room_member_timeouts', 'config_snapshots', 'config_keys', 'chat_activity',
'agent_play_requests', 'agent_play_votes', 'agent_play_log',
++ 'party_pool_settings', 'party_pool_machines', 'party_pool_holds', // 0064: OpenVibe's own computers for parties
];
const INDEXES = [
'idx_chat_ts_deleted', 'idx_chat_page_live', 'idx_chat_channel_user_ts', 'idx_chat_stream_ts', 'idx_emotes_channel_code', 'idx_chat_tl_dedup',