// 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" ) // Subscription is one alarm.subscriptions row: a device that receives the alarm. // // A subscription is the SUBSCRIBER IDENTITY (contract rev 10). Either it hangs // off an account (`SubscriberUserID` set, the pre-rev-10 shape) or it stands // alone with a device token the service minted at join (`DeviceTokenHash` set, // `SubscriberUserID` is uuid.Nil). A subscriber is therefore never a fake user: // there is no auth.users row behind an anonymous one. type Subscription struct { ID uuid.UUID AlertID uuid.UUID SubscriberUserID uuid.UUID DeviceID string DeviceLabel *string Platform *string Enabled bool LastSeenAt *time.Time Created time.Time // DeviceTokenHash is the SHA-256 of the token handed to the device once, at // join. Nil for account-backed subscriptions. DeviceTokenHash *string } const subscriptionColumns = `id, alert_id, subscriber_user_id, device_id, device_label, platform, enabled, last_seen_at, created, device_token_hash` func scanSubscription(row pgx.Row) (*Subscription, error) { var s Subscription var user *uuid.UUID err := row.Scan(&s.ID, &s.AlertID, &user, &s.DeviceID, &s.DeviceLabel, &s.Platform, &s.Enabled, &s.LastSeenAt, &s.Created, &s.DeviceTokenHash) if user != nil { s.SubscriberUserID = *user } if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } if err != nil { return nil, err } return &s, nil } // CreateSubscription inserts a subscription. A duplicate // (alert_id, subscriber_user_id, device_id) is reported as ErrConflict, which // the join action turns into the ERR_ALERT_ALREADY_JOINED race fallback — the // sequential re-scan path is handled by the caller's lookup and never reaches // this error. func CreateSubscription(ctx context.Context, s *Subscription) (*Subscription, error) { tx, err := Pool().Begin(ctx) if err != nil { return nil, err } defer func() { _ = tx.Rollback(ctx) }() var actor *uuid.UUID if s.SubscriberUserID != uuid.Nil { actor = &s.SubscriberUserID } const q = `INSERT INTO alarm.subscriptions (alert_id, subscriber_user_id, device_id, device_label, platform, device_token_hash) VALUES ($1, $2, $3, $4, $5, $6) RETURNING ` + subscriptionColumns row, err := scanSubscription(tx.QueryRow(ctx, q, s.AlertID, actor, s.DeviceID, s.DeviceLabel, s.Platform, s.DeviceTokenHash)) if isUniqueViolation(err) { return nil, ErrConflict } if err != nil { return nil, err } if err := recordEvent(ctx, tx, s.AlertID, actor, EventJoined, ""); err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } return row, nil } // GetSubscriptionByDevice is the idempotency lookup for joinAlert. func GetSubscriptionByDevice(ctx context.Context, alertID, userID uuid.UUID, deviceID string) (*Subscription, error) { const q = `SELECT ` + subscriptionColumns + ` FROM alarm.subscriptions WHERE alert_id = $1 AND subscriber_user_id = $2 AND device_id = $3` return scanSubscription(Pool().QueryRow(ctx, q, alertID, userID, deviceID)) } // FindSubscriptionByDevice looks a subscription up by alert and device digest, // whichever path created it. It is the idempotency lookup for an anonymous // join, where there is no user id to key on. func FindSubscriptionByDevice(ctx context.Context, alertID uuid.UUID, deviceID string) (*Subscription, error) { const q = `SELECT ` + subscriptionColumns + ` FROM alarm.subscriptions WHERE alert_id = $1 AND device_id = $2 ORDER BY created LIMIT 1` return scanSubscription(Pool().QueryRow(ctx, q, alertID, deviceID)) } // GetSubscriptionByToken resolves the subscriber behind a device token. The // token itself is never stored: the caller hashes what it received and this // lookup runs against the unique index. func GetSubscriptionByToken(ctx context.Context, tokenHash string) (*Subscription, error) { const q = `SELECT ` + subscriptionColumns + ` FROM alarm.subscriptions WHERE device_token_hash = $1` return scanSubscription(Pool().QueryRow(ctx, q, tokenHash)) } // RotateDeviceToken replaces the token hash of a subscription. A re-join from // the same device rotates rather than inserts, so a lost token is recoverable // with the subscribe token while the old one stops working immediately. func RotateDeviceToken(ctx context.Context, subscriptionID uuid.UUID, tokenHash string) error { tag, err := Pool().Exec(ctx, `UPDATE alarm.subscriptions SET device_token_hash = $2 WHERE id = $1`, subscriptionID, tokenHash) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // GetSubscription fetches one subscription scoped to its alert. func GetSubscription(ctx context.Context, alertID, subscriptionID uuid.UUID) (*Subscription, error) { const q = `SELECT ` + subscriptionColumns + ` FROM alarm.subscriptions WHERE alert_id = $1 AND id = $2` return scanSubscription(Pool().QueryRow(ctx, q, alertID, subscriptionID)) } // ListSubscriptions returns every device subscribed to an alert, oldest first. func ListSubscriptions(ctx context.Context, alertID uuid.UUID) ([]Subscription, error) { const q = `SELECT ` + subscriptionColumns + ` FROM alarm.subscriptions WHERE alert_id = $1 ORDER BY created, id` rows, err := Pool().Query(ctx, q, alertID) if err != nil { return nil, err } defer rows.Close() out := make([]Subscription, 0, 4) for rows.Next() { var s Subscription if err := rows.Scan(&s.ID, &s.AlertID, &s.SubscriberUserID, &s.DeviceID, &s.DeviceLabel, &s.Platform, &s.Enabled, &s.LastSeenAt, &s.Created); err != nil { return nil, err } out = append(out, s) } return out, rows.Err() } // DeleteSubscription removes one subscription. ErrNotFound when it was already // gone, so a repeated removal is visible to the caller. func DeleteSubscription(ctx context.Context, alertID, subscriptionID uuid.UUID) error { tag, err := Pool().Exec(ctx, `DELETE FROM alarm.subscriptions WHERE alert_id = $1 AND id = $2`, alertID, subscriptionID) if err != nil { return err } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // IsSubscriber reports whether the user has at least one subscription row on // the alert. It is the authorisation check behind getAlert/alarmStatus. func IsSubscriber(ctx context.Context, alertID, userID uuid.UUID) (bool, error) { var exists bool err := Pool().QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM alarm.subscriptions WHERE alert_id = $1 AND subscriber_user_id = $2)`, alertID, userID).Scan(&exists) return exists, err } // UserSubscriptionIDs lists the user's subscription ids on an alert, so a WS // connection can be tied to the right device rows. func UserSubscriptionIDs(ctx context.Context, alertID, userID uuid.UUID) ([]uuid.UUID, error) { rows, err := Pool().Query(ctx, `SELECT id FROM alarm.subscriptions WHERE alert_id = $1 AND subscriber_user_id = $2 ORDER BY created`, alertID, userID) if err != nil { return nil, err } defer rows.Close() var out []uuid.UUID for rows.Next() { var id uuid.UUID if err := rows.Scan(&id); err != nil { return nil, err } out = append(out, id) } return out, rows.Err() } // TouchSubscription stamps last_seen_at. It is called on WS connect and on // every heartbeat; a no-op update (the row vanished mid-connection) is not an // error worth propagating. func TouchSubscription(ctx context.Context, subscriptionID uuid.UUID) error { _, err := Pool().Exec(ctx, `UPDATE alarm.subscriptions SET last_seen_at = NOW() WHERE id = $1`, subscriptionID) return err }