From d596e49efeace9dd2aa9b6981c5e3ea33dd3df84 Mon Sep 17 00:00:00 2001 From: scanash00 Date: Wed, 22 Jul 2026 00:12:53 -0800 Subject: [PATCH] renew sessions --- server/src/notifier.ts | 66 ++++++++++++++++++++++++++++++++++---- server/src/pdsClient.ts | 71 +++++++++++++++++++++++++++++++++++------ 2 files changed, 120 insertions(+), 17 deletions(-) diff --git a/server/src/notifier.ts b/server/src/notifier.ts index 64ba98b..db1cabb 100644 --- a/server/src/notifier.ts +++ b/server/src/notifier.ts @@ -44,7 +44,8 @@ async function resolvePdsEndpoint(did: string): Promise { } export class BskyDmNotifier { - private session: { pds: string; accessJwt: string } | null = null; + private session: { pds: string; accessJwt: string; refreshJwt: string } | null = null; + private sessionPromise: Promise | null = null; private recipientDid: string | null = null; private convoId: string | null = null; @@ -54,8 +55,7 @@ export class BskyDmNotifier { ) {} /** Resolve the sender's DID + PDS from their handle, then create a session there. */ - private async ensureSession(): Promise<{ pds: string; accessJwt: string }> { - if (this.session) return this.session; + private async createSession(): Promise { const did = await resolveHandleToDid(this.cfg.handle); const pds = await resolvePdsEndpoint(did); const res = await fetch(`${pds}/xrpc/com.atproto.server.createSession`, { @@ -67,9 +67,61 @@ export class BskyDmNotifier { const body = await res.text().catch(() => ""); throw new Error(`createSession for ${this.cfg.handle} failed: ${res.status} ${body}`); } - const { accessJwt } = (await res.json()) as { accessJwt: string }; - this.session = { pds, accessJwt }; - return this.session; + const { accessJwt, refreshJwt } = (await res.json()) as { + accessJwt: string; + refreshJwt: string; + }; + this.session = { pds, accessJwt, refreshJwt }; + } + + private async updateSession(update: () => Promise): Promise { + let promise = this.sessionPromise; + if (!promise) { + promise = update(); + this.sessionPromise = promise; + } + try { + await promise; + } finally { + if (this.sessionPromise === promise) this.sessionPromise = null; + } + } + + private async ensureSession(): Promise<{ pds: string; accessJwt: string; refreshJwt: string }> { + if (!this.session) await this.updateSession(() => this.createSession()); + return this.session!; + } + + private async refreshSession(session: { + pds: string; + accessJwt: string; + refreshJwt: string; + }): Promise { + const res = await fetch(`${session.pds}/xrpc/com.atproto.server.refreshSession`, { + method: "POST", + headers: { Authorization: `Bearer ${session.refreshJwt}` }, + }); + if (res.status === 400 || res.status === 401) return false; + if (!res.ok) { + const body = await res.text().catch(() => ""); + throw new Error(`refreshSession for ${this.cfg.handle} failed: ${res.status} ${body}`); + } + const { accessJwt, refreshJwt } = (await res.json()) as { + accessJwt: string; + refreshJwt: string; + }; + this.session = { pds: session.pds, accessJwt, refreshJwt }; + return true; + } + + private async renewSession(expiredAccessJwt: string): Promise { + if (this.session?.accessJwt !== expiredAccessJwt) return; + await this.updateSession(async () => { + if (this.session?.accessJwt !== expiredAccessJwt) return; + if (await this.refreshSession(this.session)) return; + this.session = null; + await this.createSession(); + }); } /** Chat XRPC via the sender's PDS, service-proxied to the Bluesky chat appview. */ @@ -85,7 +137,7 @@ export class BskyDmNotifier { }, }); if (res.status === 401 && retry) { - this.session = null; // token expired — re-auth once + await this.renewSession(accessJwt); return this.chatXrpc(path, opts, false); } if (!res.ok) { diff --git a/server/src/pdsClient.ts b/server/src/pdsClient.ts index 409e677..aea7e14 100644 --- a/server/src/pdsClient.ts +++ b/server/src/pdsClient.ts @@ -37,6 +37,8 @@ export interface AdminAccount { export class PdsClient { private accessJwt: string | null = null; + private refreshJwt: string | null = null; + private sessionPromise: Promise | null = null; constructor( public readonly hostname: string, @@ -50,8 +52,7 @@ export class PdsClient { private adminIdentifier?: string, ) {} - private async ensureSession(): Promise { - if (this.accessJwt) return this.accessJwt; + private async createSession(): Promise { const res = await fetch(`https://${this.hostname}/xrpc/com.atproto.server.createSession`, { method: "POST", headers: { "Content-Type": "application/json" }, @@ -61,28 +62,78 @@ export class PdsClient { const body = await res.text().catch(() => ""); throw new Error(`PDS admin sign-in as ${this.adminIdentifier} failed: ${res.status} ${body}`); } - const { accessJwt } = (await res.json()) as { accessJwt: string }; + const { accessJwt, refreshJwt } = (await res.json()) as { + accessJwt: string; + refreshJwt: string; + }; this.accessJwt = accessJwt; - return accessJwt; + this.refreshJwt = refreshJwt; + } + + private async updateSession(update: () => Promise): Promise { + let promise = this.sessionPromise; + if (!promise) { + promise = update(); + this.sessionPromise = promise; + } + try { + await promise; + } finally { + if (this.sessionPromise === promise) this.sessionPromise = null; + } } - private async authHeader() { - if (this.adminIdentifier) return `Bearer ${await this.ensureSession()}`; - const token = Buffer.from(`admin:${this.adminPassword}`).toString("base64"); - return `Basic ${token}`; + private async ensureSession(): Promise { + if (!this.accessJwt) await this.updateSession(() => this.createSession()); + return this.accessJwt!; + } + + private async refreshSession(): Promise { + if (!this.refreshJwt) return false; + const res = await fetch(`https://${this.hostname}/xrpc/com.atproto.server.refreshSession`, { + method: "POST", + headers: { Authorization: `Bearer ${this.refreshJwt}` }, + }); + if (res.status === 400 || res.status === 401) return false; + if (!res.ok) { + const body = await res.text().catch(() => ""); + throw new Error(`PDS admin session refresh failed: ${res.status} ${body}`); + } + const { accessJwt, refreshJwt } = (await res.json()) as { + accessJwt: string; + refreshJwt: string; + }; + this.accessJwt = accessJwt; + this.refreshJwt = refreshJwt; + return true; + } + + private async renewSession(expiredAccessJwt: string): Promise { + if (this.accessJwt !== expiredAccessJwt) return; + await this.updateSession(async () => { + if (this.accessJwt !== expiredAccessJwt) return; + if (await this.refreshSession()) return; + this.accessJwt = null; + this.refreshJwt = null; + await this.createSession(); + }); } private async xrpc(path: string, opts: RequestInit = {}, retry = true): Promise { + const accessJwt = this.adminIdentifier ? await this.ensureSession() : null; + const authorization = accessJwt + ? `Bearer ${accessJwt}` + : `Basic ${Buffer.from(`admin:${this.adminPassword}`).toString("base64")}`; const res = await fetch(`https://${this.hostname}/xrpc/${path}`, { ...opts, headers: { ...opts.headers, - Authorization: await this.authHeader(), + Authorization: authorization, "Content-Type": "application/json", }, }); if (res.status === 401 && this.adminIdentifier && retry) { - this.accessJwt = null; // token expired — re-auth once + await this.renewSession(accessJwt!); return this.xrpc(path, opts, false); } if (!res.ok) { -- 2.51.2