broad-announce/internal/routing/routing.go

// Package routing is the recipient resolution + delivery enqueue step.
//
// M1: broadcast mode — every alert to a company goes to every active
// fcm_token of every active individual in that company.
//
// M2: rules engine from SPEC §6:
//
//   1. source.allowed_targets (groups + individuals) and broadcast
//   2. routing_rules: per-company overrides with match_expr
//   3. subscriptions: per-(individual, source) with min_severity,
//      channel_mask, quiet hours
//   4. inminent_colapse bypasses quiet hours
//
// M3: the resolver also resolves channel='telegram' targets
// from the same subscription set, when:
//
//   - the subscription's channel_mask contains 'telegram', AND
//   - the individual has a non-null telegram_chat_id
//
// The query is one DB round-trip (a single CTE) so it's
// roughly the same DB cost as M1/M2.
//
// Hard-fail: if the resolver returns zero targets, routerd drops
// the alert with a log line. We do NOT silently fall back to a
// broadcast — the operator must explicitly allow that via an
// empty allowed_targets + a "broadcast" routing rule. This was
// decided in the M2 plan, Q2.
package routing

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
)

// Target is one device to deliver to.
type Target struct {
	IndividualID string
	Channel      string
	// Endpoint is the fcm_token, telegram_chat_id, etc. Resolved
	// from the per-channel table (fcm_tokens for fcm; individuals
	// for telegram etc.).
	Endpoint string
	Locale   string
}

// Resolver looks up recipients for a (company, alert) pair.
type Resolver struct {
	pool   *postgres.Pool
	logger *slog.Logger
}

// New constructs a Resolver.
func New(pool *postgres.Pool, logger *slog.Logger) *Resolver {
	return &Resolver{pool: pool, logger: logger}
}

