clanker-saddle/src/Services/PushService.vala

using Invercargill;
using Invercargill.DataStructures;
using Invercargill.Expressions;
using InvercargillSql;
using InvercargillSql.Orm;
using Inversion;
using Soup;
using Astralis;
using ClankerSaddle.Data;

namespace ClankerSaddle.Services {

    // ---- OpenSSL shim (src/Push/web-push-crypto.c): raw P-256 and
    // AES-GCM primitives. Pointer parameters deliberately — no array
    // lengths cross the boundary. ----
    [CCode (cname = "ws_p256_generate")]
    internal static extern int crypto_p256_generate(uint8 *priv, uint8 *pub);

    [CCode (cname = "ws_p256_public_from_private")]
    internal static extern int crypto_p256_public_from_private(uint8 *priv, uint8 *pub);

    [CCode (cname = "ws_p256_ecdh")]
    internal static extern int crypto_p256_ecdh(uint8 *priv, uint8 *peer, uint8 *secret);

    [CCode (cname = "ws_p256_sign")]
    internal static extern int crypto_p256_sign(uint8 *priv, uint8 *digest, uint8 *sig);

    [CCode (cname = "ws_aes128gcm_encrypt")]
    internal static extern int crypto_aes128gcm_encrypt(uint8 *key, uint8 *iv,
        uint8 *plaintext, size_t pt_len, uint8 *out, out size_t out_len);

    [CCode (cname = "ws_random")]
    internal static extern int crypto_random(uint8 *buf, size_t len);

    /**
     * One outgoing notification (plan 11 task E): `title` carries the
     * task title plus the event kind, `body` the question or summary,
     * `url` the task page the service worker routes a click to. Pure
     * data — the payload builder below is deterministic so tests can
     * assert on it end-to-end (the test harness owns the client keys,
     * so it can decrypt what was sent).
     */
    public class PushNotification : Object {
        public string title { get; set; default = ""; }
        public string body { get; set; default = ""; }
        public string url { get; set; default = ""; }
    }

    /**
     * The roll-up of one {@link PushService.send_test} probe: counts
     * per outcome plus the first failure's detail for the toast.
     * `first_failure` names the HTTP status ("HTTP 401") or the
     * transport error text — the dashboard's test button toasts the
     * REAL end-to-end result, never a blind success.
     */
    public class PushTestReport : Object {
        public int delivered { get; set; }
        public int failed { get; set; }
        /** Endpoints that answered 404/410 — pruned during the probe. */
        public int pruned { get; set; }
        public string first_failure { get; set; default = ""; }
    }

    /**
     * Web Push delivery (plan 11 task E): real Web Push protocol over
     * VAPID. The keypair is generated once and persisted in the config
     * dir (a `vapid` section inside web-config.json); payloads are
     * encrypted per RFC 8291 (ECDH P-256 + HKDF + AES-128-GCM,
     * aes128gcm content coding) and authorised by an ES256 VAPID JWT.
     *
     * Targeting: BLOCK events
     * condition go to every push-subscribed user globally; a task's
     * first start, failure and completion go only to users holding a
     * `task_notify_prefs` row for it ("Notify me").
     *
     * Delivery is best-effort from a singleton queue: enqueues are
     * in-memory (they never block the request), a flush is kicked
     * immediately and re-run by the dispatcher's tick while anything
     * remains (a failed send retries next tick, up to
     * {@link MAX_ATTEMPTS}); endpoints answering 410 Gone (or 404) are
     * pruned — the browser already destroyed the subscription.
     */
    public class PushService : Object {

        private const int MAX_ATTEMPTS = 5;

        /** The persisted keypair, base64url (empty before {@link ensure_vapid_keys}). */
        public static string vapid_public_key { get; private set; default = ""; }
        private static string vapid_private_key = "";

        private Container container = inject<Container>();
        private WebConfig web_config = inject<WebConfig>();
        private Soup.Session http = new Soup.Session() {
            timeout = 10,
            idle_timeout = 10
        };

        /** {@link ensure_vapid_keys} with the injected config (main's startup call). */
        public void ensure_keys_from_config() {
            ensure_vapid_keys(web_config);
        }

        /** A resolved send: one endpoint with its keys and plaintext payload. */
        private class SendJob : Object {
            public string endpoint = "";
            public string p256dh = "";
            public string auth = "";
            public string payload = "";
            public int attempts = 0;
            /** The last HTTP status the endpoint answered (0 = transport error). */
            public int last_status = 0;
        }

