broad-announce/ARCHITECTURE.md

Broad-Announce — Architecture

Companion to SPEC.md. This doc is the "how does it actually work" view. See SPEC for requirements, entities, and the why.

1. System context

flowchart LR
    subgraph S[Sources]
        WH[Webhook / HTTP]
        WS[WebSocket clients]
        MQ[MQTT publishers]
        GR[gRPC bidi-stream]
    end
    subgraph BA[Broad-Announce]
        IG[ingestd]
        BR[(NATS JetStream)]
        RT[routerd]
        DV[deliverd]
        AD[admind]
    end
    subgraph DATA[Data tier]
        PG[(Postgres + Timescale)]
        CH[(ClickHouse archive)]
        RD[(Redis)]
    end
    subgraph OUT[Delivery sinks]
        FCM[FCM / Android]
        TG[Telegram]
        SMS[SMS / Voice]
        EM[Email]
        SL[Slack]
        MT[MS Teams]
        WH2[Custom outbound webhooks]
    end
    WH --> IG
    WS  --> IG
    MQ  --> IG
    GR  --> IG
    IG  --> RD
    IG  --> BR
    BR  --> RT
    RT  --> PG
    RT  --> BR
    BR  --> DV
    DV  --> FCM
    DV  --> TG
    DV  --> SMS
    DV  --> EM
    DV  --> SL
    DV  --> MT
    DV  --> WH2
    PG  -- archive job --> CH
    AD  --> PG
    AD  --> BR
    AD  --> DV

2. Sequence: a single alert through the system

sequenceDiagram
    autonumber
    participant Src as Source system
    participant IG as ingestd
    participant RD as Redis
    participant BR as NATS JetStream
    participant RT as routerd
    participant PG as Postgres
    participant DV as deliverd
    participant FCM as FCM

    Src->>IG: POST /v1/ingest (HMAC-signed, JSON)
    IG->>IG: Verify HMAC, validate schema
    IG->>RD: SET dedupe:{key} 1 NX EX 60
    alt first arrival
        RD-->>IG: OK
        IG->>BR: publish alerts.<company_id> {Alert, dedupe_count=1}
        IG-->>Src: 202 Accepted
    else duplicate
        RD-->>IG: nil
        RD->>RD: INCR dedupe:{key}  → n
        IG->>BR: publish alerts.<company_id> {Alert, dedupe_count=n}
        IG-->>Src: 202 Accepted (deduped)
    end

    BR->>RT: deliver Alert
    RT->>PG: SELECT recipients (batch, prepared)
    PG-->>RT: individuals + channels
    loop per (individual, channel)
        RT->>BR: publish deliveries.<channel>.<company_id>
    end

    BR->>DV: deliver job
    DV->>FCM: POST /v1/projects/.../messages:send
    alt success
        FCM-->>DV: 200, message_name
        DV->>PG: UPDATE deliveries SET status='sent'
    else transient error
        DV->>DV: schedule retry (exp backoff)
    else non-retryable
        DV->>BR: publish dlq.<channel>.<company_id>
    end

3. gRPC ingest (M11)

sequenceDiagram
    autonumber
    participant Src as Internal service
    participant GS as ingestd (gRPC :9090)
    participant RD as Redis
    participant NATS as NATS JetStream

    Src->>GS: open bidi stream (metadata: authorization=Bearer <api_key>)
    GS->>GS: verify API key, attach source_id
    loop
      Src->>GS: Alert{company_id, source_id, severity, ...}
      GS->>GS: schema validate
      GS->>RD: SET dedupe:{key} 1 NX EX 60
      alt first
        RD-->>GS: OK
        GS->>NATS: publish alerts.<company_id>
        GS-->>Src: Ack{alert_id, dedupe_count=1}
      else dup
        RD-->>GS: nil
        RD->>RD: INCR → n
        GS->>NATS: publish alerts.<company_id> {dedupe_count=n}
        GS-->>Src: Ack{alert_id, dedupe_count=n}
      end
      opt rate-limited
        GS-->>Src: Ack{error: RATE_LIMITED, retry_after_ms=1000}
      end
    end
    Src-->>GS: half-close (writes done)
    GS-->>Src: EOF

Key invariants:

