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