Reference source · 0.1.0-dev.11
subscriptions.test.ts
tests/subscriptions.test.ts · Download original file
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { fixture, PATIENT, OTHER_PATIENT, lab } from '../src/reference/fixtures.ts';
import { SCOPES, makeActorProof } from '../src/authorization.ts';
import { NotificationReceiver } from '../src/notification-receiver.ts';
import { ReferenceHost } from '../src/host.ts';
import { Client } from '../src/sdk.ts';
import { makeIdentity, signCredential } from '../src/identity.ts';
import { makeRotationProof } from '../src/rotation.ts';
import { seal, open } from '../src/envelope.ts';
import { callbackURL, publicAddress, publicCallbackTransport } from '../src/callback-transport.ts';
import { DirectoryTrust } from '../src/discovery.ts';
import { ReferenceDirectory } from '../src/reference/directory.ts';
import { HostPeer } from '../src/host-peer.ts';
const CALLBACK = 'https://requester.example/oprp/exchange';
async function authorize(f: ReturnType<typeof fixture>, scopes = Object.values(SCOPES), duration = 1800) {
const r = await f.client.requestAccess({ patient: PATIENT, scopes, requested_duration_seconds: duration, refresh_requested: true });
if (r.status === 'pending') { f.host.authorization.approve(f.app, r.request_id, PATIENT, f.clock()); await f.client.approvalStatus(r.request_id); }
assert(f.client.authorization);
}
function receiver(f: ReturnType<typeof fixture>, path = ':memory:') { return new NotificationReceiver(path, f.appIdentity, f.appCredential, f.hostCredential, f.trust, CALLBACK); }
async function register(f: ReturnType<typeof fixture>, r: NotificationReceiver, subscriptionId = 'sub-1', topics: ('Patient' | 'Observation')[] = ['Patient'], expires = f.clock() + 900) {
r.expect(subscriptionId, expires);
return f.client.subscribe(PATIENT, CALLBACK, topics, expires, subscriptionId);
}
async function activate(f: ReturnType<typeof fixture>, r: NotificationReceiver, subscriptionId = 'sub-1') {
await register(f, r, subscriptionId);
const factory = () => async (message: any) => r.handle(message, f.clock());
assert.equal((await f.host.subscriptions.deliver(f.clock, factory)).acknowledged, 1);
assert.equal((await f.client.subscriptionStatus(PATIENT, subscriptionId)).state, 'active');
await f.host.subscriptions.deliver(f.clock, factory); r.pending().forEach(t => r.complete(t.subscription_id, t.revision));
return factory;
}
test('registration is scoped, lifetime-bounded, idempotent and rate-limited without clinical or token callback contents', async () => {
const f = fixture(), r = receiver(f);
try {
await authorize(f, [SCOPES.demographicsRead]);
await assert.rejects(f.client.subscribe(PATIENT, CALLBACK, ['Observation'], f.clock() + 100, 'bad-scope'), /insufficient_scope/);
await assert.rejects(f.client.subscribe(PATIENT, CALLBACK, ['Patient'], f.clock() + 1801, 'bad-period'), /invalid_subscription_lifetime/);
await assert.rejects(f.client.subscribe(OTHER_PATIENT, CALLBACK, ['Patient'], f.clock() + 100, 'bad-patient'), /invalid_token/);
const a = await register(f,r); assert.equal(a.state, 'pending'); assert.deepEqual(await register(f,r), a);
await assert.rejects(f.client.subscribe(PATIENT, CALLBACK, ['Patient'], f.clock() + 800, 'sub-1'), /subscription_id_conflict/);
let clear: any;
const factory = () => async (message: any) => { clear = open(message, f.appIdentity, f.trust, f.clock()).body; return r.handle(message, f.clock()); };
await f.host.subscriptions.deliver(f.clock, factory); assert.equal(clear.operation, 'subscription_challenge');
await f.host.subscriptions.deliver(f.clock, factory); assert.deepEqual(Object.keys(clear), ['operation','subscription_id','delivery_id']);
for (let i = 2; i <= 10; i++) await register(f,r, `sub-${i}`);
await assert.rejects(register(f,r,'sub-11'), /subscription_rate_limited/);
} finally { r.close(); f.host.close(); f.trust.close(); }
});
test('callback ownership requires a matching PQ reply; an unsolicited challenge, wrong endpoint and wrong key fail', async () => {
const f = fixture(), r = receiver(f);
try {
await authorize(f); await register(f,r);
const fail = await f.host.subscriptions.deliver(f.clock, () => async message => {
const req = open(message, f.appIdentity, f.trust, f.clock());
return seal({ status:'ok', operation:'subscription_challenge_ack', subscription_id:'sub-1', challenge:'wrong', callback:CALLBACK }, f.appIdentity, f.appCredential, req.sender, f.clock(), req.header.messageId);
});
assert.equal(fail.retried,1); assert.equal((await f.client.subscriptionStatus(PATIENT,'sub-1')).state,'pending');
f.advance(2);
const impostor = makeIdentity(f.appIdentity.public.subject), credential = signCredential(f.applicationAuthority, impostor.public, 'application', f.clock());
const wrongKey = await f.host.subscriptions.deliver(f.clock, () => async message => {
const req = open(message, f.appIdentity, f.trust, f.clock());
return seal({ status:'ok', operation:'subscription_challenge_ack', subscription_id:'sub-1', challenge:req.body.challenge, callback:CALLBACK }, impostor, credential, req.sender, f.clock(), req.header.messageId);
});
assert.equal(wrongKey.retried,1);
const unknown = seal({operation:'subscription_challenge',subscription_id:'unsolicited',challenge:f.host.subscriptions.get('sub-1').challenge,callback:CALLBACK,expires_at:f.clock()+100},f.hostIdentity,f.hostCredential,f.app,f.clock());
assert.throws(()=>r.handle(unknown,f.clock()),/unexpected_subscription/);
const wrongCallback = seal({operation:'subscription_challenge',subscription_id:'sub-1',challenge:f.host.subscriptions.get('sub-1').challenge,callback:'https://other.example/',expires_at:f.clock()+100},f.hostIdentity,f.hostCredential,f.app,f.clock());
assert.throws(()=>r.handle(wrongCallback,f.clock()),/invalid_callback_challenge/);
f.advance(4); await f.host.subscriptions.deliver(f.clock, () => async m=>{
const reply=r.handle(m,f.clock()), request=open(m,f.appIdentity,f.trust,f.clock()), clear=open(reply,f.hostIdentity,f.trust,f.clock()).body;
const reordered=Object.fromEntries(Object.entries(clear).reverse());
return seal(reordered,f.appIdentity,f.appCredential,request.sender,f.clock(),request.header.messageId);
});
assert.equal((await f.client.subscriptionStatus(PATIENT,'sub-1')).state,'active');
} finally { r.close(); f.host.close(); f.trust.close(); }
});
test('mutation transaction enqueues only visible topics, coalesces changes and rolls back failed writes', async () => {
const f = fixture(), r = receiver(f);
try {
await authorize(f); const factory = await activate(f,r);
const count = () => (f.host.db.prepare('SELECT count(*) AS n FROM notification_outbox').get() as any).n;
await f.client.createLab(PATIENT,lab(),'hidden-lab'); assert.equal(count(),0);
f.host.authorization.transaction(()=>f.host.feed.append(OTHER_PATIENT,{resourceType:'Patient',id:OTHER_PATIENT},2,'updated')); assert.equal(count(),0);
await f.client.updateDemographics(PATIENT,'Visible',1,'edit-1'); await f.client.updateDemographics(PATIENT,'Visible Again',2,'edit-2');
assert.equal(count(),1); const revision = (f.host.db.prepare('SELECT revision FROM notification_outbox').get() as any).revision;
await f.client.updateDemographics(PATIENT,'Visible Again',2,'edit-2'); assert.equal((f.host.db.prepare('SELECT revision FROM notification_outbox').get() as any).revision,revision);
await f.host.subscriptions.deliver(f.clock,factory); r.pending().forEach(t=>r.complete(t.subscription_id,t.revision));
assert.throws(()=>f.host.authorization.transaction(()=>{f.host.feed.append(PATIENT,{resourceType:'Patient',id:PATIENT},4,'updated');throw new Error('rollback');}),/rollback/); assert.equal(count(),0);
} finally {r.close();f.host.close();f.trust.close();}
});
test('lost acknowledgments and both restarts preserve a stable delivery, exactly one pull task and fresh reply envelopes', async () => {
const dir=mkdtempSync(join(tmpdir(),'oprp-notify-')),path=join(dir,'host.sqlite'), inbox=join(dir,'inbox.sqlite'),f=fixture(path);
let h=f.host,r=receiver(f,inbox);
try {
await authorize(f);await activate(f,r);await f.client.updateDemographics(PATIENT,'Queued',1,'edit');
let delivery='';
const result=await h.subscriptions.deliver(f.clock,()=>async message=>{delivery=open(message,f.appIdentity,f.trust,f.clock()).body.delivery_id;r.handle(message,f.clock());throw new Error('lost_ack');});
assert.equal(result.retried,1);assert.equal(r.pending()[0].revision,2);
h.close();r.close();h=new ReferenceHost(path,f.hostIdentity,f.hostCredential,f.trust,f.policy,f.representatives);r=receiver(f,inbox);f.advance(2);
let repeated='';await h.subscriptions.deliver(f.clock,()=>async message=>{repeated=open(message,f.appIdentity,f.trust,f.clock()).body.delivery_id;return r.handle(message,f.clock());});
assert.equal(delivery,repeated);assert.equal(r.pending()[0].revision,2);assert.equal((h.db.prepare('SELECT count(*) AS n FROM notification_outbox').get() as any).n,0);
} finally {r.close();h.close();f.trust.close();rmSync(dir,{recursive:true,force:true});}
});
test('a change during delivery and a notification during pull completion cannot lose a wake-up', async()=>{
const f=fixture(),r=receiver(f);
try {
await authorize(f);await activate(f,r);await f.client.updateDemographics(PATIENT,'First',1,'first');
await f.host.subscriptions.deliver(f.clock,()=>async message=>{const reply=r.handle(message,f.clock());await f.client.updateDemographics(PATIENT,'Second',2,'second');return reply;});
const old=r.pending()[0];assert.equal((f.host.db.prepare('SELECT count(*) AS n FROM notification_outbox').get() as any).n,1);
await f.host.subscriptions.deliver(f.clock,()=>async m=>r.handle(m,f.clock()));assert.equal(r.pending()[0].revision,old.revision+1);
r.complete(old.subscription_id,old.revision);assert.equal(r.pending().length,1);r.complete('sub-1',old.revision+1);assert.equal(r.pending().length,0);
await f.client.updateDemographics(PATIENT,'Third',3,'third');await f.host.subscriptions.deliver(f.clock,()=>async m=>r.handle(m,f.clock()));
assert.equal(r.pending()[0].revision,old.revision+2);r.complete('sub-1',old.revision+1);assert.equal(r.pending()[0].revision,old.revision+2);
}finally{r.close();f.host.close();f.trust.close();}
});
test('delivery survives JWT expiry, but grant revocation and subscription expiry prevent any transport call',async()=>{
for(const reason of ['revoke','expire']) {
const f=fixture(),r=receiver(f);
try {
await authorize(f);await activate(f,r);f.advance(301);await f.client.updateDemographics(PATIENT,'After Refresh',1,'after');
assert.equal((await f.host.subscriptions.deliver(f.clock,()=>async m=>r.handle(m,f.clock()))).acknowledged,1);
await f.client.updateDemographics(PATIENT,'Never Delivered',2,'never');
if(reason==='revoke')f.host.authorization.revoke(f.client.authorization!.grant_id);else f.advance(600);
let calls=0;const result=await f.host.subscriptions.deliver(f.clock,()=>{calls++;throw new Error('should not prepare');});
assert.equal(calls,0);assert.equal(result.suspended,1);assert.equal(f.host.subscriptions.get('sub-1').state,'suspended');
}finally{r.close();f.host.close();f.trust.close();}
}
});
test('representative withdrawal and planned requester rotation suspend old-key callbacks',async()=>{
for(const reason of ['representative','rotation']) {
const f=fixture(),r=receiver(f);
try {
if(reason==='representative') {
const request={request_id:'actor-request',patient:PATIENT,scopes:[SCOPES.demographicsRead],requested_duration_seconds:1800,refresh_requested:true};
await f.client.requestAccess({...request,actor_proof:makeActorProof(f.parentIdentity,f.parentCredential,f.appIdentity.public.subject,PATIENT,request.request_id,f.clock())});
} else await authorize(f);
await activate(f,r);
if(reason==='representative')f.representatives[0].active=false;
else {
const next=makeIdentity(f.appIdentity.public.subject),credential=signCredential(f.applicationAuthority,next.public,'application',f.clock());
const client=new Client({identity:next,credential,hostCredential:f.hostCredential,hostId:f.hostIdentity.public.subject,trust:f.trust,transport:async m=>f.host.handle(m,f.clock()),clock:f.clock});
await client.rotateRequester(makeRotationProof(f.appIdentity,f.appCredential,next,credential,f.hostIdentity.public.subject,f.clock()));
}
let calls=0;await f.host.subscriptions.deliver(f.clock,()=>{calls++;throw new Error('should not deliver');});assert.equal(calls,0);assert.equal(f.host.subscriptions.get('sub-1').state,'suspended');
}finally{r.close();f.host.close();f.trust.close();}
}
});
test('stale trust status pauses delivery and resumes after fresh evidence without losing a notification',async()=>{
const f=fixture(),r=receiver(f);
try {
await authorize(f);await activate(f,r);await f.client.updateDemographics(PATIENT,'Pending',1,'pending');
const later=()=>f.clock()+601;let calls=0;const paused=await f.host.subscriptions.deliver(later,()=>{calls++;throw new Error('must not send');});
assert.equal(calls,0);assert.equal(paused.retried,1);assert.equal(f.host.subscriptions.get('sub-1').state,'active');
f.advance(603);assert.equal((await f.host.subscriptions.deliver(f.clock,()=>async m=>r.handle(m,f.clock()))).acknowledged,1);
}finally{r.close();f.host.close();f.trust.close();}
});
test('callback destination profile rejects private, mapped, reserved and unsafe URL destinations',async()=>{
for(const address of ['127.0.0.1','10.1.2.3','100.64.1.1','169.254.169.254','172.16.0.1','192.168.0.1','198.18.1.1','203.0.113.1','224.0.0.1','::1','::ffff:8.8.8.8','fe80::1','fc00::1','2001:db8::1','2001:1ff::1','2002:808:808::1','3fff::1'])assert.equal(publicAddress(address),false,address);
for(const address of ['8.8.8.8','1.1.1.1','2606:4700:4700::1111','2001:4860:4860::8888'])assert.equal(publicAddress(address),true,address);
for(const url of ['http://example.com','https://user:pass@example.com/','https://example.com/?token=secret','https://example.com/#fragment'])assert.throws(()=>callbackURL(url),/invalid_callback/);
for(const url of ['https://127.0.0.1/','https://[::1]/','https://8.8.8.8:8443/'])await assert.rejects(publicCallbackTransport(url),/callback_destination_denied/);
});
test('unsubscribe is final, a pending challenge expires, and delayed network preparation rechecks current authority',async()=>{
const f=fixture(),r=receiver(f);
try {
await authorize(f);await activate(f,r);await f.client.updateDemographics(PATIENT,'Cancel',1,'cancel');
assert.equal((await f.client.unsubscribe(PATIENT,'sub-1')).state,'cancelled');assert.equal((await register(f,r)).state,'cancelled');
let calls=0;await f.host.subscriptions.deliver(f.clock,()=>{calls++;throw new Error('no send');});assert.equal(calls,0);
await register(f,r,'sub-2');f.advance(301);await f.host.subscriptions.deliver(f.clock,()=>{calls++;throw new Error('expired');});assert.equal(calls,0);assert.equal(f.host.subscriptions.get('sub-2').state,'suspended');
await register(f,r,'sub-3', ['Patient'],f.clock()+500);
await f.host.subscriptions.deliver(f.clock,()=>{f.host.authorization.revoke(f.client.authorization!.grant_id);return async()=>{calls++;throw new Error('revoked while preparing');};});
assert.equal(calls,0);assert.equal(f.host.subscriptions.get('sub-3').state,'suspended');
}finally{r.close();f.host.close();f.trust.close();}
});
test('callback catch-up uses authorized polling and recovers a feed retention gap with a fresh snapshot',async()=>{
const f=fixture(),r=receiver(f);
try {
await authorize(f);const baseline=await f.client.syncSnapshot(PATIENT);await activate(f,r);await f.client.updateDemographics(PATIENT,'Catch Up',1,'catch-up');
await f.host.subscriptions.deliver(f.clock,()=>async m=>r.handle(m,f.clock()));assert.equal((await f.client.changes(PATIENT,baseline.cursor)).events.length,1);
f.host.feed.retainAfter(f.host.feed.watermark());await assert.rejects(f.client.changes(PATIENT,baseline.cursor),/cursor_expired/);
assert.equal((await f.client.syncSnapshot(PATIENT)).resources[0].version,2);assert.equal(r.pending().length,1);
}finally{r.close();f.host.close();f.trust.close();}
});
test('a live Host release retires source callbacks and transfers no subscription authority',async()=>{
const f=fixture(),r=receiver(f),identity=makeIdentity('urn:oprp:demo:destination'),credential=signCredential(f.hostingAuthority,identity.public,'host',f.clock());
const destination=new ReferenceHost(':memory:',identity,credential,f.trust,f.policy),bindings=new ReferenceDirectory(':memory:',f.hostingAuthority,['patients.example'],f.trust);
try {
await authorize(f);await activate(f,r);await f.client.updateDemographics(PATIENT,'Queued Before Move',1,'before-move');
bindings.enroll('synthetic@patients.example',PATIENT,f.hostCredential,'https://source.example/',f.clock());
const directory=new DirectoryTrust({[f.hostingAuthority.issuer]:{publicKey:f.hostingAuthority.publicKey,domains:['patients.example']}},f.trust);
const plan=f.host.migration.begin(PATIENT,credential,'https://destination.example/',bindings.document('synthetic@patients.example',f.clock()),directory,PATIENT,f.clock(),'transfer-1');destination.migration.admit(plan,PATIENT,f.clock());
const archive=f.host.migration.freeze('transfer-1',f.clock()),peer=new HostPeer(f.hostIdentity,f.hostCredential,credential,f.trust,async m=>destination.handle(m,f.clock()),f.clock);
const ready=(await peer.exchange({operation:'migration_stage_final',plan,archive})).ready,release=f.host.migration.release('transfer-1',ready,f.clock());
await peer.activate('transfer-1',release);
let calls=0;const result=await f.host.subscriptions.deliver(f.clock,()=>{calls++;throw new Error('must not send');});assert.equal(calls,0);assert.equal(result.suspended,1);
assert.equal((destination.db.prepare('SELECT count(*) AS n FROM subscriptions').get() as any).n,0);
}finally{bindings.close();destination.close();r.close();f.host.close();f.trust.close();}
});
test('database delivery leases prevent two Host instances sending the same job concurrently', async()=>{
const dir=mkdtempSync(join(tmpdir(),'oprp-lease-')),path=join(dir,'host.sqlite'),f=fixture(path),r=receiver(f);
let second:ReferenceHost|undefined;
try {
await authorize(f);await activate(f,r);await f.client.updateDemographics(PATIENT,'Leased',1,'leased');
second=new ReferenceHost(path,f.hostIdentity,f.hostCredential,f.trust,f.policy,f.representatives);
let release!:()=>void,entered!:()=>void;const gate=new Promise<void>(resolve=>release=resolve),ready=new Promise<void>(resolve=>entered=resolve);
const first=f.host.subscriptions.deliver(f.clock,()=>async m=>{entered();await gate;return r.handle(m,f.clock());});await ready;
let calls=0;const other=await second.subscriptions.deliver(f.clock,()=>{calls++;throw new Error('already leased');});assert.equal(calls,0);assert.equal(other.acknowledged,0);
release();assert.equal((await first).acknowledged,1);assert.equal(r.pending().length,1);
}finally{second?.close();r.close();f.host.close();f.trust.close();rmSync(dir,{recursive:true,force:true});}
});
test('credential revocation retires callbacks, and receiver persistence rejects changing its pinned Host',async()=>{
const dir=mkdtempSync(join(tmpdir(),'oprp-receiver-pin-')),path=join(dir,'receiver.sqlite'),f=fixture(),r=receiver(f,path);
try{
await authorize(f);await activate(f,r);await f.client.updateDemographics(PATIENT,'Revoked',1,'revoked');
f.trust.revoke(f.app.issuer,'serial',f.app.serial);
let calls=0;const result=await f.host.subscriptions.deliver(f.clock,()=>{calls++;throw new Error('revoked');});assert.equal(calls,0);assert.equal(result.suspended,1);
const other=makeIdentity('urn:oprp:demo:other-host'),credential=signCredential(f.hostingAuthority,other.public,'host',f.clock());
assert.throws(()=>new NotificationReceiver(path,f.appIdentity,f.appCredential,credential,f.trust,CALLBACK),/receiver_owner_changed/);
}finally{r.close();f.host.close();f.trust.close();rmSync(dir,{recursive:true,force:true});}
});