        /** A queued notification before target resolution. */
        private class PendingTarget : Object {
            public bool global = false;
            public int64 task_id = 0;
            public PushNotification notification;
        }

        private GLib.Queue<PendingTarget> pending = new GLib.Queue<PendingTarget>();
        private GLib.Queue<SendJob> queued = new GLib.Queue<SendJob>();
        private Mutex queue_lock = Mutex();
        private bool flushing = false;

        // ------------------------------------------------------------------
        // VAPID keys
        // ------------------------------------------------------------------

        /**
         * Loads the VAPID keypair from the config's `vapid` section,
         * generating and persisting a fresh one when missing (or when
         * the stored pair does not match — the public point is derived
         * from the private scalar and compared). Called once from
         * `main` before the server accepts requests, so every
         * subscriber always sees the same public key.
         */
        public static void ensure_vapid_keys(WebConfig config) {
            string stored_public = "";
            string stored_private = "";
            var section = config != null ? config.get_section("vapid") : null;
            if (section != null) {
                stored_public = ((!)section).get_string("public_key", "");
                stored_private = ((!)section).get_string("private_key", "");
            }
            if (keypair_valid(stored_public, stored_private)) {
                vapid_public_key = stored_public;
                vapid_private_key = stored_private;
                return;
            }

            var priv = new uint8[32];
            var pub = new uint8[65];
            if (crypto_p256_generate((uint8*) priv, (uint8*) pub) != 1) {
                warning("[ClankerSaddle] VAPID keypair generation failed;"
                    + " web push disabled for this run");
                return;
            }
            vapid_private_key = base64url_encode(priv);
            vapid_public_key = base64url_encode(pub);
            persist_vapid_keys(config);
            print("[ClankerSaddle] Generated VAPID keypair (public key: %s)\n",
                vapid_public_key);
        }

        /** True when the stored pair decodes and the public point matches. */
        private static bool keypair_valid(string public_b64, string private_b64) {
            uint8[] priv = base64url_decode(private_b64);
            uint8[] pub = base64url_decode(public_b64);
            if (priv.length != 32 || pub.length != 65 || pub[0] != 0x04) {
                return false;
            }
            var derived = new uint8[65];
            if (crypto_p256_public_from_private((uint8*) priv, (uint8*) derived) != 1) {
                return false;
            }
            for (var index = 0; index < 65; index++) {
                if (derived[index] != pub[index]) {
                    return false;
                }
            }
            return true;
        }

        /**
         * Writes the `vapid` section into the config file the app
         * loaded (first loaded file; when nothing was loaded, the first
         * discovered path — created). Existing keys and sections are
         * preserved: the file is re-parsed and only `vapid` replaced.
         */
        private static void persist_vapid_keys(WebConfig config) {
            string path = "";
            string[] loaded = config != null ? config.get_loaded_files() : new string[0];
            if (loaded.length > 0) {
                path = loaded[0];
            } else {
                try {
                    string[] discovered = WebConfigLoader.discover_config_paths();
                    if (discovered.length > 0) {
                        path = discovered[0];
                    }
                } catch (Error e) {
                    path = "";
                }
            }
            if (path.length == 0) {
                warning("[ClankerSaddle] No config file to persist the VAPID"
                    + " keypair into; keys are in-memory for this run only");
                return;
            }

            try {
                string text = "{}";
                if (FileUtils.test(path, FileTest.EXISTS)) {
                    string contents;
                    FileUtils.get_contents(path, out contents);
                    text = contents.strip().length > 0 ? contents : "{}";
                }
                var parser = new Json.Parser();
                parser.load_from_data(text);
                var root = parser.get_root();
                var document = root != null && root.get_node_type() == Json.NodeType.OBJECT
                    ? root.get_object() : new Json.Object();

                var vapid = new Json.Object();
                vapid.set_string_member("public_key", vapid_public_key);
                vapid.set_string_member("private_key", vapid_private_key);
                document.set_object_member("vapid", vapid);

                var generator = new Json.Generator() {
                    pretty = true,
                    indent = 2
                };
                generator.set_root(new Json.Node(Json.NodeType.OBJECT).init_object(document));
                FileUtils.set_contents(path, generator.to_data(null) + "\n");
            } catch (Error e) {
                warning("[ClankerSaddle] Persisting the VAPID keypair into %s"
                    + " failed: %s", path, e.message);
            }
        }

