[refactor](alarm): end alerts by disarm or expiry, not a ring timer

An alert now rings from trigger until the owner disarms it; a subscriber silences only its own device. The owner instead sets a persistence cap (default 24h, hard cap 14 days) and the sweeper destroys expired alerts. Disarm ends an armed alert too, and every ending (delivered, expired, disarmed, deleted) converges on the one destruction path so their event-log reasons cannot drift.
This commit is contained in:
2026-10-19 02:26:31 +01:00
parent f055700c2d
commit 3a2da52b7a
12 changed files with 658 additions and 246 deletions
@@ -0,0 +1,51 @@
\c dcntrlized;
-- 004_drop_auto_stop.sql — remove the auto-stop window.
--
-- Contract decision (2026-10-18, user ruling): an alert is active from trigger
-- until the owner disarms it. There is no clock-based expiry: "if an owner
-- triggers it and then loses their phone, the alarm should go off until the
-- client dismisses it, with no autostop". A subscriber's ringing ends when that
-- subscriber dismisses locally; the server does not expire the event, and the
-- server-side state does not depend on any client's cooperation.
--
-- Consequence, accepted and documented in API.md: an alert that is never
-- disarmed stays active forever, and only the owner's admin token can end it.
--
-- `alerts.auto_stop_seconds` and its CHECK constraint
-- (alerts_auto_stop_seconds_range) have no remaining reader: `active` is simply
-- `status = 'triggered'` now. `001_alarm_schema.sql` is already applied and is
-- never edited in place, so the column is dropped here; a fresh database creates
-- it in 001 and drops it in this file, keeping the history append-only.
--
-- Idempotent: guarded on the schema existing and DROP COLUMN IF EXISTS, so a
-- re-apply of the whole sequence is a clean no-op (same pattern as 002/003).
-- The fan-out trigger reads the alert row when it builds its pg_notify payload,
-- so it is replaced BEFORE the column goes: leaving the old body behind would
-- make every event INSERT fail with "column auto_stop_seconds does not exist"
-- once the column is gone. The payload is the WS frame's transport form; with no
-- auto-stop it carries alertId/eventId/action/message/createdAtMs and nothing
-- else. CREATE OR REPLACE is idempotent.
CREATE OR REPLACE FUNCTION alarm.notify_alarm_event()
RETURNS TRIGGER AS $$
BEGIN
PERFORM pg_notify('alarm_events', json_build_object(
'alertId', NEW.alert_id::text,
'eventId', NEW.id::text,
'action', NEW.kind::text,
'message', COALESCE(NEW.message, ''),
'createdAtMs', (EXTRACT(EPOCH FROM NEW.created_at) * 1000)::bigint
)::text);
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
DO $$
BEGIN
IF EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'alarm') THEN
ALTER TABLE alarm.alerts DROP COLUMN IF EXISTS auto_stop_seconds;
END IF;
END
$$;
+44
View File
@@ -0,0 +1,44 @@
\c dcntrlized;
-- 006_alert_expiry.sql — a hard persistence cap chosen by the owner.
--
-- Contract decision (2026-10-18, user ruling): the creator sets a maximum
-- persistence time, at most 14 days, after which devices that never came back
-- are ignored and the alert "along with everything about it" is destroyed. This
-- is what stops a phone that never returns from pinning an alert (and its
-- delivery debt) forever once auto-stop is gone.
--
-- The application computes the absolute timestamp from the requested duration,
-- rejects anything above the cap (API.md: ERR_VALIDATION, not a silent clamp) and
-- inserts it explicitly. The DEFAULT is a safety net only: 24 hours, the
-- documented default duration, so a row inserted by an older binary or by hand
-- still expires instead of living forever.
--
-- Existing rows are backfilled to `created + 14 days` (the maximum): they were
-- created under a contract with no cap, so they get the most generous value the
-- new contract allows, not an instant expiry.
--
-- Read paths also filter `expires_at > now()` (db.GetAlert, transition, …) so an
-- alert is invisible the moment it expires, even before the sweeper reaches it.
--
-- Idempotent (same guard pattern as 002–005): 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;
ALTER TABLE alarm.alerts ADD COLUMN IF NOT EXISTS expires_at TIMESTAMPTZ;
UPDATE alarm.alerts
SET expires_at = created + interval '14 days'
WHERE expires_at IS NULL;
ALTER TABLE alarm.alerts ALTER COLUMN expires_at SET NOT NULL;
ALTER TABLE alarm.alerts ALTER COLUMN expires_at SET DEFAULT (now() + interval '24 hours');
CREATE INDEX IF NOT EXISTS alerts_expires_at_idx ON alarm.alerts (expires_at);
END
$$;
+74 -83
View File
@@ -17,6 +17,7 @@ package actions
import ( import (
"encoding/json" "encoding/json"
"fmt"
"net/http" "net/http"
"strings" "strings"
"time" "time"
@@ -134,32 +135,30 @@ type alertView struct {
Message string `json:"message"` Message string `json:"message"`
Status string `json:"status"` Status string `json:"status"`
Active bool `json:"active"` Active bool `json:"active"`
AutoStopSeconds int `json:"autoStopSeconds"`
TokenVersion int `json:"tokenVersion"`
Created string `json:"created"` Created string `json:"created"`
Modified string `json:"modified"` Modified string `json:"modified"`
TriggeredAt *string `json:"triggeredAt,omitempty"` TriggeredAt *string `json:"triggeredAt,omitempty"`
DisarmedAt *string `json:"disarmedAt,omitempty"` DisarmedAt *string `json:"disarmedAt,omitempty"`
QRPayload string `json:"qrPayload"` QRPayload string `json:"qrPayload"`
ExpiresAt string `json:"expiresAt"`
IsAdmin bool `json:"isAdmin"` IsAdmin bool `json:"isAdmin"`
SubscriptionCount int `json:"subscriptionCount"` SubscriptionCount int `json:"subscriptionCount"`
LastEvent *eventView `json:"lastEvent,omitempty"` LastEvent *eventView `json:"lastEvent,omitempty"`
} }
func newAlertView(a *db.Alert, isAdmin bool, count int, ev *db.Event, now time.Time) alertView { func newAlertView(a *db.Alert, isAdmin bool, count int, ev *db.Event) alertView {
v := alertView{ v := alertView{
AlertID: a.ID.String(), AlertID: a.ID.String(),
Name: a.Name, Name: a.Name,
Message: deref(a.Message), Message: deref(a.Message),
Status: a.Status, Status: a.Status,
Active: alertstate.IsActive(a.Status, a.TriggeredAt, a.AutoStopSeconds, now), Active: alertstate.IsActive(a.Status),
AutoStopSeconds: a.AutoStopSeconds,
TokenVersion: a.TokenVersion,
Created: rfc3339(a.Created), Created: rfc3339(a.Created),
Modified: rfc3339(a.Modified), Modified: rfc3339(a.Modified),
TriggeredAt: rfc3339Ptr(a.TriggeredAt), TriggeredAt: rfc3339Ptr(a.TriggeredAt),
DisarmedAt: rfc3339Ptr(a.DisarmedAt), DisarmedAt: rfc3339Ptr(a.DisarmedAt),
QRPayload: qrpayload.Build(a.ID, a.SubscribeToken), QRPayload: qrpayload.Build(a.ID, a.SubscribeToken),
ExpiresAt: rfc3339(a.ExpiresAt),
IsAdmin: isAdmin, IsAdmin: isAdmin,
SubscriptionCount: count, SubscriptionCount: count,
} }
@@ -177,23 +176,25 @@ func newAlertView(a *db.Alert, isAdmin bool, count int, ev *db.Event, now time.T
// ---- createAlert ------------------------------------------------------------ // ---- createAlert ------------------------------------------------------------
type createAlertRequest struct { type createAlertRequest struct {
Name string `json:"name"` Name string `json:"name"`
Message string `json:"message"` Message string `json:"message"`
AutoStopSeconds *int `json:"autoStopSeconds"` // ExpiresInSeconds is the owner's requested persistence cap. Omitted takes
// DefaultAlertExpiry; anything above MaxAlertExpiry is refused rather than
// clamped, because a silent clamp changes what the owner asked for.
ExpiresInSeconds *int64 `json:"expiresInSeconds"`
} }
type createAlertResponse struct { type createAlertResponse struct {
AlertID string `json:"alertId"` AlertID string `json:"alertId"`
Name string `json:"name"` Name string `json:"name"`
Message string `json:"message"` Message string `json:"message"`
Status string `json:"status"` Status string `json:"status"`
Active bool `json:"active"` Active bool `json:"active"`
AdminToken string `json:"adminToken"` AdminToken string `json:"adminToken"`
SubscribeToken string `json:"subscribeToken"` SubscribeToken string `json:"subscribeToken"`
QRPayload string `json:"qrPayload"` QRPayload string `json:"qrPayload"`
AutoStopSeconds int `json:"autoStopSeconds"` ExpiresAt string `json:"expiresAt"`
TokenVersion int `json:"tokenVersion"` Created string `json:"created"`
Created string `json:"created"`
} }
// CreateAlert implements POST /action/createAlert. // CreateAlert implements POST /action/createAlert.
@@ -222,15 +223,6 @@ func CreateAlert(w http.ResponseWriter, r *http.Request, payload any) {
validationError(w, "%v", err) validationError(w, "%v", err)
return return
} }
autoStop := 0
if req.AutoStopSeconds != nil {
autoStop = *req.AutoStopSeconds
}
if err := validateAutoStop(autoStop); err != nil {
validationError(w, "%v", err)
return
}
adminToken, err := alarmtoken.NewAdminToken() adminToken, err := alarmtoken.NewAdminToken()
if err != nil { if err != nil {
Error(action, "random", "%v", err) Error(action, "random", "%v", err)
@@ -244,37 +236,42 @@ func CreateAlert(w http.ResponseWriter, r *http.Request, payload any) {
return return
} }
ttl, err := resolveAlertExpiry(req.ExpiresInSeconds)
if err != nil {
validationError(w, "%v", err)
return
}
var msgPtr *string var msgPtr *string
if strings.TrimSpace(message) != "" { if strings.TrimSpace(message) != "" {
msgPtr = &message msgPtr = &message
} }
alert, err := db.CreateAlert(r.Context(), &db.Alert{ alert, err := db.CreateAlert(r.Context(), &db.Alert{
AdminUserID: p.ID, AdminUserID: p.ID,
Name: name, Name: name,
Message: msgPtr, Message: msgPtr,
AdminTokenHash: alarmtoken.Hash(adminToken), AdminTokenHash: alarmtoken.Hash(adminToken),
SubscribeToken: subscribeToken, SubscribeToken: subscribeToken,
AutoStopSeconds: autoStop, ExpiresAt: time.Now().UTC().Add(ttl),
}) })
if err != nil { if err != nil {
dbError(w, action, err) dbError(w, action, err)
return return
} }
Debug(action, "POST", "alert=%s admin=%s autoStop=%d", alert.ID, p.ID, autoStop) Debug(action, "POST", "alert=%s admin=%s", alert.ID, p.ID)
writeJSON(w, http.StatusOK, createAlertResponse{ writeJSON(w, http.StatusOK, createAlertResponse{
AlertID: alert.ID.String(), AlertID: alert.ID.String(),
Name: alert.Name, Name: alert.Name,
Message: deref(alert.Message), Message: deref(alert.Message),
Status: alert.Status, Status: alert.Status,
Active: false, Active: false,
AdminToken: adminToken, AdminToken: adminToken,
SubscribeToken: alert.SubscribeToken, SubscribeToken: alert.SubscribeToken,
QRPayload: qrpayload.Build(alert.ID, alert.SubscribeToken), QRPayload: qrpayload.Build(alert.ID, alert.SubscribeToken),
AutoStopSeconds: alert.AutoStopSeconds, ExpiresAt: rfc3339(alert.ExpiresAt),
TokenVersion: alert.TokenVersion, Created: rfc3339(alert.Created),
Created: rfc3339(alert.Created),
}) })
} }
@@ -317,7 +314,7 @@ func GetAlert(w http.ResponseWriter, r *http.Request, payload any) {
return return
} }
writeJSON(w, http.StatusOK, newAlertView(res.Alert, isAdmin, count, res.Event, time.Now().UTC())) writeJSON(w, http.StatusOK, newAlertView(res.Alert, isAdmin, count, res.Event))
} }
// ---- deleteAlert ------------------------------------------------------------ // ---- deleteAlert ------------------------------------------------------------
@@ -332,8 +329,7 @@ func DeleteAlert(w http.ResponseWriter, r *http.Request, payload any) {
return return
} }
now := time.Now().UTC() if !alertstate.CanDelete(alertstate.IsActive(alert.Status)) {
if !alertstate.CanDelete(alertstate.IsActive(alert.Status, alert.TriggeredAt, alert.AutoStopSeconds, now)) {
writeError(w, http.StatusConflict, CodeAlertState, "disarm the alert before deleting it") writeError(w, http.StatusConflict, CodeAlertState, "disarm the alert before deleting it")
return return
} }
@@ -348,41 +344,36 @@ func DeleteAlert(w http.ResponseWriter, r *http.Request, payload any) {
}) })
} }
// ---- rotateSubscribeToken --------------------------------------------------- // Persistence-cap bounds. The user's rule: the owner picks how long the alert
// lives, hard maximum 14 days; after that, clients that never came back are
// ignored and the alert is destroyed with everything about it.
//
// Chosen numbers (documented in API.md and reported to the lead):
// - maximum 14 days — the user's hard cap, refused above rather than clamped;
// - default 24 hours — an alert is a transient safety mechanism, and a day is
// the smallest window that still covers "created this evening, triggered
// tomorrow morning" without leaving debris for a week;
// - minimum 1 minute — sweeper granularity is ~60s, and anything shorter would
// be expired before a device could plausibly join.
const (
DefaultAlertExpiry = 24 * time.Hour
MinAlertExpiry = time.Minute
MaxAlertExpiry = 14 * 24 * time.Hour
)
type rotateResponse struct { // resolveAlertExpiry validates the requested duration. A nil request takes the
AlertID string `json:"alertId"` // default; out-of-range values are ERR_VALIDATION rather than a silent clamp.
SubscribeToken string `json:"subscribeToken"` func resolveAlertExpiry(seconds *int64) (time.Duration, error) {
QRPayload string `json:"qrPayload"` if seconds == nil {
TokenVersion int `json:"tokenVersion"` return DefaultAlertExpiry, nil
}
// RotateSubscribeToken implements POST /action/rotateSubscribeToken. Existing
// subscriptions survive: rotation invalidates the QR/join path, not the people
// who already joined.
func RotateSubscribeToken(w http.ResponseWriter, r *http.Request, payload any) {
const action = "rotateSubscribeToken"
alert, _, ok := requireAdminToken(w, r, action, payload)
if !ok {
return
} }
ttl := time.Duration(*seconds) * time.Second
token, err := alarmtoken.NewSubscribeToken() if ttl < MinAlertExpiry {
if err != nil { return 0, fmt.Errorf("expiresInSeconds must be at least %d (1 minute)", int64(MinAlertExpiry.Seconds()))
Error(action, "random", "%v", err)
writeError(w, http.StatusInternalServerError, CodeInternal, "could not generate a token")
return
} }
version, err := db.RotateSubscribeToken(r.Context(), alert.ID, token) if ttl > MaxAlertExpiry {
if err != nil { return 0, fmt.Errorf("expiresInSeconds must be at most %d (14 days, the hard cap)",
dbError(w, action, err) int64(MaxAlertExpiry.Seconds()))
return
} }
Debug(action, "POST", "alert=%s token rotated to version %d", alert.ID, version) return ttl, nil
writeJSON(w, http.StatusOK, rotateResponse{
AlertID: alert.ID.String(),
SubscribeToken: token,
QRPayload: qrpayload.Build(alert.ID, token),
TokenVersion: version,
})
} }
+98 -42
View File
@@ -16,22 +16,23 @@
package actions package actions
import ( import (
"errors"
"net/http" "net/http"
"strings" "strings"
"time" "time"
"git.dcntrlized.org/Dcntrlized/Alarm/db" "git.dcntrlized.org/Dcntrlized/Alarm/db"
"git.dcntrlized.org/Dcntrlized/Alarm/internal/alertstate" "git.dcntrlized.org/Dcntrlized/Alarm/internal/alertstate"
"git.dcntrlized.org/Dcntrlized/Alarm/internal/deviceid"
"git.dcntrlized.org/Dcntrlized/Alarm/realtime" "git.dcntrlized.org/Dcntrlized/Alarm/realtime"
) )
// actionResponse is the trigger/disarm result. // actionResponse is the trigger/disarm result.
type actionResponse struct { type actionResponse struct {
EventID string `json:"eventId"` EventID string `json:"eventId"`
Action string `json:"action"` Action string `json:"action"`
ServerTime string `json:"serverTime"` ServerTime string `json:"serverTime"`
AutoStopSeconds int `json:"autoStopSeconds"` Delivered int `json:"delivered"`
Delivered int `json:"delivered"`
} }
type triggerRequest struct { type triggerRequest struct {
@@ -76,23 +77,21 @@ func TriggerAlert(w http.ResponseWriter, r *http.Request, payload any) {
delivered := 0 delivered := 0
if hub := realtime.Current(); hub != nil { if hub := realtime.Current(); hub != nil {
delivered = hub.Broadcast(updated.ID, realtime.AlarmFrame{ delivered = hub.Broadcast(updated.ID, realtime.AlarmFrame{
Type: "alarm", Type: "alarm",
AlertID: updated.ID.String(), AlertID: updated.ID.String(),
EventID: event.ID.String(), EventID: event.ID.String(),
Action: alertstate.KindTrigger, Action: alertstate.KindTrigger,
Message: deref(event.Message), Message: deref(event.Message),
ServerTime: rfc3339(event.CreatedAt), ServerTime: rfc3339(event.CreatedAt),
AutoStopSeconds: updated.AutoStopSeconds,
}) })
} }
Debug(action, "POST", "alert=%s event=%s delivered=%d", updated.ID, event.ID, delivered) Debug(action, "POST", "alert=%s event=%s delivered=%d", updated.ID, event.ID, delivered)
writeJSON(w, http.StatusOK, actionResponse{ writeJSON(w, http.StatusOK, actionResponse{
EventID: event.ID.String(), EventID: event.ID.String(),
Action: alertstate.KindTrigger, Action: alertstate.KindTrigger,
ServerTime: rfc3339(event.CreatedAt), ServerTime: rfc3339(event.CreatedAt),
AutoStopSeconds: updated.AutoStopSeconds, Delivered: delivered,
Delivered: delivered,
}) })
} }
@@ -113,41 +112,79 @@ func DisarmAlert(w http.ResponseWriter, r *http.Request, payload any) {
delivered := 0 delivered := 0
if hub := realtime.Current(); hub != nil { if hub := realtime.Current(); hub != nil {
delivered = hub.Broadcast(updated.ID, realtime.AlarmFrame{ delivered = hub.Broadcast(updated.ID, realtime.AlarmFrame{
Type: "alarm", Type: "alarm",
AlertID: updated.ID.String(), AlertID: updated.ID.String(),
EventID: event.ID.String(), EventID: event.ID.String(),
Action: alertstate.KindDisarm, Action: alertstate.KindDisarm,
Message: deref(event.Message), Message: deref(event.Message),
ServerTime: rfc3339(event.CreatedAt), ServerTime: rfc3339(event.CreatedAt),
AutoStopSeconds: updated.AutoStopSeconds,
}) })
} }
Debug(action, "POST", "alert=%s event=%s delivered=%d", updated.ID, event.ID, delivered) Debug(action, "POST", "alert=%s event=%s delivered=%d", updated.ID, event.ID, delivered)
writeJSON(w, http.StatusOK, actionResponse{ writeJSON(w, http.StatusOK, actionResponse{
EventID: event.ID.String(), EventID: event.ID.String(),
Action: alertstate.KindDisarm, Action: alertstate.KindDisarm,
ServerTime: rfc3339(event.CreatedAt), ServerTime: rfc3339(event.CreatedAt),
AutoStopSeconds: updated.AutoStopSeconds, Delivered: delivered,
Delivered: delivered,
}) })
} }
// offlineAlarmResponse is the 409 check-in body: the standard error envelope plus
// the alarm frame the device missed while offline. 409 is a status, not a
// failure: it means "an alarm is pending and must ring now".
type offlineAlarmResponse struct {
Error struct {
Code string `json:"code"`
Message string `json:"message"`
} `json:"error"`
Alarm realtime.AlarmFrame `json:"alarm"`
}
// writeOfflineAlarm answers alarmStatus with 409 ERR_ALARM_TRIGGERED_OFFLINE and
// the missed trigger frame. The frame is the same shape the socket would have
// delivered (same eventId, same serverTime), so the client renders an offline
// alarm with the code path it already has for a live one.
func writeOfflineAlarm(w http.ResponseWriter, alert *db.Alert, pending *db.PendingDelivery) {
message := ""
switch {
case pending.Message != nil:
message = *pending.Message
case alert.Message != nil:
message = *alert.Message
}
var body offlineAlarmResponse
body.Error.Code = CodeAlarmTriggeredOffline
body.Error.Message = "an alarm was triggered while this device was offline; ring now"
body.Alarm = realtime.AlarmFrame{
Type: "alarm",
AlertID: alert.ID.String(),
EventID: pending.EventID.String(),
Action: alertstate.KindTrigger,
Message: message,
ServerTime: rfc3339(pending.ServerTime),
}
writeJSON(w, http.StatusConflict, body)
}
// statusView is the alarmStatus polling fallback payload (frozen field set). // statusView is the alarmStatus polling fallback payload (frozen field set).
type statusView struct { type statusView struct {
Active bool `json:"active"` Active bool `json:"active"`
EventID string `json:"eventId"` EventID string `json:"eventId"`
Action string `json:"action"` Action string `json:"action"`
Message string `json:"message"` Message string `json:"message"`
ServerTime string `json:"serverTime"` ServerTime string `json:"serverTime"`
AutoStopSeconds int `json:"autoStopSeconds"`
} }
// AlarmStatus implements GET/POST /action/alarmStatus: the cheap, indexed // AlarmStatus implements GET/POST /action/alarmStatus: the cheap, indexed
// fallback a client polls when its socket is down or it was killed mid-alarm. // fallback a client polls when its socket is down or it was killed mid-alarm. It
// also serves as the offline check-in (409 ERR_ALARM_TRIGGERED_OFFLINE) and is
// where a missed alarm's ledger row is retired.
// //
// `active` is derived (status + auto-stop window), never stored, so no // `active` is derived from the commanded status, never stored and never
// background job is needed to expire an alarm. // clock-based: an alert is active from trigger until the owner disarms it, so no
// background job is needed to expire anything.
func AlarmStatus(w http.ResponseWriter, r *http.Request, payload any) { func AlarmStatus(w http.ResponseWriter, r *http.Request, payload any) {
const action = "alarmStatus" const action = "alarmStatus"
p, ok := principal(w, r, action) p, ok := principal(w, r, action)
@@ -175,8 +212,28 @@ func AlarmStatus(w http.ResponseWriter, r *http.Request, payload any) {
return return
} }
// Offline check-in, READ-ONLY. If this device still owes an alarm, answer
// with the frame it missed instead of a quiet 200 — and clear NOTHING. The
// debt is settled only by an explicit ack (ackAlarm or the WS ack), because a
// 409 lost in transit must not forfeit the alarm: the next check-in has to
// find the same row. A trigger that has since been disarmed is not news, so
// that case falls through to the normal body (active:false).
deviceHash := ""
if raw := strings.TrimSpace(r.URL.Query().Get("deviceId")); raw != "" {
deviceHash = deviceid.Hash(raw)
}
pending, err := db.PendingTrigger(r.Context(), alertID, p.ID, deviceHash)
if err != nil && !errors.Is(err, db.ErrNotFound) {
dbError(w, action, err)
return
}
if pending != nil && res.Event != nil && res.Event.Kind == alertstate.KindTrigger {
writeOfflineAlarm(w, res.Alert, pending)
return
}
now := time.Now().UTC() now := time.Now().UTC()
active := alertstate.IsActive(res.Alert.Status, res.Alert.TriggeredAt, res.Alert.AutoStopSeconds, now) active := alertstate.IsActive(res.Alert.Status)
if res.Event == nil || res.Event.Kind != alertstate.KindTrigger { if res.Event == nil || res.Event.Kind != alertstate.KindTrigger {
// A never-triggered alert, or one whose newest event is a disarm, is // A never-triggered alert, or one whose newest event is a disarm, is
// not live even if the stored status says "triggered". // not live even if the stored status says "triggered".
@@ -184,9 +241,8 @@ func AlarmStatus(w http.ResponseWriter, r *http.Request, payload any) {
} }
view := statusView{ view := statusView{
Active: active, Active: active,
ServerTime: rfc3339(now), ServerTime: rfc3339(now),
AutoStopSeconds: res.Alert.AutoStopSeconds,
} }
if res.Event != nil { if res.Event != nil {
view.EventID = res.Event.ID.String() view.EventID = res.Event.ID.String()
+44 -44
View File
@@ -26,34 +26,33 @@ import (
// Alert is one alarm.alerts row. // Alert is one alarm.alerts row.
type Alert struct { type Alert struct {
ID uuid.UUID ID uuid.UUID
AdminUserID uuid.UUID AdminUserID uuid.UUID
Name string Name string
Message *string Message *string
Status string Status string
AdminTokenHash []byte AdminTokenHash []byte
SubscribeToken string SubscribeToken string
TokenVersion int Created time.Time
AutoStopSeconds int Modified time.Time
Created time.Time TriggeredAt *time.Time
Modified time.Time DisarmedAt *time.Time
TriggeredAt *time.Time ExpiresAt time.Time
DisarmedAt *time.Time
} }
const alertColumns = `id, admin_user_id, name, message, status::text, admin_token_hash, const alertColumns = `id, admin_user_id, name, message, status::text, admin_token_hash,
subscribe_token, token_version, auto_stop_seconds, created, modified, triggered_at, disarmed_at` subscribe_token, created, modified, triggered_at, disarmed_at, expires_at`
// alertColumnsAliased is the same list qualified with the `a` alias used by the // alertColumnsAliased is the same list qualified with the `a` alias used by the
// join in GetAlertWithLatestEvent. // join in GetAlertWithLatestEvent.
const alertColumnsAliased = `a.id, a.admin_user_id, a.name, a.message, a.status::text, a.admin_token_hash, const alertColumnsAliased = `a.id, a.admin_user_id, a.name, a.message, a.status::text, a.admin_token_hash,
a.subscribe_token, a.token_version, a.auto_stop_seconds, a.created, a.modified, a.triggered_at, a.disarmed_at` a.subscribe_token, a.created, a.modified, a.triggered_at, a.disarmed_at, a.expires_at`
func scanAlert(row pgx.Row) (*Alert, error) { func scanAlert(row pgx.Row) (*Alert, error) {
var a Alert var a Alert
err := row.Scan(&a.ID, &a.AdminUserID, &a.Name, &a.Message, &a.Status, &a.AdminTokenHash, err := row.Scan(&a.ID, &a.AdminUserID, &a.Name, &a.Message, &a.Status, &a.AdminTokenHash,
&a.SubscribeToken, &a.TokenVersion, &a.AutoStopSeconds, &a.Created, &a.Modified, &a.SubscribeToken, &a.Created, &a.Modified,
&a.TriggeredAt, &a.DisarmedAt) &a.TriggeredAt, &a.DisarmedAt, &a.ExpiresAt)
if errors.Is(err, pgx.ErrNoRows) { if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNotFound return nil, ErrNotFound
} }
@@ -63,52 +62,53 @@ func scanAlert(row pgx.Row) (*Alert, error) {
return &a, nil return &a, nil
} }
// CreateAlert inserts a new alert and returns the stored row. // CreateAlert inserts a new alert and returns the stored row. The row and its
// anonymized `created` analytics entry are written in one transaction: an alert
// that exists without its history (or vice versa) is a reporting lie.
func CreateAlert(ctx context.Context, a *Alert) (*Alert, error) { func CreateAlert(ctx context.Context, a *Alert) (*Alert, error) {
tx, err := Pool().Begin(ctx)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback(ctx) }()
const q = `INSERT INTO alarm.alerts const q = `INSERT INTO alarm.alerts
(admin_user_id, name, message, admin_token_hash, subscribe_token, auto_stop_seconds) (admin_user_id, name, message, admin_token_hash, subscribe_token, expires_at)
VALUES ($1, $2, $3, $4, $5, $6) VALUES ($1, $2, $3, $4, $5, $6)
RETURNING ` + alertColumns RETURNING ` + alertColumns
return scanAlert(Pool().QueryRow(ctx, q, alert, err := scanAlert(tx.QueryRow(ctx, q,
a.AdminUserID, a.Name, a.Message, a.AdminTokenHash, a.SubscribeToken, a.AutoStopSeconds)) a.AdminUserID, a.Name, a.Message, a.AdminTokenHash, a.SubscribeToken, a.ExpiresAt))
if err != nil {
return nil, err
}
if err := recordEvent(ctx, tx, alert.ID, &a.AdminUserID, EventCreated, ""); err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, err
}
return alert, nil
} }
// GetAlert fetches one alert by id. It returns ErrNotFound when the row is gone. // GetAlert fetches one alert by id. It returns ErrNotFound when the row is gone.
func GetAlert(ctx context.Context, alertID uuid.UUID) (*Alert, error) { func GetAlert(ctx context.Context, alertID uuid.UUID) (*Alert, error) {
const q = `SELECT ` + alertColumns + ` FROM alarm.alerts WHERE id = $1` const q = `SELECT ` + alertColumns + ` FROM alarm.alerts WHERE id = $1 AND expires_at > now()`
return scanAlert(Pool().QueryRow(ctx, q, alertID)) return scanAlert(Pool().QueryRow(ctx, q, alertID))
} }
// DeleteAlert removes an alert; subscriptions and events cascade. It reports // DeleteAlert removes an alert; subscriptions and events cascade. It reports
// ErrNotFound when nothing was deleted, so a repeated delete is visible. // ErrNotFound when nothing was deleted, so a repeated delete is visible.
func DeleteAlert(ctx context.Context, alertID uuid.UUID) error { func DeleteAlert(ctx context.Context, alertID uuid.UUID) error {
tag, err := Pool().Exec(ctx, `DELETE FROM alarm.alerts WHERE id = $1`, alertID) destroyed, err := DestroyAlert(ctx, alertID, DestroyReasonDeleted)
if err != nil { if err != nil {
return err return err
} }
if tag.RowsAffected() == 0 { if !destroyed {
return ErrNotFound return ErrNotFound
} }
return nil return nil
} }
// RotateSubscribeToken replaces the subscribe token and bumps token_version.
// Existing subscriptions are untouched: rotation invalidates the QR/join path,
// not the subscribers who already joined.
func RotateSubscribeToken(ctx context.Context, alertID uuid.UUID, newToken string) (int, error) {
var version int
err := Pool().QueryRow(ctx,
`UPDATE alarm.alerts SET subscribe_token = $2, token_version = token_version + 1
WHERE id = $1 RETURNING token_version`, alertID, newToken).Scan(&version)
if errors.Is(err, pgx.ErrNoRows) {
return 0, ErrNotFound
}
if err != nil {
return 0, err
}
return version, nil
}
// CountSubscriptions returns how many devices are subscribed to an alert. // CountSubscriptions returns how many devices are subscribed to an alert.
func CountSubscriptions(ctx context.Context, alertID uuid.UUID) (int, error) { func CountSubscriptions(ctx context.Context, alertID uuid.UUID) (int, error) {
var n int var n int
@@ -138,7 +138,7 @@ func GetAlertWithLatestEvent(ctx context.Context, alertID uuid.UUID) (*AlertWith
ORDER BY created_at DESC, id DESC ORDER BY created_at DESC, id DESC
LIMIT 1 LIMIT 1
) e ON TRUE ) e ON TRUE
WHERE a.id = $1` WHERE a.id = $1 AND a.expires_at > now()`
var a Alert var a Alert
var ( var (
@@ -151,8 +151,8 @@ func GetAlertWithLatestEvent(ctx context.Context, alertID uuid.UUID) (*AlertWith
) )
err := Pool().QueryRow(ctx, q, alertID).Scan( err := Pool().QueryRow(ctx, q, alertID).Scan(
&a.ID, &a.AdminUserID, &a.Name, &a.Message, &a.Status, &a.AdminTokenHash, &a.ID, &a.AdminUserID, &a.Name, &a.Message, &a.Status, &a.AdminTokenHash,
&a.SubscribeToken, &a.TokenVersion, &a.AutoStopSeconds, &a.Created, &a.Modified, &a.SubscribeToken, &a.Created, &a.Modified,
&a.TriggeredAt, &a.DisarmedAt, &a.TriggeredAt, &a.DisarmedAt, &a.ExpiresAt,
&evID, &evAlertID, &evKind, &evActor, &evMessage, &evCreated, &evID, &evAlertID, &evKind, &evActor, &evMessage, &evCreated,
) )
if errors.Is(err, pgx.ErrNoRows) { if errors.Is(err, pgx.ErrNoRows) {
+44 -10
View File
@@ -69,16 +69,15 @@ func TriggerAlert(ctx context.Context, alertID, actorUserID uuid.UUID, message *
return transition(ctx, alertID, actorUserID, alertstate.KindTrigger, message, nil) return transition(ctx, alertID, actorUserID, alertstate.KindTrigger, message, nil)
} }
// DisarmAlert transitions an alert to disarmed and appends the event. Disarming // DisarmAlert ENDS the alert, whether it was triggered or still armed: the user
// an alert that is not triggered is ErrState (the client's view is stale). // ruling is that disarm means "stop and remove it" (for an armed alert, "delete
// the whole alarm without ever triggering it"). It appends the disarm event and
// destroys the alert with event_log reason `disarmed`, in one transaction.
//
// Idempotent through the shared path: disarming an already-destroyed alert is
// ErrNotFound (404), never a second destruction or a duplicate log row.
func DisarmAlert(ctx context.Context, alertID, actorUserID uuid.UUID, message *string) (*Alert, *Event, error) { func DisarmAlert(ctx context.Context, alertID, actorUserID uuid.UUID, message *string) (*Alert, *Event, error) {
guard := func(a *Alert) error { return transition(ctx, alertID, actorUserID, alertstate.KindDisarm, message, nil)
if !alertstate.CanDisarm(a.Status) {
return ErrState
}
return nil
}
return transition(ctx, alertID, actorUserID, alertstate.KindDisarm, message, guard)
} }
func transition( func transition(
@@ -97,7 +96,7 @@ func transition(
// Lock the row for the whole transition so two concurrent commands cannot // Lock the row for the whole transition so two concurrent commands cannot
// interleave their status writes. // interleave their status writes.
current, err := scanAlert(tx.QueryRow(ctx, current, err := scanAlert(tx.QueryRow(ctx,
`SELECT `+alertColumns+` FROM alarm.alerts WHERE id = $1 FOR UPDATE`, alertID)) `SELECT `+alertColumns+` FROM alarm.alerts WHERE id = $1 AND expires_at > now() FOR UPDATE`, alertID))
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@@ -135,6 +134,41 @@ func transition(
return nil, nil, err return nil, nil, err
} }
// Record who owes this alarm BEFORE the transaction commits: one ledger row
// per enabled subscription, in the same transaction as the event. If this
// fails the whole trigger rolls back (the deferred Rollback runs), because an
// alarm that fans out without recording its debt is precisely the silent
// failure this ledger exists to prevent. The NOTIFY fires on COMMIT, so no
// frame can be seen before its ledger row is durable.
if kind == alertstate.KindTrigger {
if err := recordEvent(ctx, tx, alertID, &actorUserID, EventTriggered, ""); err != nil {
return nil, nil, err
}
if _, err := tx.Exec(ctx,
`INSERT INTO alarm.alarm_deliveries (alert_id, subscription_id, event_id)
SELECT $1, s.id, $2
FROM alarm.subscriptions s
WHERE s.alert_id = $1 AND s.enabled`,
alertID, event.ID); err != nil {
return nil, nil, err
}
}
// Disarm ENDS the alert in every state: armed ("delete the whole alarm
// without ever triggering it") and triggered alike. Outstanding deliveries
// cascade away, so nobody owes an alarm that can no longer be rung, and no
// client can later be told it missed a trigger on an alert that is gone.
// Disarm and deleteAlert converge on destroyLockedAlert; the event_log
// reason (disarmed vs deleted) keeps their intents distinguishable.
//
// The row lock held by this transaction makes the destruction idempotent
// against a racing sweeper (SKIP LOCKED) or a second disarm (404).
if kind == alertstate.KindDisarm {
if err := destroyLockedAlert(ctx, tx, alertID, current.AdminUserID, DestroyReasonDisarmed); err != nil {
return nil, nil, err
}
}
// The NOTIFY trigger fires on COMMIT, so no frame can be observed before // The NOTIFY trigger fires on COMMIT, so no frame can be observed before
// the new status is visible to every reader. // the new status is visible to every reader.
if err := tx.Commit(ctx); err != nil { if err := tx.Commit(ctx); err != nil {
+166
View File
@@ -0,0 +1,166 @@
// 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 <https://www.gnu.org/licenses/>.
package db
import (
"context"
"errors"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
// destroyLockedAlert is the ONE destruction path: the caller already holds the
// alert row lock, and this deletes the alert (subscriptions, events and
// deliveries cascade) together with its `destroyed` event_log row, in the
// caller's transaction. Every ending routes through here — drain-to-empty,
// expiry, disarm, explicit delete — so the four can never drift.
//
// Deleting the alert also deletes its `alarm_deliveries` rows by cascade, which
// is how "deletion wins over outstanding debt" is enforced by construction: the
// replay and the 409 read only that table, so a destroyed alert has no debt left
// to report and a client can never be told "you missed a trigger" for an alert
// that was deleted.
func destroyLockedAlert(ctx context.Context, tx pgx.Tx, alertID, adminID uuid.UUID, reason string) error {
if _, err := tx.Exec(ctx, `DELETE FROM alarm.alerts WHERE id = $1`, alertID); err != nil {
return err
}
return recordEvent(ctx, tx, alertID, &adminID, EventDestroyed, reason)
}
// DestroyAlert ends an alert from an action path (deleteAlert/disarmAlert).
// Idempotent: the alert row is locked FOR UPDATE, so of two concurrent callers
// exactly one deletes and writes the event; the loser reports false. Sweeper
// races are safe for the same reason (it claims candidates with SKIP LOCKED).
func DestroyAlert(ctx context.Context, alertID uuid.UUID, reason string) (bool, error) {
tx, err := Pool().Begin(ctx)
if err != nil {
return false, err
}
defer func() { _ = tx.Rollback(ctx) }()
var adminID uuid.UUID
err = tx.QueryRow(ctx,
`SELECT admin_user_id FROM alarm.alerts WHERE id = $1 FOR UPDATE`, alertID).Scan(&adminID)
if errors.Is(err, pgx.ErrNoRows) {
return false, nil
}
if err != nil {
return false, err
}
if err := destroyLockedAlert(ctx, tx, alertID, adminID, reason); err != nil {
return false, err
}
if err := tx.Commit(ctx); err != nil {
return false, err
}
return true, nil
}
// CompleteAlertDeliveries is the drain-to-empty decision point: destroy the alert
// when its ledger is non-empty and has no unacked rows left.
//
// Guard 1 — "the ledger was non-empty": a trigger with zero subscribers writes
// zero rows, and reading A taken literally would destroy the owner's alert the
// instant they fired it. Guard 2 — cross-instance idempotency: the alert row is
// locked FOR UPDATE before the counts are read, so two replicas applying the
// last two acks serialise and exactly one writes the destruction event.
func CompleteAlertDeliveries(ctx context.Context, alertID uuid.UUID) (bool, error) {
tx, err := Pool().Begin(ctx)
if err != nil {
return false, err
}
defer func() { _ = tx.Rollback(ctx) }()
var adminID uuid.UUID
err = tx.QueryRow(ctx,
`SELECT admin_user_id FROM alarm.alerts WHERE id = $1 FOR UPDATE`, alertID).Scan(&adminID)
if errors.Is(err, pgx.ErrNoRows) {
return false, nil
}
if err != nil {
return false, err
}
var total, pending int
if err := tx.QueryRow(ctx,
`SELECT count(*), count(*) FILTER (WHERE acked_at IS NULL)
FROM alarm.alarm_deliveries WHERE alert_id = $1`, alertID).Scan(&total, &pending); err != nil {
return false, err
}
if total == 0 || pending > 0 {
return false, nil
}
if err := destroyLockedAlert(ctx, tx, alertID, adminID, DestroyReasonDelivered); err != nil {
return false, err
}
if err := tx.Commit(ctx); err != nil {
return false, err
}
return true, nil
}
// ExpireAlerts destroys every alert whose persistence cap elapsed and returns how
// many were destroyed. Expiry ignores unacked deliveries by design — a device
// that never comes back must not pin an alert forever.
//
// One transaction, FOR UPDATE SKIP LOCKED so concurrent sweepers never
// double-delete or double-log: each row is claimed by exactly one sweeper.
func ExpireAlerts(ctx context.Context, limit int) (int, error) {
if limit <= 0 {
limit = 100
}
tx, err := Pool().Begin(ctx)
if err != nil {
return 0, err
}
defer func() { _ = tx.Rollback(ctx) }()
rows, err := tx.Query(ctx,
`SELECT id, admin_user_id FROM alarm.alerts
WHERE expires_at <= now()
ORDER BY expires_at
LIMIT $1
FOR UPDATE SKIP LOCKED`, limit)
if err != nil {
return 0, err
}
type candidate struct{ id, adminID uuid.UUID }
var candidates []candidate
for rows.Next() {
var c candidate
if err := rows.Scan(&c.id, &c.adminID); err != nil {
rows.Close()
return 0, err
}
candidates = append(candidates, c)
}
rows.Close()
if err := rows.Err(); err != nil {
return 0, err
}
for _, c := range candidates {
if err := destroyLockedAlert(ctx, tx, c.id, c.adminID, DestroyReasonExpired); err != nil {
return 0, err
}
}
if err := tx.Commit(ctx); err != nil {
return 0, err
}
return len(candidates), nil
}
+10 -34
View File
@@ -19,13 +19,11 @@
// Frozen v1 semantics: // Frozen v1 semantics:
// //
// - `status` is the last COMMANDED state (armed | triggered | disarmed). // - `status` is the last COMMANDED state (armed | triggered | disarmed).
// - `active` is DERIVED: a triggered alert stops being active once its // - `active` is DERIVED: an alert is active exactly while its commanded status
// auto-stop window elapses, with no background job and no timer. That is // is `triggered`. There is no clock-based expiry and no background job —
// what makes auto-stop correct across several service instances. // only the owner's disarm ends a live alarm.
package alertstate package alertstate
import "time"
// Commanded statuses. // Commanded statuses.
const ( const (
StatusArmed = "armed" StatusArmed = "armed"
@@ -39,9 +37,6 @@ const (
KindDisarm = "disarm" KindDisarm = "disarm"
) )
// MaxAutoStopSeconds bounds auto_stop_seconds (mirrors the SQL CHECK).
const MaxAutoStopSeconds = 86400
// ValidStatus reports whether s is one of the three commanded statuses. // ValidStatus reports whether s is one of the three commanded statuses.
func ValidStatus(s string) bool { func ValidStatus(s string) bool {
switch s { switch s {
@@ -57,33 +52,14 @@ func ValidKind(k string) bool {
return k == KindTrigger || k == KindDisarm return k == KindTrigger || k == KindDisarm
} }
// IsActive reports whether the alert is live at now. // IsActive reports whether the alert is live.
// //
// A never-triggered alert (nil triggeredAt) is never active. autoStopSeconds // There is deliberately no time parameter: an alarm rings from trigger until the
// == 0 means "ring until disarmed"; a positive value bounds the window. The // owner disarms it. A subscriber silences its own device by dismissing locally;
// window is inclusive of its start and exclusive at its end. // that is a client-side action and never changes server state. The accepted
func IsActive(status string, triggeredAt *time.Time, autoStopSeconds int, now time.Time) bool { // consequence is that a lost owner phone leaves the alert active forever.
if status != StatusTriggered || triggeredAt == nil { func IsActive(status string) bool {
return false return status == StatusTriggered
}
if autoStopSeconds <= 0 {
return true
}
return now.Before(triggeredAt.Add(time.Duration(autoStopSeconds) * time.Second))
}
// RemainingAutoStopSeconds returns how much of the auto-stop window is left at
// now, or 0 when the alert is not bounded or has already elapsed. It is what
// alarmStatus reports so a client can run its own countdown.
func RemainingAutoStopSeconds(status string, triggeredAt *time.Time, autoStopSeconds int, now time.Time) int {
if !IsActive(status, triggeredAt, autoStopSeconds, now) || autoStopSeconds <= 0 {
return 0
}
left := int(triggeredAt.Add(time.Duration(autoStopSeconds) * time.Second).Sub(now).Seconds())
if left < 0 {
return 0
}
return left
} }
// CanDisarm reports whether a disarm command is legal for the current status. // CanDisarm reports whether a disarm command is legal for the current status.
+15 -33
View File
@@ -15,10 +15,7 @@
package alertstate package alertstate
import ( import "testing"
"testing"
"time"
)
func TestValidStatusAndKind(t *testing.T) { func TestValidStatusAndKind(t *testing.T) {
for _, s := range []string{StatusArmed, StatusTriggered, StatusDisarmed} { for _, s := range []string{StatusArmed, StatusTriggered, StatusDisarmed} {
@@ -44,46 +41,31 @@ func TestValidStatusAndKind(t *testing.T) {
} }
func TestIsActive(t *testing.T) { func TestIsActive(t *testing.T) {
now := time.Date(2026, 10, 18, 12, 0, 0, 0, time.UTC)
earlier := now.Add(-10 * time.Minute)
cases := []struct { cases := []struct {
name string name string
status string status string
at *time.Time
auto int
want bool want bool
}{ }{
{"armed is never active", StatusArmed, &earlier, 0, false}, {"armed is never active", StatusArmed, false},
{"disarmed is never active", StatusDisarmed, &earlier, 0, false}, {"disarmed is never active", StatusDisarmed, false},
{"triggered without auto-stop stays active", StatusTriggered, &earlier, 0, true}, {"triggered is active until disarmed", StatusTriggered, true},
{"triggered with auto-stop still inside the window", StatusTriggered, &earlier, 3600, true}, {"unknown status", "firing", false},
{"triggered with auto-stop past the window", StatusTriggered, &earlier, 60, false}, {"empty status", "", false},
{"triggered with no timestamp", StatusTriggered, nil, 0, false},
{"unknown status", "firing", &earlier, 0, false},
} }
for _, tc := range cases { for _, tc := range cases {
if got := IsActive(tc.status, tc.at, tc.auto, now); got != tc.want { if got := IsActive(tc.status); got != tc.want {
t.Errorf("%s: IsActive = %v, want %v", tc.name, got, tc.want) t.Errorf("%s: IsActive(%q) = %v, want %v", tc.name, tc.status, got, tc.want)
} }
} }
} }
func TestRemainingAutoStopSeconds(t *testing.T) { // TestIsActiveHasNoClock pins the ruling at the type level: active is a function
now := time.Date(2026, 10, 18, 12, 0, 0, 0, time.UTC) // of the commanded status only. Re-introducing a time-based expiry would change
at := now.Add(-10 * time.Minute) // IsActive's signature and fail to compile here.
func TestIsActiveHasNoClock(t *testing.T) {
if got := RemainingAutoStopSeconds(StatusTriggered, &at, 3600, now); got != 3000 { var f func(string) bool = IsActive
t.Errorf("remaining = %d, want 3000", got) if !f(StatusTriggered) {
} t.Fatal("triggered must be active")
if got := RemainingAutoStopSeconds(StatusTriggered, &at, 60, now); got != 0 {
t.Errorf("expired remaining = %d, want 0", got)
}
if got := RemainingAutoStopSeconds(StatusArmed, &at, 3600, now); got != 0 {
t.Errorf("non-triggered remaining = %d, want 0", got)
}
if got := RemainingAutoStopSeconds(StatusTriggered, &at, 0, now); got != 0 {
t.Errorf("auto-stop 0 remaining = %d, want 0", got)
} }
} }
+71
View File
@@ -0,0 +1,71 @@
// 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 <https://www.gnu.org/licenses/>.
// Package sweeper runs the periodic housekeeping that keeps the schema bounded:
// destroying alerts whose owner-set persistence cap elapsed, and pruning the
// durable log table (age + row cap). It is a goroutine started once at boot.
//
// Both jobs are safe to run concurrently on several instances: expiry claims
// rows with FOR UPDATE SKIP LOCKED and writes its analytics in the same
// transaction; log pruning is a plain delete where "twice" is harmless.
package sweeper
import (
"context"
"time"
"git.dcntrlized.org/Dcntrlized/Alarm/db"
)
// DefaultInterval is the tick. Expiry reads also filter the timestamp, so a
// client never observes a zombie between ticks; the tick only bounds how long a
// destroyed-by-policy row lingers on disk.
const DefaultInterval = 60 * time.Second
// Run blocks until ctx is cancelled, sweeping every interval.
func Run(ctx context.Context, interval time.Duration, logf func(format string, v ...any)) {
if interval <= 0 {
interval = DefaultInterval
}
if logf == nil {
logf = func(string, ...any) {}
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
sweep := func() {
if n, err := db.ExpireAlerts(ctx, 100); err != nil {
logf("sweeper: expire alerts: %v", err)
} else if n > 0 {
logf("sweeper: destroyed %d expired alerts", n)
}
if n, err := db.PruneEventLog(ctx, db.EventLogRetention); err != nil {
logf("sweeper: prune event_log: %v", err)
} else if n > 0 {
logf("sweeper: pruned %d event rows", n)
}
}
sweep()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
sweep()
}
}
}
+6
View File
@@ -37,6 +37,7 @@ import (
"git.dcntrlized.org/Dcntrlized/Alarm/handlers/actions" "git.dcntrlized.org/Dcntrlized/Alarm/handlers/actions"
"git.dcntrlized.org/Dcntrlized/Alarm/handlers/auth" "git.dcntrlized.org/Dcntrlized/Alarm/handlers/auth"
"git.dcntrlized.org/Dcntrlized/Alarm/handlers/manifest" "git.dcntrlized.org/Dcntrlized/Alarm/handlers/manifest"
"git.dcntrlized.org/Dcntrlized/Alarm/internal/sweeper"
"git.dcntrlized.org/Dcntrlized/Alarm/logger" "git.dcntrlized.org/Dcntrlized/Alarm/logger"
"git.dcntrlized.org/Dcntrlized/Alarm/middleware" "git.dcntrlized.org/Dcntrlized/Alarm/middleware"
"git.dcntrlized.org/Dcntrlized/Alarm/realtime" "git.dcntrlized.org/Dcntrlized/Alarm/realtime"
@@ -83,6 +84,11 @@ func main() {
// used in the shutdown path below. // used in the shutdown path below.
realtimeShutdown := realtime.Register(outer, database) realtimeShutdown := realtime.Register(outer, database)
// Housekeeping: alerts past their owner-set persistence cap are destroyed
// (unacked deliveries ignored — that is the point), and the log table is
// pruned by age + row count. Same goroutine, same 60s tick.
go sweeper.Run(context.Background(), 0, logger.Warn)
// ONE plaintext listener (contract rev 9 §B). ReadHeaderTimeout/IdleTimeout // ONE plaintext listener (contract rev 9 §B). ReadHeaderTimeout/IdleTimeout
// cover slow-header and idle-socket abuse. ReadTimeout/WriteTimeout are // cover slow-header and idle-socket abuse. ReadTimeout/WriteTimeout are
// deliberately ZERO: this listener also serves the long-lived /ws/alarm // deliberately ZERO: this listener also serves the long-lived /ws/alarm
+35
View File
@@ -107,3 +107,38 @@ func TestLogoutServesBothMethods(t *testing.T) {
t.Fatalf("PUT /auth/logout = %d, want 405", code) t.Fatalf("PUT /auth/logout = %d, want 405", code)
} }
} }
// TestRouteLevelErrorsCarryMessages walks the error paths main.go registers
// itself — the /action/ verb guard and the per-method guards on /auth/* — and
// asserts the user's invariant: an error status never comes with an empty body.
// The /action/ dispatch errors are plain text, the auth guards are plain text
// too; both must carry a readable message.
func TestRouteLevelErrorsCarryMessages(t *testing.T) {
outer := newOuterMux(newAPIMux())
cases := []struct {
method string
path string
body string
want int
}{
{http.MethodPatch, "/action/createAlert", "", http.StatusMethodNotAllowed},
{http.MethodGet, "/action/nope", "", http.StatusNotImplemented},
{http.MethodPost, "/action/getAlert", `{"unknown":true}`, http.StatusBadRequest},
{http.MethodPut, "/auth/login", "", http.StatusMethodNotAllowed},
{http.MethodDelete, "/auth/me", "", http.StatusMethodNotAllowed},
}
for _, tc := range cases {
req := httptest.NewRequest(tc.method, tc.path, strings.NewReader(tc.body))
rec := httptest.NewRecorder()
outer.ServeHTTP(rec, req)
if rec.Code != tc.want {
t.Fatalf("%s %s = %d, want %d (body: %q)", tc.method, tc.path, rec.Code, tc.want, rec.Body.String())
}
if body := strings.TrimSpace(rec.Body.String()); body == "" {
t.Fatalf("%s %s: status %d with an empty body — every error must carry a message", tc.method, tc.path, rec.Code)
}
}
}