feat(relay): Phase1 foundation — P3 Postgres store + async capability verifier + compose
RELAY-PHASE1 Wave A (2/3) + E1 infra: - A1: createPgStores() Postgres adapter for all 9 P3 store ports + runMigrations(); writable-CTE atomic INV8 status swaps; 0002_routes.sql. 23/23 pg tests, 96.72% cov. - A3: real capability verifier delegating to relay-auth verifyCapabilityToken; sync->async seam across authz/provision/main. Full CP suite 15 files/101 pass, tsc clean. - E1: deploy/docker-compose.yml (Postgres16+Redis7, loopback-only) + .env.example. - docs: PLAN_RELAY_PHASE1.md file-level execution spec; PROGRESS_LOG RELAY-PHASE1 section.
This commit is contained in:
Yaojia Wang
@@ -20,9 +20,12 @@ export class AuthzError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/** P5's verifier, injected. Mirrors the frozen `verifyCapabilityToken` signature (§4.3). */
|
||||
/**
|
||||
* P5's verifier, injected. Mirrors the frozen `verifyCapabilityToken` signature (§4.3), which is
|
||||
* ASYNC (Ed25519 verify over WebCrypto) — so this seam returns a Promise and all call sites await.
|
||||
*/
|
||||
export interface CapabilityVerifier {
|
||||
verify(raw: string, expectedAud: string, now: number): CapabilityToken
|
||||
verify(raw: string, expectedAud: string, now: number): Promise<CapabilityToken>
|
||||
}
|
||||
|
||||
/** Minimal request shape we read for the bearer token (never a body account field). */
|
||||
@@ -49,19 +52,19 @@ function extractRawToken(req: AuthRequest): string | null {
|
||||
}
|
||||
|
||||
export interface Authorizer {
|
||||
principalFromRequest(req: AuthRequest): AdminPrincipal
|
||||
principalFromRequest(req: AuthRequest): Promise<AdminPrincipal>
|
||||
requireRight(principal: AdminPrincipal, right: CapabilityRight): void
|
||||
}
|
||||
|
||||
export function createAuthorizer(deps: AuthzDeps): Authorizer {
|
||||
const now = deps.now ?? (() => Math.floor(Date.now() / 1000))
|
||||
return {
|
||||
principalFromRequest(req) {
|
||||
async principalFromRequest(req) {
|
||||
const raw = extractRawToken(req)
|
||||
if (raw === null) throw new AuthzError(401, 'missing capability token')
|
||||
let token: CapabilityToken
|
||||
try {
|
||||
token = deps.verifier.verify(raw, deps.expectedAud, now())
|
||||
token = await deps.verifier.verify(raw, deps.expectedAud, now())
|
||||
} catch (err: unknown) {
|
||||
throw new AuthzError(401, `invalid capability token: ${err instanceof Error ? err.message : 'rejected'}`)
|
||||
}
|
||||
|
||||
@@ -56,12 +56,12 @@ function assertOwnAccount(principal: AdminPrincipal, pathAccountId: string): voi
|
||||
|
||||
export function buildRouter(deps: ProvisionDeps): FastifyPluginAsync {
|
||||
return async (app) => {
|
||||
const principal = (req: FastifyRequest): AdminPrincipal =>
|
||||
const principal = (req: FastifyRequest): Promise<AdminPrincipal> =>
|
||||
deps.authorizer.principalFromRequest({ headers: req.headers })
|
||||
|
||||
app.post('/accounts', async (req, reply) => {
|
||||
try {
|
||||
const p = principal(req)
|
||||
const p = await principal(req)
|
||||
deps.authorizer.requireRight(p, 'manage')
|
||||
const plan: PlanTier = PlanSchema.parse((req.body as { plan?: unknown })?.plan ?? 'free')
|
||||
const account = await deps.accounts.createAccount(plan)
|
||||
@@ -73,7 +73,7 @@ export function buildRouter(deps: ProvisionDeps): FastifyPluginAsync {
|
||||
|
||||
app.post('/accounts/:id/pairing-codes', async (req, reply) => {
|
||||
try {
|
||||
const p = principal(req)
|
||||
const p = await principal(req)
|
||||
deps.authorizer.requireRight(p, 'manage')
|
||||
assertOwnAccount(p, (req.params as { id: string }).id)
|
||||
const issued = await deps.pairingIssuer.issuePairingCode(p.accountId)
|
||||
@@ -85,7 +85,7 @@ export function buildRouter(deps: ProvisionDeps): FastifyPluginAsync {
|
||||
|
||||
app.post('/accounts/:id/status', async (req, reply) => {
|
||||
try {
|
||||
const p = principal(req)
|
||||
const p = await principal(req)
|
||||
deps.authorizer.requireRight(p, 'manage')
|
||||
assertOwnAccount(p, (req.params as { id: string }).id)
|
||||
const { status } = StatusSchema.parse(req.body)
|
||||
@@ -98,7 +98,7 @@ export function buildRouter(deps: ProvisionDeps): FastifyPluginAsync {
|
||||
|
||||
app.get('/accounts/:id/hosts', async (req, reply) => {
|
||||
try {
|
||||
const p = principal(req)
|
||||
const p = await principal(req)
|
||||
assertOwnAccount(p, (req.params as { id: string }).id)
|
||||
const hosts = await deps.hosts.listHosts(p.accountId) // ownership-scoped
|
||||
await reply.send(hosts.map((h) => ({ ...h, agentPubkey: Buffer.from(h.agentPubkey).toString('base64') })))
|
||||
@@ -109,7 +109,7 @@ export function buildRouter(deps: ProvisionDeps): FastifyPluginAsync {
|
||||
|
||||
app.delete('/hosts/:hostId', async (req, reply) => {
|
||||
try {
|
||||
const p = principal(req)
|
||||
const p = await principal(req)
|
||||
deps.authorizer.requireRight(p, 'manage')
|
||||
// account_id in the body is IGNORED — authz uses ONLY the token principal (INV3).
|
||||
await deps.deprovisioner.deprovisionHost(p, (req.params as { hostId: string }).hostId)
|
||||
|
||||
38
control-plane/src/boot/verifier.ts
Normal file
38
control-plane/src/boot/verifier.ts
Normal file
@@ -0,0 +1,38 @@
|
||||
/**
|
||||
* A3 — REAL capability verifier, backed by relay-auth's `verifyCapabilityToken` (P5). Replaces the
|
||||
* fail-closed `refuseAllVerifier` stub. The frozen §4.3 verify signature is ASYNC (Ed25519 verify
|
||||
* over WebCrypto) and reads its verifying key from relay-auth's startup registry (config/keys.ts),
|
||||
* never as a parameter — so we (a) adapt it to the async `CapabilityVerifier` seam and (b) load the
|
||||
* key at boot from the CP env's already-validated pubkey.
|
||||
*
|
||||
* KEY BRIDGE (INV9): the CP env var is `CAPABILITY_SIGN_PUBKEY_B64` (base64), parsed+validated in
|
||||
* `env.ts` to `env.capabilitySignPubkey` (32 raw Ed25519 bytes). relay-auth's own env loader expects
|
||||
* a DIFFERENT var name (`RELAY_AUTH_VERIFY_PUBKEY`, base64url), so we do NOT use it — we import the
|
||||
* 32 raw bytes to a non-exportable verify-only CryptoKey via WebCrypto and call relay-auth's
|
||||
* `configureVerifyKey(CryptoKey)` directly.
|
||||
*/
|
||||
import { verifyCapabilityToken, configureVerifyKey } from 'relay-auth'
|
||||
import type { CapabilityToken } from 'relay-contracts'
|
||||
import type { CapabilityVerifier } from '../api/authz.js'
|
||||
|
||||
/** The real verifier: delegates verbatim to P5's frozen §4.3 `verifyCapabilityToken`. */
|
||||
export function createCapabilityVerifier(): CapabilityVerifier {
|
||||
return {
|
||||
verify(raw: string, expectedAud: string, now: number): Promise<CapabilityToken> {
|
||||
return verifyCapabilityToken(raw, expectedAud, now)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the §4.3 verifying key into relay-auth's startup registry from the CP env's raw 32-byte
|
||||
* Ed25519 public key (`env.capabilitySignPubkey`). Non-exportable, `verify`-only. Must run before
|
||||
* the real verifier serves any request, else `verifyCapabilityToken` throws `KeyConfigError`.
|
||||
*/
|
||||
export async function configureCapabilityVerifyKey(pubkeyRaw: Uint8Array): Promise<void> {
|
||||
// Copy into a fresh ArrayBuffer-backed view (WebCrypto's BufferSource typing / no shared pool).
|
||||
const bytes = new Uint8Array(pubkeyRaw.length)
|
||||
bytes.set(pubkeyRaw)
|
||||
const key = await globalThis.crypto.subtle.importKey('raw', bytes, { name: 'Ed25519' }, false, ['verify'])
|
||||
configureVerifyKey(key)
|
||||
}
|
||||
48
control-plane/src/db/migrate.ts
Normal file
48
control-plane/src/db/migrate.ts
Normal file
@@ -0,0 +1,48 @@
|
||||
/**
|
||||
* A1 — migration runner. Reads every `db/migrations/*.sql` in filename order and executes it
|
||||
* over the parameterized `query` wrapper (db/pool.ts). All migrations are `IF NOT EXISTS`, so
|
||||
* `runMigrations` is idempotent and safe to run at every boot / test setup.
|
||||
*
|
||||
* `createQuery` ALWAYS routes through the extended (parameterized) protocol — even with `[]`
|
||||
* params — which rejects multi-statement command strings. So each file is split into its
|
||||
* top-level statements (on `;`, after stripping `-- line comments`) and each statement is run
|
||||
* as its own single-command `query(stmt, [])`. Our migrations are plain IF-NOT-EXISTS DDL with
|
||||
* no `;` inside string literals and no dollar-quoted bodies, so this split is safe.
|
||||
*/
|
||||
import { readdir, readFile } from 'node:fs/promises'
|
||||
import { dirname, join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import type { QueryFn } from './pool.js'
|
||||
|
||||
/** Absolute path to `control-plane/db/migrations`, resolved relative to this source file. */
|
||||
function migrationsDir(): string {
|
||||
const here = dirname(fileURLToPath(import.meta.url)) // .../control-plane/src/db
|
||||
return join(here, '..', '..', 'db', 'migrations') // .../control-plane/db/migrations
|
||||
}
|
||||
|
||||
/** Strip `-- line comments`, then split into non-empty top-level statements on `;`. */
|
||||
export function splitStatements(sql: string): readonly string[] {
|
||||
const withoutComments = sql
|
||||
.split('\n')
|
||||
.map((line) => {
|
||||
const idx = line.indexOf('--')
|
||||
return idx >= 0 ? line.slice(0, idx) : line
|
||||
})
|
||||
.join('\n')
|
||||
return withoutComments
|
||||
.split(';')
|
||||
.map((stmt) => stmt.trim())
|
||||
.filter((stmt) => stmt.length > 0)
|
||||
}
|
||||
|
||||
/** Apply all `*.sql` migrations in filename order. Idempotent (every migration is IF NOT EXISTS). */
|
||||
export async function runMigrations(query: QueryFn): Promise<void> {
|
||||
const dir = migrationsDir()
|
||||
const files = (await readdir(dir)).filter((f) => f.endsWith('.sql')).sort()
|
||||
for (const file of files) {
|
||||
const sql = await readFile(join(dir, file), 'utf8')
|
||||
for (const stmt of splitStatements(sql)) {
|
||||
await query(stmt, [])
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -30,6 +30,7 @@ import { createDeprovisioner } from './deprovision/deprovision.js'
|
||||
import { createAuthorizer, type CapabilityVerifier } from './api/authz.js'
|
||||
import { buildRouter } from './api/provision.js'
|
||||
import { buildCaSigner, inProcessCaSigner, type KmsResolver } from './boot/ca-wiring.js'
|
||||
import { configureCapabilityVerifyKey } from './boot/verifier.js'
|
||||
import type { RevocationBus } from 'relay-contracts'
|
||||
|
||||
export interface ControlPlaneOverrides {
|
||||
@@ -40,9 +41,13 @@ export interface ControlPlaneOverrides {
|
||||
readonly caChainDer?: readonly Uint8Array[]
|
||||
}
|
||||
|
||||
/** Default fail-closed verifier — refuses everything until P5 is wired (INV6). */
|
||||
/**
|
||||
* Default fail-closed verifier — refuses everything until a real (P5) verifier is injected (INV6).
|
||||
* Async-shaped to match `CapabilityVerifier`: an async body that throws rejects the promise, so the
|
||||
* authorizer's `await` surfaces it as a 401.
|
||||
*/
|
||||
const refuseAllVerifier: CapabilityVerifier = {
|
||||
verify() {
|
||||
async verify(): Promise<never> {
|
||||
throw new Error('capability verification not configured (P5 integration point)')
|
||||
},
|
||||
}
|
||||
@@ -92,8 +97,16 @@ export async function buildControlPlane(
|
||||
const deprovisioner = createDeprovisioner({ hosts, routing })
|
||||
void bus // reserved for the node-coordinator / revoker wiring (drain + revoke publishers)
|
||||
|
||||
// A real (injected) verifier is P5's async `verifyCapabilityToken`, which reads its Ed25519 key
|
||||
// from relay-auth's startup registry — so load that key from env at boot. The fail-closed default
|
||||
// never reads a key, so leave relay-auth's registry untouched when nothing is injected (INV6).
|
||||
const verifier = overrides.verifier ?? refuseAllVerifier
|
||||
if (overrides.verifier !== undefined) {
|
||||
await configureCapabilityVerifyKey(env.capabilitySignPubkey)
|
||||
}
|
||||
|
||||
const authorizer = createAuthorizer({
|
||||
verifier: overrides.verifier ?? refuseAllVerifier,
|
||||
verifier,
|
||||
expectedAud: env.baseDomain,
|
||||
})
|
||||
|
||||
|
||||
600
control-plane/src/store/pg.ts
Normal file
600
control-plane/src/store/pg.ts
Normal file
@@ -0,0 +1,600 @@
|
||||
/**
|
||||
* A1 — Postgres adapter for the repository ports (store/ports.ts). Each store maps its port
|
||||
* methods to PARAMETERIZED SQL over the injected `query` wrapper (db/pool.ts) — values ALWAYS
|
||||
* travel as bound params, never string-interpolated (SQLi structurally impossible, §7).
|
||||
*
|
||||
* This adapter MUST reproduce store/memory.ts observable semantics EXACTLY:
|
||||
* - INV8 versioning: `swapStatus` appends a `*_status_versions` row AND bumps the single-row
|
||||
* `status`/`status_version` pointer ATOMICALLY. Postgres has real transactions, but we get
|
||||
* the same all-or-nothing with a single writable-CTE statement (no explicit tx / PoolClient).
|
||||
* - `insert` duplicates throw (PK / UNIQUE violation, pg code 23505 → `throw new Error`).
|
||||
* - `casRedeem` is a single-winner CAS of `redeemed_at` from null.
|
||||
* - `RouteStore` fails closed on TTL: `expires_at <= now()` ⇒ absent (INV7), lazily DELETEd.
|
||||
* - metering/audit are append-only with inclusive `[from,to]` time-window queries.
|
||||
*
|
||||
* Timestamps: DB columns are `timestamptz` (node-pg returns them as JS `Date`); records use ISO
|
||||
* strings. `toIso` normalizes Date|string → ISO so returned records match memory.ts string shapes.
|
||||
*/
|
||||
import type {
|
||||
AccountRecord,
|
||||
AccountStatus,
|
||||
AccountStatusVersionRow,
|
||||
HostRecord,
|
||||
HostStatus,
|
||||
HostStatusVersionRow,
|
||||
MeteringSampleRow,
|
||||
PairingCodeRecord,
|
||||
PlanTier,
|
||||
RouteEntry,
|
||||
SessionRecord,
|
||||
} from '../model/records.js'
|
||||
import type { QueryFn } from '../db/pool.js'
|
||||
import type {
|
||||
AccountStore,
|
||||
AuditRow,
|
||||
AuditStore,
|
||||
CasOutcome,
|
||||
HostStore,
|
||||
MeteringStore,
|
||||
NodeRow,
|
||||
NodeStatus,
|
||||
NodeStore,
|
||||
PairingRow,
|
||||
PairingStore,
|
||||
RouteStore,
|
||||
SessionStore,
|
||||
Stores,
|
||||
SubdomainStore,
|
||||
} from './ports.js'
|
||||
|
||||
// ---- helpers ----------------------------------------------------------------------------------
|
||||
|
||||
const toIso = (v: Date | string): string => (v instanceof Date ? v : new Date(v)).toISOString()
|
||||
const toIsoN = (v: Date | string | null): string | null => (v === null ? null : toIso(v))
|
||||
|
||||
/** pg unique/PK violation. */
|
||||
function isUniqueViolation(err: unknown): err is { code: string; constraint?: string } {
|
||||
return typeof err === 'object' && err !== null && (err as { code?: unknown }).code === '23505'
|
||||
}
|
||||
|
||||
// ---- raw DB row shapes ------------------------------------------------------------------------
|
||||
|
||||
interface AccountRow {
|
||||
account_id: string
|
||||
plan: PlanTier
|
||||
created_at: Date | string
|
||||
status: AccountStatus
|
||||
}
|
||||
interface HostRow {
|
||||
host_id: string
|
||||
account_id: string
|
||||
subdomain: string
|
||||
agent_pubkey: Buffer
|
||||
enroll_fpr: string
|
||||
status: HostStatus
|
||||
last_seen: Date | string
|
||||
created_at: Date | string
|
||||
revoked_at: Date | string | null
|
||||
}
|
||||
interface SessionRow {
|
||||
session_id: string
|
||||
host_id: string
|
||||
account_id: string
|
||||
created_at: Date | string
|
||||
last_attach_at: Date | string
|
||||
}
|
||||
interface AccountVersionRow {
|
||||
account_id: string
|
||||
version: number
|
||||
status: AccountStatus
|
||||
changed_at: Date | string
|
||||
changed_by: string
|
||||
}
|
||||
interface HostVersionRow {
|
||||
host_id: string
|
||||
version: number
|
||||
status: HostStatus
|
||||
revoked_at: Date | string | null
|
||||
changed_at: Date | string
|
||||
changed_by: string
|
||||
}
|
||||
|
||||
function mapAccount(r: AccountRow): AccountRecord {
|
||||
return { accountId: r.account_id, plan: r.plan, createdAt: toIso(r.created_at), status: r.status }
|
||||
}
|
||||
function mapHost(r: HostRow): HostRecord {
|
||||
return {
|
||||
hostId: r.host_id,
|
||||
accountId: r.account_id,
|
||||
subdomain: r.subdomain,
|
||||
agentPubkey: new Uint8Array(r.agent_pubkey),
|
||||
enrollFpr: r.enroll_fpr,
|
||||
status: r.status,
|
||||
lastSeen: toIso(r.last_seen),
|
||||
createdAt: toIso(r.created_at),
|
||||
revokedAt: toIsoN(r.revoked_at),
|
||||
}
|
||||
}
|
||||
|
||||
// ---- AccountStore -----------------------------------------------------------------------------
|
||||
|
||||
function pgAccountStore(query: QueryFn): AccountStore {
|
||||
return {
|
||||
async insert(rec) {
|
||||
// Writable CTE: create the row AND its INV8 version-1 companion in ONE statement.
|
||||
// version-1 changed_at = created_at, changed_by = 'system' (matches memory.ts).
|
||||
try {
|
||||
await query(
|
||||
`WITH ins AS (
|
||||
INSERT INTO accounts (account_id, plan, created_at, status, status_version)
|
||||
VALUES ($1, $2, $3, $4, 1)
|
||||
RETURNING account_id, created_at, status
|
||||
)
|
||||
INSERT INTO account_status_versions (account_id, version, status, changed_at, changed_by)
|
||||
SELECT account_id, 1, status, created_at, 'system' FROM ins`,
|
||||
[rec.accountId, rec.plan, rec.createdAt, rec.status],
|
||||
)
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) throw new Error('duplicate accountId')
|
||||
throw err
|
||||
}
|
||||
},
|
||||
async get(accountId) {
|
||||
const rows = await query<AccountRow>(
|
||||
`SELECT account_id, plan, created_at, status FROM accounts WHERE account_id = $1`,
|
||||
[accountId],
|
||||
)
|
||||
const row = rows[0]
|
||||
return row === undefined ? null : mapAccount(row)
|
||||
},
|
||||
async swapStatus(accountId, status, changedBy) {
|
||||
// Atomic INV8: bump pointer + append version row in one writable-CTE statement.
|
||||
const rows = await query<AccountRow>(
|
||||
`WITH upd AS (
|
||||
UPDATE accounts SET status = $2, status_version = status_version + 1
|
||||
WHERE account_id = $1
|
||||
RETURNING account_id, plan, created_at, status, status_version
|
||||
),
|
||||
ins AS (
|
||||
INSERT INTO account_status_versions (account_id, version, status, changed_by)
|
||||
SELECT account_id, status_version, status, $3 FROM upd
|
||||
)
|
||||
SELECT account_id, plan, created_at, status FROM upd`,
|
||||
[accountId, status, changedBy],
|
||||
)
|
||||
const row = rows[0]
|
||||
if (row === undefined) throw new Error('account not found')
|
||||
return mapAccount(row)
|
||||
},
|
||||
async versions(accountId) {
|
||||
const rows = await query<AccountVersionRow>(
|
||||
`SELECT account_id, version, status, changed_at, changed_by
|
||||
FROM account_status_versions WHERE account_id = $1 ORDER BY version ASC`,
|
||||
[accountId],
|
||||
)
|
||||
return rows.map(
|
||||
(r): AccountStatusVersionRow => ({
|
||||
accountId: r.account_id,
|
||||
version: r.version,
|
||||
status: r.status,
|
||||
changedAt: toIso(r.changed_at),
|
||||
changedBy: r.changed_by,
|
||||
}),
|
||||
)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- HostStore --------------------------------------------------------------------------------
|
||||
|
||||
function pgHostStore(query: QueryFn): HostStore {
|
||||
return {
|
||||
async insert(rec) {
|
||||
try {
|
||||
await query(
|
||||
`WITH ins AS (
|
||||
INSERT INTO hosts
|
||||
(host_id, account_id, subdomain, agent_pubkey, enroll_fpr, status, last_seen, created_at, revoked_at, status_version)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 1)
|
||||
RETURNING host_id, created_at, status, revoked_at
|
||||
)
|
||||
INSERT INTO host_status_versions (host_id, version, status, revoked_at, changed_at, changed_by)
|
||||
SELECT host_id, 1, status, revoked_at, created_at, 'system' FROM ins`,
|
||||
[
|
||||
rec.hostId,
|
||||
rec.accountId,
|
||||
rec.subdomain,
|
||||
Buffer.from(rec.agentPubkey),
|
||||
rec.enrollFpr,
|
||||
rec.status,
|
||||
rec.lastSeen,
|
||||
rec.createdAt,
|
||||
rec.revokedAt,
|
||||
],
|
||||
)
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) {
|
||||
// subdomain UNIQUE is the subdomain single-winner substrate; PK is host_id.
|
||||
throw new Error(err.constraint?.includes('subdomain') ? 'duplicate subdomain' : 'duplicate hostId')
|
||||
}
|
||||
throw err
|
||||
}
|
||||
},
|
||||
async get(hostId) {
|
||||
const rows = await query<HostRow>(`SELECT * FROM hosts WHERE host_id = $1`, [hostId])
|
||||
const row = rows[0]
|
||||
return row === undefined ? null : mapHost(row)
|
||||
},
|
||||
async getBySubdomain(subdomain) {
|
||||
const rows = await query<HostRow>(`SELECT * FROM hosts WHERE subdomain = $1`, [subdomain])
|
||||
const row = rows[0]
|
||||
return row === undefined ? null : mapHost(row)
|
||||
},
|
||||
async listByAccount(accountId) {
|
||||
const rows = await query<HostRow>(`SELECT * FROM hosts WHERE account_id = $1 ORDER BY created_at ASC`, [
|
||||
accountId,
|
||||
])
|
||||
return rows.map(mapHost)
|
||||
},
|
||||
async swapStatus(hostId, status, revokedAt, changedBy) {
|
||||
const rows = await query<HostRow>(
|
||||
`WITH upd AS (
|
||||
UPDATE hosts SET status = $2, revoked_at = $3, status_version = status_version + 1
|
||||
WHERE host_id = $1
|
||||
RETURNING host_id, account_id, subdomain, agent_pubkey, enroll_fpr, status, last_seen, created_at, revoked_at, status_version
|
||||
),
|
||||
ins AS (
|
||||
INSERT INTO host_status_versions (host_id, version, status, revoked_at, changed_by)
|
||||
SELECT host_id, status_version, status, revoked_at, $4 FROM upd
|
||||
)
|
||||
SELECT host_id, account_id, subdomain, agent_pubkey, enroll_fpr, status, last_seen, created_at, revoked_at FROM upd`,
|
||||
[hostId, status, revokedAt, changedBy],
|
||||
)
|
||||
const row = rows[0]
|
||||
if (row === undefined) throw new Error('host not found')
|
||||
return mapHost(row)
|
||||
},
|
||||
async touchLastSeen(hostId, ts) {
|
||||
// Advisory cache — NO version row (INV7). Silent no-op if absent (matches memory.ts).
|
||||
await query(`UPDATE hosts SET last_seen = $2 WHERE host_id = $1`, [hostId, ts])
|
||||
},
|
||||
async versions(hostId) {
|
||||
const rows = await query<HostVersionRow>(
|
||||
`SELECT host_id, version, status, revoked_at, changed_at, changed_by
|
||||
FROM host_status_versions WHERE host_id = $1 ORDER BY version ASC`,
|
||||
[hostId],
|
||||
)
|
||||
return rows.map(
|
||||
(r): HostStatusVersionRow => ({
|
||||
hostId: r.host_id,
|
||||
version: r.version,
|
||||
status: r.status,
|
||||
revokedAt: toIsoN(r.revoked_at),
|
||||
changedAt: toIso(r.changed_at),
|
||||
changedBy: r.changed_by,
|
||||
}),
|
||||
)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- SessionStore -----------------------------------------------------------------------------
|
||||
|
||||
function pgSessionStore(query: QueryFn): SessionStore {
|
||||
return {
|
||||
async insert(rec) {
|
||||
try {
|
||||
await query(
|
||||
`INSERT INTO sessions (session_id, host_id, account_id, created_at, last_attach_at)
|
||||
VALUES ($1, $2, $3, $4, $5)`,
|
||||
[rec.sessionId, rec.hostId, rec.accountId, rec.createdAt, rec.lastAttachAt],
|
||||
)
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) throw new Error('duplicate sessionId')
|
||||
throw err
|
||||
}
|
||||
},
|
||||
async get(sessionId) {
|
||||
const rows = await query<SessionRow>(`SELECT * FROM sessions WHERE session_id = $1`, [sessionId])
|
||||
const row = rows[0]
|
||||
if (row === undefined) return null
|
||||
return {
|
||||
sessionId: row.session_id,
|
||||
hostId: row.host_id,
|
||||
accountId: row.account_id,
|
||||
createdAt: toIso(row.created_at),
|
||||
lastAttachAt: toIso(row.last_attach_at),
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- SubdomainStore ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* No separate reservations table exists in the schema — the `hosts.subdomain UNIQUE` constraint
|
||||
* is the single-winner substrate (the true winner is decided at host insert). So `reserve`/`isTaken`
|
||||
* just report whether a host already owns the label (matches memory.ts intent). Two concurrent
|
||||
* `reserve()`s can both see "free" before any host row exists; that is fine, because the actual
|
||||
* single-winner is enforced when the losing insert hits the UNIQUE violation (INV1).
|
||||
*/
|
||||
function pgSubdomainStore(query: QueryFn): SubdomainStore {
|
||||
const taken = async (subdomain: string): Promise<boolean> => {
|
||||
const rows = await query<{ one: number }>(`SELECT 1 AS one FROM hosts WHERE subdomain = $1 LIMIT 1`, [
|
||||
subdomain,
|
||||
])
|
||||
return rows.length > 0
|
||||
}
|
||||
return {
|
||||
async reserve(subdomain) {
|
||||
return !(await taken(subdomain))
|
||||
},
|
||||
async isTaken(subdomain) {
|
||||
return taken(subdomain)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- PairingStore -----------------------------------------------------------------------------
|
||||
|
||||
interface PairingCodeRow {
|
||||
code_hash: string
|
||||
account_id: string
|
||||
expires_at: Date | string
|
||||
redeemed_at: Date | string | null
|
||||
redeem_attempts: number
|
||||
}
|
||||
|
||||
function pgPairingStore(query: QueryFn): PairingStore {
|
||||
return {
|
||||
async insert(rec) {
|
||||
try {
|
||||
await query(
|
||||
`INSERT INTO pairing_codes (code_hash, account_id, expires_at, redeemed_at, redeem_attempts)
|
||||
VALUES ($1, $2, $3, $4, 0)`,
|
||||
[rec.codeHash, rec.accountId, rec.expiresAt, rec.redeemedAt],
|
||||
)
|
||||
} catch (err) {
|
||||
if (isUniqueViolation(err)) throw new Error('duplicate code_hash')
|
||||
throw err
|
||||
}
|
||||
},
|
||||
async get(codeHash): Promise<PairingRow | null> {
|
||||
const rows = await query<PairingCodeRow>(`SELECT * FROM pairing_codes WHERE code_hash = $1`, [codeHash])
|
||||
const row = rows[0]
|
||||
if (row === undefined) return null
|
||||
const record: PairingCodeRecord = {
|
||||
codeHash: row.code_hash,
|
||||
accountId: row.account_id,
|
||||
expiresAt: toIso(row.expires_at),
|
||||
redeemedAt: toIsoN(row.redeemed_at),
|
||||
}
|
||||
return { record, redeemAttempts: row.redeem_attempts }
|
||||
},
|
||||
async casRedeem(codeHash, nowIso): Promise<CasOutcome> {
|
||||
// Single-winner CAS: only the row with redeemed_at IS NULL is updated; exactly one wins.
|
||||
const won = await query<{ code_hash: string }>(
|
||||
`UPDATE pairing_codes SET redeemed_at = $2
|
||||
WHERE code_hash = $1 AND redeemed_at IS NULL
|
||||
RETURNING code_hash`,
|
||||
[codeHash, nowIso],
|
||||
)
|
||||
if (won.length === 1) return 'ok'
|
||||
// 0 rows: either already redeemed or unknown — distinguish by existence.
|
||||
const exists = await query<{ one: number }>(`SELECT 1 AS one FROM pairing_codes WHERE code_hash = $1`, [
|
||||
codeHash,
|
||||
])
|
||||
return exists.length > 0 ? 'already_redeemed' : 'unknown'
|
||||
},
|
||||
async registerFailure(codeHash) {
|
||||
const rows = await query<{ redeem_attempts: number }>(
|
||||
`UPDATE pairing_codes SET redeem_attempts = redeem_attempts + 1
|
||||
WHERE code_hash = $1
|
||||
RETURNING redeem_attempts`,
|
||||
[codeHash],
|
||||
)
|
||||
const row = rows[0]
|
||||
return row === undefined ? 0 : row.redeem_attempts // 0 for unknown code (matches memory.ts)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- RouteStore -------------------------------------------------------------------------------
|
||||
|
||||
interface RouteRow {
|
||||
host_id: string
|
||||
relay_node_id: string
|
||||
updated_at: Date | string
|
||||
}
|
||||
|
||||
function pgRouteStore(query: QueryFn): RouteStore {
|
||||
const mapEntry = (r: RouteRow): RouteEntry => ({ relayNodeId: r.relay_node_id, updatedAt: toIso(r.updated_at) })
|
||||
return {
|
||||
async set(hostId, entry, ttlSec) {
|
||||
// Postgres has no per-key TTL: store an explicit expires_at = now() + ttl (server clock).
|
||||
await query(
|
||||
`INSERT INTO routes (host_id, relay_node_id, updated_at, expires_at)
|
||||
VALUES ($1, $2, $3, now() + make_interval(secs => $4))
|
||||
ON CONFLICT (host_id) DO UPDATE SET
|
||||
relay_node_id = EXCLUDED.relay_node_id,
|
||||
updated_at = EXCLUDED.updated_at,
|
||||
expires_at = EXCLUDED.expires_at`,
|
||||
[hostId, entry.relayNodeId, entry.updatedAt, ttlSec],
|
||||
)
|
||||
},
|
||||
async refreshTtl(hostId, ttlSec) {
|
||||
// Refresh only a still-live key (fail closed on already-expired/absent, INV7).
|
||||
const rows = await query<{ host_id: string }>(
|
||||
`UPDATE routes SET expires_at = now() + make_interval(secs => $2)
|
||||
WHERE host_id = $1 AND expires_at > now()
|
||||
RETURNING host_id`,
|
||||
[hostId, ttlSec],
|
||||
)
|
||||
return rows.length === 1
|
||||
},
|
||||
async get(hostId) {
|
||||
// Lazily reap the expired row, then a surviving row is guaranteed live (INV7 fail-closed).
|
||||
await query(`DELETE FROM routes WHERE host_id = $1 AND expires_at <= now()`, [hostId])
|
||||
const rows = await query<RouteRow>(
|
||||
`SELECT host_id, relay_node_id, updated_at FROM routes WHERE host_id = $1`,
|
||||
[hostId],
|
||||
)
|
||||
const row = rows[0]
|
||||
return row === undefined ? null : mapEntry(row)
|
||||
},
|
||||
async delete(hostId) {
|
||||
await query(`DELETE FROM routes WHERE host_id = $1`, [hostId])
|
||||
},
|
||||
async listByNode(nodeId) {
|
||||
await query(`DELETE FROM routes WHERE expires_at <= now()`, [])
|
||||
const rows = await query<RouteRow>(
|
||||
`SELECT host_id, relay_node_id, updated_at FROM routes
|
||||
WHERE relay_node_id = $1 AND expires_at > now() ORDER BY host_id ASC`,
|
||||
[nodeId],
|
||||
)
|
||||
return rows.map((r) => ({ hostId: r.host_id, entry: mapEntry(r) }))
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- NodeStore --------------------------------------------------------------------------------
|
||||
|
||||
interface NodeDbRow {
|
||||
node_id: string
|
||||
addr: string
|
||||
status: NodeStatus
|
||||
last_seen: Date | string
|
||||
}
|
||||
|
||||
function pgNodeStore(query: QueryFn): NodeStore {
|
||||
return {
|
||||
async upsert(row) {
|
||||
await query(
|
||||
`INSERT INTO relay_nodes (node_id, addr, status, last_seen)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (node_id) DO UPDATE SET
|
||||
addr = EXCLUDED.addr, status = EXCLUDED.status, last_seen = EXCLUDED.last_seen`,
|
||||
[row.nodeId, row.addr, row.status, row.lastSeen],
|
||||
)
|
||||
},
|
||||
async get(nodeId) {
|
||||
const rows = await query<NodeDbRow>(`SELECT * FROM relay_nodes WHERE node_id = $1`, [nodeId])
|
||||
const row = rows[0]
|
||||
if (row === undefined) return null
|
||||
return { nodeId: row.node_id, addr: row.addr, status: row.status, lastSeen: toIso(row.last_seen) }
|
||||
},
|
||||
async setStatus(nodeId, status) {
|
||||
const rows = await query<{ node_id: string }>(
|
||||
`UPDATE relay_nodes SET status = $2 WHERE node_id = $1 RETURNING node_id`,
|
||||
[nodeId, status],
|
||||
)
|
||||
if (rows.length === 0) throw new Error('node not found')
|
||||
},
|
||||
async touch(nodeId, ts) {
|
||||
// Silent no-op if absent (matches memory.ts).
|
||||
await query(`UPDATE relay_nodes SET last_seen = $2 WHERE node_id = $1`, [nodeId, ts])
|
||||
},
|
||||
async delete(nodeId) {
|
||||
await query(`DELETE FROM relay_nodes WHERE node_id = $1`, [nodeId])
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- MeteringStore ----------------------------------------------------------------------------
|
||||
|
||||
interface MeteringDbRow {
|
||||
host_id: string
|
||||
account_id: string
|
||||
node_id: string
|
||||
concurrent_viewers: number
|
||||
sampled_at: Date | string
|
||||
}
|
||||
|
||||
function pgMeteringStore(query: QueryFn): MeteringStore {
|
||||
return {
|
||||
async append(row) {
|
||||
await query(
|
||||
`INSERT INTO metering_samples (host_id, account_id, node_id, concurrent_viewers, sampled_at)
|
||||
VALUES ($1, $2, $3, $4, $5)`,
|
||||
[row.hostId, row.accountId, row.nodeId, row.concurrentViewers, row.sampledAt],
|
||||
)
|
||||
},
|
||||
async query(accountId, fromIso, toIso2) {
|
||||
const rows = await query<MeteringDbRow>(
|
||||
`SELECT host_id, account_id, node_id, concurrent_viewers, sampled_at
|
||||
FROM metering_samples
|
||||
WHERE account_id = $1 AND sampled_at >= $2 AND sampled_at <= $3
|
||||
ORDER BY id ASC`,
|
||||
[accountId, fromIso, toIso2],
|
||||
)
|
||||
return rows.map(
|
||||
(r): MeteringSampleRow => ({
|
||||
hostId: r.host_id,
|
||||
accountId: r.account_id,
|
||||
nodeId: r.node_id,
|
||||
concurrentViewers: r.concurrent_viewers,
|
||||
sampledAt: toIso(r.sampled_at),
|
||||
}),
|
||||
)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- AuditStore -------------------------------------------------------------------------------
|
||||
|
||||
interface AuditDbRow {
|
||||
action: string
|
||||
principal_id: string
|
||||
account_id: string
|
||||
host_id: string | null
|
||||
ts: Date | string
|
||||
meta: Record<string, string>
|
||||
}
|
||||
|
||||
function pgAuditStore(query: QueryFn): AuditStore {
|
||||
return {
|
||||
async append(row) {
|
||||
await query(
|
||||
`INSERT INTO audit_log (action, principal_id, account_id, host_id, ts, meta)
|
||||
VALUES ($1, $2, $3, $4, $5, $6::jsonb)`,
|
||||
[row.action, row.principalId, row.accountId, row.hostId, row.ts, JSON.stringify(row.meta)],
|
||||
)
|
||||
},
|
||||
async query(accountId, fromIso, toIso2) {
|
||||
const rows = await query<AuditDbRow>(
|
||||
`SELECT action, principal_id, account_id, host_id, ts, meta
|
||||
FROM audit_log
|
||||
WHERE account_id = $1 AND ts >= $2 AND ts <= $3
|
||||
ORDER BY id ASC`,
|
||||
[accountId, fromIso, toIso2],
|
||||
)
|
||||
return rows.map(
|
||||
(r): AuditRow => ({
|
||||
action: r.action,
|
||||
principalId: r.principal_id,
|
||||
accountId: r.account_id,
|
||||
hostId: r.host_id,
|
||||
ts: toIso(r.ts),
|
||||
meta: r.meta,
|
||||
}),
|
||||
)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- assembly ---------------------------------------------------------------------------------
|
||||
|
||||
/** Build a full Postgres-backed store set (all ports mapped to parameterized SQL over `query`). */
|
||||
export function createPgStores(query: QueryFn): Stores {
|
||||
return {
|
||||
accounts: pgAccountStore(query),
|
||||
hosts: pgHostStore(query),
|
||||
sessions: pgSessionStore(query),
|
||||
subdomains: pgSubdomainStore(query),
|
||||
pairing: pgPairingStore(query),
|
||||
routes: pgRouteStore(query),
|
||||
nodes: pgNodeStore(query),
|
||||
metering: pgMeteringStore(query),
|
||||
audit: pgAuditStore(query),
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user