Release: proximity voice rework + server fixes

fly.staging.toml

@@ -12,6 +12,7 @@ primary_region = 'iad'
   HOST = '0.0.0.0'
   NODE_ENV = 'production'
   PORT = '3000'
+  SENTRY_ENVIRONMENT = 'staging'
 
 [http_service]
   internal_port = 3000

packages/server/.env.example

@@ -25,6 +25,7 @@ CORS_ORIGIN=http://localhost:5173
 
 # Sentry (optional — error tracking disabled when absent)
 # SENTRY_DSN=https://examplePublicKey@o0.ingest.sentry.io/0
+# SENTRY_ENVIRONMENT=staging  # defaults to NODE_ENV
 
 # Logging (optional — defaults: info in production, debug in development)
 # LOG_LEVEL=info

packages/server/src/community/queries.test.ts

@@ -0,0 +1,60 @@
+import { describe, it, expect } from 'vitest';
+import type { Sql } from '../db/client.js';
+import { syncCommunityMembers } from './queries.js';
+
+// Minimal postgres.js stand-in: tracks how many transactions are open at once.
+function fakeSql(opts: { failFirst?: boolean } = {}) {
+  let open = 0;
+  let maxOpen = 0;
+  let calls = 0;
+  const order: string[] = [];
+  const sql = Object.assign((rows: unknown) => rows, {
+    begin: async (fn: (tx: unknown) => Promise<void>) => {
+      const call = calls++;
+      open++;
+      maxOpen = Math.max(maxOpen, open);
+      try {
+        const tx = async (strings: TemplateStringsArray, ...values: unknown[]) => {
+          await new Promise((r) => setTimeout(r, 1));
+          if (strings.join('?').startsWith('DELETE')) order.push(String(values[0]));
+          if (opts.failFirst && call === 0) throw new Error('boom');
+        };
+        await fn(tx);
+      } finally {
+        open--;
+      }
+    },
+  });
+  return { sql: sql as unknown as Sql, stats: () => ({ maxOpen, order }) };
+}
+
+const members = [{ did: 'did:plc:bob', addedAt: '2026-01-01T00:00:00Z' }];
+
+describe('syncCommunityMembers', () => {
+  it('runs concurrent syncs for the same owner one at a time', async () => {
+    const { sql, stats } = fakeSql();
+    await Promise.all(
+      Array.from({ length: 10 }, () => syncCommunityMembers(sql, 'did:plc:alice', members)),
+    );
+    expect(stats().maxOpen).toBe(1);
+  });
+
+  it('runs syncs for different owners in parallel', async () => {
+    const { sql, stats } = fakeSql();
+    await Promise.all([
+      syncCommunityMembers(sql, 'did:plc:alice', members),
+      syncCommunityMembers(sql, 'did:plc:carol', members),
+    ]);
+    expect(stats().maxOpen).toBe(2);
+  });
+
+  it('keeps the queue moving after a failed sync', async () => {
+    const { sql, stats } = fakeSql({ failFirst: true });
+    const first = syncCommunityMembers(sql, 'did:plc:alice', members);
+    const second = syncCommunityMembers(sql, 'did:plc:alice', members);
+    await expect(first).rejects.toThrow('boom');
+    await expect(second).resolves.toBeUndefined();
+    // First sync dies on the lock query before its DELETE; only the second gets that far.
+    expect(stats().order).toEqual(['did:plc:alice']);
+  });
+});

packages/server/src/community/queries.ts

