Reference source · 0.1.0-dev.11
change-feed.test.ts
tests/change-feed.test.ts · Download original file
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { fixture, PATIENT, OTHER_PATIENT, lab } from '../src/reference/fixtures.ts';
import { SCOPES } from '../src/authorization.ts';
import { Client } from '../src/sdk.ts';
import { ReferenceHost } from '../src/host.ts';
import { b64, unb64 } from '../src/core.ts';
async function authorizeAll(f: ReturnType<typeof fixture>): Promise<void> {
const request = await f.client.requestAccess({ patient: PATIENT, scopes: Object.values(SCOPES), requested_duration_seconds: 86400, refresh_requested: true });
if (request.status !== 'pending') throw new Error('Expected approval');
f.host.authorization.approve(f.app, request.request_id, PATIENT, f.clock()); await f.client.approvalStatus(request.request_id);
}
function anotherClient(f: ReturnType<typeof fixture>, host = f.host): Client {
return new Client({ identity: f.appIdentity, credential: f.appCredential, hostCredential: f.hostCredential, hostId: f.hostIdentity.public.subject, trust: f.trust, transport: async envelope => host.handle(envelope, f.clock()), clock: f.clock });
}
test('snapshot watermark, immutable paged events and write receipts prevent synchronization gaps or duplicate events', async () => {
const f = fixture();
try {
await authorizeAll(f);
const initial = await f.client.syncSnapshot(PATIENT);
assert.equal(initial.resources.length, 1); assert.equal(initial.resources[0].version, 1);
const updated = await f.client.updateDemographics(PATIENT, 'First Accepted Name', 1, 'name-1');
await f.client.updateDemographics(PATIENT, 'Second Accepted Name', 2, 'name-2');
const created = await f.client.createLab(PATIENT, lab(), 'lab-1');
await f.client.createLab(PATIENT, lab(), 'lab-1');
assert.equal((f.host.db.prepare('SELECT count(*) AS count FROM change_events').get() as any).count, 3);
const first = await f.client.changes(PATIENT, initial.cursor, 1);
assert.equal(first.events.length, 1); assert.equal(first.has_more, true);
assert.equal(first.events[0].resource.name[0].text, updated.resource.name[0].text);
const repeated = await f.client.changes(PATIENT, initial.cursor, 1);
assert.equal(first.events[0].event_id, repeated.events[0].event_id);
const second = await f.client.changes(PATIENT, first.cursor, 1);
const third = await f.client.changes(PATIENT, second.cursor, 1);
assert.equal(second.events[0].version, 3); assert.equal(second.has_more, true);
assert.equal(third.events[0].resource.id, created.resource.id); assert.equal(third.has_more, false);
const complete = await f.client.changes(PATIENT, third.cursor, 1);
assert.deepEqual(complete.events, []); assert.equal(complete.has_more, false);
const current = await f.client.syncSnapshot(PATIENT);
assert.equal(current.resources.length, 2); assert.equal(current.resources[0].version, 3);
assert.deepEqual((await f.client.changes(PATIENT, current.cursor)).events, []);
} finally { f.host.close(); }
});
test('restricted categories and other patients do not leak through events, paging or retention errors', async () => {
const f = fixture();
try {
await authorizeAll(f); const reader = anotherClient(f);
await reader.requestAccess({ patient: PATIENT, scopes: [SCOPES.demographicsRead], requested_duration_seconds: 86400, refresh_requested: true });
const baseline = await reader.syncSnapshot(PATIENT); assert.equal(baseline.resources.length, 1);
await f.client.createLab(PATIENT, lab(), 'restricted-lab');
f.host.authorization.transaction(() => f.host.feed.append(OTHER_PATIENT, { resourceType: 'Patient', id: OTHER_PATIENT, name: [{ text: 'Hidden Other Patient' }] }, 2, 'updated'));
const hidden = await reader.changes(PATIENT, baseline.cursor, 1);
assert.deepEqual(hidden.events, []); assert.equal(hidden.has_more, false);
f.host.authorization.transaction(() => {
for (let version = 3; version <= 13; version++) f.host.feed.append(OTHER_PATIENT, { resourceType: 'Patient', id: OTHER_PATIENT, name: [{ text: 'Unrelated Change' }] }, version, 'updated');
});
const laterHidden = await reader.changes(PATIENT, hidden.cursor, 1);
assert.deepEqual(laterHidden.events, []); assert.equal(laterHidden.has_more, false);
assert.equal(baseline.cursor.length, hidden.cursor.length); assert.equal(hidden.cursor.length, laterHidden.cursor.length);
// A demographics-only reader does not lose its visible history when only hidden events are pruned.
f.host.feed.retainAfter(f.host.feed.watermark());
const retained = await reader.changes(PATIENT, baseline.cursor, 1);
assert.deepEqual(retained.events, []); assert.equal(retained.has_more, false);
await f.client.updateDemographics(PATIENT, 'Visible Demographic Change', 1, 'visible');
const visible = await reader.changes(PATIENT, baseline.cursor, 1);
assert.equal(visible.events.length, 1); assert.equal(visible.events[0].resource.resourceType, 'Patient'); assert.equal(visible.has_more, false);
const bytes = unb64(visible.cursor);
assert(!bytes.includes(Buffer.from(PATIENT))); assert(!bytes.includes(Buffer.from('after')));
await assert.rejects(reader.changes(OTHER_PATIENT, baseline.cursor), /invalid_token/);
} finally { f.host.close(); }
});
test('cursors reject modification, grant changes, retention gaps and revocation; snapshot recovers a gap', async () => {
const f = fixture();
try {
await authorizeAll(f); const initial = await f.client.syncSnapshot(PATIENT);
const modified = unb64(initial.cursor); modified[modified.length - 1] ^= 1;
await assert.rejects(f.client.changes(PATIENT, b64(modified)), /invalid_opaque_value/);
await assert.rejects(f.client.changes(PATIENT, initial.cursor, 0), /invalid_page_size/);
await f.client.updateDemographics(PATIENT, 'Retained State', 1, 'edit');
f.host.feed.retainAfter(f.host.feed.watermark());
await assert.rejects(f.client.changes(PATIENT, initial.cursor), /cursor_expired/);
const recovered = await f.client.syncSnapshot(PATIENT);
assert.equal(recovered.resources[0].version, 2);
assert.deepEqual((await f.client.changes(PATIENT, recovered.cursor)).events, []);
await f.client.requestAccess({ patient: PATIENT, scopes: [SCOPES.demographicsRead], requested_duration_seconds: 86400, refresh_requested: true });
await assert.rejects(f.client.changes(PATIENT, recovered.cursor), /cursor_context_changed/);
const newGrantCursor = (await f.client.syncSnapshot(PATIENT)).cursor;
assert(f.client.authorization); f.host.authorization.revoke(f.client.authorization.grant_id);
await assert.rejects(f.client.changes(PATIENT, newGrantCursor), /authorization_expired_or_revoked/);
} finally { f.host.close(); }
});
test('feed cursor and pending events survive Host restart and remain usable after token refresh', async () => {
const dir = mkdtempSync(join(tmpdir(), 'oprp-feed-')); const path = join(dir, 'host.sqlite'); const f = fixture(path); let restored: ReferenceHost | undefined;
try {
await authorizeAll(f); const baseline = await f.client.syncSnapshot(PATIENT);
await f.client.createLab(PATIENT, lab(), 'restart-lab'); f.host.close();
restored = new ReferenceHost(path, f.hostIdentity, f.hostCredential, f.trust, f.policy, f.representatives);
const client = anotherClient(f, restored); client.restoreState(f.client.exportState());
f.advance(301); const page = await client.changes(PATIENT, baseline.cursor, 1);
assert.equal(page.events.length, 1); assert.equal(page.events[0].resource.resourceType, 'Observation');
assert.equal(client.authorization?.grant_id, f.client.authorization?.grant_id);
f.advance(3600); await assert.rejects(client.changes(PATIENT, page.cursor), /cursor_expired/);
} finally { if (restored) restored.close(); else { try { f.host.close(); } catch {} } rmSync(dir, { recursive: true, force: true }); }
});