        // ------------------------------------------------------------------
        // Payload builder
        // ------------------------------------------------------------------

        /** The human label of a block kind in notification titles. */
        public static string block_kind_label(string block_type) {
            switch (block_type) {
                case BLOCK_RFC:
                    return "RFC raised";
                case BLOCK_REVIEW:
                    return "review requested";
                case BLOCK_PLAN:
                    return "plan review";
                default:
                    return "question raised";
            }
        }

        /**
         * The block-event payload (global audience): title = task title
         * + kind, body = the question, click-through = the task page.
         */
        public static PushNotification block_notification(string task_title,
                int64 task_id, string block_type, string question) {
            return new PushNotification() {
                title = @"$(task_title) — $(block_kind_label(block_type))",
                body = question,
                url = @"/tasks/$(task_id)"
            };
        }

        /**
         * The opted-in lifecycle payload: `started` (first start),
         * `completed` or `failed`, with the task's summary as the body.
         */
        public static PushNotification lifecycle_notification(string task_title,
                int64 task_id, string kind, string? summary) {
            return new PushNotification() {
                title = @"$(task_title) — $(kind)",
                body = summary != null && ((!)summary).length > 0
                    ? (!)summary : @"Task $(task_id) $(kind)",
                url = @"/tasks/$(task_id)"
            };
        }

        /**
         * The goal block-event payload (plan 12, global audience —
         * goal blocks always notify): title = goal title + kind, body
         * = the question, click-through = the goal page.
         */
        public static PushNotification goal_block_notification(string goal_title,
                int64 goal_id, string block_type, string question) {
            string label = block_kind_label(block_type);
            if (block_type == BLOCK_COMPLETION) {
                label = "completion review";
            }
            return new PushNotification() {
                title = @"$(goal_title) — $(label)",
                body = question,
                url = @"/goals/$(goal_id)"
            };
        }

        /**
         * The goal terminal payload (plan 12, global audience — goal
         * success/failure always notify): `succeeded` picks the
         * wording; the report body is passed by the caller.
         */
        public static PushNotification goal_terminal_notification(string goal_title,
                int64 goal_id, bool succeeded) {
            return new PushNotification() {
                title = @"$(goal_title) — $(succeeded ? "complete" : "failed")",
                body = succeeded
                    ? @"Goal $(goal_id) reached its completion review approval"
                    : @"Goal $(goal_id) was failed by its agent",
                url = @"/goals/$(goal_id)"
            };
        }

        /**
         * The "Send test notification" payload (per-browser bell): a
         * REAL Web Push through the full RFC 8291 + VAPID path whose
         * arrival the human can see — the end-to-end probe that was
         * missing entirely.
         */
        public static PushNotification test_notification() {
            return new PushNotification() {
                title = "Test notification",
                body = "A live Web Push from Clanker Saddle — if you can"
                    + " read this, delivery works.",
                url = "/"
            };
        }

        // ------------------------------------------------------------------
        // Enqueue + delivery
        // ------------------------------------------------------------------

        /** Queues a notification for every push-subscribed user (block events). */
        public void notify_global(PushNotification notification) {
            enqueue_target(new PendingTarget() {
                global = true,
                notification = notification
            });
        }

        /**
         * Queues a notification for the users holding a "Notify me"
         * pref for `task_id` (first start / failure / completion).
         */
        public void notify_opted_in(int64 task_id, PushNotification notification) {
            enqueue_target(new PendingTarget() {
                global = false,
                task_id = task_id,
                notification = notification
            });
        }

        private void enqueue_target(PendingTarget target) {
            if (vapid_private_key.length == 0) {
                return;
            }
            queue_lock.lock();
            pending.push_tail(target);
            queue_lock.unlock();
            flush.begin();
        }

