speed up auth and site loading
1 unmodified line2345678910562 unmodified linesExpand all57357457557657757857958058158258358413 unmodified lines1 unmodified lineprocess.getBuiltinModule = require
import { cors } from '@elysiajs/cors'import { staticPlugin } from '@elysiajs/static'import { closePinnedKeepAliveAgents } from '@wispplace/atproto-utils'import { BASE_HOST, MAX_PUBLIC_UPLOAD_REQUEST_SIZE } from '@wispplace/constants'import { createLogger, initializeGrafanaExporters, logCollector, redactSecretPath } from '@wispplace/observability'import { observabilityMiddleware } from '@wispplace/observability/middleware/elysia'import type { Context } from 'elysia'562 unmodified linesExpand all }
const clientTasks = [ { name: 'Redis', promise: Promise.resolve().then(() => closeRedisClient()) }, { name: 'pinned HTTP connection pool', promise: Promise.resolve().then(() => closePinnedKeepAliveAgents()), }, { name: 'OAuth lock database', promise: closeOAuthLockDatabase() }, { name: 'site upload lock database', promise: closeSiteUploadLockDatabase() }, { name: 'database', promise: closeDatabase() }, ]13 unmodified lines12344567885 unmodified lines94959697989998991001011021031041051061072 unmodified linesimport type { NodeOAuthClient, OAuthSession } from '@atproto/oauth-client-node'import { describeCapability, missingCapabilities, wispAppRequiredCapabilities } from '@wispplace/constants'import { createLogger } from '@wispplace/observability'import { OAUTH_LEGACY_SCOPE, OAUTH_SCOPE } from './oauth-client'import { OAUTH_LEGACY_SCOPE, OAUTH_SCOPE, recentGrantedScope } from './oauth-client'
const logger = createLogger('main-app')
/**85 unmodified lines * contains before minting the token — so this compares meaning, not strings. */export const missingGrantedCapabilities = async (session: OAuthSession): Promise<string[]> => { try { const tokenInfo = await session.getTokenInfo(false) return missingCapabilities(tokenInfo.scope, wispAppRequiredCapabilities()).map(describeCapability) // This process stored the session moments ago, so the scope is usually // already known. Asking the session for it instead would take the // cluster-wide advisory lock and read the primary to recover a value that // never changes for the life of the grant. const scope = recentGrantedScope(session.did) ?? (await session.getTokenInfo(false)).scope return missingCapabilities(scope, wispAppRequiredCapabilities()).map(describeCapability) } catch (err) { // Never block a login on an introspection failure. logger.warn('[Auth] Could not read granted scope', { err: err instanceof Error ? err.message : String(err) }) return []2 unmodified lines5 unmodified lines67891011121314111 unmodified linesExpand all12612712812913013113213313413513613713813914014114214314414514614714814915015115215323 unmodified lines177178179180181182183184185114 unmodified linesExpand all3003013023033043053063073083093103113123133143153163173186 unmodified lines3253263273283293303313323333343353363373383395 unmodified linesimport { createClientMetadata } from './oauth-client-metadata'import { createOAuthFetch } from './oauth-fetch'import { withReservedOAuthLock } from './oauth-lock'import { SlingshotHandleResolver } from './slingshot-handle-resolver'import { createTtlMemoryStore } from './ttl-memory-store'
export { createClientMetadata, OAUTH_LEGACY_SCOPE, OAUTH_SCOPE } from './oauth-client-metadata'
// Cluster-wide lock backed by Postgres advisory locks. Replaces requestLocalLock111 unmodified linesExpand all await db`DELETE FROM oauth_states WHERE key = ${key}` },}
/** * The scope granted by the most recently stored session, per subject. * * The OAuth callback has to know what the authorization server actually granted * so it can detect a server that accepted the `include:place.wisp.*` permission * sets and then dropped them. Asking the session for it costs a full advisory * lock cycle and a read on the primary, microseconds after this process wrote * the very value being read. Only the scope string is remembered — never a * token — so nothing here can serve a stale credential to a refresh. */const grantedScopes = createTtlMemoryStore<string, string>({ ttlMs: 60_000, max: 500 })
/** The granted scope recorded when this process last stored a session. */export const recentGrantedScope = (sub: string): string | undefined => grantedScopes.get(sub)
const sessionStore = { async set(sub: string, data: any) { console.debug('[sessionStore] set', sub) const scope = data?.tokenSet?.scope if (typeof scope === 'string') grantedScopes.set(sub, scope) const expiresAt = Math.floor(Date.now() / 1000) + SESSION_TIMEOUT await db` INSERT INTO oauth_sessions (sub, data, updated_at, expires_at) VALUES (${sub}, ${JSON.stringify(data)}, EXTRACT(EPOCH FROM NOW()), ${expiresAt})23 unmodified lines return JSON.parse(result[0].data) }, async del(sub: string) { console.debug('[sessionStore] del', sub) grantedScopes.del(sub) await db`DELETE FROM oauth_sessions WHERE sub = ${sub}` },}
114 unmodified linesExpand all return false }}
// A cache miss on any of these costs a request to a PDS that may be a hemisphere// away, and none of the cached values change on a scale of minutes. The library// defaults to 60s, which guarantees a cold fetch on every login because the// consent screen alone takes longer than that.const METADATA_CACHE_TTL_MS = 15 * 60_000const METADATA_CACHE_MAX = 500// Nonces are single-origin and rotate on the server's schedule. A stale one// costs one extra round trip and self-heals, so it may be held as long as the// metadata; too short a TTL guarantees the challenge on every login instead.const DPOP_NONCE_CACHE_TTL_MS = 10 * 60_000
export const getOAuthClient = async (config: { domain: `http://${string}` | `https://${string}` clientName: string}) => {6 unmodified lines stateStore, sessionStore, requestLock: requestPgLock, handleResolver: new SlingshotHandleResolver(), authorizationServerMetadataCache: createTtlMemoryStore({ ttlMs: METADATA_CACHE_TTL_MS, max: METADATA_CACHE_MAX, }), protectedResourceMetadataCache: createTtlMemoryStore({ ttlMs: METADATA_CACHE_TTL_MS, max: METADATA_CACHE_MAX, }), dpopNonceCache: createTtlMemoryStore({ ttlMs: DPOP_NONCE_CACHE_TTL_MS, max: METADATA_CACHE_MAX }), })}13 unmodified lines141516171819202122201 unmodified linesExpand all224225226227227228229230231232233234235236290 unmodified linesExpand all13 unmodified linesimport { type IdentityResolvedAddress, isIdentityLoopbackDevelopmentAllowed, isPublicIdentityAddress, pinnedKeepAliveAgent,} from '@wispplace/atproto-utils'
const DEFAULT_TIMEOUT_MS = 10_000const MAX_URL_LENGTH = 8192201 unmodified linesExpand all port: url.port ? Number(url.port) : undefined, path: `${url.pathname}${url.search}`, method, headers: Object.fromEntries(headers.entries()), agent: false, // Pooled per validated address. An OAuth login makes five to eight // calls to the same authorization server; without reuse each one paid // a full TLS handshake, which dominated login latency from a distant // region. See @wispplace/atproto-utils/pinned-agent. agent: pinnedKeepAliveAgent(url, address), servername: url.protocol === 'https:' && isIP(normalizeHostname(url.hostname)) === 0 ? normalizeHostname(url.hostname) : undefined,290 unmodified linesExpand all12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364/** * A bounded in-memory TTL store shaped like `@atproto-labs/simple-store`. * * `NodeOAuthClient` defaults its authorization-server metadata, protected- * resource metadata, and DPoP nonce caches to a 60 second TTL. That is far too * short when a cache miss costs a round trip to a distant PDS: a user sits on * the consent screen for longer than 60 seconds, so every callback re-fetched * metadata it had just read, and the first request after each expiry paid the * DPoP nonce challenge again. All three are cheap to hold and safe to reuse for * minutes rather than seconds. */
export interface TtlMemoryStoreOptions { ttlMs: number max: number}
interface Entry<V> { value: V expiresAt: number}
export interface TtlMemoryStore<K, V> { get(key: K): V | undefined set(key: K, value: V): void del(key: K): void clear(): void}
export const createTtlMemoryStore = <K, V>({ ttlMs, max }: TtlMemoryStoreOptions): TtlMemoryStore<K, V> => { const entries = new Map<K, Entry<V>>()
const evictOldest = (): void => { // Map iteration is insertion-ordered and every hit re-inserts, so the // first entry is the least recently used. const oldest = entries.keys().next() if (!oldest.done) entries.delete(oldest.value) }
return { get(key) { const entry = entries.get(key) if (!entry) return undefined if (Date.now() >= entry.expiresAt) { entries.delete(key) return undefined } entries.delete(key) entries.set(key, entry) return entry.value }, set(key, value) { entries.delete(key) if (entries.size >= max) evictOldest() entries.set(key, { value, expiresAt: Date.now() + ttlMs }) }, del(key) { entries.delete(key) }, clear() { entries.clear() }, }}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143import { afterEach, describe, expect, test } from 'bun:test'import type { NodeOAuthClient, OAuthSession } from '@atproto/oauth-client-node'import type { Cookie } from 'elysia'import { authenticateRequest, clearSessionCache, invalidateSessionCache, requireAuth, SESSION_COOKIE_NAME,} from './wisp-auth'
const DID = 'did:plc:aaaaaaaaaaaaaaaaaaaaaaaa'
const cookies = (did?: string): Record<string, Cookie<unknown>> => (did ? { [SESSION_COOKIE_NAME]: { value: did } } : {}) as unknown as Record<string, Cookie<unknown>>
interface StubClient { client: NodeOAuthClient restores: string[] settle: (sub: string) => void}
/** * A client whose `restore` only resolves when the test releases it, so a burst * of callers can be observed while the first restore is still in flight. */const stubClient = (options: { fail?: boolean; session?: OAuthSession | null } = {}): StubClient => { const restores: string[] = [] const waiting = new Map<string, () => void>()
const client = { restore: (sub: string) => { restores.push(sub) return new Promise((resolve, reject) => { waiting.set(sub, () => { if (options.fail) reject(new Error('restore failed')) else resolve(options.session === undefined ? ({ did: sub, sub } as unknown as OAuthSession) : options.session) }) }) }, } as unknown as NodeOAuthClient
return { client, restores, settle: (sub) => { waiting.get(sub)?.() waiting.delete(sub) }, }}
afterEach(() => { clearSessionCache()})
describe('authenticateRequest', () => { test('returns null without a session cookie and never restores', async () => { const stub = stubClient() expect(await authenticateRequest(stub.client, cookies())).toBeNull() expect(stub.restores).toEqual([]) })
test('rejects a request carrying the session cookie twice', async () => { const stub = stubClient() const header = `${SESSION_COOKIE_NAME}=${DID}; ${SESSION_COOKIE_NAME}=other`
expect(await authenticateRequest(stub.client, cookies(DID), header)).toBeNull() expect(stub.restores).toEqual([]) })
test('collapses a burst of concurrent requests into one restore', async () => { const stub = stubClient() const pending = Array.from({ length: 8 }, () => authenticateRequest(stub.client, cookies(DID)))
// All eight are now waiting on the same in-flight restore. expect(stub.restores).toEqual([DID])
stub.settle(DID) const results = await Promise.all(pending)
expect(stub.restores).toEqual([DID]) for (const result of results) expect(result?.did).toBe(DID) })
test('answers later requests from the cache', async () => { const stub = stubClient() const first = authenticateRequest(stub.client, cookies(DID)) stub.settle(DID) await first
expect(await authenticateRequest(stub.client, cookies(DID))).not.toBeNull() expect(stub.restores).toEqual([DID]) })
test('invalidating forces the next request to restore again', async () => { const stub = stubClient() const first = authenticateRequest(stub.client, cookies(DID)) stub.settle(DID) await first
invalidateSessionCache(DID) const second = authenticateRequest(stub.client, cookies(DID)) stub.settle(DID) await second
expect(stub.restores).toEqual([DID, DID]) })
test('does not cache a failed restore', async () => { const stub = stubClient({ fail: true }) const first = authenticateRequest(stub.client, cookies(DID)) stub.settle(DID) expect(await first).toBeNull()
const second = authenticateRequest(stub.client, cookies(DID)) stub.settle(DID) expect(await second).toBeNull()
// A user signing in right now must not be held out by a cached failure. expect(stub.restores).toEqual([DID, DID]) })
test('does not cache a missing session', async () => { const stub = stubClient({ session: null }) const first = authenticateRequest(stub.client, cookies(DID)) stub.settle(DID) expect(await first).toBeNull()
const second = authenticateRequest(stub.client, cookies(DID)) stub.settle(DID) expect(await second).toBeNull()
expect(stub.restores).toEqual([DID, DID]) })})
describe('requireAuth', () => { test('throws when there is no session', async () => { const stub = stubClient() await expect(requireAuth(stub.client, cookies())).rejects.toThrow('Authentication required') })})10 unmodified lines11121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001015 unmodified lines107108109110282911111211311411513 unmodified lines10 unmodified lines did: Did session: OAuthSession}
/** * How long one restored session answers further requests from the same DID. * * `client.restore` takes a cluster-wide Postgres advisory lock and reads * `oauth_sessions`, both on the primary. From a region that is a few hundred * milliseconds from the primary that is roughly three serialized round trips * per request, and because every request for one user contends on the same lock * name they could not overlap: opening the editor fires seven API calls plus one * per site, and they queued behind each other. * * This caches only the answer to "does this DID have a live session", which is * all most routes need. It does not cache tokens: the session object still * resolves its token set through the lock on every call it makes to a PDS, so * refresh-token rotation stays serialized cluster-wide. The cost is that a * session revoked on another node keeps answering here until the entry lapses. */const SESSION_CACHE_TTL_MS = 30_000const SESSION_CACHE_MAX = 2_000
interface SessionCacheEntry { pending: Promise<AuthenticatedContext | null> expiresAt: number}
const sessionCache = new Map<string, SessionCacheEntry>()
const dropEntry = (did: string, entry: SessionCacheEntry): void => { // Only drop our own entry; a later request may already have replaced it. if (sessionCache.get(did) === entry) sessionCache.delete(did)}
/** * Restore a session, collapsing concurrent requests for the same DID. * * The in-flight promise is what gets cached, not just its result. A cold page * load arrives as a burst of parallel requests, and without this they would all * miss and then serialize on the advisory lock one after another. */const restoreSession = (client: NodeOAuthClient, did: Did): Promise<AuthenticatedContext | null> => { const now = Date.now() const cached = sessionCache.get(did) if (cached && cached.expiresAt > now) { sessionCache.delete(did) sessionCache.set(did, cached) return cached.pending }
const entry: SessionCacheEntry = { expiresAt: now + SESSION_CACHE_TTL_MS, pending: client.restore(did, 'auto').then( (session) => { // A DID with no session must not stay cached as a negative answer: // the user may be signing in right now. if (!session) dropEntry(did, entry) return session ? { did, session } : null }, (err) => { dropEntry(did, entry) throw err }, ), }
sessionCache.delete(did) if (sessionCache.size >= SESSION_CACHE_MAX) { // Insertion-ordered, and every hit re-inserts, so this is the oldest. const oldest = sessionCache.keys().next() if (!oldest.done) sessionCache.delete(oldest.value) } sessionCache.set(did, entry) return entry.pending}
/** Forget a cached session. Call this whenever a session is revoked here. */export const invalidateSessionCache = (did: string): void => { sessionCache.delete(did)}
/** Forget every cached session. */export const clearSessionCache = (): void => { sessionCache.clear()}
export const authenticateRequest = async ( client: NodeOAuthClient, cookies: Record<string, Cookie<unknown>>, rawCookieHeader?: string | null,5 unmodified lines
const did = cookies[SESSION_COOKIE_NAME]?.value as Did if (!did) return null
const session = await client.restore(did, 'auto') return session ? { did, session } : null return await restoreSession(client, did) } catch (err) { logger.error('[Auth] Authentication error', err) return null }13 unmodified lines7 unmodified lines89101112121314151676 unmodified lines9394959697989910010110210355 unmodified lines15916016116216316416516616716816936 unmodified lines7 unmodified lines isLegacyScopeState, missingGrantedCapabilities, unmarkLegacyScopeState,} from '../lib/oauth-authorize'import { authenticateRequest, SESSION_COOKIE_NAME } from '../lib/wisp-auth'import { authenticateRequest, invalidateSessionCache, SESSION_COOKIE_NAME } from '../lib/wisp-auth'import { resolvePrivateShareState } from './private-redeem'
const logger = createLogger('main-app')
76 unmodified lines c.cookie[SESSION_COOKIE_NAME].remove() return c.redirect('/?error=auth_failed') }
// A new grant supersedes whatever this node last restored for the DID. invalidateSessionCache(session.did)
const cookieSession = c.cookie cookieSession[SESSION_COOKIE_NAME].set({ value: session.did, httpOnly: true,55 unmodified lines cookieSession[SESSION_COOKIE_NAME].remove()
// If we have a DID, try to revoke the OAuth session if (did && typeof did === 'string') { // Drop the cached session first: a concurrent request must not be // handed a session this logout is about to revoke. invalidateSessionCache(did) try { await client.revoke(did) logger.debug('[Auth] Revoked OAuth session for', did as any) } catch (err) {36 unmodified lines14 unmodified lines1516171819192021222324252627282930313222 unmodified lines5556575859606162636465666768697071727374757677787980818 unmodified lines14 unmodified linesmock.module('../lib/db', () => ({ eventualRead: { getDomainsForDid: async () => ({ customDomains: [], wispDomains: [] }), getDomainsForSite: async () => [], getSitesForDid: async () => [], getSitesWithDomainsForDid: async () => [ { created_at: 1, did: DID, display_name: 'site-a', domains: [{ domain: 'site-a.wisp.place', type: 'wisp' }], rkey: 'site-a', updated_at: 1, }, ], getSupporterStatus: async () => true, getUserStatus: async () => ({ domain: null, sites: [] }), }, getSitesByDid: async () => [],22 unmodified lines expect(await responseJson(response)).toEqual({ did: DID, handle: 'alice.example', isSupporter: true }) expect(requests).toEqual(['https://example.com/.well-known/did.json']) })
test('serves each site with its domains so the list view needs one request', async () => { const app = userRoutes({} as never, 'test-cookie-secret', async () => new Response('{}'))
const response = await app.handle(new Request('http://localhost/api/user/sites')) expect(response.status).toBe(200) expect(await responseJson(response)).toEqual({ sites: [ { created_at: 1, did: DID, display_name: 'site-a', domains: [{ domain: 'site-a.wisp.place', type: 'wisp' }], rkey: 'site-a', updated_at: 1, }, ], }) })
test('returns an unknown handle without logging a raw identity error', async () => { logs.length = 0 const app = userRoutes({} as never, 'test-cookie-secret', async () => { throw new Error('https://token:secret@private.example')8 unmodified lines87 unmodified lines888990919292939495969697989910010160 unmodified lines87 unmodified lines } }) /** * GET /api/user/sites * Success: { sites } * Success: { sites } — each site carries its own `domains` array, so the * list view needs one request rather than one per site. */ .get('/sites', async ({ auth }) => { try { const sites = await eventualRead.getSitesForDid(auth.did) const sites = await eventualRead.getSitesWithDomainsForDid(auth.did) return { sites } } catch { logger.error('[User] Sites error') throw new Error('Failed to get sites')60 unmodified lines7 unmodified lines8910111213141516305 unmodified linesExpand all322323324325325326327328329330331332745 unmodified linesExpand all7 unmodified linesimport { request as httpRequest } from 'node:http'import { request as httpsRequest } from 'node:https'import { isIP } from 'node:net'import { Readable } from 'node:stream'import { pinnedKeepAliveAgent } from './pinned-agent'
const HANDLE_RESOLVER = 'https://slingshot.microcosm.blue/xrpc/com.atproto.identity.resolveHandle'const PLC_DIRECTORY = 'https://plc.directory'export const MAX_IDENTITY_JSON_BYTES = 1024 * 1024305 unmodified linesExpand all port: url.port ? Number(url.port) : undefined, path: `${url.pathname}${url.search}`, method: 'GET', headers: { Accept: 'application/json' }, agent: false, // Pooled per validated address, so a reused socket is always already // connected to an answer that passed validation. See pinned-agent.ts. agent: pinnedKeepAliveAgent(url, address), servername: url.protocol === 'https:' && isIP(hostname(url.hostname)) === 0 ? hostname(url.hostname) : undefined, // This lookup is the connection path, not merely a preflight check. lookup: (_name, options, callback) =>745 unmodified linesExpand all31 unmodified lines323334353637383940414247 unmodified lines31 unmodified lines resolvePdsFromHandle, unsafeRawIdentityGet, validatePdsEndpoint,} from './identity'// DNS-pinned keep-alive connection poolingexport type { PinnedAgentAddress, PinnedAgentFamily } from './pinned-agent'export { closePinnedKeepAliveAgents, pinnedKeepAliveAgent } from './pinned-agent'export type { ExpandedSubfs, ExpandSubfsOptions, FetchSubfsRecord,47 unmodified lines49 unmodified lines5051525354545556575859606156575859606162636465666768697071727374756263646594 unmodified lines49 unmodified lines fetch('/api/user/sites'), fetch('/api/user/private-sites').catch(() => null), ]) const data = await response.json() const sitesData: Site[] = data.sites || [] // /api/user/sites already returns each site's domains. It used to be one // extra request per site, and every one of those had to re-establish the // caller's session before it could answer. const sitesWithDomains: SiteWithDomains[] = (data.sites || []).map((site: SiteWithDomains) => ({ ...site, domains: site.domains || [], }))
// Fetch domain info for each site const sitesWithDomains = await Promise.all( sitesData.map(async (site) => { try { const domainsResponse = await fetch(`/api/user/site/${site.rkey}/domains`) const domainsData = await domainsResponse.json() return { ...site, domains: domainsData.domains || [], } } catch (err) { console.error(`Failed to fetch domains for site ${site.rkey}:`, err) return { ...site, domains: [], } } }), )
let privateSites: SiteWithDomains[] = [] if (privateResponse?.ok) { try { const privateData = await privateResponse.json()94 unmodified lines