• Stream is the unit of auth + rate-limit (one stream = one source).
• Per-stream in-flight cap = 256 → natural backpressure when the source outpaces us.
• After M11, gRPC handles the same dedupe_count contract as HTTP.

4. Recipient resolution algorithm

flowchart TD
    A[Alert arrives in routerd] --> B{Resolve targets}
    B --> C[source.allowed_targets]
    C --> D{routing_rules overrides?}
    D -- yes --> E[apply rules in priority order]
    D -- no  --> F[use source targets]
    E --> G[expand groups → members]
    F --> G
    G --> H{active subscription?}
    H -- no --> X[skip]
    H -- yes --> I{severity ≥ sub.min_severity?}
    I -- no --> X
    I -- yes --> J{in quiet_hours AND severity != inminent_colapse?}
    J -- yes --> X
    J -- no --> K[emit one Delivery per channel in sub.channel_mask]

5. Storage topology

flowchart LR
    APP[App services] --> PG[(Postgres 16)]
    PG --- TS[(Timescale 2.x extension)]
    TS --- HG1[hypertables: alerts, deliveries<br/>7d retention]
    PG --- CORE[regular tables: companies,<br/>individuals, groups, fcm_tokens,<br/>sources, subscriptions, bots]
    APP --> RD[(Redis 7)]
    RD --- DEDUPE[dedupe keys, TTL 60s]
    RD --- RL[rate-limit token buckets]
    PG -- nightly cron --> CH[(ClickHouse 24.x)]
    CH --- AR1[alerts_archive]
    CH --- AR2[deliveries_archive]
    CH --- AGG[materialized aggregates<br/>per-company, per-channel]

6. NATS subject layout

subject	producer	consumer	retention
alerts.<company_id>	ingestd	routerd	24h (JetStream stream ALERTS)
deliveries.fcm.<company_id>	routerd	deliverd-fcm	1h
deliveries.telegram.<company_id>	routerd	deliverd-telegram	1h
deliveries.sms.<company_id>	routerd	deliverd-sms	1h
deliveries.email.<company_id>	routerd	deliverd-email	1h
deliveries.slack.<company_id>	routerd	deliverd-slack	1h
deliveries.teams.<company_id>	routerd	deliverd-teams	1h
deliveries.webhook.<company_id>	routerd	deliverd-webhook	1h
dlq.<channel>.<company_id>	deliverd-*	(operator)	7d

Subject-based partitioning keeps a single company's alert stream in order (mostly) and lets us add per-company worker affinity in K8s later.

7. FCM delivery paths

flowchart TD
    A[deliverd-fcm job] --> B{group_fcm_topic set<br/>AND group size > 50?}
    B -- yes --> C[POST to /topics/&lt;topic&gt; : send<br/>1 FCM call]
    B -- no --> D[For each device token in target]
    D --> E[POST : send<br/>1 FCM call per token]
    C --> F[log delivery aggregate only]
    E --> G[log per-token delivery report]

FCM payload shape

{
  "message": {
    "token": "<device_token>",
    "notification": {
      "title": "Disk full on db-prod-03",
      "body": "92% used (×12 in 60s)"
    },
    "data": {
      "company_id": "acme",
      "alert_id": "01HXYZ...",
      "severity": "critical",
      "category": "storage",
      "dedupe_count": "12",
      "deep_link": "broadannounce://alert/01HXYZ"
    },
    "android": {
      "priority": "HIGH",
      "notification": {
        "sound": "siren_storage",
        "channel_id": "alerts.critical"
      }
    }
  }
}

The Android app reads data.category and data.severity to pick the right sound + channel.

8. Telegram bot surface