        /**
         * Delivers everything queued: resolves pending targets into
         * per-endpoint jobs (a fresh database scope — subscriptions may
         * have changed since the event), then sends each job. Failures
         * stay queued for the next run (the dispatcher's tick calls
         * this); 410/404 endpoints are pruned instead of retried.
         * Single-flight: a concurrent call returns immediately while a
         * flush is in progress.
         */
        public async void flush() {
            if (flushing) {
                return;
            }
            flushing = true;
            try {
                yield resolve_pending();
                while (true) {
                    queue_lock.lock();
                    SendJob? job = queued.pop_head();
                    queue_lock.unlock();
                    if (job == null) {
                        break;
                    }
                    bool requeued = yield deliver((!)job);
                    if (requeued) {
                        // A failed send goes back for a LATER pass:
                        // draining in place would burn every attempt in
                        // one tight loop. The dispatcher tick (or the
                        // next enqueue's flush) picks the job back up —
                        // that spacing IS the retry backoff.
                        break;
                    }
                }
            } finally {
                flushing = false;
            }
        }

        /** Moves pending targets into per-endpoint send jobs. */
        private async void resolve_pending() {
            queue_lock.lock();
            var targets = new GLib.List<PendingTarget>();
            while (!pending.is_empty()) {
                targets.append(pending.pop_head());
            }
            queue_lock.unlock();
            if (targets.length() == 0) {
                return;
            }

            try {
                var scope = container.create_scope();
                var db = scope.resolve<OrmSession>();
                var subscriptions = yield db.query<PushSubscription>().materialise_async();
                var jobs = new GLib.Queue<SendJob>();
                foreach (var target in targets) {
                    foreach (var row in subscriptions) {
                        if (!target.global
                                && !(yield pref_held(db, row.user_id, target.task_id))) {
                            continue;
                        }
                        jobs.push_tail(new SendJob() {
                            endpoint = row.endpoint,
                            p256dh = row.p256dh,
                            auth = row.auth,
                            payload = payload_json(target.notification)
                        });
                    }
                }
                if (jobs.is_empty()) {
                    return;
                }
                queue_lock.lock();
                while (!jobs.is_empty()) {
                    queued.push_tail(jobs.pop_head());
                }
                queue_lock.unlock();
            } catch (Error e) {
                warning("[ClankerSaddle] Push target resolution failed: %s", e.message);
                // Targets stay lost this pass; the event is best-effort.
            }
        }

        /** The wire payload JSON (compact, exactly what the SW parses). */
        private static string payload_json(PushNotification notification) {
            var builder = new Json.Builder();
            builder.begin_object();
            builder.set_member_name("title");
            builder.add_string_value(notification.title);
            builder.set_member_name("body");
            builder.add_string_value(notification.body);
            builder.set_member_name("url");
            builder.add_string_value(notification.url);
            builder.end_object();
            var generator = new Json.Generator();
            generator.set_root(builder.get_root());
            return generator.to_data(null);
        }

        private enum SendOutcome {
            DELIVERED,
            RETRY,
            GONE
        }

        /**
         * Applies one job's outcome: DELIVERED drops it, GONE prunes
         * the subscription row, anything else requeues it for a later
         * tick. Returns true when the job was requeued — the flush
         * drain then STOPS (the retry waits for the next tick instead
         * of burning every attempt back-to-back), which is what makes
         * MAX_ATTEMPTS a real backoff rather than one hot loop.
         */
        private async bool deliver(SendJob job) {
            SendOutcome outcome = SendOutcome.RETRY;
            try {
                outcome = yield deliver_once(job);
            } catch (Error e) {
                outcome = SendOutcome.RETRY;
            }
            if (outcome == SendOutcome.DELIVERED) {
                return false;
            }
            if (outcome == SendOutcome.GONE) {
                yield prune_subscription(job.endpoint);
                return false;
            }
            job.attempts++;
            if (job.attempts >= MAX_ATTEMPTS) {
                // Giving up is loud on purpose: a silently dropped
                // notification looks exactly like a delivered one.
                warning("[ClankerSaddle] Push to %s gave up after %d attempts"
                    + " — notification dropped undelivered",
                    job.endpoint, job.attempts);
                return false;
            }
            queue_lock.lock();
            queued.push_tail(job);
            queue_lock.unlock();
            return true;
        }

