Reference source · 0.1.0-dev.11
change-feed.ts
src/change-feed.ts · Download original file
import type { Visibility } from './visibility.ts';
import { randomBytes } from 'node:crypto';
import type { HostDatabase } from './host-storage.ts';
import type { CredentialBody } from './identity.ts';
import { AuthorizationService, SCOPES } from './authorization.ts';
import { b64, unb64, jsonBytes, exact, integer, fingerprint, newId, requireThat } from './core.ts';
import { RESOURCE_TYPES, RESOURCE_SCOPES } from './resource-profile.ts';
import { protect, reveal } from './opaque.ts';
export type ResourceVersion = { resource: Record<string, any>; version: number };
export type Change = ResourceVersion & { event_id: string; kind: 'created' | 'updated' | 'deleted' };
export type Snapshot = { status: 'ok'; resources: ResourceVersion[]; cursor: string; has_more: boolean; page_cursor: string | null; snapshot_id: string; total: number; visibility_epoch:string };
export type ChangePage = { status: 'ok'; events: Change[]; cursor: string; has_more: boolean; visibility_epoch:string };
export class ChangeFeed {
visibility!: Visibility;
db: HostDatabase; authorization: AuthorizationService; host: string; secret: Buffer; epoch: string;
constructor(db: HostDatabase, authorization: AuthorizationService, host: string) {
this.db = db; this.authorization = authorization; this.host = host;
db.exec(`CREATE TABLE IF NOT EXISTS change_events (sequence INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT UNIQUE NOT NULL, patient TEXT NOT NULL, resource_type TEXT NOT NULL, resource_id TEXT NOT NULL, version INTEGER NOT NULL, kind TEXT NOT NULL, resource TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS change_event_visibility ON change_events(patient,resource_type,sequence);
CREATE TABLE IF NOT EXISTS feed_retention (patient TEXT NOT NULL, resource_type TEXT NOT NULL, sequence INTEGER NOT NULL, PRIMARY KEY(patient,resource_type));
CREATE TABLE IF NOT EXISTS visibility_retention(grant_id TEXT PRIMARY KEY,sequence INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS feed_meta (id INTEGER PRIMARY KEY CHECK(id=1), epoch TEXT NOT NULL, secret TEXT NOT NULL, low_water INTEGER NOT NULL);`);
db.prepare('INSERT INTO feed_meta VALUES (1, ?, ?, 0) ON CONFLICT(id) DO NOTHING').run(newId(), b64(randomBytes(32)));
const meta = db.prepare('SELECT * FROM feed_meta WHERE id=1').get() as any;
this.secret = unb64(meta.secret, 32); this.epoch = meta.epoch;
}
context(): Buffer { return jsonBytes(['OPRP-FEED-CURSOR-0.1', this.host]); }
visible(claims: ReturnType<AuthorizationService['authorize']>): string[] {
const types: string[] = [];
if (claims.scope.includes(SCOPES.demographicsRead)) types.push('Patient');
if (claims.scope.includes(SCOPES.labsRead)) types.push('Observation');
for (const type of RESOURCE_TYPES) if ((claims.scope as string[]).includes(RESOURCE_SCOPES[type].read)) types.push(type);
requireThat(types.length, 'insufficient_scope'); return types;
}
watermark(): number {
// sqlite_sequence preserves the high watermark even when all retained events are pruned.
const row = this.db.prepare('SELECT seq FROM sqlite_sequence WHERE name=?').get('change_events') as any;
const value = row?.seq ?? 0; integer(value); return value;
}
cursor(claims: ReturnType<AuthorizationService['authorize']>, types: string[], after: number, now: number): string {
const payload = { v: '0.1', epoch: this.epoch, host: this.host, app: claims.sub, kid: claims.cnf.oprp_signing_kid, patient: claims.patient, grant: claims.grant_id, visibility: fingerprint(jsonBytes([types,this.visibility.epoch(claims.grant_id)])), after, expires_at: Math.min(now + 3600, this.authorization.grant(claims.grant_id).until) };
const wrapped = { payload, padding: '' };
const length = jsonBytes(wrapped).length; requireThat(length <= 2048, 'cursor_context_too_large');
wrapped.padding = '0'.repeat(2048 - length);
// Fixed plaintext length also hides growth in the number of digits of the private global counter.
return protect(wrapped, this.secret, this.context());
}
snapshot(token: string, app: CredentialBody, patient: string, now: number, inTransaction=false): Pick<Snapshot,'status'|'resources'|'cursor'|'visibility_epoch'> {
const claims = this.authorization.authorize(token, app, patient, undefined, now), types = this.visible(claims);
const build = () => {
const resources: ResourceVersion[] = [];
if (types.includes('Patient')) {
const row = this.db.prepare('SELECT version,body FROM patients WHERE id=?').get(patient) as any;
requireThat(row, 'not_found'); resources.push({ resource: JSON.parse(row.body), version: row.version });
}
if (types.includes('Observation')) {
const rows = this.db.prepare('SELECT body FROM labs WHERE patient=? ORDER BY id').all(patient) as any[];
rows.forEach(row => resources.push({ resource: JSON.parse(row.body), version: 1 }));
}
const clinical = this.db.prepare('SELECT body,version,resource_type FROM clinical_resources WHERE patient=? AND deleted=0 ORDER BY resource_type,id').all(patient) as any[];
clinical.filter(row => types.includes(row.resource_type)).forEach(row => resources.push({resource:JSON.parse(row.body),version:row.version}));
const filtered=resources.flatMap(entry=>{const resource=this.visibility.project(claims.grant_id,entry.resource);return resource?[{...entry,resource}]:[];});
return { status: 'ok' as const, resources:filtered,visibility_epoch:this.visibility.epoch(claims.grant_id),cursor: this.cursor(claims, types, this.watermark(), now) };
};
return inTransaction?build():this.authorization.transaction(build);
}
read(token: string, app: CredentialBody, patient: string, cursor: string, limit: number, now: number): ChangePage {
const claims = this.authorization.authorize(token, app, patient, undefined, now), types = this.visible(claims);
integer(limit); requireThat(limit >= 1 && limit <= 100, 'invalid_page_size');
const wrapped = reveal(cursor, this.secret, this.context()); exact(wrapped, ['payload', 'padding']);
requireThat(typeof wrapped.padding === 'string' && /^0*$/.test(wrapped.padding) && jsonBytes(wrapped).length === 2048, 'invalid_cursor');
const c = wrapped.payload;
exact(c, ['v','epoch','host','app','kid','patient','grant','visibility','after','expires_at']); integer(c.after); integer(c.expires_at);
requireThat(c.v === '0.1' && c.epoch === this.epoch && c.host === this.host && c.app === app.subject && c.kid === app.signingKid && c.patient === patient && c.grant === claims.grant_id, 'cursor_context_changed');
requireThat(c.visibility===fingerprint(jsonBytes([types,this.visibility.epoch(claims.grant_id)])),'visibility_changed');
requireThat(now < c.expires_at, 'cursor_expired');
return this.authorization.transaction(() => {
this.authorization.authorize(token,app,patient,undefined,now);
requireThat(c.visibility===fingerprint(jsonBytes([types,this.visibility.epoch(claims.grant_id)])),'visibility_changed');
const retention=this.db.prepare('SELECT sequence FROM visibility_retention WHERE grant_id=?').get(claims.grant_id) as any;
requireThat(c.after >= (retention?.sequence ?? 0), 'cursor_expired');
const watermark = this.watermark(); requireThat(c.after <= watermark, 'invalid_cursor');
const rows = this.db.prepare(`SELECT * FROM change_events WHERE patient=? AND resource_type IN (${types.map(() => '?').join(',')}) AND sequence>? AND sequence<=? ORDER BY sequence`).all(patient, ...types, c.after, watermark) as any[];
const visibleRows=rows.flatMap(row=>{const resource=this.visibility.event(claims.grant_id,patient,JSON.parse(row.resource),row.kind);return resource?[{...row,resource:JSON.stringify(resource)}]:[];});
const page:any[]=[];let bytes=0;for(const row of visibleRows.slice(0,limit)){const size=Buffer.byteLength(row.resource)+512;if(bytes+size>700000)break;page.push(row);bytes+=size;}
requireThat(page.length>0||visibleRows.length===0,'resource_too_large');
const hasMore = visibleRows.length > page.length;
const next = hasMore ? page[page.length - 1].sequence : watermark;
return { status: 'ok',visibility_epoch:this.visibility.epoch(claims.grant_id),events: page.map(row => ({ event_id: row.event_id, kind: row.kind, version: row.version, resource: JSON.parse(row.resource) })), cursor: this.cursor(claims, types, next, now), has_more: hasMore };
});
}
// Call inside the mutation transaction, after the resource was accepted.
append(patient: string, resource: Record<string, any>, version: number, kind: Change['kind']): void {
integer(version);
this.db.prepare('INSERT INTO change_events(event_id,patient,resource_type,resource_id,version,kind,resource) VALUES (?, ?, ?, ?, ?, ?, ?)').run(newId(), patient, resource.resourceType, resource.id, version, kind, JSON.stringify(resource));
this.visibility?.notify(patient,resource,kind);
}
// Host-only retention administration. Clients recover from expired cursors using a new snapshot.
retainAfter(sequence: number): void {
integer(sequence); requireThat(sequence <= this.watermark(), 'invalid_retention_watermark');
this.authorization.transaction(() => {
const events=this.db.prepare('SELECT * FROM change_events WHERE sequence<=? ORDER BY sequence').all(sequence) as any[];
for(const row of this.db.prepare('SELECT body FROM grants').all() as any[]){const g=JSON.parse(row.body);if(!g.active)continue;let types:string[];try{types=this.visible({scope:g.scopes} as any);}catch{continue;}let lostSequence=0;for(const e of events)if(e.patient===g.patient&&types.includes(e.resource_type)&&this.visibility.event(g.grant_id,g.patient,JSON.parse(e.resource),e.kind))lostSequence=e.sequence;if(lostSequence)this.db.prepare('INSERT INTO visibility_retention VALUES (?,?) ON CONFLICT(grant_id) DO UPDATE SET sequence=max(sequence,excluded.sequence)').run(g.grant_id,lostSequence);}
const lost = this.db.prepare('SELECT patient,resource_type,max(sequence) AS sequence FROM change_events WHERE sequence<=? GROUP BY patient,resource_type').all(sequence) as any[];
lost.forEach(row => this.db.prepare('INSERT INTO feed_retention VALUES (?, ?, ?) ON CONFLICT(patient,resource_type) DO UPDATE SET sequence=max(sequence,excluded.sequence)').run(row.patient, row.resource_type, row.sequence));
this.db.prepare('DELETE FROM change_events WHERE sequence<=?').run(sequence);
this.db.prepare('UPDATE feed_meta SET low_water=max(low_water, ?) WHERE id=1').run(sequence);
});
}
close(): void { this.secret.fill(0); }
}