@@ -21,13 +21,38 @@ export async function upsertCommunityList(
   `;
 }
 
+// Tail of the pending member syncs per owner DID. Queuing in-process means a burst of syncs for
+// one user (many tabs, reconnect storms) waits here instead of each holding a pool connection
+// while blocked on the advisory lock below.
+const memberSyncQueues = new Map<string, Promise<void>>();
+
 export async function syncCommunityMembers(
   sql: Sql,
   ownerDid: string,
   members: Array<{ did: string; addedAt: string }>,
+): Promise<void> {
+  const prev = memberSyncQueues.get(ownerDid) ?? Promise.resolve();
+  const run = prev.then(() => replaceCommunityMembers(sql, ownerDid, members));
+  const tail = run.catch(() => undefined);
+  memberSyncQueues.set(ownerDid, tail);
+  try {
+    await run;
+  } finally {
+    if (memberSyncQueues.get(ownerDid) === tail) memberSyncQueues.delete(ownerDid);
+  }
+}
+
+async function replaceCommunityMembers(
+  sql: Sql,
+  ownerDid: string,
+  members: Array<{ did: string; addedAt: string }>,
 ): Promise<void> {
   /* eslint-disable @typescript-eslint/no-explicit-any, @typescript-eslint/no-unsafe-call -- postgres.js TransactionSql type loses tagged template call signature via Omit */
   await sql.begin(async (tx: any) => {
+    // Serialize syncs per owner across machines — the WS sync_community handler and the Jetstream
+    // consumer can sync the same list concurrently, and the second DELETE can't see rows the
+    // first just inserted, so its INSERT hits the primary key. Released at commit/rollback.
+    await tx`SELECT pg_advisory_xact_lock(hashtext(${'community_members:' + ownerDid}))`;
     await tx`DELETE FROM community_members WHERE owner_did = ${ownerDid}`;
 
     if (members.length > 0) {

packages/server/src/config.ts

@@ -25,6 +25,8 @@ const envSchema = z.object({
   OAUTH_CLIENT_ID: z.string().optional(),
   OAUTH_REDIRECT_URI: z.string().url().optional(),
   SENTRY_DSN: z.string().url().optional(),
+  // Overrides NODE_ENV as the Sentry environment tag (staging runs with NODE_ENV=production).
+  SENTRY_ENVIRONMENT: z.string().optional(),
   REDIS_URL: z
     .string()
     .regex(/^rediss?:\/\//)

packages/server/src/sentry.ts

@@ -2,11 +2,13 @@ import * as Sentry from '@sentry/node';
 import type { Config } from './config.js';
 import { createLogger } from './logger.js';
 
-export function initSentry(config: Pick<Config, 'SENTRY_DSN' | 'NODE_ENV'>): void {
+export function initSentry(
+  config: Pick<Config, 'SENTRY_DSN' | 'SENTRY_ENVIRONMENT' | 'NODE_ENV'>,
+): void {
   if (!config.SENTRY_DSN) return;
   Sentry.init({
     dsn: config.SENTRY_DSN,
-    environment: config.NODE_ENV,
+    environment: config.SENTRY_ENVIRONMENT ?? config.NODE_ENV,
   });
   const log = createLogger('sentry');
   log.info('Sentry initialized');

packages/server/src/ws/handlers.ts

@@ -806,7 +806,10 @@ export async function handleClientMessage(
     }
 
     case 'town_join': {
-      await townRoom.enter(ws, did, data.scene, data.x, data.y, data.dir, data.fof);
+      // No `voice` (e.g. walking through a door): keep what the socket had.
+      const voiceJoin =
+        data.voice === undefined ? undefined : { on: data.voice, pid: data.voicePid };
+      await townRoom.enter(ws, did, data.scene, data.x, data.y, data.dir, data.fof, voiceJoin);
       break;
     }
 
@@ -865,6 +868,16 @@ export async function handleClientMessage(
       break;
     }
 
+    case 'town_voice_join': {
+      townRoom.setVoiceWanted(ws, data.on);
+      break;
+    }
+
+    case 'town_stay': {
+      townRoom.setStay(ws, data.on);
+      break;
+    }
+
     case 'town_leave': {
       townRoom.leave(ws);
       break;

packages/server/src/ws/server.ts

@@ -368,7 +368,7 @@ export function createWsServer(
         });
     });
 
-    ws.on('close', () => {
+    ws.on('close', (code: number) => {
       const ip = (ws as WebSocket & { remoteIp?: string }).remoteIp;
       if (ip) connectionTracker.decrement(ip);
       clearTimeout(authTimer);
@@ -378,7 +378,9 @@ export function createWsServer(
         // Remove this socket first so we can check remaining connections
         userSockets.remove(did, ws);
         communityWatchers.unwatchAll(ws);
-        townRoom.leave(ws);
+        // A network drop (not a server-side close like a ban or auth failure):
+        // keep its voice identity briefly so a reconnect resumes it.
+        townRoom.leave(ws, code === 1006 || code === 1001);
         botService?.handleClose(ws);
 
         const abandonedConvos = dmSubs.unsubscribeAll(ws);

packages/server/src/ws/town-proximity.test.ts

@@ -167,3 +167,60 @@ describe('buildClusters keeps conversations stable', () => {
     expect(clusters.map((c) => c.map((n) => n.id).sort())).toEqual([[2, 3]]);
   });
 });
+
+describe('buildClusters inside a home (wholeScene)', () => {
+  it('links everyone in the house regardless of distance, dwell or span', () => {
+    const far = BASE + maxSpan * 3;
+    const clusters = buildClusters(
+      [node(1, BASE), node(2, far), node(3, BASE, far)],
+      rules({ newLinkReady: () => false }),
+      true,
+    );
+    expect(ids(clusters)).toEqual([[1, 2, 3]]);
+  });
+
+  it('still keeps incompatible people apart', () => {
+    const blocked = (a: ProxNode, b: ProxNode) =>
+      a.did !== b.did && !([a.id, b.id].includes(1) && [a.id, b.id].includes(3));
+    const clusters = buildClusters(
+      [node(1, BASE), node(2, BASE + 500), node(3, BASE + 900)],
+      rules({ compatible: blocked }),
+      true,
+    );
+    expect(ids(clusters)).toEqual([[1, 2]]);
+  });
+});
+
+describe('buildClusters grace and "stay together"', () => {
+  const far = BASE + leaveRadius + 50;
+
+  it('holds an existing link past the leave radius while the grace lasts', () => {
+    const pair = [node(1, BASE, BASE, 'r'), node(2, far, BASE, 'r')];
+    expect(ids(buildClusters(pair, rules({ holdLink: () => true })))).toEqual([[1, 2]]);
+    expect(buildClusters(pair, rules({ holdLink: () => false }))).toEqual([]);
+  });
+
+  it('never extends grace to people who were not already linked', () => {
+    const pair = [node(1, BASE), node(2, far)];
+    expect(buildClusters(pair, rules({ holdLink: () => true }))).toEqual([]);
+  });
+
+  it('keeps a stay-together link at any distance only when both chose it', () => {
+    const veryFar = BASE + maxSpan * 4;
+    const both = [
+      { ...node(1, BASE, BASE, 'r'), stay: true },
+      { ...node(2, veryFar, BASE, 'r'), stay: true },
+    ];
+    expect(ids(buildClusters(both, rules()))).toEqual([[1, 2]]);
+    const one = [{ ...node(1, BASE, BASE, 'r'), stay: true }, node(2, veryFar, BASE, 'r')];
+    expect(buildClusters(one, rules())).toEqual([]);
+  });
+
+  it('a stay-together link still respects blocks', () => {
+    const both = [
+      { ...node(1, BASE, BASE, 'r'), stay: true },
+      { ...node(2, far, BASE, 'r'), stay: true },
+    ];
+    expect(buildClusters(both, rules({ compatible: () => false }))).toEqual([]);
+  });
+});

packages/server/src/ws/town-proximity.ts

@@ -6,8 +6,10 @@ export interface ProxNode {
   did: string;
   x: number;
   y: number;
-  /** Voice room this node is in right now, if any. */
+  /** Conversation this node is in right now, if any. */
   roomId: string | null;
+  /** Chose "stay together" in its current conversation. */
+  stay?: boolean;
 }
 
 export interface ClusterRules {
@@ -21,6 +23,11 @@ export interface ClusterRules {
   compatible(a: ProxNode, b: ProxNode): boolean;
   /** A brand-new link (not yet sharing a room) has dwelled long enough. */
   newLinkReady(a: ProxNode, b: ProxNode): boolean;
+  /**
+   * An existing link that drifted past the leave radius: still within its
+   * grace period? (Absent: no grace.)
+   */
+  holdLink?(a: ProxNode, b: ProxNode): boolean;
 }
 
 function dist2(a: ProxNode, b: ProxNode): number {
@@ -34,10 +41,24 @@ export function isNewLinkCandidate(a: ProxNode, b: ProxNode): boolean {
   );
 }
 
-function linked(a: ProxNode, b: ProxNode, rules: ClusterRules): boolean {
+function sameRoom(a: ProxNode, b: ProxNode): boolean {
+  return a.roomId !== null && a.roomId === b.roomId;
+}
+
+/** Both chose to stay together in the conversation they share. */
+function stayLink(a: ProxNode, b: ProxNode): boolean {
+  return sameRoom(a, b) && a.stay === true && b.stay === true;
+}
+
+function linked(a: ProxNode, b: ProxNode, rules: ClusterRules, wholeScene: boolean): boolean {
+  if (wholeScene) return true;
   const d2 = dist2(a, b);
-  // Hysteresis — an existing link holds until the wider leave radius.
-  if (a.roomId !== null && a.roomId === b.roomId) return d2 < TOWN_PROXIMITY.leaveRadius ** 2;
+  if (sameRoom(a, b)) {
+    if (stayLink(a, b)) return true;
+    // Hysteresis — an existing link holds until the wider leave radius, and
+    // past it for a short grace.
+    return d2 < TOWN_PROXIMITY.leaveRadius ** 2 || (rules.holdLink?.(a, b) ?? false);
+  }
   return d2 < TOWN_PROXIMITY.joinRadius ** 2 && rules.newLinkReady(a, b);
 }
 
@@ -67,6 +88,31 @@ function nearbyPairs(nodes: ProxNode[]): Array<[ProxNode, ProxNode]> {
   return pairs;
 }
 
+/**
+ * Pairs already sharing a conversation, at any distance — grace and "stay
+ * together" links can outlast the spatial hash's reach. Bounded by the
+ * cluster cap.
+ */
+function roomPairs(nodes: ProxNode[]): Array<[ProxNode, ProxNode]> {
+  const byRoom = new Map<string, ProxNode[]>();
+  for (const n of nodes) {
+    if (!n.roomId) continue;
+    const list = byRoom.get(n.roomId);
+    if (list) list.push(n);
+    else byRoom.set(n.roomId, [n]);
+  }
+  return [...byRoom.values()].flatMap(allPairs);
+}
+
+/** Every pair — for scenes where distance doesn't matter. */
+function allPairs(nodes: ProxNode[]): Array<[ProxNode, ProxNode]> {
+  const pairs: Array<[ProxNode, ProxNode]> = [];
+  for (const a of nodes) {
+    for (const b of nodes) if (a.id < b.id) pairs.push([a, b]);
+  }
+  return pairs;
+}
+
 /**
  * Group avatars into voice clusters: connected components of the proximity
  * graph ("the mesh"), so A–B–C can all talk even if A and C are farther apart.
@@ -76,17 +122,32 @@ function nearbyPairs(nodes: ProxNode[]): Array<[ProxNode, ProxNode]> {
  * Clusters are capped in size and in span (distance from the seed), so a
  * daisy chain can't stretch audio across the map. Only clusters of 2+ are
  * returned.
+ *
+ * `wholeScene` (inside a home) links everyone regardless of distance or
+ * dwell: the house is one conversation. Compatibility and the size cap
+ * still apply.
  */
-export function buildClusters(nodes: ProxNode[], rules: ClusterRules): ProxNode[][] {
+export function buildClusters(
+  nodes: ProxNode[],
+  rules: ClusterRules,
+  wholeScene = false,
+): ProxNode[][] {
   // People already in a conversation seed first, so a newcomer can never
   // push an existing member out of theirs.
   const eligible = nodes
     .filter((n) => rules.eligible(n))
     .sort((a, b) => Number(a.roomId === null) - Number(b.roomId === null) || a.id - b.id);
   const adj = new Map<number, ProxNode[]>();
   for (const n of eligible) adj.set(n.id, []);
-  for (const [a, b] of nearbyPairs(eligible)) {
-    if (!linked(a, b, rules)) continue;
+  const seen = new Set<string>();
+  const pairs = wholeScene
+    ? allPairs(eligible)
+    : [...nearbyPairs(eligible), ...roomPairs(eligible)];
+  for (const [a, b] of pairs) {
+    const key = a.id < b.id ? `${String(a.id)}:${String(b.id)}` : `${String(b.id)}:${String(a.id)}`;
+    if (seen.has(key)) continue;
+    seen.add(key);
+    if (!linked(a, b, rules, wholeScene)) continue;
     adj.get(a.id)?.push(b);
     adj.get(b.id)?.push(a);
   }
@@ -111,7 +172,8 @@ export function buildClusters(nodes: ProxNode[], rules: ClusterRules): ProxNode[
       for (const nb of neighbours) {
         if (cluster.length >= TOWN_PROXIMITY.maxClusterSize) break;
         if (visited.has(nb.id)) continue;
-        if (dist2(seed, nb) > maxSpan2) continue;
+        // "Stay together" is a choice both made — distance doesn't apply.
+        if (!wholeScene && !stayLink(cur, nb) && dist2(seed, nb) > maxSpan2) continue;
         if (!cluster.every((m) => rules.compatible(m, nb))) continue;
         visited.add(nb.id);
         cluster.push(nb);