        /**
         * The dashboard's "Send test notification" probe (per-browser
         * bell, shown only when THIS browser is subscribed): one REAL
         * delivery attempt to each of the user's endpoints through the
         * same {@link deliver_once} crypto + VAPID path the queue
         * uses, awaited inline so the action can report per-endpoint
         * outcomes. The shared queue's flush is single-flight and
         * fire-and-forget — its outcomes are unobservable from a
         * request — so the probe resolves and sends synchronously
         * instead of enqueueing. A refused endpoint is reported rather
         * than retried here (the queue's own backoff still governs
         * real events); 404/410 endpoints are pruned exactly as the
         * queue would, so the probe also heals dead rows on demand.
         */
        public async PushTestReport send_test(OrmSession db, int64 user_id) throws Error {
            var report = new PushTestReport();
            var jobs = new GLib.List<SendJob>();
            foreach (var row in yield db.query<PushSubscription>()
                    .where(expr("user_id == $0", elem<int64?>(user_id)))
                    .materialise_async()) {
                jobs.append(new SendJob() {
                    endpoint = row.endpoint,
                    p256dh = row.p256dh,
                    auth = row.auth,
                    payload = payload_json(test_notification())
                });
            }
            foreach (var job in jobs) {
                SendOutcome outcome = SendOutcome.RETRY;
                string problem = "";
                try {
                    outcome = yield deliver_once(job);
                } catch (Error e) {
                    problem = e.message;
                }
                if (outcome == SendOutcome.DELIVERED) {
                    report.delivered++;
                    continue;
                }
                if (outcome == SendOutcome.GONE) {
                    report.pruned++;
                    yield prune_subscription(job.endpoint);
                    continue;
                }
                report.failed++;
                if (report.first_failure.length == 0) {
                    report.first_failure = job.last_status > 0
                        ? @"HTTP $(job.last_status)" : problem;
                }
            }
            return report;
        }

        /**
         * One Web Push protocol POST: RFC 8291 aes128gcm body, ES256
         * VAPID JWT in the Authorization header. Transport errors and
         * 5xx throw (retry); 404/410 report {@link SendOutcome.GONE}
         * (prune); any OTHER 4xx is a refusal — logged and retried,
         * never silently delivered; remaining 2xx/3xx counts as
         * delivered (the push service owns delivery from here).
         */
        private async SendOutcome deliver_once(SendJob job) throws Error {
            uint8[] client_public = base64url_decode(job.p256dh);
            uint8[] auth_secret = base64url_decode(job.auth);
            if (client_public.length != 65 || auth_secret.length != 16) {
                return SendOutcome.GONE;
            }

            // RFC 8291: ephemeral ECDH -> HKDF -> AES-128-GCM.
            var eph_priv = new uint8[32];
            var eph_pub = new uint8[65];
            if (crypto_p256_generate((uint8*) eph_priv, (uint8*) eph_pub) != 1) {
                throw new Error(Quark.from_string("clanker-saddle"), 60,
                    "ephemeral P-256 generation failed");
            }
            var secret = new uint8[32];
            if (crypto_p256_ecdh((uint8*) eph_priv, (uint8*) client_public, (uint8*) secret) != 1) {
                throw new Error(Quark.from_string("clanker-saddle"), 61,
                    "ECDH with the subscription key failed");
            }
            var ikm = new uint8[48];
            for (var index = 0; index < 32; index++) {
                ikm[index] = secret[index];
            }
            for (var index = 0; index < 16; index++) {
                ikm[32 + index] = auth_secret[index];
            }

            var salt = new uint8[16];
            if (crypto_random((uint8*) salt, 16) != 1) {
                throw new Error(Quark.from_string("clanker-saddle"), 62,
                    "salt generation failed");
            }
            // "WebPush: info" || 0x00 || eph_pub || client_pub
            var context = new uint8[14 + 65 + 65];
            var label = "WebPush: info".data;
            for (var index = 0; index < 13; index++) {
                context[index] = label[index];
            }
            context[13] = 0x00;
            for (var index = 0; index < 65; index++) {
                context[14 + index] = eph_pub[index];
                context[14 + 65 + index] = client_public[index];
            }
            uint8[] cek = hkdf(salt, ikm, context, 16);
            uint8[] nonce = hkdf(salt, ikm, context, 12);

            uint8[] plaintext = job.payload.data;
            var framed = new uint8[plaintext.length + 1];
            for (var index = 0; index < plaintext.length; index++) {
                framed[index] = plaintext[index];
            }
            framed[plaintext.length] = 0x02; // last-record delimiter
            var ciphertext = new uint8[framed.length + 16];
            size_t written = 0;
            if (crypto_aes128gcm_encrypt((uint8*) cek, (uint8*) nonce, (uint8*) framed,
                    framed.length, (uint8*) ciphertext, out written) != 1) {
                throw new Error(Quark.from_string("clanker-saddle"), 63,
                    "payload encryption failed");
            }

            var body = new uint8[16 + 4 + 1 + 65 + written];
            for (var index = 0; index < 16; index++) {
                body[index] = salt[index];
            }
            int64 record_size = plaintext.length + 17;
            body[16] = (uint8) ((record_size >> 24) & 0xff);
            body[17] = (uint8) ((record_size >> 16) & 0xff);
            body[18] = (uint8) ((record_size >> 8) & 0xff);
            body[19] = (uint8) (record_size & 0xff);
            body[20] = 65;
            for (var index = 0; index < 65; index++) {
                body[21 + index] = eph_pub[index];
            }
            for (var index = 0; index < written; index++) {
                body[21 + 65 + index] = ciphertext[index];
            }

            var message = new Soup.Message("POST", job.endpoint);
            message.set_request_body_from_bytes("application/octet-stream",
                new Bytes(body));
            message.request_headers.append("TTL", "60");
            message.request_headers.append("Urgency", "high");
            message.request_headers.append("Authorization",
                @"vapid t=$(vapid_jwt(job.endpoint)), k=$(vapid_public_key)");
            try {
                yield http.send_and_read_async(message, Priority.DEFAULT, null);
            } catch (Error e) {
                throw new Error(Quark.from_string("clanker-saddle"), 64,
                    @"push POST failed: $(e.message)");
            }
            if (message.status_code == 404 || message.status_code == 410) {
                return SendOutcome.GONE;
            }
            if (message.status_code >= 500) {
                job.last_status = (int) message.status_code;
                throw new Error(Quark.from_string("clanker-saddle"), 65,
                    @"push service answered HTTP $(message.status_code)");
            }
            if (message.status_code >= 400) {
                // 401/403 (VAPID rejected) and every other 4xx that is
                // not the terminal 404/410 handled above: the push
                // service did NOT accept the message. Truthful
                // delivery — log it with the endpoint and status,
                // retry on later ticks, give up loudly after
                // MAX_ATTEMPTS, and never count it as delivered. The
                // subscription row stays: the endpoint may recover
                // (e.g. a VAPID key rotation heals on a later send).
                job.last_status = (int) message.status_code;
                warning("[ClankerSaddle] Push endpoint %s refused delivery:"
                    + " HTTP %d — not delivered, retrying",
                    job.endpoint, (int) message.status_code);
                return SendOutcome.RETRY;
            }
            return SendOutcome.DELIVERED;
        }

