Stian/Shade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

release(v4.7.0): peer-presence events for instant BroadcastChannel revoke

Browse Source 
Adds the bridge-connection-lifecycle signal that closes Prism's
~45s revoke window down to one server→client round-trip (~50ms).

Server (`@shade/inbox-server`):
- `inbox.peer_connected` / `inbox.peer_disconnected` events on the
  0↔1 boundary across WS + SSE bridges. Long-poll deliberately not
  tracked (every poll boundary would flap; push transports are also
  the only ones where instant revoke matters).
- `PresenceTracker` collapses two parallel bridges (e.g. WS + SSE
  during fallback handover) into one connect/disconnect pair.
- `GET /v1/bridge/presence` SSE endpoint: signed query with
  `kind: 'presence'`, `watched: string[]`; on open streams a
  per-address snapshot, then change frames filtered server-side.
  MAX_WATCHED_ADDRESSES = 64. Subscribing does not itself count as
  a peer-bridge connection.
- `createBridgeRoutes` now returns `{ app, websocket, presence }`.

Client (`@shade/transport-bridge`):
- `PresenceBridge.subscribe({ watch, onPresenceChange })` →
  `{ addPeer, removePeer, watching, unsubscribe }`. addPeer/removePeer
  mutate via reconnect with a fresh signed query.
- `signPresenceQuery` helper for non-PresenceBridge consumers.

Tests cover all four acceptance criteria from the Prism request:
server-event smoke, online→offline subscription, address scoping
(carol invisible to a [alice]-only sub), reconnect, plus an
addPeer/removePeer regression.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sterister
2026-05-07 23:16:35 +02:00
parent 8746571d2a
commit 594992a183
34 changed files with 1042 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
packages/shade-inbox-server/src/bridge.ts
View File

