Fix qc throttling and 'session expired': Bearer auth for events, no refresh storm (0.5.1, qc 0.4.2) (#284)

- authenticateRequest's token client called setSession() with an empty
  refresh token, which auth-js rejects, so every Bearer request through it
  (qc's /api/events) was a 401. Use a client carrying the Authorization
  header and getUser(token).
- qc reconnected /api/events on 401 straight after a refresh with no
  backoff, rotating refresh tokens in a tight loop (GoTrue logged several
  per 60ms) until the session died. Now one refresh per rejection, then stop.
  Parallel 401s share one refresh; a stale-token 401 just retries.
- qc waits out 429s (Retry-After, which the middleware now sends).
- The TUI no longer loads 40 chats on start to join their live rooms: the
  events stream joins all of the user's conversations on connect.
- Closing one SSE connection no longer drops the user from every room while
  another of their connections is open.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: ralyodio

package.json

@@ -1,7 +1,7 @@
 {
   "name": "qryptchat-web",
   "private": true,
-  "version": "0.5.0",
+  "version": "0.5.1",
   "type": "module",
   "bin": {
     "qc": "./bin/qc.js",

packages/qryptchat/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@profullstack/qryptchat",
-  "version": "0.4.1",
+  "version": "0.4.2",
   "description": "qc: qrypt.chat in your terminal. A full-screen end-to-end encrypted chat client (ML-KEM-1024), plus a scriptable CLI and an MCP server.",
   "type": "module",
   "bin": {

src/app/api/events/route.js

@@ -40,6 +40,14 @@ export async function GET(request, { params } = {}) {
 		const userId = userData.id;
 		console.log(`📡 [SSE] Internal user ID: ${userId}`);
 
+		// Join every live conversation up front, so news from any chat arrives
+		// without the client first loading each one's messages.
+		const { data: memberships } = await supabase
+			.from('conversation_participants')
+			.select('conversation_id')
+			.eq('user_id', userId)
+			.is('left_at', null);
+
 		// Create a readable stream for SSE
 		const stream = new ReadableStream({
 			start(controller) {
@@ -63,6 +71,7 @@ export async function GET(request, { params } = {}) {
 
 				// Add connection to SSE manager with internal user ID
 				sseManager.addConnection(responseWriter, userId);
+				for (const m of memberships || []) sseManager.joinRoom(userId, m.conversation_id);
 
 				// Send initial connection success message
 				const welcomeMessage = sseManager.formatSSEMessage('CONNECTED', {

src/cli/api.js

@@ -59,19 +59,34 @@ export class QcClient {
 		return this.session.access_token;
 	}
 
-	async request(path, { method = 'GET', body, signal, retry = true } = {}) {
+	/**
+	 * A 401 means the access token is stale, unless another request already
+	 * swapped it out while this one was in flight: then just retry with the new
+	 * one. Refresh tokens rotate, so refreshing twice for one expiry burns a token.
+	 */
+	async reauth(usedToken) {
+		if (this.session.access_token === usedToken) await this.refresh();
+	}
+
+	async request(path, { method = 'GET', body, signal, retry = true, throttled = 0 } = {}) {
+		const used = await this.token();
 		const res = await this.fetch(`${this.base}${path}`, {
 			method,
 			signal,
 			headers: {
-				Authorization: `Bearer ${await this.token()}`,
+				Authorization: `Bearer ${used}`,
 				...(body !== undefined ? { 'Content-Type': 'application/json' } : {}),
 			},
 			body: body !== undefined ? JSON.stringify(body) : undefined,
 		});
 		if (res.status === 401 && retry) {
-			await this.refresh();
-			return this.request(path, { method, body, signal, retry: false });
+			await this.reauth(used);
+			return this.request(path, { method, body, signal, retry: false, throttled });
+		}
+		// Rate limited: wait as told (or back off) and try again a few times.
+		if (res.status === 429 && throttled < 3) {
+			await sleep(retryAfterMs(res, throttled), signal);
+			return this.request(path, { method, body, signal, retry, throttled: throttled + 1 });
 		}
 		const data = await res.json().catch(() => ({}));
 		if (!res.ok) throw new QcError(data.error_description || data.error || `HTTP ${res.status}`, res.status);
@@ -162,20 +177,31 @@ export class QcClient {
 	 */
 	async events(onEvent, { signal, onStatus = () => {} } = {}) {
 		let delay = 1000;
+		let reauthed = false;
 		while (!signal?.aborted) {
 			onStatus('connecting');
 			try {
+				const used = await this.token();
 				const res = await this.fetch(`${this.base}/api/events`, {
-					headers: { Authorization: `Bearer ${await this.token()}`, Accept: 'text/event-stream' },
+					headers: { Authorization: `Bearer ${used}`, Accept: 'text/event-stream' },
 					signal,
 				});
 				if (res.status === 401) {
-					await this.refresh();
+					// One refresh per rejection. If a fresh token is refused too, stop:
+					// looping here would rotate refresh tokens until the session dies.
+					if (reauthed) throw new QcError('The server refused this session. Run qc login.', 401);
+					reauthed = true;
+					await this.reauth(used);
+					continue;
+				}
+				if (res.status === 429) {
+					await sleep(retryAfterMs(res, 2), signal);
 					continue;
 				}
 				if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);
 				onStatus('live');
 				delay = 1000;
+				reauthed = false;
 				await readSse(res.body, onEvent);
 			} catch (err) {
 				if (signal?.aborted) break;
@@ -217,6 +243,12 @@ export async function readSse(body, onEvent) {
 	}
 }
 
+/** How long a 429 says to wait: Retry-After seconds, else 2s, 4s, 8s ... */
+export function retryAfterMs(res, attempt = 0) {
+	const s = Number(res.headers?.get?.('retry-after'));
+	return Number.isFinite(s) && s > 0 ? Math.min(s, 60) * 1000 : 2000 * 2 ** attempt;
+}
+
 function sleep(ms, signal) {
 	return new Promise((resolve) => {
 		const t = setTimeout(resolve, ms);

src/cli/tui.js

@@ -278,12 +278,9 @@ export async function runTui(client, { initialChat } = {}) {
 			} else {
 				state.selected = Math.max(0, state.conversations.findIndex((c) => c.id === before));
 			}
+			// The events stream joins every chat's live room itself, so unread
+			// counts arrive without loading each chat (which tripped the rate limit).
 			redraw();
-			// Loading a conversation's messages joins its live room on the server,
-			// so warm every room: that is what makes unread counts arrive.
-			for (const c of state.conversations.slice(0, 40)) {
-				if (c.id !== state.activeId && !state.messages[c.id]) await loadMessages(c.id);
-			}
 		} catch (err) {
 			state.loading = false;
 			note(err.status === 401 ? 'Session ended: run qc login' : `Could not load chats: ${err.message}`);

src/lib/api/middleware/auth.js

@@ -20,7 +20,8 @@ export async function authenticateRequest(request) {
       ? await createSupabaseServerClientWithToken(token)
       : await createSupabaseServerClient();
 
-    const { data: { user }, error } = await supabase.auth.getUser();
+    // With a Bearer token the client holds no session, so name the JWT to verify.
+    const { data: { user }, error } = token ? await supabase.auth.getUser(token) : await supabase.auth.getUser();
 
     if (error || !user) {
       return { success: false, error: 'Unauthorized' };

src/lib/api/middleware/auth.test.js

@@ -52,6 +52,8 @@ describe('authenticateRequest bearer token parsing', () => {
     expect(auth.user).toEqual({ id: 'user-1' });
     expect(mocks.createSupabaseServerClientWithToken).toHaveBeenCalledWith('access-token-123');
     expect(mocks.createSupabaseServerClient).not.toHaveBeenCalled();
+    // The token client holds no session, so the JWT must be named explicitly.
+    expect(mocks.tokenClient.auth.getUser).toHaveBeenCalledWith('access-token-123');
   });
 
   it('falls back to cookies when the authorization header has no token', async () => {

src/lib/api/sse-manager.js

@@ -55,7 +55,9 @@ class SSEManager {
 
 		this.connectionUsers.delete(response);
 
-		// Remove from all rooms
+		// Leave the rooms only with the user's last connection: a closing qc
+		// session must not silence the same user's open browser tab.
+		if (this.userConnections.has(userId)) return;
 		for (const [conversationId, users] of this.conversationRooms.entries()) {
 			users.delete(userId);
 			if (users.size === 0) {

src/lib/api/sse-manager.test.js

@@ -0,0 +1,21 @@
+import { describe, expect, it, vi } from 'vitest';
+import { sseManager } from './sse-manager.js';
+
+describe('sseManager rooms', () => {
+	it('keeps a user in their rooms while another of their connections is open', () => {
+		vi.spyOn(console, 'log').mockImplementation(() => {});
+		const browser = { write: vi.fn() };
+		const terminal = { write: vi.fn() };
+		sseManager.addConnection(browser, 'u-rooms');
+		sseManager.addConnection(terminal, 'u-rooms');
+		sseManager.joinRoom('u-rooms', 'conv-rooms');
+
+		// qc reconnecting must not silence the open browser tab.
+		sseManager.removeConnection(terminal);
+		sseManager.broadcastToRoom('conv-rooms', 'NEW_MESSAGE', { id: 1 });
+		expect(browser.write).toHaveBeenCalled();
+
+		sseManager.removeConnection(browser);
+		expect(sseManager.conversationRooms.has('conv-rooms')).toBe(false);
+	});
+});

src/lib/supabase.js

@@ -1,4 +1,5 @@
 import { createBrowserClient, createServerClient } from '@supabase/ssr';
+import { createClient } from '@supabase/supabase-js';
 
 const SUPABASE_URL = process.env.NEXT_PUBLIC_SUPABASE_URL;
 const SUPABASE_ANON_KEY = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY;
@@ -35,12 +36,19 @@ export async function createSupabaseServerClient() {
   });
 }
 
+/**
+ * A server client that acts as the bearer of `token` (qc, MCP, passkey and
+ * CoinPay sessions, which carry no cookie). The token rides on every request,
+ * so RLS sees that user. setSession() is not an option: auth-js refuses a
+ * session without a refresh token, which silently turned every Bearer request
+ * into a 401.
+ */
 export async function createSupabaseServerClientWithToken(token) {
-  const client = await createSupabaseServerClient();
-  if (token) {
-    await client.auth.setSession({ access_token: token, refresh_token: '' });
-  }
-  return client;
+  if (!token) return createSupabaseServerClient();
+  return createClient(SUPABASE_URL, SUPABASE_ANON_KEY, {
+    global: { headers: { Authorization: `Bearer ${token}` } },
+    auth: { persistSession: false, autoRefreshToken: false, detectSessionInUrl: false },
+  });
 }
 
 export class SupabaseHelpers {