Automerge backed Markdown editor for the Atmosphere
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108import pg from 'pg'
const failure = (code, status = 503) => Object.assign(new Error(code), { code, status })function detach(value) { if (value === null || typeof value !== 'object') return value if (value instanceof URLSearchParams) return new URLSearchParams(value) if (Array.isArray(value)) return value.map(detach) if (![Object.prototype, null].includes(Object.getPrototypeOf(value))) throw failure('store_unavailable') return Object.fromEntries(Object.entries(value).map(([key, child]) => [key, detach(child)]))}function validate(state, limits) { if (!state || typeof state !== 'object' || Array.isArray(state) || state.version !== 1) throw failure('store_unavailable') for (const kind of ['sessions', 'transactions', 'grants', 'proofs']) { const value = state[kind] if (value !== undefined && (!value || typeof value !== 'object' || Array.isArray(value))) throw failure('store_unavailable') if (Object.keys(value ?? {}).length > limits[kind]) throw failure('store_capacity_exceeded') } const serialized = JSON.stringify(state) if (Buffer.byteLength(serialized) > limits.storeBytes) throw failure('store_capacity_exceeded') return serialized}
/** A short, row-locked transaction is the mutation boundary. No OAuth upstream * I/O belongs inside the callback. Intentional protocol failures commit their * replay/nonce bookkeeping; unexpected failures roll back. Never retry a commit. */export async function createPostgresBrokerStore({ connectionString, namespace, max = 4, timeoutMs = 6000 }) { if (!connectionString || !namespace || !Number.isSafeInteger(timeoutMs) || timeoutMs < 100 || timeoutMs > 10_000) throw new TypeError('Invalid broker database configuration') const pool = new pg.Pool({ connectionString, max, connectionTimeoutMillis: Math.min(2000, timeoutMs), query_timeout: timeoutMs, application_name: 'reams-auth', idleTimeoutMillis: 30_000 }) pool.on('error', () => { console.error('Broker database idle connection failed') }) let pending = 0, closed = false, closing
async function transaction(operation) { const expires = Date.now() + timeoutMs let client, committed = false, begun = false, discard = false const query = (text, values) => { const remaining = expires - Date.now() if (remaining < 1) throw failure('store_unavailable') return client.query({ text, values, query_timeout: remaining }) } try { client = await pool.connect() await query('BEGIN') begun = true await query("SELECT set_config('statement_timeout', $1, true), set_config('lock_timeout', $2, true), set_config('idle_in_transaction_session_timeout', $1, true)", [String(Math.max(1, expires - Date.now())), String(Math.min(1000, timeoutMs))]) const result = await operation(query, expires) await query('COMMIT') committed = true if (result.protocolFailure) throw result.protocolFailure return result.value } catch (error) { if (client && begun && !committed) { try { await client.query({ text: 'ROLLBACK', query_timeout: 1000 }) } catch { discard = true } } if (!committed) discard = true // Timed-out or ambiguous clients never reenter the pool. if (committed && error.brokerProtocolFailure) throw error if (error.status && ['store_capacity_exceeded', 'store_unavailable', 'store_busy'].includes(error.code)) throw error throw failure('store_unavailable') } finally { client?.release(discard) } }
try { await transaction(async query => { // Serialize first schema creation too; a nonexistent row cannot be locked. await query("SELECT pg_advisory_xact_lock(hashtext('reams.oauth.schema.v1'))") await query('CREATE TABLE IF NOT EXISTS reams_oauth_state (namespace text PRIMARY KEY, state jsonb NOT NULL)') await query('INSERT INTO reams_oauth_state(namespace, state) VALUES ($1, $2::jsonb) ON CONFLICT DO NOTHING', [namespace, JSON.stringify({ version: 1, transactions: {}, grants: {} })]) return { value: undefined } }) } catch (error) { await pool.end(); throw error }
return { async transact(operation, limits) { if (closed) throw failure('store_unavailable') if (pending >= limits.pending) throw failure('store_busy') pending++ try { return await transaction(async (query, expires) => { const row = await query('SELECT state FROM reams_oauth_state WHERE namespace = $1 FOR UPDATE', [namespace]) if (row.rows.length !== 1) throw failure('store_unavailable') const state = row.rows[0].state const original = validate(state, limits) let value, protocolFailure, deadline try { value = await Promise.race([ Promise.resolve().then(() => operation(state)), new Promise((_, reject) => { deadline = setTimeout(() => reject(failure('store_unavailable')), Math.max(1, expires - Date.now())) }), ]) } catch (error) { if (!error.brokerProtocolFailure) throw error protocolFailure = error } finally { clearTimeout(deadline) } const serialized = validate(state, limits) if (serialized !== original) await query('UPDATE reams_oauth_state SET state = $2::jsonb WHERE namespace = $1', [namespace, serialized]) // Results are detached from the committed mutable snapshot. return { value: detach(value), protocolFailure } }) } finally { pending-- } }, async health() { if (closed) return false try { await pool.query('SELECT 1'); return true } catch { return false } }, async close() { closed = true; closing ??= pool.end(); await closing }, }}