From 6e6a1e9c8a4d3d2d5522ae61860c32941c22b5da Mon Sep 17 00:00:00 2001
From: Admin <1+admin@noreply.git.dcntrlized.org>
Date: Mon, 19 Oct 2026 02:26:46 +0100
Subject: [PATCH] [feat](alarm): add a durable per-device delivery ledger
pg_notify and the socket fan-out are not durable, so an offline phone used to miss a trigger forever and no-auto-stop would leave it ringing at nobody. Write one ledger row per enabled subscription in the trigger transaction, replay what a subscription still owes on every socket connect, and clear a row only on an explicit ack (live socket ack or ackAlarm); alarmStatus is a read-only check-in that answers 409 ERR_ALARM_TRIGGERED_OFFLINE with the missed frame. Ack is idempotent, and a destroyed alert cascades its debt away so a client is never told it missed an alarm that no longer exists.
---
database/migrations/005_alarm_deliveries.sql | 52 ++++
src/actions/ack.go | 136 +++++++++
src/actions/deliveries.go | 50 ++++
src/actions/deliveries_test.go | 95 ++++++
src/db/deliveries.go | 168 +++++++++++
src/db/deliveries_db_test.go | 287 +++++++++++++++++++
src/handlers/actions/action__actionName_.go | 73 +++--
src/realtime/client.go | 41 +++
src/realtime/hub.go | 123 ++++++--
src/realtime/hub_test.go | 141 ++++++++-
src/realtime/latency_test.go | 13 +-
src/realtime/notify.go | 26 +-
12 files changed, 1128 insertions(+), 77 deletions(-)
create mode 100644 database/migrations/005_alarm_deliveries.sql
create mode 100644 src/actions/ack.go
create mode 100644 src/actions/deliveries.go
create mode 100644 src/actions/deliveries_test.go
create mode 100644 src/db/deliveries.go
create mode 100644 src/db/deliveries_db_test.go
diff --git a/database/migrations/005_alarm_deliveries.sql b/database/migrations/005_alarm_deliveries.sql
new file mode 100644
index 0000000..c96ad11
--- /dev/null
+++ b/database/migrations/005_alarm_deliveries.sql
@@ -0,0 +1,52 @@
+\c dcntrlized;
+
+-- 005_alarm_deliveries.sql — a durable per-device delivery ledger.
+--
+-- Why this exists: pg_notify and the WebSocket fan-out are not durable. A phone
+-- that is offline at trigger time misses the frame forever, and the in-memory
+-- catch-up cache only lives in the process that produced the event. This table is
+-- the durable record of who still owes an alarm, so "no auto-stop" is safe: the
+-- alarm stays live until the owner disarms it, and every subscribed device has a
+-- row here until it has been told.
+--
+-- Shape: one row per (alert, subscription, event), written in the SAME
+-- transaction as the alarm_events insert and therefore before the NOTIFY on
+-- commit. If this insert fails the trigger fails — a fan-out that records no debt
+-- would be exactly the silent failure this design prevents.
+--
+-- Cascades are in the schema, not in application code: deleting an alert or a
+-- subscription removes its ledger rows. event_id deliberately has no foreign key:
+-- the ledger is the record of what a device owes, and it must outlive nothing but
+-- its alert/subscription (both of which cascade).
+--
+-- Idempotent (same guard pattern as 002/003/004): a re-apply of the whole
+-- sequence is a clean no-op.
+
+DO $$
+BEGIN
+ IF NOT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'alarm') THEN
+ RETURN;
+ END IF;
+
+ CREATE TABLE IF NOT EXISTS alarm.alarm_deliveries (
+ alert_id UUID NOT NULL REFERENCES alarm.alerts(id) ON DELETE CASCADE,
+ subscription_id UUID NOT NULL REFERENCES alarm.subscriptions(id) ON DELETE CASCADE,
+ event_id UUID NOT NULL,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ acked_at TIMESTAMPTZ,
+ PRIMARY KEY (alert_id, subscription_id, event_id)
+ );
+
+ -- The check-in reads by alert + subscription; the pending index keeps
+ -- "who still owes this alert" a small partial scan instead of a table scan.
+ CREATE INDEX IF NOT EXISTS alarm_deliveries_alert_idx
+ ON alarm.alarm_deliveries (alert_id);
+ CREATE INDEX IF NOT EXISTS alarm_deliveries_pending_idx
+ ON alarm.alarm_deliveries (alert_id, subscription_id)
+ WHERE acked_at IS NULL;
+
+ IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'alarm_user') THEN
+ GRANT SELECT, INSERT, UPDATE, DELETE ON alarm.alarm_deliveries TO alarm_user;
+ END IF;
+END
+$$;
diff --git a/src/actions/ack.go b/src/actions/ack.go
new file mode 100644
index 0000000..1429c97
--- /dev/null
+++ b/src/actions/ack.go
@@ -0,0 +1,136 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package actions
+
+import (
+ "net/http"
+ "strings"
+
+ "git.dcntrlized.org/Dcntrlized/Alarm/db"
+ "git.dcntrlized.org/Dcntrlized/Alarm/internal/deviceid"
+ "github.com/google/uuid"
+)
+
+// ackAlarmRequest is the HTTP dismissal of a ringing alarm. The device rides the
+// payload (or the query string) exactly like joinAlert's, so the caller is
+// addressed by (alert, user, device) and can only clear its own debt.
+type ackAlarmRequest struct {
+ AlertID string `json:"alertId"`
+ EventID string `json:"eventId"`
+ DeviceID string `json:"deviceId"`
+}
+
+type ackAlarmResponse struct {
+ AlertID string `json:"alertId"`
+ EventID string `json:"eventId"`
+ Acked int64 `json:"acked"`
+ Destroyed bool `json:"destroyed"`
+}
+
+// AckAlarm implements POST /action/ackAlarm.
+//
+// User ruling: "On client dismiss of a triggered alarm, a request is sent back to
+// the server ACKing the alarm, that's what triggers the clear." Clearing is
+// therefore ack-driven and never response-driven: the alarmStatus check-in is
+// read-only, and this action (or the WebSocket `ack` frame) is what retires the
+// delivery-ledger row.
+//
+// Idempotent: the second ack for the same event clears nothing, returns
+// acked:0, and is not an error. When the ack clears the alert's last unacked
+// row, the alert is destroyed (reading A) and `destroyed:true` reports it — the
+// alert no longer exists, which is the intended outcome, not a failure.
+func AckAlarm(w http.ResponseWriter, r *http.Request, payload any) {
+ const action = "ackAlarm"
+ p, ok := principal(w, r, action)
+ if !ok {
+ return
+ }
+ alertID, err := alertIDFrom(r, payload)
+ if err != nil {
+ validationError(w, "%v", err)
+ return
+ }
+ eventRaw := fieldOrQuery(r, payload, "eventId")
+ if eventRaw == "" {
+ validationError(w, "eventId is required")
+ return
+ }
+ eventID, err := uuid.Parse(eventRaw)
+ if err != nil {
+ validationError(w, "eventId is not a valid uuid")
+ return
+ }
+
+ alert, err := db.GetAlert(r.Context(), alertID)
+ if err != nil {
+ dbError(w, action, err)
+ return
+ }
+ allowed, err := isAdminOrSubscriber(r, alert, p.ID)
+ if err != nil {
+ dbError(w, action, err)
+ return
+ }
+ if !allowed {
+ writeError(w, http.StatusNotFound, CodeAlertNotSubscribed, "not subscribed to this alert")
+ return
+ }
+
+ resp := ackAlarmResponse{AlertID: alertID.String(), EventID: eventID.String()}
+
+ // The owner has no delivery row (only subscribers are owed alarms), so an
+ // owner ack is a well-formed no-op rather than a validation error.
+ if alert.AdminUserID != p.ID {
+ raw, err := validateDeviceID(fieldOrQuery(r, payload, "deviceId"))
+ if err != nil {
+ validationError(w, "%v", err)
+ return
+ }
+ cleared, err := db.AckAlarmDelivery(r.Context(), alertID, p.ID, deviceid.Hash(raw), eventID)
+ if err != nil {
+ dbError(w, action, err)
+ return
+ }
+ resp.Acked = cleared
+ if cleared > 0 {
+ if err := db.RecordDelivered(r.Context(), alertID, p.ID); err != nil {
+ dbError(w, action, err)
+ return
+ }
+ destroyed, err := db.CompleteAlertDeliveries(r.Context(), alertID)
+ if err != nil {
+ dbError(w, action, err)
+ return
+ }
+ resp.Destroyed = destroyed
+ }
+ }
+
+ Debug(action, "POST", "alert=%s event=%s acked=%d destroyed=%t", alertID, eventID, resp.Acked, resp.Destroyed)
+ writeJSON(w, http.StatusOK, resp)
+}
+
+// fieldOrQuery reads a string from the payload and falls back to the query
+// string, matching how the other actions accept both transports.
+func fieldOrQuery(r *http.Request, payload any, field string) string {
+ if v := strings.TrimSpace(r.URL.Query().Get(field)); v != "" {
+ return v
+ }
+ if payload == nil {
+ return ""
+ }
+ return strings.TrimSpace(lenientString(payload, field))
+}
diff --git a/src/actions/deliveries.go b/src/actions/deliveries.go
new file mode 100644
index 0000000..51e6a00
--- /dev/null
+++ b/src/actions/deliveries.go
@@ -0,0 +1,50 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package actions
+
+import (
+ "context"
+ "errors"
+
+ "github.com/google/uuid"
+)
+
+// completeAlertDelivery is the decision point for the second half of the
+// delivery design: "once every client is removed for an alarm, the alarm is then
+// deleted with whichever cascade deletions are relevant".
+//
+// It is DELIBERATELY UNWIRED — nothing calls it. That sentence has two readings
+// and they are different products, so the user has been asked which one is meant:
+//
+// (a) delete the ALERT. The definition, both tokens, the subscriptions and the
+// event history go with it (ON DELETE CASCADE). An alert becomes a
+// one-shot event: once every device has acknowledged, it cannot be
+// triggered again and the admin token becomes dead.
+// (b) retire only the DELIVERY record (and/or the event). The alert survives,
+// stays armed, and can be triggered again later; only the ledger bookkeeping
+// about this alarm is completed.
+//
+// Implementing either now would delete or preserve data on a reading nobody
+// confirmed, so this function reports that the decision is pending. When the
+// answer arrives it gets implemented in exactly one place and called from the
+// paths that observe the last outstanding delivery (the alarmStatus check-in and
+// the WebSocket ack), where "every client has been told" is decided atomically
+// from the ledger rather than from a client's claim.
+func completeAlertDelivery(ctx context.Context, alertID uuid.UUID) error {
+ _ = ctx
+ _ = alertID
+ return errors.New("completeAlertDelivery: pending user decision between (a) deleting the alert as a one-shot and (b) retiring only the delivery record")
+}
diff --git a/src/actions/deliveries_test.go b/src/actions/deliveries_test.go
new file mode 100644
index 0000000..9b6c6bf
--- /dev/null
+++ b/src/actions/deliveries_test.go
@@ -0,0 +1,95 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package actions
+
+import (
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+ "time"
+
+ "git.dcntrlized.org/Dcntrlized/Alarm/db"
+ "git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
+ "github.com/google/uuid"
+)
+
+// TestWriteOfflineAlarmShape pins the 409 check-in contract: the status, the
+// code, and that the body carries a real alarm frame with the SAME event id the
+// device missed. This is the one response an integrator must not treat as an
+// error, so its shape is not allowed to drift silently.
+func TestWriteOfflineAlarmShape(t *testing.T) {
+ alertID, eventID := uuid.New(), uuid.New()
+ override := "Leave the building now"
+ alert := &db.Alert{ID: alertID, Message: &override}
+ at := time.Date(2026, 10, 18, 17, 5, 0, 123000000, time.UTC)
+ pending := &db.PendingDelivery{EventID: eventID, Action: "trigger", Message: &override, ServerTime: at}
+
+ rec := httptest.NewRecorder()
+ writeOfflineAlarm(rec, alert, pending)
+
+ if rec.Code != http.StatusConflict {
+ t.Fatalf("status = %d, want 409", rec.Code)
+ }
+ if ct := rec.Header().Get("Content-Type"); ct != "application/json" {
+ t.Fatalf("Content-Type = %q, want application/json", ct)
+ }
+
+ var body struct {
+ Error struct {
+ Code string `json:"code"`
+ Message string `json:"message"`
+ } `json:"error"`
+ Alarm struct {
+ Type string `json:"type"`
+ AlertID string `json:"alertId"`
+ EventID string `json:"eventId"`
+ Action string `json:"action"`
+ Message string `json:"message"`
+ ServerTime string `json:"serverTime"`
+ } `json:"alarm"`
+ }
+ if err := json.NewDecoder(rec.Body).Decode(&body); err != nil {
+ t.Fatalf("decode body: %v", err)
+ }
+ if body.Error.Code != apierr.AlarmTriggeredOffline {
+ t.Fatalf("error.code = %q, want %q", body.Error.Code, apierr.AlarmTriggeredOffline)
+ }
+ if body.Error.Message == "" {
+ t.Fatal("error.message must not be empty")
+ }
+ if body.Alarm.Type != "alarm" || body.Alarm.Action != "trigger" {
+ t.Fatalf("alarm frame = %+v, want type=alarm action=trigger", body.Alarm)
+ }
+ if body.Alarm.EventID != eventID.String() || body.Alarm.AlertID != alertID.String() {
+ t.Fatalf("alarm ids = %q/%q, want %q/%q", body.Alarm.AlertID, body.Alarm.EventID, alertID, eventID)
+ }
+ if body.Alarm.Message != override {
+ t.Fatalf("alarm.message = %q, want %q", body.Alarm.Message, override)
+ }
+ if body.Alarm.ServerTime != "2026-10-18T17:05:00.123Z" {
+ t.Fatalf("alarm.serverTime = %q", body.Alarm.ServerTime)
+ }
+}
+
+// TestCompleteAlertDeliveryIsUnwired guards the deliberate hole: the function
+// must keep saying "decision pending" rather than quietly deleting or preserving
+// data on one reading of the user's sentence.
+func TestCompleteAlertDeliveryIsUnwired(t *testing.T) {
+ if err := completeAlertDelivery(nil, uuid.New()); err == nil {
+ t.Fatal("completeAlertDelivery must report the pending decision, not succeed")
+ }
+}
diff --git a/src/db/deliveries.go b/src/db/deliveries.go
new file mode 100644
index 0000000..56663f4
--- /dev/null
+++ b/src/db/deliveries.go
@@ -0,0 +1,168 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package db
+
+import (
+ "context"
+ "errors"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/jackc/pgx/v5"
+)
+
+// PendingDelivery is an unacked delivery together with the event it points at.
+// It is the payload of the "you missed an alarm" check-in.
+type PendingDelivery struct {
+ EventID uuid.UUID
+ Action string
+ Message *string
+ ServerTime time.Time
+}
+
+// PendingTrigger is the READ-ONLY offline check-in: it reports the newest
+// trigger this device still owes, or ErrNotFound when it owes nothing. It marks
+// NO row — clearing is ack-driven (see AckAlarmDelivery / AckDelivery), because a
+// check-in answer can be lost in transit and must not forfeit the alarm.
+//
+// deviceID selects the subscription the same way the socket does: the exact
+// stored digest when given, otherwise the caller's oldest subscription on the
+// alert. deviceID is expected to be ALREADY HASHED (actions hash the raw query
+// value before calling); passing a raw id would simply match nothing.
+func PendingTrigger(ctx context.Context, alertID, userID uuid.UUID, deviceID string) (*PendingDelivery, error) {
+ const q = `
+ WITH target AS (
+ SELECT s.id
+ FROM alarm.subscriptions s
+ WHERE s.alert_id = $1
+ AND s.subscriber_user_id = $2
+ AND ($3 = '' OR s.device_id = $3)
+ ORDER BY s.created, s.id
+ LIMIT 1
+ )
+ SELECT e.id, e.kind::text, e.message, e.created_at
+ FROM alarm.alarm_deliveries d
+ JOIN alarm.alarm_events e ON e.id = d.event_id
+ WHERE d.alert_id = $1
+ AND d.subscription_id = (SELECT id FROM target)
+ AND d.acked_at IS NULL
+ ORDER BY e.created_at DESC, e.id DESC
+ LIMIT 1`
+
+ var out PendingDelivery
+ err := Pool().QueryRow(ctx, q, alertID, userID, deviceID).Scan(
+ &out.EventID, &out.Action, &out.Message, &out.ServerTime)
+ if errors.Is(err, pgx.ErrNoRows) {
+ return nil, ErrNotFound
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &out, nil
+}
+
+// PendingReplay returns the trigger frames a subscription still owes, oldest
+// first — the WebSocket reconnect path. It marks nothing: a replay is not a
+// delivery, and the debt only clears when the client acks (or checks in), so a
+// frame the client never processed stays owed.
+//
+// A trigger that has since been disarmed is not replayed: ringing a stale alarm
+// on reconnect would be worse than staying silent, and the check-in clears that
+// debt instead (see ClaimPendingDelivery). The NOT EXISTS keeps this one indexed
+// anti-join rather than a scan of the event history.
+//
+// The LIMIT bounds the work a pathological ledger can put on a handshake; more
+// than a handful of owed events for one device means something is very wrong.
+func PendingReplay(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]PendingDelivery, error) {
+ const q = `
+ SELECT e.id, e.kind::text, e.message, e.created_at
+ FROM alarm.alarm_deliveries d
+ JOIN alarm.alarm_events e ON e.id = d.event_id
+ WHERE d.alert_id = $1
+ AND d.subscription_id = $2
+ AND d.acked_at IS NULL
+ AND NOT EXISTS (
+ SELECT 1 FROM alarm.alarm_events x
+ WHERE x.alert_id = d.alert_id
+ AND x.kind = 'disarm'
+ AND x.created_at > e.created_at
+ )
+ ORDER BY e.created_at, e.id
+ LIMIT 20`
+
+ rows, err := Pool().Query(ctx, q, alertID, subscriptionID)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ out := make([]PendingDelivery, 0, 2)
+ for rows.Next() {
+ var d PendingDelivery
+ if err := rows.Scan(&d.EventID, &d.Action, &d.Message, &d.ServerTime); err != nil {
+ return nil, err
+ }
+ out = append(out, d)
+ }
+ return out, rows.Err()
+}
+
+// AckDelivery marks one delivery acknowledged (acked_at = now()) and reports how
+// many rows it cleared. It is the WebSocket parity path: a client that received
+// and acked the live frame never also gets the offline check-in. Idempotent by
+// construction — a repeated ack (or an ack for an event the device never owed)
+// updates nothing and is not an error; the zero count is what keeps analytics
+// honest and is why callers only fire the delivered row when n > 0.
+func AckDelivery(ctx context.Context, alertID, subscriptionID, eventID uuid.UUID) (int64, error) {
+ tag, err := Pool().Exec(ctx,
+ `UPDATE alarm.alarm_deliveries
+ SET acked_at = now()
+ WHERE alert_id = $1 AND subscription_id = $2 AND event_id = $3
+ AND acked_at IS NULL`,
+ alertID, subscriptionID, eventID)
+ if err != nil {
+ return 0, err
+ }
+ return tag.RowsAffected(), nil
+}
+
+// AckAlarmDelivery clears the caller's subscription rows for one event and
+// reports how many were cleared — the explicit `ackAlarm` action, sent when the
+// person holding the phone dismisses a ringing alarm.
+//
+// deviceID is the ALREADY HASHED digest of the raw client value; the actor is
+// addressed through (alert, user, device) exactly like the socket, so a caller can
+// only ever clear its own device's debt. Idempotent: a repeat ack matches no row
+// and returns 0, never an error.
+func AckAlarmDelivery(ctx context.Context, alertID, userID uuid.UUID, deviceID string, eventID uuid.UUID) (int64, error) {
+ const q = `
+ UPDATE alarm.alarm_deliveries d
+ SET acked_at = now()
+ WHERE d.alert_id = $1
+ AND d.event_id = $2
+ AND d.acked_at IS NULL
+ AND d.subscription_id IN (
+ SELECT s.id FROM alarm.subscriptions s
+ WHERE s.alert_id = $1
+ AND s.subscriber_user_id = $3
+ AND s.device_id = $4
+ )`
+ tag, err := Pool().Exec(ctx, q, alertID, eventID, userID, deviceID)
+ if err != nil {
+ return 0, err
+ }
+ return tag.RowsAffected(), nil
+}
diff --git a/src/db/deliveries_db_test.go b/src/db/deliveries_db_test.go
new file mode 100644
index 0000000..1c77734
--- /dev/null
+++ b/src/db/deliveries_db_test.go
@@ -0,0 +1,287 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package db
+
+// The lifecycle probe. It is skipped unless DATABASE_URL points at a database
+// with migrations 001–008 applied; the default suite stays DB-free. It covers the
+// four ways an alert can end plus the ledger and log-table mechanics that only
+// SQL can prove:
+//
+// - drain-to-empty (every subscriber acked) destroys the alert and writes
+// `destroyed/delivered` in one transaction;
+// - a trigger with zero subscribers does NOT destroy (guard 1);
+// - expiry destroys with unacked rows and the alert reads as not-found;
+// - disarm of a triggered alert destroys with reason `disarmed`;
+// - acks are idempotent; the second ack clears nothing;
+// - event-log retention pruning works.
+//
+// Run it with:
+//
+// DATABASE_URL='postgres://postgres:dcntrlized@127.0.0.1:55432/dcntrlized?sslmode=disable' \
+// go test -run TestLedgerProbe -count=1 -v ./db/
+//
+// It uses its own freshly-created alerts and deletes them at the end.
+
+import (
+ "context"
+ "errors"
+ "os"
+ "strings"
+ "testing"
+ "time"
+
+ "git.dcntrlized.org/Dcntrlized/Alarm/internal/alertstate"
+ "github.com/google/uuid"
+)
+
+func TestLedgerProbe(t *testing.T) {
+ dsn := os.Getenv("DATABASE_URL")
+ if dsn == "" {
+ t.Skip("DATABASE_URL not set; lifecycle probe needs a migrated database")
+ }
+ ctx := context.Background()
+ _ = dsn
+ NewDB()
+ t.Cleanup(Close)
+
+ admin := uuid.New()
+ sub1 := uuid.New()
+ sub2 := uuid.New()
+ sub3 := uuid.New()
+
+ newAlert := func(name string, ttl time.Duration) *Alert {
+ msg := "probe"
+ a, err := CreateAlert(ctx, &Alert{
+ AdminUserID: admin,
+ Name: name,
+ Message: &msg,
+ AdminTokenHash: []byte("probe"),
+ SubscribeToken: "probe-" + uuid.NewString(),
+ ExpiresAt: time.Now().UTC().Add(ttl),
+ })
+ if err != nil {
+ t.Fatalf("CreateAlert: %v", err)
+ }
+ return a
+ }
+ join := func(alertID, user uuid.UUID) *Subscription {
+ s, err := CreateSubscription(ctx, &Subscription{
+ AlertID: alertID, SubscriberUserID: user, DeviceID: "sha256:" + strings.Repeat("a", 64),
+ })
+ if err != nil {
+ t.Fatalf("CreateSubscription: %v", err)
+ }
+ return s
+ }
+ countAnalytics := func(alertID uuid.UUID, event, reason string) int {
+ var n int
+ err := Pool().QueryRow(ctx,
+ `SELECT count(*) FROM alarm.event_log WHERE alert_hash = $1 AND event = $2 AND coalesce(reason,'') = $3`,
+ HashUUID(alertID), event, reason).Scan(&n)
+ if err != nil {
+ t.Fatalf("analytics count: %v", err)
+ }
+ return n
+ }
+
+ // ---- drain-to-empty destroys the alert with reason "delivered" ----
+ drain := newAlert("probe drain", time.Hour)
+ join(drain.ID, sub1)
+ join(drain.ID, sub2)
+ join(drain.ID, sub3) // third enabled subscriber: acking a different device's row must not count
+ disabled, err := CreateSubscription(ctx, &Subscription{
+ AlertID: drain.ID, SubscriberUserID: uuid.New(), DeviceID: "sha256:" + strings.Repeat("b", 64),
+ })
+ if err != nil {
+ t.Fatalf("disabled subscription: %v", err)
+ }
+ if _, err := Pool().Exec(ctx, `UPDATE alarm.subscriptions SET enabled = false WHERE id = $1`, disabled.ID); err != nil {
+ t.Fatalf("disable sub: %v", err)
+ }
+ _, ev, err := TriggerAlert(ctx, drain.ID, admin, nil)
+ if err != nil {
+ t.Fatalf("TriggerAlert: %v", err)
+ }
+ var rows, enabled int
+ if err := Pool().QueryRow(ctx,
+ `SELECT count(*) FROM alarm.alarm_deliveries WHERE alert_id = $1`, drain.ID).Scan(&rows); err != nil {
+ t.Fatalf("ledger rows: %v", err)
+ }
+ if err := Pool().QueryRow(ctx,
+ `SELECT count(*) FROM alarm.subscriptions WHERE alert_id = $1 AND enabled`, drain.ID).Scan(&enabled); err != nil {
+ t.Fatalf("enabled count: %v", err)
+ }
+ if rows != 3 || enabled != 3 {
+ t.Fatalf("trigger produced %d ledger rows for %d enabled subscriptions, want 3/3", rows, enabled)
+ }
+ if got := countAnalytics(drain.ID, EventTriggered, ""); got != 1 {
+ t.Fatalf("triggered analytics rows = %d, want 1", got)
+ }
+
+ pending, err := PendingTrigger(ctx, drain.ID, sub1, "")
+ if err != nil || pending.EventID != ev.ID {
+ t.Fatalf("PendingTrigger = %+v, %v; want event %s", pending, err, ev.ID)
+ }
+
+ // Ack all three, idempotently. Only the last one completes the drain.
+ for i, user := range []uuid.UUID{sub1, sub2, sub3} {
+ sub, err := deviceDigest(ctx, drain.ID, user)
+ if err != nil {
+ t.Fatalf("digest %d: %v", i, err)
+ }
+ n, err := AckAlarmDelivery(ctx, drain.ID, user, sub, ev.ID)
+ if err != nil {
+ t.Fatalf("AckAlarmDelivery %d: %v", i, err)
+ }
+ if n != 1 {
+ t.Fatalf("ack %d cleared %d rows, want 1", i, n)
+ }
+ if again, err := AckAlarmDelivery(ctx, drain.ID, user, sub, ev.ID); err != nil || again != 0 {
+ t.Fatalf("repeat ack %d = %d, %v; want 0, nil", i, again, err)
+ }
+ }
+
+ destroyed, err := CompleteAlertDeliveries(ctx, drain.ID)
+ if err != nil {
+ t.Fatalf("CompleteAlertDeliveries: %v", err)
+ }
+ if !destroyed {
+ t.Fatal("drained alert was not destroyed")
+ }
+ if _, err := GetAlert(ctx, drain.ID); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("drained alert read = %v, want ErrNotFound", err)
+ }
+ if got := countAnalytics(drain.ID, EventDestroyed, DestroyReasonDelivered); got != 1 {
+ t.Fatalf("destroyed/delivered analytics rows = %d, want 1", got)
+ }
+
+ // ---- guard 1: a zero-subscriber trigger must NOT destroy ----
+ empty := newAlert("probe empty", time.Hour)
+ if _, _, err := TriggerAlert(ctx, empty.ID, admin, nil); err != nil {
+ t.Fatalf("TriggerAlert(empty): %v", err)
+ }
+ ok, err := CompleteAlertDeliveries(ctx, empty.ID)
+ if err != nil {
+ t.Fatalf("CompleteAlertDeliveries(empty): %v", err)
+ }
+ if ok {
+ t.Fatal("zero-subscriber trigger destroyed the owner's alert")
+ }
+ if _, err := GetAlert(ctx, empty.ID); err != nil {
+ t.Fatalf("zero-subscriber alert should still exist: %v", err)
+ }
+
+ // ---- expiry destroys with unacked rows, and reads 404 before the sweeper ----
+ expiring := newAlert("probe expire", time.Hour)
+ join(expiring.ID, sub1)
+ if _, _, err := TriggerAlert(ctx, expiring.ID, admin, nil); err != nil {
+ t.Fatalf("TriggerAlert(expiring): %v", err)
+ }
+ if _, err := Pool().Exec(ctx, `UPDATE alarm.alerts SET expires_at = now() - interval '1 second' WHERE id = $1`, expiring.ID); err != nil {
+ t.Fatalf("force expiry: %v", err)
+ }
+ if _, err := GetAlert(ctx, expiring.ID); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("expired alert read = %v, want ErrNotFound", err)
+ }
+ n, err := ExpireAlerts(ctx, 100)
+ if err != nil || n < 1 {
+ t.Fatalf("ExpireAlerts = %d, %v; want >= 1", n, err)
+ }
+ if got := countAnalytics(expiring.ID, EventDestroyed, DestroyReasonExpired); got != 1 {
+ t.Fatalf("destroyed/expired analytics rows = %d, want 1", got)
+ }
+ var leftovers int
+ if err := Pool().QueryRow(ctx,
+ `SELECT count(*) FROM alarm.alarm_deliveries WHERE alert_id = $1`, expiring.ID).Scan(&leftovers); err != nil {
+ t.Fatalf("leftover rows: %v", err)
+ }
+ if leftovers != 0 {
+ t.Fatalf("expiry left %d delivery rows, want 0 (cascade)", leftovers)
+ }
+
+ // ---- disarm of a triggered alert ends it ----
+ disarmed := newAlert("probe disarm", time.Hour)
+ join(disarmed.ID, sub1)
+ if _, _, err := TriggerAlert(ctx, disarmed.ID, admin, nil); err != nil {
+ t.Fatalf("TriggerAlert(disarm): %v", err)
+ }
+ if _, _, err := DisarmAlert(ctx, disarmed.ID, admin, nil); err != nil {
+ t.Fatalf("DisarmAlert: %v", err)
+ }
+ if _, err := GetAlert(ctx, disarmed.ID); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("disarmed alert read = %v, want ErrNotFound", err)
+ }
+ if got := countAnalytics(disarmed.ID, EventDestroyed, DestroyReasonDisarmed); got != 1 {
+ t.Fatalf("destroyed/disarmed analytics rows = %d, want 1", got)
+ }
+
+ // ---- deletion wins over debt: cascade removes the ledger, no replay/409 ----
+ deleted := newAlert("probe delete", time.Hour)
+ join(deleted.ID, sub1)
+ if _, _, err := TriggerAlert(ctx, deleted.ID, admin, nil); err != nil {
+ t.Fatalf("TriggerAlert(deleted): %v", err)
+ }
+ if err := DeleteAlert(ctx, deleted.ID); err != nil {
+ t.Fatalf("DeleteAlert: %v", err)
+ }
+ if _, err := PendingTrigger(ctx, deleted.ID, sub1, ""); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("deleted alert still owes a delivery: %v", err)
+ }
+ if _, err := GetAlert(ctx, deleted.ID); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("deleted alert read = %v, want ErrNotFound", err)
+ }
+ if got := countAnalytics(deleted.ID, EventDestroyed, DestroyReasonDeleted); got != 1 {
+ t.Fatalf("destroyed/deleted event rows = %d, want 1", got)
+ }
+
+ // ---- disarm of an ARMED alert also ends it (delete-without-trigger) ----
+ armed := newAlert("probe armed", time.Hour)
+ if _, _, err := DisarmAlert(ctx, armed.ID, admin, nil); err != nil {
+ t.Fatalf("DisarmAlert(armed): %v", err)
+ }
+ if _, err := GetAlert(ctx, armed.ID); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("armed disarm left the alert alive: %v", err)
+ }
+ if got := countAnalytics(armed.ID, EventDestroyed, DestroyReasonDisarmed); got != 1 {
+ t.Fatalf("armed disarm event rows = %d, want 1", got)
+ }
+
+ // ---- event-log retention: age-based delete ----
+ if _, err := Pool().Exec(ctx,
+ `INSERT INTO alarm.event_log (occurred_at, event, alert_hash) VALUES (now() - interval '30 days', 'created', $1)`,
+ HashUUID(uuid.New())); err != nil {
+ t.Fatalf("insert old event: %v", err)
+ }
+ if n, err := PruneEventLog(ctx, EventLogRetention); err != nil || n < 1 {
+ t.Fatalf("PruneEventLog = %d, %v; want >= 1", n, err)
+ }
+
+ // Clean any leftover probe alerts (drain/empty) so repeat runs are clean.
+ _, _ = Pool().Exec(ctx, `DELETE FROM alarm.alerts WHERE name LIKE 'probe %'`)
+
+ t.Log("lifecycle probe ok: 3/3 ledger rows, idempotent acks, drain/zero-sub/expiry/disarm/deletion endings, event-log retention")
+}
+
+// deviceDigest reads back the stored (already hashed) device id of a subscription.
+func deviceDigest(ctx context.Context, alertID, userID uuid.UUID) (string, error) {
+ var d string
+ err := Pool().QueryRow(ctx,
+ `SELECT device_id FROM alarm.subscriptions WHERE alert_id = $1 AND subscriber_user_id = $2
+ ORDER BY created, id LIMIT 1`, alertID, userID).Scan(&d)
+ return d, err
+}
+
+var _ = alertstate.KindTrigger
diff --git a/src/handlers/actions/action__actionName_.go b/src/handlers/actions/action__actionName_.go
index a058b1a..4b1f57a 100644
--- a/src/handlers/actions/action__actionName_.go
+++ b/src/handlers/actions/action__actionName_.go
@@ -1,3 +1,18 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
package actions
import (
@@ -5,9 +20,35 @@ import (
"net/http"
"git.dcntrlized.org/Dcntrlized/Alarm/actions"
+ "git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
"git.dcntrlized.org/Dcntrlized/Alarm/types"
)
+// This layer routes /action/{actionName} by HTTP method. It sits in front of the
+// action handlers, so its errors are plain text (apierr.WriteText) rather than
+// the JSON envelope the handlers emit; API.md documents that boundary. Both
+// emitters guarantee a non-empty message.
+
+const notImplementedMessage = "Action not implemented"
+
+// notImplemented answers an action/verb pair the dispatcher does not route.
+func notImplemented(w http.ResponseWriter) {
+ apierr.WriteText(w, http.StatusNotImplemented, notImplementedMessage)
+}
+
+// decodeEnvelope reads the {payload} wrapper. Unknown top-level fields are
+// rejected (the inner payload is decoded leniently by the action itself).
+func decodeEnvelope(w http.ResponseWriter, r *http.Request) (types.ActionRequest, bool) {
+ var req types.ActionRequest
+ dec := json.NewDecoder(r.Body)
+ dec.DisallowUnknownFields()
+ if err := dec.Decode(&req); err != nil {
+ apierr.WriteText(w, http.StatusBadRequest, err.Error())
+ return req, false
+ }
+ return req, true
+}
+
// ActionActionNameGet dispatches the GET /action/{actionName} request.
//
// GET carries no body: the read actions take their arguments from the query
@@ -30,7 +71,7 @@ func ActionActionNameGet(w http.ResponseWriter, r *http.Request, params map[stri
case "me":
actions.Me(w, r)
default:
- http.Error(w, "Action not implemented", http.StatusNotImplemented)
+ notImplemented(w)
return
}
}
@@ -38,11 +79,8 @@ func ActionActionNameGet(w http.ResponseWriter, r *http.Request, params map[stri
// ActionActionNamePost dispatches the POST /action/{actionName} request.
func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[string]string) {
action := params["actionName"]
- var req types.ActionRequest
- dec := json.NewDecoder(r.Body)
- dec.DisallowUnknownFields()
- if err := dec.Decode(&req); err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
+ req, ok := decodeEnvelope(w, r)
+ if !ok {
return
}
@@ -64,12 +102,12 @@ func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[str
actions.DisarmAlert(w, r, req.Payload)
case "alarmStatus":
actions.AlarmStatus(w, r, req.Payload)
+ case "ackAlarm":
+ actions.AckAlarm(w, r, req.Payload)
case "deleteAlert":
actions.DeleteAlert(w, r, req.Payload)
case "qrImage":
actions.QRImage(w, r, req.Payload)
- case "rotateSubscribeToken":
- actions.RotateSubscribeToken(w, r, req.Payload)
// Shared-auth proxies.
case "login":
@@ -83,7 +121,7 @@ func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[str
case "logout":
actions.Logout(w, r, req.Payload)
default:
- http.Error(w, "Action not implemented", http.StatusNotImplemented)
+ notImplemented(w)
return
}
}
@@ -91,17 +129,13 @@ func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[str
// ActionActionNamePut dispatches the PUT /action/{actionName} request.
func ActionActionNamePut(w http.ResponseWriter, r *http.Request, params map[string]string) {
action := params["actionName"]
- var req types.ActionRequest
- dec := json.NewDecoder(r.Body)
- dec.DisallowUnknownFields()
- if err := dec.Decode(&req); err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
+ if _, ok := decodeEnvelope(w, r); !ok {
return
}
switch action {
default:
- http.Error(w, "Action not implemented", http.StatusNotImplemented)
+ notImplemented(w)
return
}
}
@@ -111,11 +145,8 @@ func ActionActionNamePut(w http.ResponseWriter, r *http.Request, params map[stri
// verb and carries the same {payload} envelope.
func ActionActionNameDelete(w http.ResponseWriter, r *http.Request, params map[string]string) {
action := params["actionName"]
- var req types.ActionRequest
- dec := json.NewDecoder(r.Body)
- dec.DisallowUnknownFields()
- if err := dec.Decode(&req); err != nil {
- http.Error(w, err.Error(), http.StatusBadRequest)
+ req, ok := decodeEnvelope(w, r)
+ if !ok {
return
}
@@ -125,7 +156,7 @@ func ActionActionNameDelete(w http.ResponseWriter, r *http.Request, params map[s
case "removeSubscription":
actions.RemoveSubscription(w, r, req.Payload)
default:
- http.Error(w, "Action not implemented", http.StatusNotImplemented)
+ notImplemented(w)
return
}
}
diff --git a/src/realtime/client.go b/src/realtime/client.go
index 16c0b8d..362b8d6 100644
--- a/src/realtime/client.go
+++ b/src/realtime/client.go
@@ -16,10 +16,13 @@
package realtime
import (
+ "context"
"encoding/json"
"sync"
"time"
+ "git.dcntrlized.org/Dcntrlized/Alarm/db"
+ "github.com/google/uuid"
"github.com/gorilla/websocket"
)
@@ -107,6 +110,11 @@ func (c *client) readPump() {
c.acks++
c.ackMu.Unlock()
c.hub.recordAck(f.EventID)
+ // Parity with the offline check-in: the live frame was received and
+ // acknowledged, so this device no longer owes it. Without this an
+ // online client would also be told 409 ERR_ALARM_TRIGGERED_OFFLINE
+ // later — i.e. told twice about one alarm.
+ c.ackDelivery(f.EventID)
case "ping":
c.touch()
// Answer at the protocol level: the frozen contract defines no
@@ -116,6 +124,39 @@ func (c *client) readPump() {
}
}
+// ackDelivery marks the ledger row for a client ack. It is best-effort and
+// idempotent: a malformed event id, an admin connection (no subscription) or a
+// repeated ack are all silent no-ops. A database failure is logged, never fatal —
+// the client already has the alarm, and the worst case is one duplicate offline
+// signal, not a dropped ring.
+func (c *client) ackDelivery(eventID string) {
+ if c.identity.SubscriptionID == nil {
+ return
+ }
+ id, err := uuid.Parse(eventID)
+ if err != nil {
+ return
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ cleared, err := db.AckDelivery(ctx, c.identity.AlertID, *c.identity.SubscriptionID, id)
+ if err != nil {
+ c.hub.opts.Logf("ws: ledger ack failed alert=%s event=%s: %v", c.identity.AlertID, id, err)
+ return
+ }
+ if cleared == 0 {
+ // A repeat ack (or one for an event this device never owed) is a no-op:
+ // nothing to record and no drain to check.
+ return
+ }
+ if err := db.RecordDelivered(ctx, c.identity.AlertID, c.identity.UserID); err != nil {
+ c.hub.opts.Logf("ws: delivered analytics alert=%s: %v", c.identity.AlertID, err)
+ }
+ if _, err := db.CompleteAlertDeliveries(ctx, c.identity.AlertID); err != nil {
+ c.hub.opts.Logf("ws: drain check alert=%s: %v", c.identity.AlertID, err)
+ }
+}
+
// writePump owns all writes except the WriteControl calls in readPump (which
// gorilla explicitly allows concurrently).
func (c *client) writePump() {
diff --git a/src/realtime/hub.go b/src/realtime/hub.go
index ea59830..d77941d 100644
--- a/src/realtime/hub.go
+++ b/src/realtime/hub.go
@@ -43,7 +43,9 @@ import (
"sync"
"time"
+ "git.dcntrlized.org/Dcntrlized/Alarm/db"
"git.dcntrlized.org/Dcntrlized/Alarm/internal/alertstate"
+ "git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
"github.com/google/uuid"
"github.com/gorilla/websocket"
)
@@ -56,19 +58,21 @@ const (
DefaultSendBuffer = 16
DefaultMaxFrameSize = 4096
+ // replayTimeout bounds the ledger read done during a handshake.
+ replayTimeout = 3 * time.Second
+
// ackCap bounds the per-event ack counters kept for observability.
ackCap = 256
)
// AlarmFrame is the server -> client alarm frame.
type AlarmFrame struct {
- Type string `json:"type"` // always "alarm"
- AlertID string `json:"alertId"`
- EventID string `json:"eventId"`
- Action string `json:"action"` // "trigger" | "disarm"
- Message string `json:"message"`
- ServerTime string `json:"serverTime"` // RFC3339, UTC
- AutoStopSeconds int `json:"autoStopSeconds"`
+ Type string `json:"type"` // always "alarm"
+ AlertID string `json:"alertId"`
+ EventID string `json:"eventId"`
+ Action string `json:"action"` // "trigger" | "disarm"
+ Message string `json:"message"`
+ ServerTime string `json:"serverTime"` // RFC3339, UTC
}
// AckFrame is the server -> client acknowledgement of a client ack.
@@ -114,6 +118,10 @@ type Options struct {
Authenticate func(r *http.Request) (*Identity, error)
// Touch refreshes subscriptions.last_seen_at. Optional.
Touch func(ctx context.Context, subscriptionID uuid.UUID) error
+ // PendingReplay returns the trigger frames a subscription still owes (the
+ // durable delivery ledger). Nil uses the real ledger; tests inject a fake.
+ // A replay marks nothing: the debt clears on the client's ack/check-in.
+ PendingReplay func(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]AlarmFrame, error)
PingInterval time.Duration
PongWait time.Duration
@@ -142,22 +150,50 @@ func (o Options) withDefaults() Options {
if o.Logf == nil {
o.Logf = log.Printf
}
+ if o.PendingReplay == nil {
+ o.PendingReplay = ledgerReplay
+ }
return o
}
+// ledgerReplay is the production PendingReplay: the durable delivery ledger,
+// converted into the same frames the hub broadcasts. Keeping the conversion here
+// (rather than in db) avoids a db -> realtime cycle; the ledger stays the source
+// of truth either way.
+func ledgerReplay(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]AlarmFrame, error) {
+ rows, err := db.PendingReplay(ctx, alertID, subscriptionID)
+ if err != nil {
+ return nil, err
+ }
+ frames := make([]AlarmFrame, 0, len(rows))
+ for _, r := range rows {
+ message := ""
+ if r.Message != nil {
+ message = *r.Message
+ }
+ frames = append(frames, AlarmFrame{
+ Type: "alarm",
+ AlertID: alertID.String(),
+ EventID: r.EventID.String(),
+ Action: alertstate.KindTrigger,
+ Message: message,
+ ServerTime: r.ServerTime.UTC().Format(time.RFC3339Nano),
+ })
+ }
+ return frames, nil
+}
+
type lastEvent struct {
frame AlarmFrame
at time.Time
}
-func (e lastEvent) active(now time.Time) bool {
- if e.frame.Action != alertstate.KindTrigger {
- return false
- }
- if e.frame.AutoStopSeconds <= 0 {
- return true
- }
- return now.Before(e.at.Add(time.Duration(e.frame.AutoStopSeconds) * time.Second))
+// active reports whether the cached event is still live. With no auto-stop an
+// alarm is live exactly while it is a trigger that has not yet been superseded
+// by a disarm — and the cache only ever holds the newest event, so checking the
+// action is the whole test.
+func (e lastEvent) active() bool {
+ return e.frame.Action == alertstate.KindTrigger
}
// Hub fans alarm frames out to every WebSocket client of an alert.
@@ -204,7 +240,7 @@ func (h *Hub) Handler() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
w.Header().Set("Allow", http.MethodGet)
- writeJSONError(w, http.StatusMethodNotAllowed, "ERR_METHOD_NOT_ALLOWED", "method not allowed")
+ writeJSONError(w, http.StatusMethodNotAllowed, apierr.MethodNotAllowed, "method not allowed")
return
}
@@ -215,7 +251,7 @@ func (h *Hub) Handler() http.HandlerFunc {
writeJSONError(w, authErr.Status, authErr.Code, authErr.Message)
return
}
- writeJSONError(w, http.StatusUnauthorized, "ERR_UNAUTHORIZED", "unauthorized")
+ writeJSONError(w, http.StatusUnauthorized, apierr.Unauthorized, "unauthorized")
return
}
@@ -223,13 +259,13 @@ func (h *Hub) Handler() http.HandlerFunc {
closed := h.closed
h.mu.Unlock()
if closed {
- writeJSONError(w, http.StatusServiceUnavailable, "ERR_UNAVAILABLE", "server is shutting down")
+ writeJSONError(w, http.StatusServiceUnavailable, apierr.Unavailable, "server is shutting down")
return
}
conn, err := h.upgrader.Upgrade(w, r, nil)
if err != nil {
- // Upgrade already wrote the HTTP error response.
+ // Upgrade writes its own HTTP error with a non-empty status text.
h.opts.Logf("ws: upgrade failed alert=%s: %v", identity.AlertID, err)
return
}
@@ -237,10 +273,26 @@ func (h *Hub) Handler() http.HandlerFunc {
c := newClient(h, conn, *identity)
h.register(c)
+ // Durable replay: frames this subscription still owes from the delivery
+ // ledger. This is what makes a native service holding the socket (with
+ // the JS layer dead) fire an alarm it missed — the FGS has no access
+ // token and therefore cannot call the alarmStatus check-in at all. Sent
+ // after register/upgrade, into the buffered queue, bounded by the ledger
+ // query's LIMIT.
+ replayed := make(map[string]bool, 2)
+ for _, f := range h.replayPending(*identity) {
+ if b, err := json.Marshal(f); err == nil {
+ replayed[f.EventID] = true
+ c.trySend(b)
+ }
+ }
+
// Reconnect catch-up: replay the last event only while it is still
// live, so a client that reconnects mid-alarm fires immediately and a
- // client reconnecting after the alarm is never re-alarmed.
- if f, ok := h.lastActive(identity.AlertID); ok {
+ // client reconnecting after the alarm is never re-alarmed. Skipped when
+ // the ledger replay above already sent that event (a live trigger whose
+ // ledger row is still unacked would otherwise arrive twice).
+ if f, ok := h.lastActive(identity.AlertID); ok && !replayed[f.EventID] {
if b, err := json.Marshal(f); err == nil {
c.trySend(b)
}
@@ -260,6 +312,24 @@ func (h *Hub) Handler() http.HandlerFunc {
}
}
+// replayPending queries the ledger for this exact (alert, subscription) and
+// returns the frames owed. Admin connections have no subscription and owe
+// nothing. A failure is logged and degrades to the last-event catch-up: a broken
+// ledger read must not refuse a socket that can still receive live alarms.
+func (h *Hub) replayPending(identity Identity) []AlarmFrame {
+ if identity.SubscriptionID == nil {
+ return nil
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), replayTimeout)
+ defer cancel()
+ frames, err := h.opts.PendingReplay(ctx, identity.AlertID, *identity.SubscriptionID)
+ if err != nil {
+ h.opts.Logf("ws: pending delivery replay failed alert=%s: %v", identity.AlertID, err)
+ return nil
+ }
+ return frames
+}
+
func (h *Hub) register(c *client) {
h.mu.Lock()
defer h.mu.Unlock()
@@ -344,7 +414,7 @@ func (h *Hub) lastActive(alertID uuid.UUID) (AlarmFrame, bool) {
h.mu.Lock()
defer h.mu.Unlock()
e, ok := h.last[alertID]
- if !ok || !e.active(time.Now().UTC()) {
+ if !ok || !e.active() {
return AlarmFrame{}, false
}
return e.frame, true
@@ -444,10 +514,9 @@ func (h *Hub) touch(c *client) {
}
}
+// writeJSONError emits the handshake rejection envelope. apierr.Write supplies
+// a canonical message when code/message are empty, so no rejection can answer a
+// bare status.
func writeJSONError(w http.ResponseWriter, status int, code, message string) {
- w.Header().Set("Content-Type", "application/json")
- w.WriteHeader(status)
- _ = json.NewEncoder(w).Encode(map[string]any{
- "error": map[string]string{"code": code, "message": message},
- })
+ apierr.Write(w, status, code, message)
}
diff --git a/src/realtime/hub_test.go b/src/realtime/hub_test.go
index 96be95e..ab28a43 100644
--- a/src/realtime/hub_test.go
+++ b/src/realtime/hub_test.go
@@ -21,9 +21,11 @@ import (
"net/http"
"net/http/httptest"
"strings"
+ "sync"
"testing"
"time"
+ "git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
"github.com/google/uuid"
"github.com/gorilla/websocket"
)
@@ -191,13 +193,12 @@ func TestReconnectReplaysLiveTrigger(t *testing.T) {
// Broadcast with nobody connected: the hub keeps it as the last event.
frame := AlarmFrame{
- Type: "alarm",
- AlertID: alertID.String(),
- EventID: uuid.NewString(),
- Action: "trigger",
- Message: "still ringing",
- ServerTime: time.Now().UTC().Format(time.RFC3339Nano),
- AutoStopSeconds: 300,
+ Type: "alarm",
+ AlertID: alertID.String(),
+ EventID: uuid.NewString(),
+ Action: "trigger",
+ Message: "still ringing",
+ ServerTime: time.Now().UTC().Format(time.RFC3339Nano),
}
if delivered := hub.Broadcast(alertID, frame); delivered != 0 {
t.Fatalf("Broadcast with no clients delivered to %d, want 0", delivered)
@@ -226,7 +227,8 @@ func TestHandlerRejectsNonGet(t *testing.T) {
}
var body struct {
Error struct {
- Code string `json:"code"`
+ Code string `json:"code"`
+ Message string `json:"message"`
} `json:"error"`
}
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
@@ -235,4 +237,127 @@ func TestHandlerRejectsNonGet(t *testing.T) {
if body.Error.Code == "" {
t.Fatalf("error body has no code: %+v", body)
}
+ if strings.TrimSpace(body.Error.Message) == "" {
+ t.Fatalf("error body has no message: %+v", body)
+ }
+}
+
+// TestHandshakeRejectionAlwaysHasMessage proves the structural fallback: the
+// Authenticate hook can hand the hub an AuthError with an empty or whitespace
+// message and the handshake rejection still carries a readable one. A bare
+// status on a failed upgrade is exactly the bug this pins.
+func TestHandshakeRejectionAlwaysHasMessage(t *testing.T) {
+ alertID := uuid.New()
+ cases := []*AuthError{
+ {Status: http.StatusUnauthorized, Code: apierr.Unauthorized, Message: ""},
+ {Status: http.StatusForbidden, Code: apierr.AlertTokenInvalid, Message: " "},
+ {Status: http.StatusNotFound, Code: apierr.AlertNotSubscribed, Message: ""},
+ {Status: http.StatusServiceUnavailable, Code: apierr.Unavailable, Message: ""},
+ }
+ for _, authErr := range cases {
+ authErr := authErr
+ hub := NewHub(Options{
+ Authenticate: func(r *http.Request) (*Identity, error) { return nil, authErr },
+ Logf: func(format string, v ...any) { t.Logf(format, v...) },
+ })
+ srv := httptest.NewServer(hub.Handler())
+ resp, err := http.Get(srv.URL + "/ws/alarm?alertId=" + alertID.String())
+ if err != nil {
+ srv.Close()
+ t.Fatalf("GET handshake: %v", err)
+ }
+ var body struct {
+ Error struct {
+ Code string `json:"code"`
+ Message string `json:"message"`
+ } `json:"error"`
+ }
+ decodeErr := json.NewDecoder(resp.Body).Decode(&body)
+ resp.Body.Close()
+ srv.Close()
+
+ if resp.StatusCode != authErr.Status {
+ t.Fatalf("status = %d, want %d", resp.StatusCode, authErr.Status)
+ }
+ if decodeErr != nil {
+ t.Fatalf("rejection body is not JSON: %v", decodeErr)
+ }
+ if body.Error.Code != authErr.Code {
+ t.Fatalf("code = %q, want %q", body.Error.Code, authErr.Code)
+ }
+ if strings.TrimSpace(body.Error.Message) == "" {
+ t.Fatalf("%d %s: rejection carried an empty message", authErr.Status, authErr.Code)
+ }
+ }
+}
+
+// TestConnectReplaysPendingDeliveriesBeforeAnyTrigger is the durable-delivery
+// half of task-4: a subscription that owes a trigger from the ledger receives it
+// on connect, before anyone broadcasts anything. After the client acks (and the
+// ledger row is cleared, simulated by the fake flipping `owed`), a second
+// reconnect replays nothing.
+func TestConnectReplaysPendingDeliveriesBeforeAnyTrigger(t *testing.T) {
+ alertID := uuid.New()
+ subscriptionID := uuid.New()
+ eventID := uuid.New()
+ serverTime := time.Date(2026, 10, 18, 17, 5, 0, 123000000, time.UTC)
+
+ var mu sync.Mutex
+ owed := true
+ hub := NewHub(Options{
+ Authenticate: func(*http.Request) (*Identity, error) {
+ id := Identity{AlertID: alertID, UserID: uuid.New(), SubscriptionID: &subscriptionID}
+ return &id, nil
+ },
+ PendingReplay: func(_ context.Context, gotAlert, gotSub uuid.UUID) ([]AlarmFrame, error) {
+ if gotAlert != alertID || gotSub != subscriptionID {
+ t.Errorf("replay queried (%s, %s), want (%s, %s)", gotAlert, gotSub, alertID, subscriptionID)
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if !owed {
+ return nil, nil
+ }
+ return []AlarmFrame{{
+ Type: "alarm",
+ AlertID: alertID.String(),
+ EventID: eventID.String(),
+ Action: "trigger",
+ Message: "missed while offline",
+ ServerTime: serverTime.Format(time.RFC3339Nano),
+ }}, nil
+ },
+ Logf: func(format string, v ...any) { t.Logf(format, v...) },
+ })
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ go hub.Run(ctx)
+ srv := httptest.NewServer(hub.Handler())
+ t.Cleanup(srv.Close)
+ wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
+
+ first := dial(t, wsURL)
+ got := readAlarm(t, first)
+ if got.EventID != eventID.String() || got.Message != "missed while offline" || got.Action != "trigger" {
+ t.Fatalf("replay frame = %+v, want the owed trigger %s", got, eventID)
+ }
+
+ if err := first.WriteJSON(ClientFrame{Type: "ack", EventID: eventID.String()}); err != nil {
+ t.Fatalf("ack: %v", err)
+ }
+ // Give the ack a moment to be processed by the read pump, then simulate the
+ // ledger row having been cleared by it (AckDelivery is covered against a real
+ // database in db/deliveries_db_test.go).
+ time.Sleep(50 * time.Millisecond)
+ mu.Lock()
+ owed = false
+ mu.Unlock()
+
+ second := dial(t, wsURL)
+ if err := second.SetReadDeadline(time.Now().Add(300 * time.Millisecond)); err != nil {
+ t.Fatalf("set deadline: %v", err)
+ }
+ if _, _, err := second.ReadMessage(); err == nil {
+ t.Fatal("reconnect after ack replayed a frame, want silence")
+ }
}
diff --git a/src/realtime/latency_test.go b/src/realtime/latency_test.go
index 35a50d6..a6f58a4 100644
--- a/src/realtime/latency_test.go
+++ b/src/realtime/latency_test.go
@@ -44,13 +44,12 @@ func TestBroadcastLatency(t *testing.T) {
for i := 0; i < iterations; i++ {
eventID := uuid.NewString()
frame := AlarmFrame{
- Type: "alarm",
- AlertID: alertID.String(),
- EventID: eventID,
- Action: "trigger",
- Message: "latency probe",
- ServerTime: time.Now().UTC().Format(time.RFC3339Nano),
- AutoStopSeconds: 300,
+ Type: "alarm",
+ AlertID: alertID.String(),
+ EventID: eventID,
+ Action: "trigger",
+ Message: "latency probe",
+ ServerTime: time.Now().UTC().Format(time.RFC3339Nano),
}
start := time.Now()
if delivered := hub.Broadcast(alertID, frame); delivered != 2 {
diff --git a/src/realtime/notify.go b/src/realtime/notify.go
index e573693..f8af18e 100644
--- a/src/realtime/notify.go
+++ b/src/realtime/notify.go
@@ -36,26 +36,24 @@ const NotifyChannel = "alarm_events"
// the one that served the trigger request — fans the event out to its own
// subscribers.
type NotifyPayload struct {
- AlertID string `json:"alertId"`
- EventID string `json:"eventId"`
- Action string `json:"action"`
- Message string `json:"message"`
- AutoStopSeconds int `json:"autoStopSeconds"`
- CreatedAtMs int64 `json:"createdAtMs"`
+ AlertID string `json:"alertId"`
+ EventID string `json:"eventId"`
+ Action string `json:"action"`
+ Message string `json:"message"`
+ CreatedAtMs int64 `json:"createdAtMs"`
}
// Frame converts the transport payload into the wire frame. ServerTime is the
// EVENT time, not the delivery time, so every instance and every client agrees
-// on when the alarm started (and therefore on the auto-stop countdown).
+// on when the alarm started.
func (p NotifyPayload) Frame() AlarmFrame {
return AlarmFrame{
- Type: "alarm",
- AlertID: p.AlertID,
- EventID: p.EventID,
- Action: p.Action,
- Message: p.Message,
- ServerTime: time.UnixMilli(p.CreatedAtMs).UTC().Format(time.RFC3339Nano),
- AutoStopSeconds: p.AutoStopSeconds,
+ Type: "alarm",
+ AlertID: p.AlertID,
+ EventID: p.EventID,
+ Action: p.Action,
+ Message: p.Message,
+ ServerTime: time.UnixMilli(p.CreatedAtMs).UTC().Format(time.RFC3339Nano),
}
}