Alarm/src/realtime/notify.go

// 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"`
	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.
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),
	}
}

// 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)
		}
	}
}