|
|
@@ -1,64 +1,383 @@
|
|
|
// 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 replaces
|
|
|
-// this with the rules engine from SPEC §6 (subscriptions, opt-in,
|
|
|
-// quiet hours, routing rules, …).
|
|
|
+// 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 interface is small: ResolveTokens returns the set of
|
|
|
-// (individual_id, fcm_token, locale) tuples that should receive
|
|
|
-// the alert. deliverd-fcm is the only consumer in M1.
|
|
|
+// 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
|
|
|
- FCMToken string
|
|
|
- Locale 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
|
|
|
+ pool *postgres.Pool
|
|
|
+ logger *slog.Logger
|
|
|
}
|
|
|
|
|
|
// New constructs a Resolver.
|
|
|
-func New(pool *postgres.Pool) *Resolver { return &Resolver{pool: pool} }
|
|
|
+func New(pool *postgres.Pool, logger *slog.Logger) *Resolver {
|
|
|
+ return &Resolver{pool: pool, logger: logger}
|
|
|
+}
|
|
|
|
|
|
-// ResolveTokens returns every active FCM token for every active
|
|
|
-// individual in the given company. This is the M1 broadcast path.
|
|
|
-// M2 swaps this for the rules engine.
|
|
|
+// ResolveTargets returns every (individual, channel, endpoint) tuple
|
|
|
+// that should receive the alert. M2 contract:
|
|
|
//
|
|
|
-// One DB round-trip via a single join. Add a per-company limit
|
|
|
-// here if a malicious company can ever register 1M tokens.
|
|
|
-func (r *Resolver) ResolveTokens(ctx context.Context, companyID string) ([]Target, error) {
|
|
|
- rows, err := r.pool.Query(ctx, `
|
|
|
- SELECT i.id, t.token, COALESCE(t.locale, i.locale, 'en')
|
|
|
- FROM individuals i
|
|
|
- JOIN fcm_tokens t ON t.individual_id = i.id
|
|
|
- WHERE i.company_id = $1
|
|
|
- AND i.status = 'active'
|
|
|
- AND t.status = 'active'
|
|
|
- ORDER BY i.id, t.id
|
|
|
- `, companyID)
|
|
|
+// - 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 tokens: %w", err)
|
|
|
+ return nil, fmt.Errorf("resolve targets: %w", err)
|
|
|
}
|
|
|
defer rows.Close()
|
|
|
|
|
|
+ now := time.Now().UTC()
|
|
|
var out []Target
|
|
|
for rows.Next() {
|
|
|
var t Target
|
|
|
- if err := rows.Scan(&t.IndividualID, &t.FCMToken, &t.Locale); err != nil {
|
|
|
+ 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)
|
|
|
}
|
|
|
- return out, rows.Err()
|
|
|
+ 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
|
|
|
}
|
Luis Rosales