        /**
         * The ES256 VAPID JWT for an endpoint: aud = its origin, exp =
         * now + 12h, sub = a mailto contact (`SADDLE_VAPID_SUBJECT`
         * overrides), signature = raw r||s.
         */
        private static string vapid_jwt(string endpoint) {
            string origin = endpoint;
            int scheme = endpoint.index_of("://");
            if (scheme > 0) {
                int path = endpoint.index_of("/", scheme + 3);
                origin = path > 0 ? endpoint.substring(0, path) : endpoint;
            }
            string subject = Environment.get_variable("SADDLE_VAPID_SUBJECT")
                ?? "mailto:clanker-saddle@localhost";
            string header = "{\"typ\":\"JWT\",\"alg\":\"ES256\"}";
            string claims = "{\"aud\":\"" + origin + "\",\"exp\":"
                + (new DateTime.now_local().to_unix() + 43200).to_string()
                + ",\"sub\":\"" + subject + "\"}";
            string header_b64 = base64url_encode(header.data);
            string claims_b64 = base64url_encode(claims.data);
            string signing_input = header_b64 + "." + claims_b64;

            var checksum = new Checksum(ChecksumType.SHA256);
            size_t digest_len = 32;
            var digest = new uint8[32];
            checksum.update(signing_input.data, signing_input.length);
            checksum.get_digest(digest, ref digest_len);

            uint8[] priv = base64url_decode(vapid_private_key);
            var signature = new uint8[64];
            if (crypto_p256_sign((uint8*) priv, (uint8*) digest, (uint8*) signature) != 1) {
                warning("[ClankerSaddle] VAPID JWT signing failed");
                return "";
            }
            return signing_input + "." + base64url_encode(signature);
        }

