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