openvibers/openvibe.chat Commits

A room's shared clock and control lease (B3/B4) (#387)

OP OpenVibers committed 6 days ago · b061d0d6d2 · 7 files, +1188 −2

* Rooms get a shared clock and a control lease (common.sync-clock@1, common.control-lease@1)

* Party foundation review: OpenVibe media only, a minute's seat, the owner hands an AI the remote; the test runs

- The clock's media: an https URL must be on a host a Contracts service manifest names. Party v1 plays OpenVibe
  media, and a holder must not point every viewer's player (and their addresses) at an outside server.
- A person holder whose last socket closes keeps the remote for a minute (a reload, a phone between networks):
  expires_ms becomes at most a minute away; back in time restores their own expiry; after it the remote is free.
- lease.grant to an AI: the room's owner (or chat staff) only; an AI acts with the room's authority.
- test/room-sync-clock.test.js never called t.run(), so it passed having run nothing. It runs now (5 checks), with
  the outside-media refusal, the minute's seat both ways and the guest-to-AI refusal.

Changes

ddiff --git a/docs/room-sync-clock.md b/docs/room-sync-clock.md
nnew file mode 100644
iindex 0000000..39d9f41
---- /dev/null
++++ b/docs/room-sync-clock.md
@@@ -0,0 +1,153 @@
++# A room's shared clock and control lease (B3/B4)
++
++Two common blocks Party stands on, both carried by Chat's room socket (`/ws/chat`, `join_room`):
++
++- **B4 `common.sync-clock@1`** — one timeline for every player in the room.
++- **B3 `common.control-lease@1`** — who holds the room's controls (the remote).
++
++The state shapes are in OpenVibe.Contracts (`contracts/common/sync-clock.v1.json`,
++`contracts/common/control-lease.v1.json`). This file is the Wire protocol and the decisions Chat makes; the browser
++half of B4 is `ov-sync.js` in OpenVibe.Shared.
++
++## Storage: memory and the durable copy
++
++| State | Live copy | Durable copy |
++|---|---|---|
++| Clock | `server/rooms/sync-clock.js`, a `Map` keyed by room id | `room_sync_clocks` (migration `0042_room_shared_state.sql`) |
++| Lease | `server/rooms/control-lease.js`, a `Map` keyed by room id | `room_control_leases` (same migration) |
++
++Every accepted clock set writes the whole row in the same operation that returns the stamped state, so a restart
++reads the latest timeline back. The lease is written on every control change (grant, release, take, expiry); its
++`watchers` are live presence recomputed from the sockets in the room and are deliberately not a durable membership
++list. `seq` is the version of the control change, so a client ignores a state with a lower `seq`; watcher-only
++changes do not bump it.
++
++`evict(roomId)` in either module drops the memory copy and reads the row back; tests use it to prove the durable
++copy.
++
++## B4: the shared clock
++
++State:
++
++```json
++{ "room": "friday-film", "media": "https://media.openvibe.network/vod/friday-film.webm",
++ "position_ms": 61234, "rate": 1, "paused": false, "at_ms": 1794067200123,
++ "set_by": "usr_01JAB2C3D4E5F6G7H8J9K0MNPS", "seq": 7 }
++```
++
++`media` is a `media.media-ref@1` object or an `https` URL on an OpenVibe host (a domain a Contracts service manifest
++names; Party v1 plays OpenVibe media only, so a holder cannot point every viewer's player at an outside server). A player is at
++`position_ms + (serverNow - at_ms) * rate` while `paused` is false, and at `position_ms` while it is true. `at_ms`
++is always Chat's clock when Chat accepted the set; a player measures that clock with `time` (below).
++
++### Setting the clock
++
++Client → Chat:
++
++```json
++{ "type": "sync", "media": "https://…", "position_ms": 61234, "rate": 1, "paused": false }
++```
++
++`media` may be left out to keep the room's current media. `position_ms` may be left out to pause or resume where the
++timeline is now. `rate` and `paused` may be left out to keep their current values. The first set needs `media` and
++`position_ms`.
++
++Chat → room, and to a member when they join:
++
++```json
++{ "type": "sync", "room": "friday-film", "state": { … } }
++```
++
++Only the current **person** holder of the room's control lease may set it. Chat refuses everyone else with:
++
++```json
++{ "type": "sync_error", "room": "friday-film", "code": "sync.not_holder", "message": "Only the person with the remote can move the room" }
++```
++
++Other codes: `sync.sign_in`, `sync.not_joined`, `sync.media`, `sync.position`, `sync.rate`, `sync.paused`, `rate`.
++
++### Clock measurement
++
++Client → Chat (five of them, about 120 ms apart, once on attach and again every 60 s):
++
++```json
++{ "type": "time", "id": "t1-0-ab12cd", "t0": 1794067198123 }
++```
++
++Chat → that socket:
++
++```json
++{ "type": "time", "id": "t1-0-ab12cd", "t0": 1794067198123, "t_server": 1794067198127 }
++```
++
++`ov-sync.js` computes `offset = t_server - (t0 + t3) / 2` for each sample, keeps the five with the smallest round
++trip, and uses the best one. `serverNow()` is `Date.now() + offset`.
++
++## B3: the control lease
++
++State:
++
++```json
++{ "thing": "chat.room:friday-film",
++ "holder": { "kind": "person", "subject": "usr_01JAB2C3D4E5F6G7H8J9K0MNPS" },
++ "since_ms": 1794067200000, "expires_ms": null,
++ "watchers": ["usr_01JAB2C3D4E5F6G7H8J9K0MNPS", "usr_01JAB2C3D4E5F6G7H8J9K0MNPX"], "seq": 4 }
++```
++
++A room is one thing; Chat names it `chat.room:<slug>`. Every room has a lease: a room with no control history is
++`nobody` at `seq` 1, with the signed-in people connected to the room as `watchers`.
++
++### Messages
++
++All on the same room socket, all fanned out to the room by Chat:
++
++| Client sends | Chat broadcasts | Who may send it |
++|---|---|---|
++| `{ "type": "lease.request", "expires_ms": … }` | `{ "type": "lease", "room", "state", "granted": true }` when the controls were free; otherwise `{ "type": "lease.request", "room", "request": { requester, holder, since_ms, expires_ms, seq } }` | any signed-in member |
++| `{ "type": "lease.grant", "to": { "kind": "person", "subject": "usr_…" }, "expires_ms": … }` | `{ "type": "lease", "room", "state", "granted": true }` | the current holder |
++| `{ "type": "lease.release" }` | `{ "type": "lease", "room", "state", "granted": true }` | the current holder |
++| `{ "type": "lease.take", "expires_ms": … }` (also `lease.take_back` and `lease.revoke`) | `{ "type": "lease", "room", "state", "taken": true }` | the room's owner, or chat staff |
++
++`to` may also be `{ "kind": "ai", "subject": "ai:vibe" }` or `{ "kind": "nobody", "subject": null }`. A `person`
++target must be one of the room's current watchers, so a remote never goes to somebody who is not there. Only the
++room's owner (or chat staff) may hand the remote to an `ai` (`lease.not_owner` otherwise): an AI acts with the room's
++authority. Chat has no socket for it, so the owner takes it back to end the AI's turn.
++
++`expires_ms` is an absolute epoch instant within the next 24 hours. At that instant Chat makes the holder `nobody`,
++increments `seq` and broadcasts the state with `"reason": "expired"`. A person holder who leaves the room (their
++last socket closes, or they are removed) keeps the remote for one minute: the state's `expires_ms` becomes at most a
++minute away, and coming back inside it restores their own expiry. After the minute the remote is free, the same way.
++Take-back always wins and needs no consent.
++
++Errors:
++
++```json
++{ "type": "lease_error", "room": "friday-film", "code": "lease.not_holder", "message": "Only the current holder can hand the controls on" }
++```
++
++Codes: `lease.sign_in`, `lease.not_joined`, `lease.not_holder`, `lease.not_owner`, `lease.not_watching`,
++`lease.target`, `lease.expiry`, `rate`.
++
++### Decisions
++
++- **Free controls go to the first request.** When `holder.kind` is `nobody`, `lease.request` grants at once; this is
++ how a party's host picks up the remote on arrival and how the next person takes it after a release.
++- **A held remote is never grabbed by request**, only granted by its holder or taken back by the owner.
++- **Watchers are the room, live.** They are the signed-in subjects connected to the room now; joining or leaving
++ refreshes them, and they do not bump `seq`.
++- **A leave keeps the seat for a minute.** A reload or a phone between networks does not lose the remote; after a
++ minute away it is free, and the owner can always take it at once.
++
++## Client use
++
++```js
++const a = OVSync.attach(video, { room: 'friday-film', socket, self: subject });
++// a.drift() ms ahead of the room (positive = ahead)
++// a.isHolder() whether the control lease says this subject holds the room
++// a.clock() { offset, rtt, ready }
++// a.destroy()
++```
++
++Shared serves the file as `/shared/ov-sync.js` and exports it as `openvibe-shared/ov-sync`. The page opens one
++`socket` that has already sent `join_room` (Chat sends the first `sync` and `lease` frames in answer to that join);
++`self` is optional when the socket's `room_joined` frame carries `subject`.
ddiff --git a/migrations/0042_room_shared_state.sql b/migrations/0042_room_shared_state.sql
nnew file mode 100644
iindex 0000000..370a37b
---- /dev/null
++++ b/migrations/0042_room_shared_state.sql
@@@ -0,0 +1,34 @@
++-- phase: expand
++-- Shared party state (seven domains plan B3/B4, 2026-11-07): one control lease and one sync clock per Chat room.
++--
++-- A room's latest sync clock is kept in memory by server/rooms/sync-clock.js and copied here after every accepted
++-- change; a restart reads it back. A room's control lease is kept the same way in server/rooms/control-lease.js.
++-- Both tables are expand-only: new tables, no existing column changes, and an older release ignores them.
++--
++-- The clock's media is JSON text: either a Media ref ({ media_id, role, variant }) or an https URL string.
++-- The lease's watchers are JSON text too, but they are live presence (the signed-in subjects connected to the room),
++-- not a membership list: they are refreshed on every join and leave and do not increment seq.
++CREATE TABLE IF NOT EXISTS room_sync_clocks (
++ room_id bigint PRIMARY KEY REFERENCES rooms(id) ON DELETE CASCADE,
++ media text COLLATE "C" NOT NULL,
++ position_ms bigint NOT NULL CHECK (position_ms >= 0 AND position_ms <= 2592000000),
++ rate double precision NOT NULL DEFAULT 1 CHECK (rate > 0 AND rate <= 16),
++ paused boolean NOT NULL DEFAULT false,
++ at_ms bigint NOT NULL CHECK (at_ms >= 0),
++ set_by text COLLATE "C" NOT NULL,
++ seq bigint NOT NULL DEFAULT 1 CHECK (seq >= 1),
++ updated_at timestamptz NOT NULL DEFAULT now()
++);
++
++CREATE TABLE IF NOT EXISTS room_control_leases (
++ room_id bigint PRIMARY KEY REFERENCES rooms(id) ON DELETE CASCADE,
++ thing text COLLATE "C" NOT NULL,
++ holder_kind text COLLATE "C" NOT NULL CHECK (holder_kind IN ('person', 'ai', 'nobody')),
++ holder_subject text COLLATE "C",
++ since_ms bigint NOT NULL CHECK (since_ms >= 0),
++ expires_ms bigint CHECK (expires_ms IS NULL OR expires_ms >= since_ms),
++ watchers text NOT NULL DEFAULT '[]',
++ seq bigint NOT NULL DEFAULT 1 CHECK (seq >= 1),
++ updated_at timestamptz NOT NULL DEFAULT now(),
++ CONSTRAINT room_control_leases_holder CHECK ((holder_kind = 'nobody') = (holder_subject IS NULL))
++);
ddiff --git a/server/chat/chat-server.js b/server/chat/chat-server.js
iindex d5d20ff..9092964 100644
---- a/server/chat/chat-server.js
++++ b/server/chat/chat-server.js
@@@ -36,6 +36,8 @@ const dm = require('./dm');
const networkBlocks = require('./network-blocks');
const presence = require('../presence');
const vipBadges = require('../vip/badges');
++const syncClock = require('../rooms/sync-clock');
++const controlLease = require('../rooms/control-lease');
// Cosmetics and tags are Live's (monetization/cosmetics, game/tags); read through live-context.
const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };
const tags = { getTagProfile: (userId) => ctx.getTagProfile(userId) };
@@@ -205,6 +207,11 @@ class ChatServer {
async init(server) {
this.wss = new WebSocket.Server({ noServer: true, maxPayload: 64 * 1024, perMessageDeflate: false });
++ // A room's control lease that runs out announces itself even if nobody is typing.
++ controlLease.setExpiryHandler(({ room, state }) => {
++ if (room && room.slug) this.broadcastToRoom(room.id, { type: 'lease', room: room.slug, state, reason: 'expired' });
++ });
++
// Word filter
wordFilter.load();
@@@ -407,15 +414,19 @@ class ChatServer {
ws.on('close', () => {
release();
++ const room = clientInfo.roomId ? { id: clientInfo.roomId, slug: clientInfo.roomSlug } : null;
this.clients.delete(ws);
this.broadcastUserCount(streamId);
this.broadcastUsersList(streamId);
++ if (room) this._afterPresence(room.id, room.slug).catch(() => {});
});
ws.on('error', (err) => {
console.warn('[Chat] WebSocket error for', ws._clientIp || 'unknown', ':', err.message);
release();
++ const room = clientInfo.roomId ? { id: clientInfo.roomId, slug: clientInfo.roomSlug } : null;
this.clients.delete(ws);
++ if (room) this._afterPresence(room.id, room.slug).catch(() => {});
});
}
@@@ -563,6 +574,7 @@ class ChatServer {
newcomer: this._isNewcomer(client),
username: displayName,
core_username: client.user?.username || null,
++ subject: client.user?.subject_id || null,
role: client.user ? client.user.role : 'anon',
user_id: client.user?.id || null,
slowmode_seconds: streamSlowSec,
@@@ -603,15 +615,100 @@ class ChatServer {
const room = await rooms.bySlug(msg.room);
const a = room ? await rooms.access(room, client.user) : null;
if (!room || !a.read) { this.sendTo(ws, { type: 'room_error', room: msg.room || null, code: 'rooms.not_found', message: 'No such room' }); break; }
++ const previous = client.roomId && Number(client.roomId) !== Number(room.id) ? { id: client.roomId, slug: client.roomSlug } : null;
client.roomId = room.id;
client.roomSlug = room.slug;
-- this.sendTo(ws, { type: 'room_joined', room: room.slug, kind: a.kind, role: a.role, can: { post: a.post, moderate: a.moderate, manage: a.manage, join: a.join, talk: a.talk } });
++ if (previous) await this._afterPresence(previous.id, previous.slug);
++ this.sendTo(ws, { type: 'room_joined', room: room.slug, kind: a.kind, role: a.role, subject: client.user?.subject_id || null, can: { post: a.post, moderate: a.moderate, manage: a.manage, join: a.join, talk: a.talk } });
++ // The shared clock and the control lease are the room's, not the socket's: a member joining
++ // gets the current timeline and the current holder (common.sync-clock@1, common.control-lease@1).
++ await this.sendRoomState(room, ws);
break;
}
-- case 'leave_room':
++ case 'leave_room': {
++ const left = client.roomId ? { id: client.roomId, slug: client.roomSlug } : null;
client.roomId = null;
client.roomSlug = null;
++ if (left) await this._afterPresence(left.id, left.slug);
++ break;
++ }
++ // ── The shared clock (B4) ───────────────────────────────────────
++ // Only the current B3 holder may set it. Chat stamps at_ms and seq, keeps the state in
++ // server/rooms/sync-clock.js (memory + room_sync_clocks) and fans it out to the room.
++ case 'sync': {
++ if (!this._withinRate(client, 'sync', 20, 1000)) { this.sendTo(ws, { type: 'sync_error', room: client.roomSlug || null, code: 'rate', message: 'Too many clock changes; slow down.' }); break; }
++ const room = await this._joinedRoom(client);
++ if (!room) { this.sendTo(ws, { type: 'sync_error', room: client.roomSlug || null, code: 'sync.not_joined', message: 'Open a room first' }); break; }
++ if (!client.user) { this.sendTo(ws, { type: 'sync_error', room: room.slug, code: 'sync.sign_in', message: 'Sign in to change the room\'s clock' }); break; }
++ const rooms = require('../rooms/rooms');
++ const access = await rooms.access(room, client.user);
++ const subject = client.user.subject_id;
++ if (!access.read || !subject) { this.sendTo(ws, { type: 'sync_error', room: room.slug, code: 'sync.sign_in', message: 'Sign in to change the room\'s clock' }); break; }
++ let holder;
++ try { holder = (await controlLease.state(room, this.roomWatchers(room.id))).holder; } catch (err) { this.sendTo(ws, { type: 'sync_error', room: room.slug, code: 'sync.lease', message: 'The room\'s remote could not be read' }); break; }
++ if (holder.kind !== 'person' || holder.subject !== subject) {
++ this.sendTo(ws, { type: 'sync_error', room: room.slug, code: 'sync.not_holder', message: 'Only the person with the remote can move the room' });
++ break;
++ }
++ try {
++ const state = await syncClock.set(room, subject, msg);
++ this.broadcastToRoom(room.id, { type: 'sync', room: room.slug, state });
++ } catch (err) {
++ this.sendTo(ws, { type: 'sync_error', room: room.slug, code: err.code || 'sync.error', message: err.code ? err.message : 'The room\'s clock did not change' });
++ }
++ break;
++ }
++ // NTP-style clock measurement on the same socket: the client sends t0, Chat answers with its
++ // own clock at the same instant. Five samples and the smallest round trip give the offset.
++ case 'time': {
++ if (!this._withinRate(client, 'time', 60, 1000)) break;
++ const id = typeof msg.id === 'string' ? msg.id.slice(0, 80) : (Number.isFinite(Number(msg.id)) ? Number(msg.id) : null);
++ const t0 = Number.isFinite(Number(msg.t0)) ? Number(msg.t0) : null;
++ this.sendTo(ws, { type: 'time', id, t0, t_server: Date.now() });
break;
++ }
++ // ── The control lease (B3) ──────────────────────────────────────
++ case 'lease.request':
++ case 'lease.grant':
++ case 'lease.release':
++ case 'lease.take':
++ case 'lease.take_back':
++ case 'lease.revoke': {
++ if (!this._withinRate(client, 'lease', 10, 1000)) { this.sendTo(ws, { type: 'lease_error', room: client.roomSlug || null, code: 'rate', message: 'Too many remote changes; slow down.' }); break; }
++ const room = await this._joinedRoom(client);
++ if (!room) { this.sendTo(ws, { type: 'lease_error', room: client.roomSlug || null, code: 'lease.not_joined', message: 'Open a room first' }); break; }
++ if (!client.user) { this.sendTo(ws, { type: 'lease_error', room: room.slug, code: 'lease.sign_in', message: 'Sign in to use the room\'s remote' }); break; }
++ const rooms = require('../rooms/rooms');
++ const access = await rooms.access(room, client.user);
++ const subject = client.user.subject_id;
++ if (!access.read || !subject) { this.sendTo(ws, { type: 'lease_error', room: room.slug, code: 'lease.sign_in', message: 'Sign in to use the room\'s remote' }); break; }
++ const watchers = this.roomWatchers(room.id);
++ try {
++ if (msg.type === 'lease.request') {
++ const result = await controlLease.request(room, subject, { watchers, expires_ms: msg.expires_ms });
++ if (result.granted && result.already) this.sendTo(ws, { type: 'lease', room: room.slug, state: result.state });
++ else if (result.granted) this.broadcastToRoom(room.id, { type: 'lease', room: room.slug, state: result.state, granted: true });
++ else this.broadcastToRoom(room.id, { type: 'lease.request', room: room.slug, request: result.request });
++ } else if (msg.type === 'lease.grant') {
++ const to = msg.to || msg.holder;
++ // An AI holder acts with the room's authority, so only the owner (or chat staff) hands it the remote.
++ if (to && to.kind === 'ai' && !access.manage) { this.sendTo(ws, { type: 'lease_error', room: room.slug, code: 'lease.not_owner', message: 'Only the room\'s owner can hand the remote to an AI' }); break; }
++ const result = await controlLease.grant(room, subject, to, { watchers, expires_ms: msg.expires_ms });
++ this.broadcastToRoom(room.id, { type: 'lease', room: room.slug, state: result.state, granted: true });
++ } else if (msg.type === 'lease.release') {
++ const result = await controlLease.release(room, subject, { watchers });
++ this.broadcastToRoom(room.id, { type: 'lease', room: room.slug, state: result.state, granted: true });
++ } else {
++ // Take-back always wins: the room's owner (or chat staff) needs no holder's consent.
++ if (!access.manage) { this.sendTo(ws, { type: 'lease_error', room: room.slug, code: 'lease.not_owner', message: 'Only the room\'s owner can take the remote back' }); break; }
++ const result = await controlLease.take(room, subject, { watchers, expires_ms: msg.expires_ms });
++ this.broadcastToRoom(room.id, { type: 'lease', room: room.slug, state: result.state, taken: true });
++ }
++ } catch (err) {
++ this.sendTo(ws, { type: 'lease_error', room: room.slug, code: err.code || 'lease.error', message: err.code ? err.message : 'The remote did not change' });
++ }
++ break;
++ }
// Support inboxes (server/inbox/inbox.js): a visitor follows their own conversation (its id and their visitor
// token), one of the inbox's people follows the whole inbox.
case 'inbox_watch': {
@@@ -2717,6 +2814,68 @@ class ChatServer {
}
}
++ /** A socket's room, re-read so a vanished room or a changed slug is not trusted from the client. */
++ async _joinedRoom(client) {
++ if (!client.roomId || !client.roomSlug) return null;
++ const room = await require('../rooms/rooms').bySlug(client.roomSlug);
++ return room && Number(room.id) === Number(client.roomId) ? room : null;
++ }
++
++ /** The signed-in subjects in a room now: the control lease's watchers, read from the sockets, not stored. */
++ roomWatchers(roomId) {
++ const subjects = new Set();
++ for (const [, client] of this.clients) {
++ if (client.roomId === roomId && client.user && typeof client.user.subject_id === 'string' && client.user.subject_id) subjects.add(client.user.subject_id);
++ }
++ return [...subjects].sort();
++ }
++
++ /** Per-socket token bucket for the room clock, clock pings and remote messages. */
++ _withinRate(client, key, max, windowMs, now = Date.now()) {
++ if (!client._rateBuckets) client._rateBuckets = new Map();
++ const bucket = client._rateBuckets.get(key);
++ if (!bucket || now - bucket.start >= windowMs) {
++ client._rateBuckets.set(key, { start: now, count: 1 });
++ return true;
++ }
++ bucket.count++;
++ return bucket.count <= max;
++ }
++
++ /** The room's current lease state, with current presence, to everyone following it. */
++ async _broadcastLease(room) {
++ try {
++ const state = await controlLease.state(room, this.roomWatchers(room.id));
++ this.broadcastToRoom(room.id, { type: 'lease', room: room.slug, state });
++ } catch (err) {
++ console.warn('[Lease] state:', err && err.message);
++ }
++ }
++
++ /** A member joined: their socket gets the current clock, the room gets the current lease with them in it. */
++ async sendRoomState(room, ws) {
++ try {
++ const state = await syncClock.get(room);
++ if (state) this.sendTo(ws, { type: 'sync', room: room.slug, state });
++ } catch (err) {
++ console.warn('[Sync] state:', err && err.message);
++ }
++ await this._broadcastLease(room);
++ }
++
++ /** Presence changed (a join, a leave, a removal): refresh the watchers and release a holder who left. */
++ async _afterPresence(roomId, roomSlug = null) {
++ let room = roomSlug ? { id: roomId, slug: roomSlug } : null;
++ if (!room) {
++ const row = await db.get('SELECT id, slug FROM rooms WHERE id = ?', [roomId]);
++ if (!row) return;
++ room = row;
++ }
++ const watchers = this.roomWatchers(room.id);
++ try { await controlLease.reconcile(room, watchers); } catch (err) { console.warn('[Lease] presence:', err && err.message); return; }
++ await this._broadcastLease(room);
++ }
++
/**
* A room's roles or visibility changed: every socket following it learns what it may do now
* (`room_access`), and one that may no longer read it leaves (`room_left`). → sockets told
@@@ -2731,6 +2890,7 @@ class ChatServer {
if (!a.read) { client.roomId = null; client.roomSlug = null; this.sendTo(ws, { type: 'room_left', reason: 'removed' }); continue; }
this.sendTo(ws, { type: 'room_access', room: room.slug, kind: a.kind, role: a.role, can: { post: a.post, moderate: a.moderate, manage: a.manage, join: a.join, talk: a.talk } });
}
++ await this._afterPresence(room.id, room.slug);
return n;
}
@@@ -2743,6 +2903,7 @@ class ChatServer {
this.sendTo(ws, { type: 'room_left', reason: 'removed' });
}
}
++ if (n) this._afterPresence(roomId).catch(() => {});
return n;
}
@@@ -3077,6 +3238,8 @@ class ChatServer {
this.wss.clients.forEach(ws => ws.close());
this.wss.close();
}
++ controlLease.close();
++ syncClock.close();
}
/**
ddiff --git a/server/rooms/control-lease.js b/server/rooms/control-lease.js
nnew file mode 100644
iindex 0000000..e4ea623
---- /dev/null
++++ b/server/rooms/control-lease.js
@@@ -0,0 +1,403 @@
++'use strict';
++/**
++ * A Chat room's control lease (seven domains plan B3, common.control-lease@1).
++ *
++ * { thing, holder: { kind: person|ai|nobody, subject }, since_ms, expires_ms?, watchers[], seq }
++ *
++ * The room is the thing: Chat names it chat.room:<slug>, so a room socket carries exactly one lease. Chat checks who
++ * may send what; this module decides and stores the state:
++ *
++ * request a signed-in member asks for the controls. Free controls (nobody) go to the requester at once; otherwise
++ * the room hears the request and the holder may grant it.
++ * grant the holder hands the controls to a person in the room, an AI, or nobody. A person target must be
++ * watching, so a lease never goes to somebody who is not there.
++ * release the holder gives the controls up (nobody).
++ * take the room's owner (or chat staff) takes the controls back; it always wins and needs no holder's consent.
++ * expire at expires_ms the holder becomes nobody and seq increments; a timer broadcasts that even if the room is
++ * quiet, and a lazy check covers a restart.
++ *
++ * The live copy is the module's Map; room_control_leases is the durable copy. Watchers are live presence from the
++ * room's sockets (server/chat/chat-server.js) and do not increment seq.
++ */
++const db = require('../db/database');
++
++const MAX_EXPIRY_MS = 24 * 60 * 60 * 1000;
++// A person holder whose last socket closed keeps the remote this long (a reload, a phone between networks); after it
++// the remote is free. Coming back inside it restores the holder's own expiry.
++let AWAY_GRACE_MS = 60 * 1000;
++/** Tests shorten the minute; nothing else calls this. */
++function setAwayGraceForTests(ms) { AWAY_GRACE_MS = Math.max(1, Number(ms) || 60 * 1000); }
++const MAX_WATCHERS = 500;
++const KINDS = ['person', 'ai', 'nobody'];
++
++class LeaseError extends Error {
++ constructor(code, message) { super(message || code); this.code = code; }
++}
++
++const _cache = new Map(); // roomId → state
++const _timers = new Map(); // roomId → expiry timer
++const _meta = new Map(); // roomId → { id, slug }
++const _away = new Map(); // roomId → { subject, expires_ms } the holder's own expiry while their seat waits
++const _queues = new Map(); // roomId → tail of that room's operations
++let _onExpire = null;
++
++function serial(roomId, fn) {
++ const previous = _queues.get(roomId) || Promise.resolve();
++ const run = previous.then(fn, fn);
++ const tail = run.catch(() => {});
++ _queues.set(roomId, tail);
++ tail.finally(() => { if (_queues.get(roomId) === tail) _queues.delete(roomId); });
++ return run;
++}
++
++function thingFor(room) { return `chat.room:${room.slug}`; }
++
++function snapshot(state) {
++ if (!state) return null;
++ return {
++ thing: state.thing,
++ holder: { ...state.holder },
++ since_ms: state.since_ms,
++ expires_ms: state.expires_ms == null ? null : state.expires_ms,
++ watchers: state.watchers.slice(),
++ seq: state.seq,
++ };
++}
++
++function watchersOf(list) {
++ if (!Array.isArray(list)) return [];
++ const out = [];
++ for (const value of list) {
++ if (typeof value !== 'string') continue;
++ const subject = value.trim().slice(0, 200);
++ if (subject && !out.includes(subject)) out.push(subject);
++ if (out.length >= MAX_WATCHERS) break;
++ }
++ return out.sort();
++}
++
++function sameWatchers(a, b) {
++ if (a.length !== b.length) return false;
++ for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false;
++ return true;
++}
++
++function rowState(row) {
++ let watchers;
++ try { watchers = JSON.parse(row.watchers || '[]'); } catch (e) { watchers = []; }
++ return {
++ thing: row.thing,
++ holder: {
++ kind: row.holder_kind,
++ subject: row.holder_subject == null ? null : row.holder_subject,
++ },
++ since_ms: Number(row.since_ms),
++ expires_ms: row.expires_ms == null ? null : Number(row.expires_ms),
++ watchers: watchersOf(watchers),
++ seq: Number(row.seq),
++ };
++}
++
++function defaultState(room, now) {
++ return {
++ thing: thingFor(room),
++ holder: { kind: 'nobody', subject: null },
++ since_ms: now,
++ expires_ms: null,
++ watchers: [],
++ seq: 1,
++ };
++}
++
++async function load(roomId) {
++ const cached = _cache.get(roomId);
++ if (cached) return cached;
++ const row = await db.get('SELECT * FROM room_control_leases WHERE room_id = ?', [roomId]);
++ if (!row) return null;
++ const state = rowState(row);
++ _cache.set(roomId, state);
++ return state;
++}
++
++async function persist(roomId, state) {
++ await db.run(
++ `INSERT INTO room_control_leases (room_id, thing, holder_kind, holder_subject, since_ms, expires_ms, watchers, seq)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT (room_id) DO UPDATE SET
++ thing = excluded.thing, holder_kind = excluded.holder_kind, holder_subject = excluded.holder_subject,
++ since_ms = excluded.since_ms, expires_ms = excluded.expires_ms, watchers = excluded.watchers,
++ seq = excluded.seq, updated_at = now()`,
++ [roomId, state.thing, state.holder.kind, state.holder.subject, state.since_ms, state.expires_ms, JSON.stringify(state.watchers), state.seq]
++ );
++}
++
++function remember(room) {
++ _meta.set(room.id, { id: room.id, slug: room.slug });
++}
++
++function expired(state, now) {
++ return state.holder.kind !== 'nobody' && state.expires_ms != null && state.expires_ms <= now;
++}
++
++function becomeNobody(state, now) {
++ return {
++ ...state,
++ holder: { kind: 'nobody', subject: null },
++ since_ms: now,
++ expires_ms: null,
++ seq: state.seq + 1,
++ };
++}
++
++function expiryOf(value, now) {
++ if (value === undefined || value === null) return null;
++ const n = Number(value);
++ if (!Number.isInteger(n) || n <= now || n > now + MAX_EXPIRY_MS) {
++ throw new LeaseError('lease.expiry', 'expires_ms is a time in the next 24 hours, in epoch milliseconds');
++ }
++ return n;
++}
++
++function targetOf(to) {
++ if (!to || typeof to !== 'object') throw new LeaseError('lease.target', 'A hand-off names a holder');
++ const kind = String(to.kind || '');
++ if (!KINDS.includes(kind)) throw new LeaseError('lease.target', 'A holder is a person, an ai or nobody');
++ if (kind === 'nobody') return { kind, subject: null };
++ const subject = typeof to.subject === 'string' ? to.subject.trim().slice(0, 200) : '';
++ if (!subject) throw new LeaseError('lease.target', 'A person or AI holder needs a subject');
++ return { kind, subject };
++}
++
++function schedule(roomId, state) {
++ const previous = _timers.get(roomId);
++ if (previous) clearTimeout(previous);
++ _timers.delete(roomId);
++ if (state.holder.kind === 'nobody' || state.expires_ms == null) return;
++ const delay = Math.max(0, state.expires_ms - Date.now()) + 5;
++ const timer = setTimeout(() => expire(roomId), delay);
++ if (timer.unref) timer.unref();
++ _timers.set(roomId, timer);
++}
++
++async function expire(roomId) {
++ let next = null;
++ try {
++ next = await serial(roomId, async () => {
++ const state = await load(roomId);
++ const now = Date.now();
++ if (!state || !expired(state, now)) return null;
++ const fresh = becomeNobody(state, now);
++ await persist(roomId, fresh);
++ _cache.set(roomId, fresh);
++ schedule(roomId, fresh);
++ return fresh;
++ });
++ } catch (err) {
++ console.warn('[Lease] expiry failed:', err && err.message);
++ return;
++ }
++ if (next && _onExpire) {
++ try { _onExpire({ room: _meta.get(roomId) || { id: roomId, slug: null }, state: snapshot(next) }); } catch (e) { /* the server's handler */ }
++ }
++}
++
++/** Normalise expiry and the person-holder-is-watching rule after a presence or time change. */
++async function current(room, watchers, now = Date.now()) {
++ remember(room);
++ return await serial(room.id, async () => {
++ let state = await load(room.id);
++ let changed = false;
++ let holderChanged = false;
++ if (!state) {
++ state = defaultState(room, now);
++ changed = true;
++ } else if (expired(state, now)) {
++ state = becomeNobody(state, now);
++ changed = holderChanged = true;
++ }
++ const list = watchers === undefined ? state.watchers : watchersOf(watchers);
++ if (!sameWatchers(state.watchers, list)) { state = { ...state, watchers: list }; changed = true; }
++ const away = _away.get(room.id);
++ if (state.holder.kind === 'person' && !state.watchers.includes(state.holder.subject)) {
++ // Gone: the seat waits AWAY_GRACE_MS (as an expiry, so a restart and the timer both honour it).
++ if (!away || away.subject !== state.holder.subject) {
++ _away.set(room.id, { subject: state.holder.subject, expires_ms: state.expires_ms });
++ const until = now + AWAY_GRACE_MS;
++ state = { ...state, expires_ms: state.expires_ms == null ? until : Math.min(state.expires_ms, until) };
++ changed = holderChanged = true;
++ }
++ } else if (away) {
++ // Back inside the minute (or the holder changed): the holder's own expiry again.
++ _away.delete(room.id);
++ if (state.holder.kind === 'person' && state.holder.subject === away.subject) {
++ state = { ...state, expires_ms: away.expires_ms };
++ changed = holderChanged = true;
++ }
++ }
++ if (holderChanged) await persist(room.id, state);
++ _cache.set(room.id, state);
++ schedule(room.id, state);
++ return { state, changed };
++ });
++}
++
++/** The room's current lease (never null; a room with no control history is nobody at seq 1). */
++async function state(room, watchers, now = Date.now()) {
++ return snapshot((await current(room, watchers, now)).state);
++}
++
++/**
++ * Ask for the controls. When nobody holds them the requester gets them at once; otherwise the answer carries the
++ * pending request for the holder (and the room) to see.
++ */
++async function request(room, subject, { watchers, expires_ms: expiresRaw, now = Date.now() } = {}) {
++ remember(room);
++ const expires = expiryOf(expiresRaw, now);
++ return await serial(room.id, async () => {
++ let entry = await load(room.id);
++ if (!entry) entry = defaultState(room, now);
++ if (expired(entry, now)) entry = becomeNobody(entry, now);
++ const list = watchersOf(watchers === undefined ? entry.watchers : watchers);
++ entry = { ...entry, watchers: list };
++ if (entry.holder.kind === 'person' && entry.holder.subject === subject) {
++ _cache.set(room.id, entry);
++ schedule(room.id, entry);
++ return { state: snapshot(entry), granted: true, already: true };
++ }
++ if (entry.holder.kind !== 'nobody') {
++ _cache.set(room.id, entry);
++ schedule(room.id, entry);
++ return {
++ state: snapshot(entry),
++ granted: false,
++ request: {
++ room: room.slug,
++ requester: { kind: 'person', subject },
++ holder: { ...entry.holder },
++ since_ms: now,
++ expires_ms: expires,
++ seq: entry.seq,
++ },
++ };
++ }
++ const next = {
++ ...entry,
++ holder: { kind: 'person', subject },
++ since_ms: now,
++ expires_ms: expires,
++ seq: entry.seq + 1,
++ };
++ await persist(room.id, next);
++ _cache.set(room.id, next);
++ schedule(room.id, next);
++ return { state: snapshot(next), granted: true };
++ });
++}
++
++/** The holder hands the controls on. A person target must be one of the room's watchers. */
++async function grant(room, actor, to, { watchers, expires_ms: expiresRaw, now = Date.now() } = {}) {
++ remember(room);
++ const target = targetOf(to);
++ const expires = target.kind === 'nobody' ? null : expiryOf(expiresRaw, now);
++ return await serial(room.id, async () => {
++ let entry = await load(room.id);
++ if (!entry) entry = defaultState(room, now);
++ if (expired(entry, now)) entry = becomeNobody(entry, now);
++ const list = watchersOf(watchers === undefined ? entry.watchers : watchers);
++ entry = { ...entry, watchers: list };
++ if (entry.holder.kind !== 'person' || entry.holder.subject !== actor) {
++ throw new LeaseError('lease.not_holder', 'Only the current holder can hand the controls on');
++ }
++ if (target.kind === 'person' && !list.includes(target.subject)) {
++ throw new LeaseError('lease.not_watching', 'That person is not in the room');
++ }
++ const next = {
++ ...entry,
++ holder: target,
++ since_ms: now,
++ expires_ms: expires,
++ seq: entry.seq + 1,
++ };
++ await persist(room.id, next);
++ _cache.set(room.id, next);
++ schedule(room.id, next);
++ return { state: snapshot(next) };
++ });
++}
++
++/** The holder gives the controls up. */
++async function release(room, actor, { watchers, now = Date.now() } = {}) {
++ remember(room);
++ return await serial(room.id, async () => {
++ let entry = await load(room.id);
++ if (!entry) entry = defaultState(room, now);
++ if (expired(entry, now)) entry = becomeNobody(entry, now);
++ const list = watchersOf(watchers === undefined ? entry.watchers : watchers);
++ entry = { ...entry, watchers: list };
++ if (entry.holder.kind !== 'person' || entry.holder.subject !== actor) {
++ throw new LeaseError('lease.not_holder', 'Only the current holder can release the controls');
++ }
++ const next = becomeNobody(entry, now);
++ await persist(room.id, next);
++ _cache.set(room.id, next);
++ schedule(room.id, next);
++ return { state: snapshot(next) };
++ });
++}
++
++/** Take the controls back. Chat checks the room owner (or staff); this always wins, whatever the holder is doing. */
++async function take(room, actor, { watchers, expires_ms: expiresRaw, now = Date.now() } = {}) {
++ remember(room);
++ const expires = expiryOf(expiresRaw, now);
++ return await serial(room.id, async () => {
++ let entry = await load(room.id);
++ if (!entry) entry = defaultState(room, now);
++ const list = watchersOf(watchers === undefined ? entry.watchers : watchers);
++ const next = {
++ ...entry,
++ holder: { kind: 'person', subject: String(actor) },
++ since_ms: now,
++ expires_ms: expires,
++ watchers: list,
++ seq: entry.seq + 1,
++ };
++ await persist(room.id, next);
++ _cache.set(room.id, next);
++ schedule(room.id, next);
++ return { state: snapshot(next) };
++ });
++}
++
++/** Presence changed: refresh watchers, and release a person holder who left the room. */
++async function reconcile(room, watchers, now = Date.now()) {
++ return await current(room, watchers, now);
++}
++
++/** Where the expiry broadcast goes (the Chat server attaches a listener once). */
++function setExpiryHandler(fn) { _onExpire = typeof fn === 'function' ? fn : null; }
++
++/** Forget a room's live copy; the next read comes from room_control_leases (tests and recovery). */
++function evict(roomId) {
++ _cache.delete(roomId);
++ _away.delete(roomId);
++ const timer = _timers.get(roomId);
++ if (timer) clearTimeout(timer);
++ _timers.delete(roomId);
++}
++
++function close() {
++ for (const timer of _timers.values()) clearTimeout(timer);
++ _timers.clear();
++ _cache.clear();
++ _queues.clear();
++ _meta.clear();
++ _away.clear();
++ _onExpire = null;
++}
++
++module.exports = {
++ MAX_EXPIRY_MS, MAX_WATCHERS, KINDS, LeaseError,
++ thingFor, state, request, grant, release, take, reconcile,
++ setExpiryHandler, evict, close,
++ _watchersOf: watchersOf, setAwayGraceForTests,
++};
ddiff --git a/server/rooms/sync-clock.js b/server/rooms/sync-clock.js
nnew file mode 100644
iindex 0000000..917a563
---- /dev/null
++++ b/server/rooms/sync-clock.js
@@@ -0,0 +1,209 @@
++'use strict';
++/**
++ * A Chat room's shared clock (seven domains plan B4, common.sync-clock@1).
++ *
++ * One state per room: { room, media, position_ms, rate, paused, at_ms, set_by, seq }. Chat stamps at_ms and seq and
++ * broadcasts the state; server/chat/chat-server.js refuses a set from anyone but the current control-lease holder.
++ * The latest state lives in the module's Map (the live copy) and in room_sync_clocks (the durable copy). The cache is
++ * per process, like Chat's other room state; a second process reads the row from the database.
++ *
++ * Media is a Media ref or an https URL on an OpenVibe host (a domain one of the Contracts service manifests names).
++ * Party v1 plays OpenVibe media only: a holder must not be able to point every viewer's player, and so their
++ * addresses, at a server outside the platform. A set may leave media out to keep the room's current one, and may
++ * leave position_ms out to pause or resume where the timeline is now.
++ */
++const db = require('../db/database');
++
++const MAX_POSITION_MS = 30 * 24 * 60 * 60 * 1000; // the contract's 30-day cap
++const MAX_RATE = 16;
++const MEDIA_ID = /^(med_[0-9A-HJKMNP-TV-Z]{26}|legacy:[a-z][a-z0-9-]{1,39}:(vod|clip|file|paste|thumbnail|avatar):[A-Za-z0-9._/-]{1,200})$/;
++const HTTP_URL = /^https:\/\/[^\s]{1,2048}$/;
++const ROLE = /^[a-z][a-z0-9_]{1,39}$/;
++const VARIANT = /^[a-z0-9][a-z0-9_-]{0,39}$/;
++
++/** The platform's own hosts: every domain the pinned Contracts service manifests name (exact names, no wildcard). */
++let _hosts = null;
++function firstPartyHosts() {
++ if (_hosts) return _hosts;
++ const fs = require('fs');
++ const path = require('path');
++ const out = new Set();
++ try {
++ const dir = path.join(path.dirname(require.resolve('openvibe-contracts/package.json')), 'manifests', 'services');
++ for (const f of fs.readdirSync(dir)) {
++ if (!f.endsWith('.json')) continue;
++ const m = JSON.parse(fs.readFileSync(path.join(dir, f), 'utf8'));
++ for (const d of Array.isArray(m.domains) ? m.domains : []) if (typeof d === 'string') out.add(d.toLowerCase());
++ }
++ } catch (err) {
++ console.warn('[Sync] the Contracts manifests could not be read; only Media refs are accepted:', err && err.message);
++ }
++ _hosts = out;
++ return out;
++}
++
++function onOpenVibe(url) {
++ try {
++ const u = new URL(url);
++ return u.protocol === 'https:' && !u.username && !u.password && !u.port && firstPartyHosts().has(u.hostname.toLowerCase());
++ } catch (e) { return false; }
++}
++
++class SyncError extends Error {
++ constructor(code, message) { super(message || code); this.code = code; }
++}
++
++const _cache = new Map(); // roomId → state
++const _queues = new Map(); // roomId → the tail of that room's set operations
++
++/** Run fn after the earlier operation of this room, so two holders' sockets cannot write the same seq. */
++function serial(roomId, fn) {
++ const previous = _queues.get(roomId) || Promise.resolve();
++ const run = previous.then(fn, fn);
++ const tail = run.catch(() => {});
++ _queues.set(roomId, tail);
++ tail.finally(() => { if (_queues.get(roomId) === tail) _queues.delete(roomId); });
++ return run;
++}
++
++function snapshot(state) {
++ if (!state) return null;
++ return {
++ room: state.room,
++ media: typeof state.media === 'object' && state.media ? { ...state.media } : state.media,
++ position_ms: state.position_ms,
++ rate: state.rate,
++ paused: state.paused,
++ at_ms: state.at_ms,
++ set_by: state.set_by,
++ seq: state.seq,
++ };
++}
++
++function rowState(row, room) {
++ let media;
++ try { media = JSON.parse(row.media); } catch (e) { media = null; }
++ if (media == null) throw new SyncError('sync.corrupt', 'The stored clock is not readable');
++ return {
++ room: room.slug,
++ media,
++ position_ms: Number(row.position_ms),
++ rate: Number(row.rate),
++ paused: !!row.paused,
++ at_ms: Number(row.at_ms),
++ set_by: row.set_by,
++ seq: Number(row.seq),
++ };
++}
++
++function mediaOf(value) {
++ if (typeof value === 'string') {
++ if (!HTTP_URL.test(value)) throw new SyncError('sync.media', 'media is an https URL or a Media ref');
++ if (!onOpenVibe(value)) throw new SyncError('sync.media', 'a party plays OpenVibe media: the link must be on an OpenVibe site');
++ return value;
++ }
++ if (!value || typeof value !== 'object' || Array.isArray(value)) {
++ throw new SyncError('sync.media', 'media is an https URL or a Media ref');
++ }
++ const keys = Object.keys(value);
++ if (keys.some((k) => !['media_id', 'role', 'variant'].includes(k))) throw new SyncError('sync.media', 'media has only media_id, role and variant');
++ if (!MEDIA_ID.test(String(value.media_id || ''))) throw new SyncError('sync.media', 'media_id is a Media ref');
++ if (value.role !== undefined && !ROLE.test(String(value.role))) throw new SyncError('sync.media', 'media role is a lowercase name');
++ if (value.variant !== undefined && !VARIANT.test(String(value.variant))) throw new SyncError('sync.media', 'media variant is lowercase letters, numbers, _ or -');
++ return {
++ media_id: value.media_id,
++ ...(value.role !== undefined ? { role: value.role } : {}),
++ ...(value.variant !== undefined ? { variant: value.variant } : {}),
++ };
++}
++
++function positionOf(value, previous, now) {
++ if (value !== undefined) {
++ const n = Number(value);
++ if (!Number.isInteger(n) || n < 0 || n > MAX_POSITION_MS) throw new SyncError('sync.position', 'position_ms is whole milliseconds from 0 to 30 days');
++ return n;
++ }
++ if (!previous) throw new SyncError('sync.position', 'the first clock needs position_ms');
++ if (previous.paused) return previous.position_ms;
++ return Math.max(0, Math.round(previous.position_ms + (now - previous.at_ms) * previous.rate));
++}
++
++function rateOf(value, previous) {
++ if (value === undefined) return previous ? previous.rate : 1;
++ const n = Number(value);
++ if (!Number.isFinite(n) || n <= 0 || n > MAX_RATE) throw new SyncError('sync.rate', `rate is a number above 0 and at most ${MAX_RATE}`);
++ return n;
++}
++
++function pausedOf(value, previous) {
++ if (value === undefined) return previous ? previous.paused : false;
++ if (typeof value !== 'boolean') throw new SyncError('sync.paused', 'paused is true or false');
++ return value;
++}
++
++async function load(room) {
++ const cached = _cache.get(room.id);
++ if (cached) return cached;
++ const row = await db.get('SELECT * FROM room_sync_clocks WHERE room_id = ?', [room.id]);
++ if (!row) return null;
++ const state = rowState(row, room);
++ _cache.set(room.id, state);
++ return state;
++}
++
++/** The room's current clock, or null before the first set. */
++async function get(room) {
++ return snapshot(await load(room));
++}
++
++/**
++ * Accept one change. `input` is the frame's media, position_ms, rate, paused (all but position may be left out; a
++ * missing position pauses/resumes where the timeline is now). Chat has already checked the caller holds the lease.
++ */
++async function set(room, subject, input, { now = Date.now() } = {}) {
++ return await serial(room.id, async () => {
++ const previous = await load(room);
++ const body = input && typeof input === 'object' ? input : {};
++ const media = body.media === undefined ? (previous ? previous.media : null) : mediaOf(body.media);
++ if (media == null) throw new SyncError('sync.media', 'the first clock needs media');
++ const position = positionOf(body.position_ms, previous, now);
++ const rate = rateOf(body.rate, previous);
++ const paused = pausedOf(body.paused, previous);
++ const state = {
++ room: room.slug,
++ media,
++ position_ms: position,
++ rate,
++ paused,
++ at_ms: now,
++ set_by: String(subject),
++ seq: (previous ? previous.seq : 0) + 1,
++ };
++ await db.run(
++ `INSERT INTO room_sync_clocks (room_id, media, position_ms, rate, paused, at_ms, set_by, seq)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?)
++ ON CONFLICT (room_id) DO UPDATE SET
++ media = excluded.media, position_ms = excluded.position_ms, rate = excluded.rate,
++ paused = excluded.paused, at_ms = excluded.at_ms, set_by = excluded.set_by,
++ seq = excluded.seq, updated_at = now()`,
++ [room.id, JSON.stringify(media), position, rate, paused, now, state.set_by, state.seq]
++ );
++ _cache.set(room.id, state);
++ return snapshot(state);
++ });
++}
++
++/**
++ * Forget the in-memory copy, as a restart or a failover would: the next read comes from room_sync_clocks. Tests use
++ * it to prove the durable copy; the server uses it if a room's row is changed outside this process.
++ */
++function evict(roomId) { _cache.delete(roomId); }
++
++/** Drop every cache and queue (a test/server restart). */
++function close() { _cache.clear(); _queues.clear(); }
++
++module.exports = {
++ MAX_POSITION_MS, MAX_RATE, SyncError, set, get, evict, close,
++ // Exposed for the WS layer and its tests; the validation stays here.
++ _mediaOf: mediaOf, _positionOf: positionOf, _rateOf: rateOf, _pausedOf: pausedOf,
++};
ddiff --git a/test/pg-schema.test.js b/test/pg-schema.test.js
iindex b419821..160f8fd 100644
---- a/test/pg-schema.test.js
++++ b/test/pg-schema.test.js
@@@ -31,6 +31,7 @@ const TABLES = [
'network_blocks', 'account_data_events', 'ticket_conversations', 'ticket_messages', 'telemetry_samples',
'chat_app_link_tokens', 'chat_app_settings', 'chat_app_callbacks', 'chat_app_deliveries', 'chat_app_rates',
'agent_room_reads', 'agent_room_polls', 'agent_room_poll_votes', 'agent_room_identities', 'agent_room_controls', 'room_parties',
++ 'room_sync_clocks', 'room_control_leases',
'room_message_reactions', 'room_member_timeouts', 'config_snapshots', 'config_keys',
'agent_play_requests', 'agent_play_votes', 'agent_play_log',
];
ddiff --git a/test/room-sync-clock.test.js b/test/room-sync-clock.test.js
nnew file mode 100644
iindex 0000000..f6ba16f
---- /dev/null
++++ b/test/room-sync-clock.test.js
@@@ -0,0 +1,223 @@
++'use strict';
++/**
++ * A room's shared clock (B4, common.sync-clock@1) and control lease (B3, common.control-lease@1):
++ *
++ * - only the current lease holder may set the clock, and Chat refuses everyone else;
++ * - Chat stamps at_ms and seq, fans the state out and gives it to a member who joins later;
++ * - `time` answers a ping with Chat's clock on the same socket;
++ * - the first request takes a free remote, the holder passes it on, take-back always wins, a time limit
++ * expires into nobody, and a person holder who leaves keeps the remote a minute (back in time: still theirs);
++ * - the clock plays OpenVibe media only, and only the owner hands the remote to an AI;
++ * - the durable rows survive dropping the modules' memory copies.
++ */
++const assert = require('assert');
++const { ids } = require('openvibe-contracts');
++const { boot, suite } = require('./helpers');
++
++const t = suite('room sync clock and control lease');
++let h, ann, bob, cat;
++let room;
++const api = (method, path, u, body) => h.http(method, `/api/chat/rooms${path}`, { token: u && u.token, body });
++const close = (ws) => ws.close();
++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
++
++async function join(ws, slug = 'watch-party') {
++ ws.sendJson({ type: 'join_room', room: slug });
++ const joined = await ws.next((m) => m.type === 'room_joined');
++ return joined;
++}
++
++/** The lease frame that follows a join (or an action); the prior frames stay available to next(). */
++const lease = (ws, pred = () => true) => ws.next((m) => m.type === 'lease' && pred(m.state));
++const sync = (ws) => ws.next((m) => m.type === 'sync');
++
++t('boot', async () => {
++ h = await boot();
++ ann = h.addUser('ann', { subject: ids.newId('user') });
++ bob = h.addUser('bob', { subject: ids.newId('user') });
++ cat = h.addUser('cat', { subject: ids.newId('user') });
++ for (const u of [ann, bob, cat]) await h.ctx.upsertUser(h.live.users.get(u.id));
++});
++
++t('a room has nobody holding the remote at seq 1, and a signed-out socket cannot set anything', async () => {
++ const made = await api('POST', '/', ann, { name: 'Watch Party', topic: 'Friday film' });
++ assert.strictEqual(made.status, 201, made.text);
++ room = made.body.room;
++
++ const annWs = await h.ws({ token: ann.token });
++ const joined = await join(annWs);
++ assert.strictEqual(joined.role, 'owner');
++ assert.strictEqual(joined.subject, ann.subject_id);
++ const first = await lease(annWs);
++ assert.deepStrictEqual([first.state.thing, first.state.holder.kind, first.state.holder.subject, first.state.seq],
++ [`chat.room:${room.slug}`, 'nobody', null, 1]);
++
++ const guest = await h.ws({});
++ await join(guest);
++ guest.sendJson({ type: 'sync', media: 'https://openvibe.media/vod/test.webm', position_ms: 0 });
++ const refused = await guest.next((m) => m.type === 'sync_error');
++ assert.strictEqual(refused.code, 'sync.sign_in');
++ guest.sendJson({ type: 'lease.request' });
++ assert.strictEqual((await guest.next((m) => m.type === 'lease_error')).code, 'lease.sign_in');
++ close(guest);
++ close(annWs);
++});
++
++t('the first request takes the free remote, and only its holder can set the clock', async () => {
++ const annWs = await h.ws({ token: ann.token });
++ const bobWs = await h.ws({ token: bob.token });
++ await join(annWs); await lease(annWs);
++ await join(bobWs); await lease(bobWs);
++
++ bobWs.sendJson({ type: 'sync', media: 'https://openvibe.media/vod/test.webm', position_ms: 1000 });
++ assert.strictEqual((await bobWs.next((m) => m.type === 'sync_error')).code, 'sync.not_holder');
++
++ annWs.sendJson({ type: 'lease.request' });
++ const granted = await lease(bobWs, (s) => s.holder.kind === 'person');
++ assert.strictEqual(granted.state.holder.subject, ann.subject_id);
++ assert.strictEqual(granted.state.seq, 2);
++ assert.deepStrictEqual(granted.state.watchers, [ann.subject_id, bob.subject_id].sort());
++ assert.ok((await lease(annWs, (s) => s.holder.subject === ann.subject_id)).granted, 'the requester is told too');
++
++ annWs.sendJson({ type: 'sync', media: 'https://openvibe.media/vod/test.webm', position_ms: 60000, rate: 1, paused: false });
++ const clock = await sync(bobWs);
++ assert.strictEqual(clock.room, room.slug);
++ assert.deepStrictEqual([clock.state.media, clock.state.position_ms, clock.state.rate, clock.state.paused],
++ ['https://openvibe.media/vod/test.webm', 60000, 1, false]);
++ assert.strictEqual(clock.state.set_by, ann.subject_id);
++ assert.strictEqual(clock.state.seq, 1);
++ assert.ok(Math.abs(clock.state.at_ms - Date.now()) < 2000, 'at_ms is the server clock');
++ assert.ok((await sync(annWs)).state.seq === 1);
++
++ // A later member gets both the current timeline and the current holder.
++ const catWs = await h.ws({ token: cat.token });
++ await join(catWs);
++ const lateSync = await sync(catWs);
++ assert.deepStrictEqual([lateSync.state.position_ms, lateSync.state.seq, lateSync.state.set_by], [60000, 1, ann.subject_id]);
++ const lateLease = await lease(catWs);
++ assert.strictEqual(lateLease.state.holder.subject, ann.subject_id);
++ assert.deepStrictEqual(lateLease.state.watchers, [ann.subject_id, bob.subject_id, cat.subject_id].sort());
++
++ // A request on a held remote is a request, not a seizure.
++ bobWs.sendJson({ type: 'lease.request', expires_ms: Date.now() + 60000 });
++ const pending = await annWs.next((m) => m.type === 'lease.request');
++ assert.deepStrictEqual([pending.request.requester.subject, pending.request.holder.subject], [bob.subject_id, ann.subject_id]);
++ assert.strictEqual(pending.request.seq, lateLease.state.seq, 'a request is not a control change');
++ assert.ok((await bobWs.next((m) => m.type === 'lease.request')).request, 'the requester sees it too');
++
++ // The holder passes it on; the new holder sets.
++ annWs.sendJson({ type: 'lease.grant', to: { kind: 'person', subject: bob.subject_id }, expires_ms: Date.now() + 60000 });
++ const passed = await lease(catWs, (s) => s.holder.subject === bob.subject_id);
++ assert.strictEqual(passed.state.seq, lateLease.state.seq + 1);
++ bobWs.sendJson({ type: 'sync', position_ms: 90000, rate: 1.5 });
++ const changed = await sync(catWs);
++ assert.deepStrictEqual([changed.state.position_ms, changed.state.rate, changed.state.set_by, changed.state.seq], [90000, 1.5, bob.subject_id, 2]);
++
++ // A member cannot take it; the owner can, and take-back always wins.
++ bobWs.sendJson({ type: 'lease.take' });
++ assert.strictEqual((await bobWs.next((m) => m.type === 'lease_error')).code, 'lease.not_owner');
++ annWs.sendJson({ type: 'lease.take' });
++ const back = await lease(catWs, (s) => s.holder.subject === ann.subject_id);
++ assert.ok(back.taken);
++ bobWs.sendJson({ type: 'sync', position_ms: 1000 });
++ assert.strictEqual((await bobWs.next((m) => m.type === 'sync_error')).code, 'sync.not_holder');
++
++ // A time limit expires into nobody even if the room is quiet.
++ annWs.sendJson({ type: 'lease.grant', to: { kind: 'person', subject: bob.subject_id }, expires_ms: Date.now() + 250 });
++ await lease(catWs, (s) => s.holder.subject === bob.subject_id);
++ const expired = await lease(catWs, (s) => s.holder.kind === 'nobody');
++ assert.strictEqual(expired.reason, 'expired');
++ assert.strictEqual(expired.state.holder.subject, null);
++ bobWs.sendJson({ type: 'sync', position_ms: 10 });
++ assert.strictEqual((await bobWs.next((m) => m.type === 'sync_error')).code, 'sync.not_holder');
++
++ // The clock is stamped with Chat's time on a time ping.
++ const before = Date.now();
++ annWs.sendJson({ type: 'time', id: 't1', t0: 12345 });
++ const pong = await annWs.next((m) => m.type === 'time');
++ assert.strictEqual(pong.id, 't1');
++ assert.strictEqual(pong.t0, 12345);
++ assert.ok(pong.t_server >= before && pong.t_server <= Date.now() + 1000);
++
++ // Drop the memory copies: a fresh join reads the durable rows back.
++ annWs.sendJson({ type: 'lease.request' });
++ await lease(catWs, (s) => s.holder.subject === ann.subject_id);
++ annWs.sendJson({ type: 'sync', position_ms: 30000 });
++ await sync(catWs);
++ require('../server/rooms/sync-clock').evict(room.id);
++ require('../server/rooms/control-lease').evict(room.id);
++ await join(catWs);
++ const fromDbSync = await sync(catWs);
++ const fromDbLease = await lease(catWs);
++ assert.deepStrictEqual([fromDbSync.state.position_ms, fromDbSync.state.set_by, fromDbLease.state.holder.subject], [30000, ann.subject_id, ann.subject_id]);
++
++ // The clock plays OpenVibe media only: a link to somebody else's server is refused.
++ annWs.sendJson({ type: 'sync', media: 'https://tracker.example.com/film.webm', position_ms: 0 });
++ assert.strictEqual((await annWs.next((m) => m.type === 'sync_error')).code, 'sync.media');
++
++ // A person holder whose last socket leaves keeps the remote a minute (shortened here), then it is free.
++ const controlLease = require('../server/rooms/control-lease');
++ controlLease.setAwayGraceForTests(400);
++ annWs.sendJson({ type: 'lease.grant', to: { kind: 'person', subject: bob.subject_id } });
++ await lease(catWs, (s) => s.holder.subject === bob.subject_id && s.expires_ms == null);
++ close(bobWs);
++ const away = await lease(catWs, (s) => s.holder.subject === bob.subject_id && s.expires_ms != null);
++ assert.ok(away.state.expires_ms <= Date.now() + 400 + 50, 'the seat waits at most the grace');
++ const released = await lease(catWs, (s) => s.holder.kind === 'nobody');
++ assert.strictEqual(released.state.holder.subject, null);
++ assert.strictEqual(released.reason, 'expired');
++
++ // Back inside the minute: still theirs, with their own (open) expiry again.
++ controlLease.setAwayGraceForTests(10000);
++ let bobBack = await h.ws({ token: bob.token });
++ await join(bobBack);
++ annWs.sendJson({ type: 'lease.take' });
++ await lease(catWs, (s) => s.holder.subject === ann.subject_id);
++ annWs.sendJson({ type: 'lease.grant', to: { kind: 'person', subject: bob.subject_id } });
++ await lease(catWs, (s) => s.holder.subject === bob.subject_id && s.expires_ms == null);
++ close(bobBack);
++ await lease(catWs, (s) => s.holder.subject === bob.subject_id && s.expires_ms != null);
++ bobBack = await h.ws({ token: bob.token });
++ await join(bobBack);
++ const kept = await lease(catWs, (s) => s.holder.subject === bob.subject_id && s.expires_ms == null);
++ assert.strictEqual(kept.state.holder.kind, 'person');
++ controlLease.setAwayGraceForTests(60 * 1000);
++ close(bobBack);
++
++ close(annWs);
++ close(catWs);
++});
++
++t('an AI can hold the remote, and only the owner takes it back', async () => {
++ const annWs = await h.ws({ token: ann.token });
++ const catWs = await h.ws({ token: cat.token });
++ await join(annWs); await lease(annWs);
++ await join(catWs); await lease(catWs);
++
++ annWs.sendJson({ type: 'lease.take' });
++ await lease(catWs, (s) => s.holder.subject === ann.subject_id);
++ // A guest holding the remote cannot hand it to an AI: that is the owner's call.
++ annWs.sendJson({ type: 'lease.grant', to: { kind: 'person', subject: cat.subject_id } });
++ await lease(catWs, (s) => s.holder.subject === cat.subject_id);
++ catWs.sendJson({ type: 'lease.grant', to: { kind: 'ai', subject: 'ai:vibe' } });
++ assert.strictEqual((await catWs.next((m) => m.type === 'lease_error')).code, 'lease.not_owner');
++ annWs.sendJson({ type: 'lease.take' });
++ await lease(catWs, (s) => s.holder.subject === ann.subject_id);
++ annWs.sendJson({ type: 'lease.grant', to: { kind: 'ai', subject: 'ai:vibe' } });
++ const ai = await lease(catWs, (s) => s.holder.kind === 'ai');
++ assert.strictEqual(ai.state.holder.subject, 'ai:vibe');
++ annWs.sendJson({ type: 'sync', position_ms: 1 });
++ const refused = await annWs.next((m) => m.type === 'sync_error');
++ assert.strictEqual(refused.code, 'sync.not_holder', 'even the owner cannot move a remote an AI holds');
++ annWs.sendJson({ type: 'lease.take' });
++ assert.strictEqual((await lease(catWs, (s) => s.holder.subject === ann.subject_id)).state.holder.kind, 'person');
++
++ close(annWs);
++ close(catWs);
++});
++
++t('close', async () => {
++ await h.close();
++});
++
++t.run();