Reference source · 0.1.0-dev.11

notification-receiver.ts

src/notification-receiver.ts · Download original file

import { DatabaseSync } from 'node:sqlite';
import type { Identity, Credential } from './identity.ts';
import { TrustStore } from './identity.ts';
import { open, seal } from './envelope.ts';
import type { Envelope } from './envelope.ts';
import { callbackURL } from './callback-transport.ts';
import { exact, fingerprint, id, integer, jsonBytes, requireThat, unb64, parseObject } from './core.ts';

// Receiver is explicitly pinned to one Host and callback. A dedicated instance/database is
// recommended per endpoint. No clinical retrieval or external side effect runs in handle().
export class NotificationReceiver {
  db: DatabaseSync; identity: Identity; credential: Credential; hostCredential: Credential; trust: TrustStore; callback: string;
  constructor(path: string, identity: Identity, credential: Credential, hostCredential: Credential, trust: TrustStore, callback: string) {
    this.db = new DatabaseSync(path); this.identity = identity; this.credential = credential; this.hostCredential = hostCredential; this.trust = trust; this.callback = callbackURL(callback).href;
    this.db.exec(`PRAGMA journal_mode=WAL;
      CREATE TABLE IF NOT EXISTS receiver_owner(id INTEGER PRIMARY KEY CHECK(id=1), digest TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS expected_subscriptions(id TEXT PRIMARY KEY, expires_at INTEGER NOT NULL);
      CREATE TABLE IF NOT EXISTS callback_subscriptions(id TEXT PRIMARY KEY, challenge_digest TEXT NOT NULL, expires_at INTEGER NOT NULL);
      CREATE TABLE IF NOT EXISTS callback_receipts(delivery_id TEXT PRIMARY KEY, subscription_id TEXT NOT NULL, digest TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS notification_inbox(subscription_id TEXT PRIMARY KEY, revision INTEGER NOT NULL, pending INTEGER NOT NULL);`);
    const host = parseObject(unb64(hostCredential.payload));
    const owner = fingerprint(jsonBytes([identity.public.subject, identity.public.signingKid, identity.public.encryptionKid, host.subject, host.signingKid, host.encryptionKid, this.callback]));
    const previous = this.db.prepare('SELECT digest FROM receiver_owner WHERE id=1').get() as any;
    if (previous && previous.digest !== owner) { this.db.close(); requireThat(false, 'receiver_owner_changed'); }
    this.db.prepare('INSERT INTO receiver_owner VALUES (1,?) ON CONFLICT(id) DO NOTHING').run(owner);
  }
  // Application records its own expected registration BEFORE asking the Host. This prevents
  // an otherwise trusted Host from allocating arbitrary unsolicited receiver subscriptions.
  expect(subscriptionId: string, expiresAt: number): void {
    id(subscriptionId); integer(expiresAt);
    const previous = this.db.prepare('SELECT expires_at FROM expected_subscriptions WHERE id=?').get(subscriptionId) as any;
    requireThat(!previous || previous.expires_at === expiresAt, 'subscription_id_conflict');
    this.db.prepare('INSERT INTO expected_subscriptions VALUES (?,?) ON CONFLICT(id) DO NOTHING').run(subscriptionId, expiresAt);
  }
  handle(envelope: Envelope, now: number): Envelope {
    const host = this.trust.validate(this.hostCredential, now, undefined, 'host');
    this.trust.validate(this.credential, now, this.identity.public.subject, 'application');
    const opened = open(envelope, this.identity, this.trust, now, host.subject, null), body = opened.body;
    requireThat(opened.sender.kind === 'host' && opened.sender.signingKid === host.signingKid && opened.sender.encryptionKid === host.encryptionKid, 'wrong_host_key');
    id(body.subscription_id);
    const expected = this.db.prepare('SELECT expires_at FROM expected_subscriptions WHERE id=?').get(body.subscription_id) as any;
    requireThat(expected && now < expected.expires_at, 'unexpected_subscription');
    let reply: Record<string, any>;
    this.db.exec('BEGIN IMMEDIATE');
    try {
      if (body.operation === 'subscription_challenge') {
        exact(body, ['operation','subscription_id','challenge','callback','expires_at']);
        integer(body.expires_at); unb64(body.challenge, 32);
        requireThat(body.callback === this.callback && now < body.expires_at && body.expires_at <= Math.min(now + 300, expected.expires_at), 'invalid_callback_challenge');
        const digest = fingerprint(jsonBytes([body.operation, body.subscription_id, body.challenge, body.callback, body.expires_at])), prior = this.db.prepare('SELECT challenge_digest FROM callback_subscriptions WHERE id=?').get(body.subscription_id) as any;
        requireThat(!prior || prior.challenge_digest === digest, 'subscription_id_conflict');
        this.db.prepare('INSERT INTO callback_subscriptions VALUES (?,?,?) ON CONFLICT(id) DO NOTHING').run(body.subscription_id, digest, expected.expires_at);
        reply = { status: 'ok', operation: 'subscription_challenge_ack', subscription_id: body.subscription_id, challenge: body.challenge, callback: body.callback };
      } else {
        exact(body, ['operation','subscription_id','delivery_id']); requireThat(body.operation === 'change_notification', 'unsupported_operation'); id(body.delivery_id);
        requireThat(this.db.prepare('SELECT id FROM callback_subscriptions WHERE id=?').get(body.subscription_id), 'unverified_subscription');
        const digest = fingerprint(jsonBytes([body.operation, body.subscription_id, body.delivery_id])), prior = this.db.prepare('SELECT digest FROM callback_receipts WHERE delivery_id=?').get(body.delivery_id) as any;
        requireThat(!prior || prior.digest === digest, 'delivery_id_conflict');
        if (!prior) {
          this.db.prepare('INSERT INTO callback_receipts VALUES (?,?,?)').run(body.delivery_id, body.subscription_id, digest);
          this.db.prepare('INSERT INTO notification_inbox VALUES (?,1,1) ON CONFLICT(subscription_id) DO UPDATE SET revision=revision+1,pending=1').run(body.subscription_id);
        }
        reply = { status: 'ok', operation: 'notification_ack', subscription_id: body.subscription_id, delivery_id: body.delivery_id };
      }
      // Commit the receipt AND durable pull task before acknowledging. Duplicates get a fresh
      // reply envelope, preserving reply binding even when the first HTTP response was lost.
      this.db.exec('COMMIT');
    } catch (e) { this.db.exec('ROLLBACK'); throw e; }
    return seal(reply, this.identity, this.credential, opened.sender, now, opened.header.messageId);
  }
  pending(): { subscription_id: string; revision: number }[] {
    return this.db.prepare('SELECT subscription_id,revision FROM notification_inbox WHERE pending=1 ORDER BY subscription_id').all() as any;
  }
  // Call only after durably applying feed pages/cursor. An intervening notification increments
  // the revision, so finishing an earlier task cannot erase a newer wake-up. Keep the
  // counter after completion: resetting it would let a repeated old completion erase new work.
  complete(subscriptionId: string, revision: number): void {
    id(subscriptionId); integer(revision);
    this.db.prepare('UPDATE notification_inbox SET pending=0 WHERE subscription_id=? AND revision=?').run(subscriptionId, revision);
  }
  close(): void { this.db.close(); }
}