        /** Deletes subscriptions whose endpoint went 404/410 Gone. */
        private async void prune_subscription(string endpoint) {
            try {
                var scope = container.create_scope();
                var db = scope.resolve<OrmSession>();
                foreach (var row in yield db.query<PushSubscription>()
                        .where(expr("endpoint == $0", elem(endpoint)))
                        .materialise_async()) {
                    yield db.delete_async<PushSubscription>(row);
                }
                print("[ClankerSaddle] Pruned gone push subscription %s\n",
                    endpoint.substring(0, int.min(endpoint.length, 72)));
            } catch (Error e) {
                warning("[ClankerSaddle] Pruning push subscription failed: %s", e.message);
            }
        }

        // ------------------------------------------------------------------
        // Subscriptions + notify prefs (caller's database scope)
        // ------------------------------------------------------------------

        /**
         * Validates a browser subscription registration: the endpoint
         * must be an absolute http(s) URL, `p256dh` a base64url 65-byte
         * uncompressed P-256 point and `auth` a base64url 16-byte
         * secret. Returns null when valid, else the shown error.
         */
        public static string? validate_subscription(string endpoint,
                string p256dh, string auth) {
            if (!(endpoint.has_prefix("http://") || endpoint.has_prefix("https://"))
                    || endpoint.length <= 8) {
                return "The endpoint must be an absolute http(s) URL";
            }
            uint8[] public_point = base64url_decode(p256dh);
            if (public_point.length != 65 || public_point[0] != 0x04) {
                return "p256dh must be a base64url 65-byte uncompressed P-256 point";
            }
            uint8[] auth_secret = base64url_decode(auth);
            if (auth_secret.length != 16) {
                return "auth must be a base64url 16-byte secret";
            }
            return null;
        }

        /**
         * Stores (or replaces — the endpoint is unique per browser) a
         * user's subscription.
         */
        public async void upsert_subscription(OrmSession db, int64 user_id,
                string endpoint, string p256dh, string auth) throws Error {
            var existing = yield db.query<PushSubscription>()
                .where(expr("endpoint == $0", elem(endpoint)))
                .first_async();
            if (existing != null) {
                ((!)existing).user_id = user_id;
                ((!)existing).p256dh = p256dh;
                ((!)existing).auth = auth;
                yield db.update_async<PushSubscription>((!)existing);
                return;
            }
            yield db.insert_async<PushSubscription>(new PushSubscription() {
                user_id = user_id,
                endpoint = endpoint,
                p256dh = p256dh,
                auth = auth,
                created_at = new DateTime.now_local()
            });
        }

        /**
         * Removes one of the user's subscription rows by endpoint (a
         * no-op when the endpoint belongs to someone else or is gone).
         */
        public async void remove_subscription(OrmSession db, int64 user_id,
                string endpoint) throws Error {
            foreach (var row in yield db.query<PushSubscription>()
                    .where(expr("endpoint == $0 && user_id == $1",
                        elem(endpoint), elem<int64?>(user_id)))
                    .materialise_async()) {
                yield db.delete_async<PushSubscription>(row);
            }
        }

        /**
         * True when the user holds any subscription. Convenience only
         * since the per-browser bell: the authoritative per-browser
         * state is {@link user_endpoint_fingerprints} matched
         * client-side against the browser's own subscription.
         */
        public async bool user_subscribed(OrmSession db, int64 user_id) throws Error {
            foreach (var row in yield db.query<PushSubscription>()
                    .where(expr("user_id == $0", elem<int64?>(user_id)))
                    .materialise_async()) {
                return true;
            }
            return false;
        }

        /**
         * This user's subscription endpoint fingerprints, newest
         * first (the per-browser bell state): each entry is
         * {@link endpoint_fingerprint} of one stored endpoint — never
         * the endpoint itself, which stays server-private. The browser
         * fingerprints its own pushManager subscription and looks it
         * up in the list: present => THIS browser is subscribed (show
         * "Turn off"), absent => "Enable" (+ a count of the others).
         */
        public async string[] user_endpoint_fingerprints(OrmSession db, int64 user_id) throws Error {
            var fingerprints = new string[0];
            foreach (var row in yield db.query<PushSubscription>()
                    .where(expr("user_id == $0", elem<int64?>(user_id)))
                    .order_by_desc(expr("created_at"))
                    .order_by_desc(expr("id"))
                    .materialise_async()) {
                fingerprints += endpoint_fingerprint(row.endpoint);
            }
            return fingerprints;
        }