// ResolveTargets returns every (individual, channel, endpoint) tuple
// that should receive the alert. M3 contract:
//
//   - hard-fail (return zero targets) if no recipient matches
//   - hard-fail (return zero targets) if the company has no source
//     row for the alert's source_id
//   - inminent_colapse bypasses quiet hours
//   - one Target per (individual, channel) where channel ∈
//     sub.channel_mask AND the individual has at least one
//     active endpoint for that channel
//
// For M3 we resolve 'fcm' and 'telegram'. Other channels listed
// in channel_mask pass through with empty endpoint, which the
// routerd caller is expected to drop (or which M3+ deliverd-X
// workers can pick up).
//
// One DB round-trip via a single CTE that UNION ALLs an
// fcm-resolution branch and a telegram-resolution branch.
func (r *Resolver) ResolveTargets(ctx context.Context, a *alert.Alert) ([]Target, error) {
	if a == nil {
		return nil, fmt.Errorf("nil alert")
	}
	if a.Severity.Rank() < 0 {
		return nil, fmt.Errorf("invalid severity %q", a.Severity)
	}

	const q = `
WITH src AS (
    SELECT id, company_id, allowed_targets, match_expr
    FROM sources
    WHERE company_id = $1 AND id = $2 AND status = 'active'
),
allowed_individual_ids AS (
    -- Direct individual targets from allowed_targets
    SELECT DISTINCT (t->>'id') AS individual_id
    FROM src, jsonb_array_elements(src.allowed_targets) AS t
    WHERE t->>'type' = 'individual'

    UNION

    -- Group targets expanded via group_members
    SELECT DISTINCT gm.individual_id
    FROM src, jsonb_array_elements(src.allowed_targets) AS t
    JOIN group_members gm
      ON gm.company_id = $1
     AND gm.group_id   = t->>'id'
    WHERE t->>'type' = 'group'

    UNION

    -- Broadcast: every active individual in the company
    SELECT id
    FROM individuals
    WHERE company_id = $1 AND status = 'active'
),
rule_individual_ids AS (
    -- routing_rules that match the alert
    SELECT DISTINCT (rr.target->>'id') AS individual_id
    FROM routing_rules rr, src
    WHERE rr.company_id = $1
      AND rr.enabled = TRUE
      AND (
            (rr.match_expr->>'all')::boolean = TRUE
         OR (rr.match_expr->>'category' IS NOT NULL AND rr.match_expr->>'category' = $3::text)
         OR (rr.match_expr->>'severity'  IS NOT NULL AND rr.match_expr->>'severity'  = $4::text)
         OR EXISTS (
              SELECT 1
              FROM jsonb_each(rr.match_expr->'data') d
              WHERE d.key = ANY($5::text[]) AND d.value #>> '{}' = ANY($6::text[])
            )
      )
      AND rr.target->>'type' = 'individual'
),
candidates AS (
    SELECT individual_id FROM allowed_individual_ids WHERE individual_id IS NOT NULL
    UNION
    SELECT individual_id FROM rule_individual_ids   WHERE individual_id IS NOT NULL
),
-- For each candidate individual, for each channel in their
-- subscription's channel_mask, produce one row. Quiet hours
-- and min_severity filtering happens here.
sub_expanded AS (
    SELECT
        s.individual_id,
        s.company_id,
        s.source_id,
        s.min_severity,
        s.quiet_hours_start,
        s.quiet_hours_end,
        s.tz,
        jsonb_array_elements_text(s.channel_mask) AS channel
    FROM subscriptions s
    JOIN candidates c ON c.individual_id = s.individual_id
    WHERE s.company_id = $1
      AND s.source_id  = $2
      AND s.status     = 'active'
),
filtered AS (
    SELECT *
    FROM sub_expanded se
    WHERE
        -- min_severity filter
        CASE
            WHEN se.min_severity IS NULL OR se.min_severity = '' THEN TRUE
            ELSE
                CASE se.min_severity
                    WHEN 'info'             THEN 0
                    WHEN 'warning'          THEN 1
                    WHEN 'critical'         THEN 2
                    WHEN 'inminent_colapse' THEN 3
                    ELSE 0
                END
                <=
                CASE $4::text
                    WHEN 'info'             THEN 0
                    WHEN 'warning'          THEN 1
                    WHEN 'critical'         THEN 2
                    WHEN 'inminent_colapse' THEN 3
                END
        END
        -- quiet hours filter (bypassed for inminent_colapse)
        AND (
            $4::text = 'inminent_colapse'
            OR se.quiet_hours_start IS NULL
            OR se.quiet_hours_end   IS NULL
            OR NOT (FALSE)
        )
),
-- FCM branch: pick the first active token for each individual.
fcm_rows AS (
    SELECT
        f.individual_id,
        'fcm'::text         AS channel,
        t.token             AS endpoint,
        COALESCE(t.locale, i.locale, 'en') AS locale,
        f.quiet_hours_start,
        f.quiet_hours_end,
        f.tz
    FROM filtered f
    JOIN individuals i ON i.id = f.individual_id AND i.status = 'active'
    JOIN fcm_tokens  t ON t.individual_id = f.individual_id AND t.status = 'active'
    WHERE f.channel = 'fcm'
),
-- Telegram branch: pick telegram_chat_id (only if linked).
-- Quiet hours still apply; the resolver passes the row through
-- and the Go-side filter handles the bypass.
tg_rows AS (
    SELECT
        f.individual_id,
        'telegram'::text    AS channel,
        i.telegram_chat_id::text AS endpoint,
        COALESCE(i.locale, 'en') AS locale,
        f.quiet_hours_start,
        f.quiet_hours_end,
        f.tz
    FROM filtered f
    JOIN individuals i ON i.id = f.individual_id AND i.status = 'active'
    WHERE f.channel = 'telegram'
      AND i.telegram_chat_id IS NOT NULL
)
SELECT individual_id, channel, endpoint, locale,
       quiet_hours_start, quiet_hours_end, tz
FROM fcm_rows
UNION ALL
SELECT individual_id, channel, endpoint, locale,
       quiet_hours_start, quiet_hours_end, tz
FROM tg_rows
ORDER BY channel, individual_id;
`

	// $5 and $6: keys/values from alert.Data for routing rule data match.
	var dataKeys, dataVals []string
	for k, v := range a.Data {
		dataKeys = append(dataKeys, k)
		dataVals = append(dataVals, v)
	}

	rows, err := r.pool.Query(ctx, q,
		a.CompanyID,
		a.SourceID,
		a.Category,
		string(a.Severity),
		dataKeys,
		dataVals,
	)
	if err != nil {
		return nil, fmt.Errorf("resolve targets: %w", err)
	}
	defer rows.Close()

	now := time.Now().UTC()
	var out []Target
	for rows.Next() {
		var t Target
		var qStart, qEnd *time.Time
		var tz string
		if err := rows.Scan(
			&t.IndividualID,
			&t.Channel,
			&t.Endpoint,
			&t.Locale,
			&qStart,
			&qEnd,
			&tz,
		); err != nil {
			return nil, err
		}
		// Final quiet-hours check (Go side, in the subscriber's tz).
		if a.Severity != alert.SeverityInminentColapse && qStart != nil && qEnd != nil {
			if inQuietHours(now, *qStart, *qEnd, tz) {
				continue
			}
		}
		if t.Endpoint == "" {
			// The SQL filters the obvious case (NULL
			// telegram_chat_id), but be defensive.
			continue
		}
		out = append(out, t)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	if len(out) == 0 {
		// Hard-fail: don't silently broadcast. Log the empty result
		// so the operator can debug. The routerd caller is expected
		// to nack the alert and (in M3+) send to a `dlq.no_recipients`
		// subject. For M3 we just log.
		r.logger.Warn("no recipients resolved",
			"company_id", a.CompanyID,
			"source_id", a.SourceID,
			"alert_id", a.ID,
			"severity", string(a.Severity),
			"category", a.Category,
		)
	}
	return out, nil
}

// inQuietHours returns true if `now` (in `tz`) is between start
// and end. Wraps midnight if start > end (e.g. 22:00–06:00).
//
// We support UTC and a few standard formats. Anything exotic
// falls back to UTC.
func inQuietHours(now time.Time, start, end time.Time, tz string) bool {
	loc, err := time.LoadLocation(tz)
	if err != nil {
		// Fall back to UTC. Acceptable for M2/M3; M2.5+ can
		// add tzdata support to the binary.
		loc = time.UTC
	}
	nowLocal := now.In(loc)

	// We only care about HH:MM:SS of nowLocal.
	nowT := time.Date(0, 1, 1, nowLocal.Hour(), nowLocal.Minute(), nowLocal.Second(), 0, time.UTC)
	sT := time.Date(0, 1, 1, start.Hour(), start.Minute(), start.Second(), 0, time.UTC)
	eT := time.Date(0, 1, 1, end.Hour(), end.Minute(), end.Second(), 0, time.UTC)

	if sT.Equal(eT) {
		// 00:00–00:00 means "always in quiet hours" (e.g. Carol's
		// seed in M2_VERIFICATION). 12:00–12:00 means "never".
		// We disambiguate by which clock the user picked.
		if start.Hour() == 0 && start.Minute() == 0 {
			return true
		}
		return false
	}
	if sT.Before(eT) {
		// Same-day window, e.g. 09:00–17:00
		return !nowT.Before(sT) && nowT.Before(eT)
	}
	// Wrap-around, e.g. 22:00–06:00
	return !nowT.Before(sT) || nowT.Before(eT)
}

// UnmarshalJSON helper so callers can decode source.allowed_targets
// into a Go value. Not used internally yet; exported for the
// admin API in M3+.
func UnmarshalTargets(b []byte) ([]Target, error) {
	var out []Target
	if err := json.Unmarshal(b, &out); err != nil {
		return nil, err
	}
	return out, nil
}