sequenceDiagram
    participant U as User
    participant TG as Telegram
    participant BOT as deliverd-telegram
    participant BR as NATS
    participant RT as routerd
    participant PG as Postgres

    U->>TG: /start INVITE_CODE
    TG->>BOT: update (chat_id, from.id, text=invite_code)
    BOT->>PG: SELECT individual WHERE invite_code=?
    Note over BOT,PG: Individual must pre-exist;
    Note over BOT,PG: admin creates row + invite code;
    Note over BOT,PG: bot matches and burns code
    BOT->>PG: UPDATE individual SET telegram_chat_id=...
    BOT->>TG: sendMessage "Linked to NAME"

    U->>TG: /mute 2h
    TG->>BOT: update
    BOT->>PG: UPDATE subscriptions SET mute_until = now()+2h

    Note over RT,PG: New alert arrives
    RT->>PG: resolve recipients incl. telegram
    RT->>BR: deliveries.telegram.COMPANY_ID
    BR->>BOT: job
    BOT->>TG: sendMessage(chat_id, text, reply_markup)

    U->>TG: taps [Acknowledge]
    TG->>BOT: callback_query
    BOT->>BR: ack.ALERT_ID.INDIVIDUAL_ID
    BR->>RT: close alert (or notify source)

9. Failure & retry

layer	failure mode	behavior
ingestd	source 5xx storm	rate-limit per source, return 429
ingestd	broker down	503 to source, source retries
routerd	DB down	nack to broker, message redelivered (no ack until DB write)
deliverd	third-party 5xx	exp backoff, max 10, then DLQ
deliverd	third-party 4xx (non-retryable)	straight to DLQ
FCM	token unregistered	mark fcm_tokens.status='unregistered', skip on next send
Telegram	chat not found	mark individuals.telegram_status='revoked', skip

10. SLOs

SLI	target	measured at
Ingest availability	99.9%	ingress LB
Ingest latency p99	≤ 100ms	ingestd
End-to-end latency p99	≤ 5s	accept → device wake
Delivery success rate	≥ 99.5% (excluding DLQ)	deliverd
DLQ rate	< 0.5% of deliveries	deliverd
Per-company data isolation	100% (no cross-tenant queries possible)	audit + e2e test

11. Capacity model (back-of-envelope)

Assumption: 50k alerts/sec peak, avg fan-out 10 recipients × 2 channels = 20 deliveries per alert.

• ingestd: 1 alert ≈ 200µs (validate + Redis + NATS publish). 4 cores handle 20k/s. → 12 replicas for 50k/s with headroom.
• routerd: recipient expansion is the hot path. 1 expansion ≈ 2ms (one batched query). 4 cores ≈ 2k/s per replica. → 25 replicas for 50k/s × 2ms ≈ 5s p99.
• deliverd-fcm: FCM HTTP v1 p95 ~150ms. Per worker, ~6 msg/s serialised, or 60 msg/s with 10 in-flight. 1M deliveries/min needs ~280 workers. We budget 500.
• deliverd-telegram: TG API p95 ~100ms. ~10 msg/s serialised per worker. 1M/min → 1700 workers. We budget 2000.

Total: ~2700 service processes for peak. Docker Compose can't carry this — that is exactly the SLO that forces K8s in v2.

For v1 we target 5k alerts/sec sustained in the docker-compose profile, and gate "50k/s" on the K8s + multi-broker milestone.

12. Security model

flowchart LR
    subgraph perimeter
        LB[TLS 1.2+ LB]
    end
    subgraph mTLS[mTLS optional per source]
        M1[client cert verify]
    end
    subgraph auth[Auth]
        K1[API key in X-BA-Key<br/>argon2id at rest]
        K2[HMAC-SHA256<br/>Stripe-style X-BA-Signature]
        K3[JWT for WS clients]
    end
    subgraph data[At rest]
        S1[AES-256-GCM<br/>BA_MASTER_KEY]
    end
    LB --> mTLS --> auth
    auth --> APP[services]
    APP --> data
    APP --> PG
    APP --> CH
    APP --> RD

13. What changes when we move to K8s

• ingestd: HPA on CPU + custom metric ingestd_queue_depth
• routerd: HPA on CPU + custom metric routerd_pending_messages
• deliverd: HPA per channel, custom metric deliverd_inflight
• NATS JetStream: 3-node cluster, replicas=3, ack-quorum=2
• Postgres: managed (RDS / Cloud SQL) + 1 read replica for archive
• Redis: managed (ElastiCache / Memorystore), cluster mode off for v1
• ClickHouse: 2 shards, 1 replica, Zookeeper-less (CH 24 supports it)
• MQTT broker: EMQX cluster, 3 nodes, per-company topic ACLs

For v1 we keep all of the above out of scope and run a single node of each on a beefy docker-compose host.