        /**
         * A privacy-safe handle for one endpoint: the first 16 hex
         * chars of its SHA-256 — enough for a browser to recognise its
         * own subscription, useless for reconstructing the URL. The
         * client mirrors this with crypto.subtle (both need a secure
         * context, so any browser that can subscribe can compare).
         */
        public static string endpoint_fingerprint(string endpoint) {
            var checksum = new Checksum(ChecksumType.SHA256);
            size_t digest_len = 32;
            var digest = new uint8[32];
            checksum.update(endpoint.data, endpoint.length);
            checksum.get_digest(digest, ref digest_len);
            var builder = new StringBuilder();
            for (var index = 0; index < 8; index++) {
                builder.append_printf("%02x", digest[index]);
            }
            return builder.str;
        }

        /** True when the user holds a "Notify me" pref for the task. */
        public async bool has_notify_pref(OrmSession db, int64 user_id,
                int64 task_id) throws Error {
            return yield pref_held(db, user_id, task_id);
        }

        /** Inserts or removes the per-task "Notify me" pref. */
        public async void set_notify_pref(OrmSession db, int64 user_id,
                int64 task_id, bool enabled) throws Error {
            foreach (var row in yield db.query<TaskNotifyPref>()
                    .where(expr("task_id == $0 && user_id == $1",
                        elem<int64?>(task_id), elem<int64?>(user_id)))
                    .materialise_async()) {
                if (!enabled) {
                    yield db.delete_async<TaskNotifyPref>(row);
                }
                return;
            }
            if (enabled) {
                yield db.insert_async<TaskNotifyPref>(new TaskNotifyPref() {
                    task_id = task_id,
                    user_id = user_id
                });
            }
        }

        private static async bool pref_held(OrmSession db, int64 user_id,
                int64 task_id) throws Error {
            var row = yield db.query<TaskNotifyPref>()
                .where(expr("task_id == $0 && user_id == $1",
                    elem<int64?>(task_id), elem<int64?>(user_id)))
                .first_async();
            return row != null;
        }

        // ------------------------------------------------------------------
        // Primitives
        // ------------------------------------------------------------------

        /** HKDF (RFC 5869, SHA-256): extract(salt, ikm) then expand(info). */
        private static uint8[] hkdf(uint8[] salt, uint8[] ikm, uint8[] info,
                size_t length) {
            var extract_hmac = new Hmac(ChecksumType.SHA256, salt);
            extract_hmac.update(ikm);
            size_t prk_len = 32;
            var prk = new uint8[32];
            extract_hmac.get_digest(prk, ref prk_len);

            var output = new uint8[length];
            var counter = new uint8[1];
            uint8[] accumulator = new uint8[0];
            size_t produced = 0;
            uint8 block_index = 1;
            while (produced < length) {
                var expand_hmac = new Hmac(ChecksumType.SHA256, prk);
                expand_hmac.update(accumulator);
                expand_hmac.update(info);
                counter[0] = block_index;
                expand_hmac.update(counter);
                size_t block_len = 32;
                var block = new uint8[32];
                expand_hmac.get_digest(block, ref block_len);
                size_t take = 32;
                if (length - produced < 32) {
                    take = length - produced;
                }
                for (var index = 0; index < take; index++) {
                    output[produced + index] = block[index];
                }
                produced += take;
                accumulator = block;
                block_index++;
            }
            return output;
        }

        /** Standard base64 -> base64url (unpadded). */
        public static string base64url_encode(uint8[] data) {
            string encoded = Base64.encode(data);
            var builder = new StringBuilder();
            for (var index = 0; index < encoded.length; index++) {
                char character = encoded[index];
                if (character == '+') {
                    builder.append_c('-');
                } else if (character == '/') {
                    builder.append_c('_');
                } else if (character != '=') {
                    builder.append_c(character);
                }
            }
            return builder.str;
        }

        /** base64url (padded or not) -> bytes; junk decodes empty. */
        public static uint8[] base64url_decode(string text) {
            var normalised = new StringBuilder();
            for (var index = 0; index < text.length; index++) {
                char character = text[index];
                if (character == '-') {
                    normalised.append_c('+');
                } else if (character == '_') {
                    normalised.append_c('/');
                } else {
                    normalised.append_c(character);
                }
            }
            while (normalised.len % 4 != 0) {
                normalised.append_c('=');
            }
            return Base64.decode(normalised.str);
        }
    }
}