init

src/realtime/notify.go

@@ -0,0 +1,141 @@
+// 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 realtime
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"git.dcntrlized.org/Dcntrlized/Alarm/internal/alertstate"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+// NotifyChannel is the PostgreSQL NOTIFY channel the alarm schema publishes
+// trigger/disarm events on (see alarm.notify_alarm_event() in
+// database/migrations/001_alarm_schema.sql).
+const NotifyChannel = "alarm_events"
+
+// NotifyPayload is the transport form of an alarm frame on the NOTIFY channel.
+// It is emitted by the database trigger, so every service instance — not just
+// 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"`
+}
+
+// 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).
+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,
+	}
+}
+
+// ErrBadNotifyPayload reports a NOTIFY payload that cannot be trusted; the
+// bridge logs and ignores it rather than delivering a malformed frame.
+var ErrBadNotifyPayload = errors.New("realtime: malformed notify payload")
+
+// DeliverNotifyPayload parses one NOTIFY payload and broadcasts the frame. It
+// is the unit-testable half of the bridge (no PostgreSQL needed).
+func (h *Hub) DeliverNotifyPayload(payload []byte) (AlarmFrame, bool, error) {
+	var p NotifyPayload
+	if err := json.Unmarshal(payload, &p); err != nil {
+		return AlarmFrame{}, false, ErrBadNotifyPayload
+	}
+	alertID, err := uuid.Parse(p.AlertID)
+	if err != nil {
+		return AlarmFrame{}, false, ErrBadNotifyPayload
+	}
+	if _, err := uuid.Parse(p.EventID); err != nil {
+		return AlarmFrame{}, false, ErrBadNotifyPayload
+	}
+	if !alertstate.ValidKind(p.Action) {
+		return AlarmFrame{}, false, ErrBadNotifyPayload
+	}
+	f := p.Frame()
+	h.Broadcast(alertID, f)
+	return f, true, nil
+}
+
+// RunNotifyBridge listens on the alarm_events channel and fans every payload
+// out to the hub, reconnecting with exponential backoff. It returns nil when
+// ctx is cancelled. Run it as `go realtime.RunNotifyBridge(ctx, hub, dsn)`.
+func RunNotifyBridge(ctx context.Context, hub *Hub, dsn string) error {
+	backoff := 500 * time.Millisecond
+	const maxBackoff = 10 * time.Second
+
+	for {
+		if ctx.Err() != nil {
+			return nil
+		}
+		err := listenOnce(ctx, hub, dsn)
+		if ctx.Err() != nil {
+			return nil
+		}
+		hub.opts.Logf("notify: bridge disconnected (%v); reconnecting in %s", err, backoff)
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(backoff):
+		}
+		if backoff < maxBackoff {
+			backoff *= 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+		}
+	}
+}
+
+func listenOnce(ctx context.Context, hub *Hub, dsn string) error {
+	conn, err := pgx.Connect(ctx, dsn)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = conn.Close(context.Background()) }()
+
+	if _, err := conn.Exec(ctx, "LISTEN "+NotifyChannel); err != nil {
+		return err
+	}
+	hub.opts.Logf("notify: listening on channel %s", NotifyChannel)
+
+	for {
+		n, err := conn.WaitForNotification(ctx)
+		if err != nil {
+			return err
+		}
+		if _, ok, err := hub.DeliverNotifyPayload([]byte(n.Payload)); err != nil {
+			hub.opts.Logf("notify: ignored payload: %v", err)
+		} else if ok {
+			hub.opts.Logf("notify: delivered event from channel %s", NotifyChannel)
+		}
+	}
+}