Reference source · 0.1.0-dev.11
subscriptions.ts
src/subscriptions.ts · Download original file
import { randomBytes } from 'node:crypto';
import type { ReferenceHost } from './host.ts';
import type { Credential, CredentialBody } from './identity.ts';
import type { Transport } from './sdk.ts';
import { seal, open } from './envelope.ts';
import { b64, exact, fingerprint, id, integer, jsonBytes, newId, ProtocolError, requireThat, unb64, parseObject } from './core.ts';
import type { ResourceType } from './resource-profile.ts';
import { callbackURL, publicCallbackTransport } from './callback-transport.ts';
import type { CallbackTransportFactory } from './callback-transport.ts';
export type Topic = 'Patient' | 'Observation' | ResourceType;
export type SubscriptionView = { status: 'ok'; subscription_id: string; state: 'pending' | 'active' | 'suspended' | 'cancelled'; expires_at: number; topics: Topic[] };
type Subscription = { id: string; app: string; patient: string; grant_id: string; credential: Credential; callback: string; topics: Topic[]; expires_at: number; state: SubscriptionView['state']; created_at: number; challenge: string; challenge_until: number; digest: string };
export class Subscriptions {
host: ReferenceHost; running = false;
constructor(host: ReferenceHost) {
this.host = host;
host.db.exec(`CREATE TABLE IF NOT EXISTS subscriptions (id TEXT PRIMARY KEY, app TEXT NOT NULL, patient TEXT NOT NULL, state TEXT NOT NULL, body TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS notification_outbox (subscription_id TEXT PRIMARY KEY REFERENCES subscriptions(id), delivery_id TEXT NOT NULL, revision INTEGER NOT NULL, attempts INTEGER NOT NULL, next_at INTEGER NOT NULL, lease_until INTEGER NOT NULL, lease_id TEXT);
CREATE INDEX IF NOT EXISTS subscription_visibility ON subscriptions(patient,state);
DROP TRIGGER IF EXISTS subscription_change;`);
}
get(subscriptionId: string): Subscription {
id(subscriptionId); const row = this.host.db.prepare('SELECT body FROM subscriptions WHERE id=?').get(subscriptionId) as any;
requireThat(row, 'unknown_subscription'); return JSON.parse(row.body);
}
save(s: Subscription): void {
this.host.db.prepare('INSERT INTO subscriptions VALUES (?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET state=excluded.state,body=excluded.body').run(s.id, s.app, s.patient, s.state, JSON.stringify(s));
}
view(s: Subscription): SubscriptionView { return { status: 'ok', subscription_id: s.id, state: s.state, expires_at: s.expires_at, topics: s.topics }; }
current(s: Subscription, now: number): CredentialBody {
const h = this.host;
requireThat(now < s.expires_at, 'subscription_expired'); h.migration.assertReadable(s.patient);
const app = h.trust.validate(s.credential, now, s.app, 'application'); h.authorization.requester(app);
const grant = h.authorization.grant(s.grant_id); h.authorization.current(grant, app, now);
requireThat(grant.patient === s.patient && s.topics.every(t => h.feed.visible({ scope: grant.scopes } as any).includes(t)), 'insufficient_scope');
h.trust.validate(h.credential, now, h.identity.public.subject, 'host'); return app;
}
stop(s: Subscription, state: 'suspended' | 'cancelled'): void {
s.state = state; this.save(s); this.host.db.prepare('DELETE FROM notification_outbox WHERE subscription_id=?').run(s.id);
}
register(body: Record<string, any>, app: CredentialBody, credential: Credential, now: number): SubscriptionView {
exact(body, ['operation','patient','access_token','subscription_id','callback','topics','expires_at']);
id(body.subscription_id); id(body.patient); integer(body.expires_at);
const claims = this.host.authorization.authorize(body.access_token, app, body.patient, undefined, now);
const visible = this.host.feed.visible(claims);
requireThat(Array.isArray(body.topics) && body.topics.length > 0 && body.topics.length <= 8 && new Set(body.topics).size === body.topics.length && body.topics.every((t: string) => visible.includes(t)), 'insufficient_scope');
const endpoint = callbackURL(body.callback).href;
requireThat(body.expires_at > now && body.expires_at <= this.host.authorization.grant(claims.grant_id).until, 'invalid_subscription_lifetime');
const digest = fingerprint(jsonBytes([app.subject, app.signingKid, app.encryptionKid, body.patient, claims.grant_id, endpoint, body.topics, body.expires_at]));
return this.host.authorization.transaction(() => {
this.host.migration.assertReadable(body.patient);
requireThat(this.host.db.prepare('SELECT id FROM patients WHERE id=?').get(body.patient), 'not_found');
const previous = this.host.db.prepare('SELECT body FROM subscriptions WHERE id=?').get(body.subscription_id) as any;
if (previous) {
const s: Subscription = JSON.parse(previous.body); requireThat(s.app === app.subject && s.digest === digest, 'subscription_id_conflict');
// Same-key credential renewal does not change subscription authority or lifetime.
s.credential = credential; this.save(s); return this.view(s);
}
const count = this.host.db.prepare("SELECT count(*) AS n FROM subscriptions WHERE app=? AND patient=? AND (state IN ('pending','active') OR json_extract(body,'$.created_at')>?)").get(app.subject, body.patient, now - 60) as any;
requireThat(count.n < 10, 'subscription_rate_limited');
const s: Subscription = { id: body.subscription_id, app: app.subject, patient: body.patient, grant_id: claims.grant_id, credential, callback: endpoint, topics: body.topics,
expires_at: body.expires_at, state: 'pending', created_at: now, challenge: b64(randomBytes(32)), challenge_until: Math.min(now + 300, body.expires_at), digest };
this.save(s); this.enqueue(s.id, now); return this.view(s);
});
}
control(body: Record<string, any>, app: CredentialBody, now: number): SubscriptionView {
exact(body, ['operation','patient','access_token','subscription_id']);
const s = this.get(body.subscription_id), registered = this.host.trust.validate(s.credential, now, s.app, 'application');
requireThat(s.app === app.subject && s.patient === body.patient && registered.signingKid === app.signingKid && registered.encryptionKid === app.encryptionKid, 'wrong_subscription_holder');
const claims = this.host.authorization.authorize(body.access_token, app, body.patient, undefined, now);
requireThat(claims.grant_id === s.grant_id, 'wrong_subscription_grant');
return this.host.authorization.transaction(() => {
if (body.operation === 'unsubscribe') this.stop(s, 'cancelled');
else if (s.state === 'active' || s.state === 'pending') {
try { this.current(s, now); } catch (e) { if (this.permanent(e)) this.stop(s, 'suspended'); else throw e; }
}
return this.view(s);
});
}
enqueue(subscriptionId: string, now: number): void {
this.host.db.prepare('INSERT INTO notification_outbox VALUES (?,?,1,0,?,0,NULL) ON CONFLICT(subscription_id) DO UPDATE SET revision=revision+1').run(subscriptionId, newId(), now);
}
permanent(error: unknown): boolean {
return error instanceof ProtocolError && ['subscription_expired','record_moved','migration_not_active','authorization_expired_or_revoked','wrong_token_holder','representative_not_authorized','insufficient_scope','requester_rotation_required','revoked_credential','disabled_issuer'].includes(error.code);
}
// Explicit operator worker; never sends from the request transaction. A transport override
// is a privileged network-policy decision, used only for synthetic loopback demos/tests here.
async deliver(clock: () => number, factory: CallbackTransportFactory = publicCallbackTransport, limit = 20): Promise<{ acknowledged: number; retried: number; suspended: number }> {
requireThat(!this.running, 'subscription_worker_busy'); integer(limit); requireThat(limit >= 1 && limit <= 100, 'invalid_page_size');
this.running = true; const result = { acknowledged: 0, retried: 0, suspended: 0 };
try {
// Retire quiet subscriptions too: stale grants must not accumulate work on later writes.
for (const row of this.host.db.prepare("SELECT body FROM subscriptions WHERE state IN ('pending','active')").all() as any[]) {
this.host.authorization.transaction(() => {
const original: Subscription = JSON.parse(row.body), s = this.get(original.id);
if (s.state !== 'pending' && s.state !== 'active') return;
try { this.current(s, clock()); } catch (e) { if (this.permanent(e)) { this.stop(s, 'suspended'); result.suspended++; } }
});
}
const rows = this.host.db.prepare('SELECT subscription_id FROM notification_outbox WHERE next_at<=? AND lease_until<=? ORDER BY next_at,subscription_id LIMIT ?').all(clock(), clock(), limit) as any[];
for (const row of rows) {
const lease = newId();
const job = this.host.authorization.transaction(() => {
const now = clock();
const changed = this.host.db.prepare('UPDATE notification_outbox SET lease_until=?,lease_id=? WHERE subscription_id=? AND next_at<=? AND lease_until<=?').run(now + 30, lease, row.subscription_id, now, now);
return changed.changes ? this.host.db.prepare('SELECT * FROM notification_outbox WHERE subscription_id=?').get(row.subscription_id) as any : undefined;
});
if (!job) continue;
let s = this.get(row.subscription_id);
try {
requireThat(s.state === 'pending' || s.state === 'active', 'subscription_expired'); this.current(s, clock());
if (s.state === 'pending') requireThat(clock() < s.challenge_until, 'subscription_expired');
const transport: Transport = await factory(s.callback);
// DNS/policy preparation may take time: recheck authority immediately before sealing/sending.
s = this.get(s.id); requireThat(s.state === 'pending' || s.state === 'active', 'subscription_expired');
const app = this.current(s, clock());
requireThat(clock() < job.lease_until, 'worker_lease_expired');
if (s.state === 'pending') requireThat(clock() < s.challenge_until, 'subscription_expired');
const body = s.state === 'pending'
? { operation: 'subscription_challenge', subscription_id: s.id, challenge: s.challenge, callback: s.callback, expires_at: s.challenge_until }
: { operation: 'change_notification', subscription_id: s.id, delivery_id: job.delivery_id };
const message = seal(body, this.host.identity, this.host.credential, app, clock());
const header = parseObject(unb64(message.protected));
const reply = open(await transport(message), this.host.identity, this.host.trust, clock(), s.app, header.messageId);
requireThat(reply.sender.kind === 'application' && reply.sender.signingKid === app.signingKid && reply.sender.encryptionKid === app.encryptionKid, 'wrong_callback_key');
const expected = s.state === 'pending'
? { status: 'ok', operation: 'subscription_challenge_ack', subscription_id: s.id, challenge: s.challenge, callback: s.callback }
: { status: 'ok', operation: 'notification_ack', subscription_id: s.id, delivery_id: job.delivery_id };
exact(reply.body, Object.keys(expected));
requireThat(Object.entries(expected).every(([key, value]) => reply.body[key] === value), 'invalid_notification_ack');
this.host.authorization.transaction(() => {
const latest = this.get(s.id); this.current(latest, clock());
if (latest.state === 'pending') requireThat(clock() < latest.challenge_until, 'subscription_expired');
requireThat(latest.state === s.state, 'subscription_expired');
const held = this.host.db.prepare('SELECT * FROM notification_outbox WHERE subscription_id=? AND lease_id=?').get(s.id, lease) as any;
if (!held) return;
if (s.state === 'pending') {
latest.state = 'active'; this.save(latest);
// Activation always schedules catch-up, covering writes during the challenge.
this.host.db.prepare('UPDATE notification_outbox SET delivery_id=?,revision=revision+1,attempts=0,next_at=?,lease_until=0,lease_id=NULL WHERE subscription_id=?').run(newId(), clock(), s.id);
} else if (held.revision === job.revision) this.host.db.prepare('DELETE FROM notification_outbox WHERE subscription_id=?').run(s.id);
else this.host.db.prepare('UPDATE notification_outbox SET delivery_id=?,attempts=0,next_at=?,lease_until=0,lease_id=NULL WHERE subscription_id=?').run(newId(), clock(), s.id);
});
result.acknowledged++;
} catch (e) {
this.host.authorization.transaction(() => {
const latest = this.get(s.id);
if (this.permanent(e)) { this.stop(latest, latest.state === 'cancelled' ? 'cancelled' : 'suspended'); result.suspended++; }
else {
this.host.db.prepare('UPDATE notification_outbox SET attempts=attempts+1,next_at=?,lease_until=0,lease_id=NULL WHERE subscription_id=? AND lease_id=?').run(clock() + Math.min(300, 2 ** Math.min(job.attempts + 1, 9)), s.id, lease);
result.retried++;
}
});
}
}
return result;
} finally { this.running = false; }
}
}