@@ -38,8 +38,15 @@ import {
import { verifyPayload, validateAddress } from '@shade/server';
import type { InboxStore } from './store.js';
import type { InboxServerEvents } from './events.js';
import { PresenceTracker, type TrackedBridgeKind } from './presence.js';
export type BridgeKind = 'stream' | 'poll' | 'ws';
/**
* Wire-protocol kind tag for `/v1/bridge/presence`. Distinct from
* `BridgeKind` because the canonical signed payload is shaped
* differently (`watched: string[]` instead of `since: number`).
*/
export type PresenceKind = 'presence';
export interface BridgeRoutesOptions {
store: InboxStore;
@@ -60,6 +67,13 @@ export interface BridgeRoutesOptions {
* Default 1_000.
*/
fallbackPollIntervalMs?: number;
/**
* Inject an existing presence tracker. Useful when multiple
* `createBridgeRoutes` calls need to share state (e.g. mounting the
* routes under several hostnames in a single process). When omitted,
* the bridge auto-creates an internal tracker bound to `events`.
*/
presenceTracker?: PresenceTracker;
}
interface VerifiedBridgeRequest {
@@ -68,6 +82,13 @@ interface VerifiedBridgeRequest {
since: number;
}
interface VerifiedPresenceRequest {
/** The watcher's address (signer of the request). */
address: string;
/** Addresses whose presence the watcher is asking to track. */
watched: string[];
}
/**
* Build the bridge Hono router and a paired Bun-WebSocket handler.
*
@@ -80,6 +101,8 @@ export function createBridgeRoutes(opts: BridgeRoutesOptions): {
app: Hono;
/** Pass to `Bun.serve({ websocket })`. Undefined if Bun adapter is missing. */
websocket: unknown;
/** Live presence tracker. Tests + observers can read it; routes update it. */
presence: PresenceTracker;
} {
const app = new Hono();
const pageLimit = opts.pageLimit ?? 50;
@@ -87,6 +110,7 @@ export function createBridgeRoutes(opts: BridgeRoutesOptions): {
const longPollDefault = opts.longPollTimeoutMs ?? 25_000;
const longPollMax = opts.longPollMaxTimeoutMs ?? 55_000;
const fallbackPollIntervalMs = opts.fallbackPollIntervalMs ?? 1_000;
const presence = opts.presenceTracker ?? new PresenceTracker(opts.events ?? null);
app.onError((err, c) => {
if (err instanceof ShadeError) {
@@ -102,6 +126,14 @@ export function createBridgeRoutes(opts: BridgeRoutesOptions): {
const verified = await verifyBridgeAuth(c, opts, 'stream');
return streamSSE(c, async (stream) => {
const address = verified.address;
const connId = presence.newConnectionId();
presence.markConnected(address, 'sse', connId);
let presenceClosed = false;
const closePresence = (reason: 'closed' | 'error'): void => {
if (presenceClosed) return;
presenceClosed = true;
presence.markDisconnected(address, 'sse', connId, reason);
};
let cursor = verified.since;
const writer = makeBlobWriter(opts.store, pageLimit);
@@ -163,6 +195,68 @@ export function createBridgeRoutes(opts: BridgeRoutesOptions): {
clearInterval(fallbackTimer);
clearInterval(heartbeat);
await pendingFlushPromise.catch(() => {});
closePresence('closed');
});
});
// ─── Presence (V4.7) ──────────────────────────────────────────
// SSE feed of `peer_connected` / `peer_disconnected` events filtered
// by a watcher-supplied address list. Subscribing does NOT count as
// a peer-bridge connection (it doesn't call `markConnected`) so
// monitoring presence doesn't make you appear online to others.
app.get('/v1/bridge/presence', async (c) => {
const verified = await verifyPresenceAuth(c, opts);
return streamSSE(c, async (stream) => {
const watched = new Set(verified.watched);
// Initial snapshot — one frame per watched address with current
// status. Lets subscribers render UI immediately rather than
// waiting for the next state change.
const now = Date.now();
for (const addr of verified.watched) {
await stream.writeSSE({
event: 'presence',
data: JSON.stringify({
address: addr,
status: presence.isOnline(addr) ? 'online' : 'offline',
at: now,
}),
});
}
let unsubscribe: (() => void) | null = null;
if (opts.events) {
unsubscribe = opts.events.on((e) => {
if (e.name !== 'inbox.peer_connected' && e.name !== 'inbox.peer_disconnected') return;
const data = e.data as { address: string; bridgeKind: TrackedBridgeKind };
if (!watched.has(data.address)) return;
const status = e.name === 'inbox.peer_connected' ? 'online' : 'offline';
// Fire-and-forget: drop the frame if the stream has gone away.
void stream
.writeSSE({
event: 'presence',
data: JSON.stringify({
address: data.address,
status,
at: e.timestamp,
via: data.bridgeKind,
}),
})
.catch(() => {});
});
}
const heartbeat = setInterval(() => {
stream.write(`: ping ${Date.now()}\n\n`).catch(() => {});
}, heartbeatIntervalMs);
await new Promise<void>((resolve) => {
const sig = c.req.raw.signal;
if (sig.aborted) return resolve();
sig.addEventListener('abort', () => resolve(), { once: true });
});
unsubscribe?.();
clearInterval(heartbeat);
});
});
@@ -230,6 +324,7 @@ export function createBridgeRoutes(opts: BridgeRoutesOptions): {
}
const address = verified.address;
const connId = presence.newConnectionId();
let cursor = verified.since;
const writer = makeBlobWriter(opts.store, pageLimit);
let unsubscribe: (() => void) | null = null;
@@ -237,12 +332,19 @@ export function createBridgeRoutes(opts: BridgeRoutesOptions): {
let pendingFlushPromise: Promise<void> = Promise.resolve();
let signalled = false;
let connected = true;
let presenceClosed = false;
const closePresence = (reason: 'closed' | 'error'): void => {
if (presenceClosed) return;
presenceClosed = true;
presence.markDisconnected(address, 'ws', connId, reason);
};
return {
onOpen(_evt: unknown, ws: {
send: (data: string) => void;
close: (code?: number, reason?: string) => void;
}) {
presence.markConnected(address, 'ws', connId);
const triggerFlush = (): void => {
signalled = true;
pendingFlushPromise = pendingFlushPromise.then(async () => {
@@ -269,12 +371,19 @@ export function createBridgeRoutes(opts: BridgeRoutesOptions): {
connected = false;
unsubscribe?.();
if (fallbackTimer) clearInterval(fallbackTimer);
closePresence('closed');
},
onError() {
connected = false;
unsubscribe?.();
if (fallbackTimer) clearInterval(fallbackTimer);
closePresence('error');
},
};
}),
);
return { app, websocket };
return { app, websocket, presence };
}
// ─── helpers ──────────────────────────────────────────────────
@@ -321,6 +430,68 @@ async function verifyBridgeAuth(
return { address, kind: kind as BridgeKind, since };
}
const MAX_WATCHED_ADDRESSES = 64;
/**
* Verify a `/v1/bridge/presence` request.
*
* Signed canonical payload: `{address, kind: 'presence', watched: string[],
* signedAt}`. The watcher's address must be a registered inbox; the
* signature is verified against the registered owner key for that
* address. The `watched` list bounds what the subscription will
* receive — server-side filtering is enforced inside the handler.
*/
async function verifyPresenceAuth(
c: Context,
opts: BridgeRoutesOptions,
): Promise<VerifiedPresenceRequest> {
const url = new URL(c.req.url);
const qs = url.searchParams;
const address = validateAddress(qs.get('address'));
const kind = qs.get('kind');
if (kind !== 'presence') {
throw new ValidationError(`bridge kind mismatch: expected presence`, 'kind');
}
const watchedRaw = qs.get('watched');
if (watchedRaw === null) throw new ValidationError('missing watched', 'watched');
// Empty subscription is allowed (subscribe to nothing — useful for a
// client that intends to call addPeer right after open). A null/
// missing param is still rejected so the canonicalization is
// unambiguous.
const watched =
watchedRaw === ''
? []
: watchedRaw.split(',').map((a) => validateAddress(a));
if (watched.length > MAX_WATCHED_ADDRESSES) {
throw new ValidationError(
`watched list too large: ${watched.length} > ${MAX_WATCHED_ADDRESSES}`,
'watched',
);
}
const signedAtStr = qs.get('signedAt');
const signature = qs.get('signature');
if (signedAtStr === null) throw new ValidationError('missing signedAt', 'signedAt');
if (!signature) throw new ValidationError('missing signature', 'signature');
const signedAt = Number(signedAtStr);
if (!Number.isFinite(signedAt)) {
throw new ValidationError('signedAt must be a number', 'signedAt');
}
const owner = await opts.store.getAddressOwner(address);
if (!owner) {
throw new UnauthorizedError(`address ${address} is not registered`);
}
await verifyPayload(opts.crypto, owner, {
address,
kind,
watched,
signedAt,
signature,
});
return { address, watched };
}
interface BlobRow {
msgId: string;
ciphertext: Uint8Array;