Reference source · 0.1.0-dev.11
queue.ts

import { DatabaseSync } from 'node:sqlite';
import type { Client } from './sdk.ts';
import type { ResourceType } from './resource-profile.ts';
import type { Transport } from './sdk.ts';
import { DirectoryTrust } from './discovery.ts';
import { newId, id, requireThat, ProtocolError, fingerprint, jsonBytes } from './core.ts';

export type QueueState = 'pending' | 'submitted' | 'committed' | 'conflicted' | 'rejected';
export type WriteIntent = { operation: 'create_lab'; patient: string; resource: Record<string, any>; operation_id: string }
  | { operation: 'update_demographics'; patient: string; resource: { name: string }; base_version: number; operation_id: string }
  | { operation:'create_resource';patient:string;resource:Record<string,any>;operation_id:string }
  | { operation:'update_resource';patient:string;resource:Record<string,any>;base_version:number;operation_id:string }
  | { operation:'delete_resource';patient:string;resource_type:ResourceType;resource_id:string;base_version:number;operation_id:string };
export type QueueEntry = { sequence: number; operation_id: string; state: QueueState; intent: WriteIntent; result: Record<string, any> | null; error: string | null };

// One worker/one process per database in this reference adapter. Local cache encryption is operator-owned.
export class DurableQueue {
  db: DatabaseSync; client: Client; pumping = false;
  constructor(path: string, client: Client) {
    this.client = client; this.db = new DatabaseSync(path);
    this.db.exec(`PRAGMA journal_mode=WAL;
      CREATE TABLE IF NOT EXISTS owner (id INTEGER PRIMARY KEY CHECK(id=1), application TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS queue (sequence INTEGER PRIMARY KEY AUTOINCREMENT, operation_id TEXT UNIQUE NOT NULL, intent TEXT NOT NULL, digest TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT);`);
    this.db.exec('CREATE TABLE IF NOT EXISTS queue_routes (patient TEXT PRIMARY KEY, host TEXT NOT NULL, generation INTEGER, address TEXT);');
    this.db.prepare('INSERT INTO owner VALUES (1, ?) ON CONFLICT(id) DO NOTHING').run(client.identity.public.subject);
    const owner = this.db.prepare('SELECT application FROM owner WHERE id=1').get() as any;
    if (owner.application !== client.identity.public.subject) { this.db.close(); throw new ProtocolError('wrong_queue_owner'); }
    // A process that stopped before saving a response cannot know whether the Host committed.
    // Retrying the unchanged operation with a new envelope resolves that uncertainty through its receipt.
    this.db.prepare('UPDATE queue SET state=? WHERE state=?').run('pending', 'submitted');
    for (const entry of this.entries()) this.pinRoute(entry.intent.patient);
  }
  pinRoute(patient: string): void {
    const location = this.client.location?.patient === patient ? this.client.location : undefined;
    this.db.prepare('INSERT INTO queue_routes VALUES (?, ?, ?, ?) ON CONFLICT(patient) DO NOTHING').run(patient, this.client.hostId, location?.generation ?? null, location?.address ?? null);
  }
  enqueue(intent: WriteIntent): string {
    id(intent.patient); id(intent.operation_id);
    this.pinRoute(intent.patient);
    const route = this.db.prepare('SELECT host FROM queue_routes WHERE patient=?').get(intent.patient) as any;
    requireThat(route.host === this.client.hostId, 'queue_route_update_required');
    const encoded = JSON.stringify(intent), digest = fingerprint(jsonBytes(intent));
    const prior = this.db.prepare('SELECT digest FROM queue WHERE operation_id=?').get(intent.operation_id) as any;
    if (prior) { requireThat(prior.digest === digest, 'queue_operation_id_conflict'); return intent.operation_id; }
    this.db.prepare('INSERT INTO queue(operation_id,intent,digest,state) VALUES (?, ?, ?, ?)').run(intent.operation_id, encoded, digest, 'pending');
    return intent.operation_id;
  }
  createLab(patient: string, resource: Record<string, any>, operationId = newId()): string {
    return this.enqueue({ operation: 'create_lab', patient, resource, operation_id: operationId });
  }
  updateDemographics(patient: string, name: string, baseVersion: number, operationId = newId()): string {
    return this.enqueue({ operation: 'update_demographics', patient, resource: { name }, base_version: baseVersion, operation_id: operationId });
  }
  createResource(patient: string, resource: Record<string, any>, operationId = newId()): string { return this.enqueue({operation:'create_resource',patient,resource,operation_id:operationId}); }
  updateResource(patient: string, resource: Record<string, any>, baseVersion: number, operationId = newId()): string { return this.enqueue({operation:'update_resource',patient,resource,base_version:baseVersion,operation_id:operationId}); }
  deleteResource(patient: string, type: ResourceType, resourceId: string, baseVersion: number, operationId = newId()): string { return this.enqueue({operation:'delete_resource',patient,resource_type:type,resource_id:resourceId,base_version:baseVersion,operation_id:operationId}); }
  entries(): QueueEntry[] {
    return (this.db.prepare('SELECT * FROM queue ORDER BY sequence').all() as any[]).map(row => ({ sequence: row.sequence, operation_id: row.operation_id, state: row.state, intent: JSON.parse(row.intent), result: row.result ? JSON.parse(row.result) : null, error: row.error }));
  }
  async followMigration(patient: string, document: Record<string, any>, address: string, directory: DirectoryTrust, transport: Transport): Promise<void> {
    requireThat(!this.pumping, 'queue_worker_busy'); id(patient);
    const route = this.db.prepare('SELECT * FROM queue_routes WHERE patient=?').get(patient) as any; requireThat(route, 'unknown_queue_route');
    const located = directory.validate(document, address, this.client.clock(), route.generation ?? 1);
    requireThat(located.record_id === patient && (!route.address || route.address === located.address), 'wrong_discovery_patient');
    requireThat(located.host === route.host || route.generation === null || located.generation > route.generation, 'stale_discovery_generation');
    // Save changed SDK state before changing the durable queue route; a failed save cannot send to a new Host.
    await this.client.followDiscovery(document, address, directory, transport, patient, route.generation ?? 1);
    this.db.prepare('UPDATE queue_routes SET host=?,generation=?,address=? WHERE patient=?').run(located.host, located.generation, located.address, patient);
  }
  async flush(): Promise<void> {
    requireThat(!this.pumping, 'queue_worker_busy'); this.pumping = true;
    try {
      for (;;) {
        const entry = this.entries().find(row => row.state !== 'committed');
        if (!entry || entry.state !== 'pending') return; // Never reorder later writes around a blocked dependency.
        this.db.prepare('UPDATE queue SET state=?, error=NULL WHERE operation_id=?').run('submitted', entry.operation_id);
        try {
          const intent = entry.intent;
          const route = this.db.prepare('SELECT host FROM queue_routes WHERE patient=?').get(intent.patient) as any;
          requireThat(route.host === this.client.hostId, 'queue_route_update_required');
          const result = intent.operation === 'create_resource' ? await this.client.createResource(intent.patient,intent.resource,intent.operation_id)
            : intent.operation === 'update_resource' ? await this.client.updateResource(intent.patient,intent.resource,intent.base_version,intent.operation_id)
            : intent.operation === 'delete_resource' ? await this.client.deleteResource(intent.patient,intent.resource_type,intent.resource_id,intent.base_version,intent.operation_id)
            : intent.operation === 'create_lab'
            ? await this.client.createLab(intent.patient, intent.resource, intent.operation_id)
            : await this.client.updateDemographics(intent.patient, intent.resource.name, intent.base_version, intent.operation_id);
          this.db.prepare('UPDATE queue SET state=?, result=?, error=NULL WHERE operation_id=?').run('committed', JSON.stringify(result), entry.operation_id);
        } catch (error) {
          const code = error instanceof ProtocolError ? error.code : 'transport_unavailable';
          const conflict = ['version_conflict', 'operation_id_conflict'].includes(code);
          const rejected = ['insufficient_scope', 'invalid_token', 'wrong_token_holder', 'authorization_expired_or_revoked', 'representative_not_authorized', 'authorization_expired', 'invalid_message', 'invalid_lab_category', 'not_found','resource_deleted','unsupported_resource_type','invalid_resource','invalid_resource_id','invalid_resource_code','invalid_resource_status','wrong_resource_patient','unsupported_resource_content','external_validation_rejected'].includes(code);
          this.db.prepare('UPDATE queue SET state=?, error=? WHERE operation_id=?').run(conflict ? 'conflicted' : rejected ? 'rejected' : 'pending', code, entry.operation_id);
          return; // Retry scheduling, backoff, and user resolution belong to the application.
        }
      }
    } finally { this.pumping = false; }
  }
  close(): void { requireThat(!this.pumping, 'queue_worker_busy'); this.db.close(); }
}