# Broad-Announce — Project Spec (v1)

> Multi-tenant notification router. Receives alerts from many source systems,
> normalizes them, routes them to the right people (companies → groups →
> individuals), and delivers via FCM, Telegram, SMS, email, voice, Slack,
> Teams, and custom webhooks.

## 1. Goals (v1)

- Accept alerts via **HTTP POST (JSON)**, **WebSockets (client→server)**,
  **MQTT**, and **gRPC bidi-streaming** (internal high-volume sources).
- Normalize every inbound alert to a single internal `Alert` shape.
- Resolve recipients by querying `companies` / `groups` / `individuals` tables
  + per-user `subscriptions` (opt-in per source/severity).
- Deliver to recipients via **FCM** (primary), **Telegram** (primary),
  **SMS**, **email**, **voice call**, **Slack**, **MS Teams**, and
  **generic outbound webhook**.
- Sustain **~50,000 alerts/sec** with **≤5s p99** end-to-end latency
  (accept → device wake).
- Multi-tenant: **~10,000 companies**, up to **1,000 devices per company**.
- Strict data isolation: a company can never see another company's alerts,
  recipients, or delivery logs.

## 2. Non-goals (v1)

- No inbound email parsing, no SMS shortcode receivers, no phone IVR trees.
- No on-prem appliance mode. We deploy as Docker Compose first, K8s later.
- No cross-company fan-out except an explicit `dev-team` supergroup.
- No iOS APNs in v1 (FCM only). APNs comes in v2.

## 3. Multi-tier architecture

```
[ Sources ]
   │  HTTP POST   (webhook)        ── HMAC-SHA256, API key, optional mTLS
   │  WebSocket   (client push)    ── JWT, TLS
   │  MQTT        (broker sub)     ── per-company topic, QoS 1
   │  gRPC bidi-stream             ── API key in metadata, optional mTLS
   │                                  (internal high-volume sources only)
   ▼
[ INGEST TIER ]  (ingestd, stateless, N replicas behind LB)
   │  - AuthN/Z (API key / mTLS / JWT)
   │  - Rate limit (Redis token bucket per source)
   │  - Schema validate (per source type)
   │  - Normalize → Alert v1
   │  - Dedupe check (Redis SET NX, TTL 60s, key=hash(source|dedupe_key))
   │  - Stamp dedupe_count on alert if already seen
   │  - Publish to broker (NATS JetStream, subject: alerts.<company_id>)
   ▼
[ BROKER ]  NATS JetStream
   │  - Persistent stream `ALERTS` (replicas=3, retention 24h)
   │  - Per-company subjects for partition affinity
   ▼
[ ROUTER TIER ]  (routerd, stateless, N replicas, JetStream consumer)
   │  - Consume alert
   │  - Resolve recipients:
   │       companies → groups → individuals
   │       ∩ subscriptions (opt-in, severity ≥ user.min_severity)
   │       ∩ quiet_hours (unless severity == "inminent_colapse")
   │  - For each recipient, enqueue 1..N Delivery jobs to:
   │       deliveries.<channel>.<company_id>  (channel-specific subjects)
   ▼
[ DELIVERY TIER ]  (deliverd, one worker pool per channel)
   │  - FCM: shared FCM project, per-token sends, FCM topic for groups>50
   │  - Telegram: per-company bot token, message + inline keyboard
   │  - SMS / Voice: pluggable provider (Twilio v1)
   │  - Email: SMTP relay or SES
   │  - Slack / Teams: incoming webhook URLs
   │  - Custom outbound webhook: HMAC sign with company key
   │  - Retry: exponential backoff (1s, 2s, 4s, 8s, 16s, 32s, 64s, 128s, 256s, 512s)
   │           max 10 attempts, then → Dead-Letter Queue (DLQ)
   ▼
[ DATA TIER ]
   - PostgreSQL: tenants, sources, companies, groups, individuals,
                 subscriptions, telegram_bots, fcm_tokens, routing_rules
   - TimescaleDB: alerts (hypertable, 7d retention), deliveries
                 (hypertable, 7d retention)
   - ClickHouse: alerts_archive, deliveries_archive, aggregates
                 (indefinite, columnar, cheap to query)
   - Redis: dedupe set, rate-limit counters, idempotency keys
```

### Why 4 tiers instead of 1 monolith

- **Ingest** scales with incoming HTTP/WS/MQTT concurrency; it's CPU-cheap
  but connection-heavy. Independent replicas behind an L7 LB.
- **Router** is CPU + DB heavy (recipient expansion can be 1 → 10,000s). It
  must not be on the request path of the source system — once accepted,
  the alert is in the broker and the source is acked.
- **Delivery** is the slowest tier (third-party API calls). Each channel is
  a separate worker pool with its own concurrency, timeouts, and circuit
  breaker.
- **Broker** is the buffer. If delivery falls behind, alerts queue up
  instead of timing out the source.

## 4. Entities (Postgres schema sketch)

```sql
-- Tenancy
companies(id, name, slug, status, fcm_shared, telegram_bot_token_enc,
          rate_limit_per_sec, created_at)

-- People & devices
individuals(id, company_id, full_name, email, phone_e164, locale, tz,
            telegram_chat_id, telegram_user_id,
            telegram_invite_code, telegram_invite_expires_at,
            status, created_at)
fcm_tokens(id, individual_id, token, device_id, platform, locale,
           app_version, last_seen, status, created_at)
groups(id, company_id, name, description, fcm_topic, telegram_chat_id,
       created_at)
group_members(group_id, individual_id, added_at, added_by)

-- Source allowlist
sources(id, company_id, name, type, hmac_secret_enc, api_key_hash,
        mtls_required, rate_limit_per_sec, allowed_ips cidr[],
        topic_prefix, created_at)

-- Routing
subscriptions(id, individual_id, source_id, min_severity,
              channel_mask, quiet_hours_start, quiet_hours_end, tz)
routing_rules(id, company_id, priority, match_expr, target_type,
              target_id, enabled)

-- Normalized alert (Timescale hypertable)
alerts(id, company_id, source_id, severity, category, title_key,
       body_key, data jsonb, dedupe_key, dedupe_count, received_at)
-- hypertable on received_at, 7d retention, then COPY to ClickHouse

-- Delivery attempts (Timescale hypertable)
deliveries(id, alert_id, individual_id, channel, target, status,
           attempts, last_error, next_attempt_at, completed_at)
-- hypertable on completed_at (NULL = in-flight), 7d retention
```

### Severity taxonomy (used everywhere)

| severity | sound | bypasses_quiet_hours | fcm_priority | example |
|---|---|---|---|---|
| `info` | default | no | normal | "deploy finished" |
| `warning` | alert | no | normal | "disk 80% full" |
| `critical` | siren | no | high | "service down" |
| `inminent_colapse` | klaxon | **yes** | high | "rack PDU overload imminent" |

`category` is free-form per source (e.g. `network`, `power`, `security`,
`hvac`, `process`). The Android app uses `category` to pick a custom sound
file (`res/raw/siren_<category>.ogg`).

## 5. Dedupe + dedupe_count

When `ingestd` receives alert with `dedupe_key`:

```
key = hash(source_id || ':' || dedupe_key)
SET key 1 NX EX 60       -- first arrival creates a 60s window
```

- If `SET` succeeded → first arrival, publish as new alert.
- If `SET` failed → existing window. `INCR` the counter on the key and
  attach `dedupe_count` to the published alert payload. The recipient
  sees e.g. *"Disk full (×12 in 60s)"* instead of 12 separate pushes.

The dedupe window default is 60s and is configurable per source.

## 6. Recipient resolution (the heart of the router)

For an inbound alert:

```
recipients = ∅
for each target (group or individual) implied by source.allowed_targets:
    if routing_rule matches alert.tags → override targets
    individuals = expand(target)   -- groups → members
    for each individual:
        sub = active_subscription(individual, source, alert.severity)
        if not sub: skip
        if in_quiet_hours(sub, now, alert.severity): skip
                     -- severity=inminent_colapse never skipped
        for each channel in sub.channel_mask:
            recipients += (individual, channel)
```

The router uses **prepared statement + batch IN-list** to avoid N+1.

For groups > 50 members, the router uses **FCM topic messaging**
(`/topics/<group_fcm_topic>`) which is one FCM API call regardless of
subscriber count. For smaller groups, it uses per-token sends.

## 7. FCM (shared project model)

- **One FCM project** for all companies. App must be multi-tenant aware
  (the Android app reads `data.company_id` and renders a "company badge").
- Per-individual **device tokens** registered by the Android app on login.
- Per-group **FCM topic** if group size > 50.
- Quota: FCM HTTP v1 default is 600k msg/min to a project, 1k msg/min per
  token recipient is the realistic soft cap. At 50k alerts/sec, even
  bursty, we stay well under if we use **topics for big groups** and
  **dedupe** to collapse storms. If a single company ever exceeds
  per-project quota, we can **shard into a second FCM project** —
  schema already supports `companies.fcm_shared=false`.

### 10k companies × 1k devices = 10M tokens

A single FCM project can hold this. Storage is on our side (the
`fcm_tokens` table), not FCM's. We shard the table by `company_id % N`
(partition by hash) for vacuum/index perf.

## 8. Telegram (primary channel for ops)

- Each company registers one or more **Telegram bots** (`@company_x_bot`).
  Token stored encrypted at rest (`telegram_bot_token_enc`).
- For 1:1 alerts we need the user's `telegram_chat_id` (resolved when
  user runs `/start` in the bot).
- For group alerts we use the bot in a company group chat
  (`telegram_chat_id` on the `groups` table).
- **Bot commands** (Telegram = management UI):
  - `/start <invite_code>` — link telegram account to the
    pre-existing `individuals.id` (admin must create the
    individual and issue the code first; unknown users are
    rejected)
  - `/mute <duration>` — e.g. `/mute 2h`, `/mute until 18:00`
  - `/unmute`
  - `/subscribe <source> <min_severity>`
  - `/unsubscribe <source>`
  - `/preferences` — show current subscriptions
  - `/status` — last N alerts received
- Telegram is bidirectional: users reply to acknowledge; ack updates
  `deliveries.status` and (optionally) closes the alert via a webhook
  callback to the source system.

## 9. Delivery retry & DLQ

```
attempt 1, fail  → wait 1s
attempt 2, fail  → wait 2s
...
attempt 10, fail → publish to dlq.<channel>.<company_id>
```

DLQ is a **JetStream stream** with no consumer in normal operation, plus
a **DLQ table** in Postgres for the admin UI. An operator can:
- View failed deliveries, with full payload + last error
- Replay (re-enqueue) one or many
- Discard

`deliverd` exposes `/health` and Prometheus metrics
(`deliverd_delivery_attempts_total{channel,status}`,
`deliverd_delivery_latency_seconds_bucket{channel}`).

## 10. Observability (operate from day 1)

- **Prometheus** metrics on every service:
  - `ingestd_alerts_received_total{source_type,result}`
  - `ingestd_dedupe_hits_total`
  - `routerd_recipient_expansion_seconds`
  - `deliverd_delivery_attempts_total{channel,status}`
  - `deliverd_dlq_total{channel}`
- **Structured logs** (slog JSON) with `trace_id`, `company_id`,
  `alert_id`, `individual_id`.
- **OpenTelemetry traces** across ingest → broker → router → delivery
  (propagate `traceparent` through NATS headers).
- **Grafana dashboards**: per-company drill-down, channel health, DLQ
  rate, p50/p95/p99 latency.

## 11. Security

- Secrets at rest: FCM service account JSON, Telegram bot tokens, HMAC
  secrets → AES-256-GCM with key from env (`BA_MASTER_KEY`).
- TLS 1.2+ everywhere; mTLS optional for HTTP sources.
- Per-source HMAC: scheme = **Stripe-style** (header
  `X-BA-Signature: t=<unix_ts>,v1=<hex_hmac>` over
  `timestamp.body`, with 5-min replay window). GitHub-style is fine too
  (header `X-Hub-Signature-256: sha256=...`) but Stripe's timestamp
  prevents replay; we ship that one as the default and accept
  GitHub-style as a per-source option.
- API keys: argon2id hashed at rest.
- Postgres row-level security optional (off by default; tenant filter
  is enforced in app code paths).

## 12. Data retention

- **alerts** (Timescale): 7 days, then COPY → ClickHouse
  `alerts_archive`, then drop chunk.
- **deliveries** (Timescale): 7 days, then COPY → ClickHouse
  `deliveries_archive`, then drop chunk.
- **companies / individuals / fcm_tokens / groups / subscriptions**:
  indefinite in Postgres. GDPR right-to-be-forgotten deletes the
  individual + cascades fcm_tokens + subscriptions.
- **ClickHouse**: indefinite. Partition by month, TTL optional later.

## 13. Region / data residency (Q24)

> "Don't understand this point, can you elaborate?"

This is about **where the data lives** and **who has to obey whose laws**:

- **Single region** (e.g. one AWS `us-east-1`): cheapest, simplest, all
  customers served from one data center. Fine if your 10k companies
  don't have legal requirements to keep data in a specific country.
- **Multi-region active-passive** (us-east-1 primary, eu-west-1
  standby): you can failover, but EU customers' data still passes
  through the US at some point. Needed if you want to claim EU
  residency for marketing.
- **Multi-region active-active** (every region runs a full stack,
  broker replicated, Postgres logical-replicated): each company pinned
  to one region. Expensive, complex, but lets you say "company X's
  alerts never leave Frankfurt."

For v1: **single region**, but pick the FCM endpoint that matches
(e.g. `fcm.googleapis.com` vs `fcm.googleapis.eu`). Document the
limitation; revisit at 1k paying companies.

## 14. Compliance (Q25)

> "Also don't know the implications of this."

Three things that matter at 10k companies:

1. **GDPR / right to be forgotten**: a `DELETE /individuals/:id` must
   hard-delete the row + cascade fcm_tokens + subscriptions. Keep a
   tombstone (id, deleted_at) in a separate `erased_individuals` table
   for 30 days so a re-created account doesn't double-deliver. ClickHouse
   archives get the individual's `id` zeroed (we keep analytics counts
   but no PII).
2. **Data Processing Agreement** (DPA) with each customer: required by
   GDPR. We are the processor; they are the controller.
3. **Encryption at rest + in transit** (already in §11). SOC2 auditors
   care; GDPR does too indirectly.

If any customer is in healthcare/finance, add HIPAA/SOC2 controls later
(audit log, BAAs). Not v1.

## 15. mTLS vs HMAC vs API key (Q8)

| mechanism | what it proves | who needs it | v1 |
|---|---|---|---|
| **API key** (`X-BA-Key: <opaque>`) | "the caller knows the secret" | any HTTP source | ✅ default, always on |
| **HMAC signature** | "the caller has the secret AND the body wasn't tampered with" | higher-trust sources, public webhooks | ✅ default-on, Stripe-style |
| **mTLS** | "the caller controls this exact client cert" | enterprise on-prem, very high trust | ✅ optional per source |

Recommendation: **API key + HMAC for everyone, mTLS opt-in for sources
that want it.** API key alone is fine for trusted internal sources. HMAC
alone is the public-webhook norm. mTLS is overkill for v1 unless a
customer asks.

## 16. MQTT (Q9) — is it the right choice?

| protocol | pros | cons | fit |
|---|---|---|---|
| **MQTT** | tiny, perfect for IoT, persistent sessions, QoS guarantees | broker to operate, clients must speak MQTT | great for sensor gateways, PLCs, on-prem devices |
| **WebSocket ingest** | no broker, plain TLS, works from any browser/curl | we have to do the reconnect logic | great for browser dashboards, ops consoles |
| **HTTP POST** | dumb simple, every device can do it | no push back, no auto-reconnect magic | great for batch sources (Prometheus, Grafana) |

**Verdict**: support all three. MQTT is the **right** choice for IoT
sources (cameras, sensors, PLCs) but not the only one. We'd use
**EMQX** (or Mosquitto) as the broker.

**Per-company topic prefix**: yes — `ba/<company_slug>/<source_id>/...`
so a misbehaving publisher can be ACL'd to its prefix only.

**QoS 1 vs 2**:
- **QoS 1** (at-least-once): broker acks on receipt, message may be
  delivered twice on reconnect. **Default for alerts.** With our dedupe
  layer this is fine.
- **QoS 2** (exactly-once): 4-step handshake, slower, ensures no dup.
  Overkill given our 60s dedupe window. Skip.

## 17. FCM topic vs per-token fan-out (Q14)

| | FCM topic messaging | Per-token send |
|---|---|---|
| 1 push to N devices | **1 FCM API call** | N FCM API calls |
| Latency at 1000 devices | ~200ms | ~30s with batching |
| Subscriptions dynamic | user subs/unsubs via app | we manage the list |
| Delivery report | aggregate only | per-token |

**Rule**: if a group has > 50 members and is the primary target,
**use FCM topic messaging**. Otherwise, per-token is fine and gives
us per-device delivery reports.

## 18. Stack

- **Language**: Go 1.22+
- **DB**: PostgreSQL 16 + TimescaleDB 2.x (active window), ClickHouse 24.x
  (archive)
- **Broker**: NATS JetStream (single-node for v1, 3-node later)
- **Cache / dedupe / rate-limit**: Redis 7
- **MQTT broker**: EMQX 5.x (Docker image)
- **Observability**: Prometheus + Grafana + Loki + OTel collector
- **Deployment**: Docker Compose (v1), Kubernetes manifests (v2)
- **Migrations**: `golang-migrate`
- **No-code admin UI** (later): React + Vite, served by `admind`

### Port conventions (project rule)

- **App HTTP services** (ingestd, routerd, deliverd, admind, loadgen,
  future services): **8800–8899**. Service picks a slot in order of
  addition: ingestd 8800, routerd 8801, deliverd 8802, admind 8803,
  loadgen 8891. New services pick the next free number.
- **Canonical ports stay**: 5432 postgres, 4222 nats, 8222 nats
  monitor, 6379 redis, 1883 mqtt, 18083 emqx admin, 8123 clickhouse
  http, 9000 clickhouse native, 9090 prometheus, 3000 grafana.
- Why: 8080+ collides with half the dev tooling on a workstation
  (airplay, jenkins, …). The 8800 band is wide enough that
  local-dev services, sidecar debuggers, and test runners all
  have headroom.

## 19. gRPC ingest (internal high-volume sources, M11)

In addition to HTTP/WS/MQTT, `ingestd` exposes a gRPC service on a
separate port (default `:9090`) for **first-party, persistent, typed**
sources. Use cases: other Go services inside our infra that need to
push thousands of alerts/sec with strict schemas and backpressure.

### Why a fourth protocol

- Strongly-typed `.proto` schemas — no JSON drift between producers
- HTTP/2 + protobuf — smaller payloads, faster serialize at 50k/s
- Native bidi streaming — no reconnect-loop code in the source
- Built-in deadlines / cancellation / metadata (auth tokens, trace ids)
- Backpressure is explicit (client-driven): a slow source can't OOM us
- Generated stubs for Go/Java/Python/Node — easy for other teams

### What it is *not*

- Not for **public SaaS webhooks** (Grafana, Stripe, Datadog …) — they
  speak HTTP. Those keep using HTTP POST.
- Not for **browsers** — gRPC needs grpc-web + Envoy. The dashboard
  story stays on WebSocket.
- Not for **IoT / PLCs** — MQTT is already the right answer there.

### API surface

```proto
syntax = "proto3";
package broadannounce.v1;

service Ingest {
  // Bidi stream: client pushes Alerts, server pushes Acks.
  rpc StreamAlerts(stream Alert) returns (stream Ack);
}

message Alert {
  string company_id     = 1;
  string source_id      = 2;
  string severity       = 3;   // info | warning | critical | inminent_colapse
  string category       = 4;
  string title          = 5;   // pre-localized
  string body           = 6;   // pre-localized
  map<string, string> data = 7;
  string dedupe_key     = 8;
  int64  client_ts_ms   = 9;
}

message Ack {
  string alert_id       = 1;
  string dedupe_key     = 2;
  uint32 dedupe_count   = 3;   // 1 = first arrival, >1 = collapsed burst
  int64  accepted_at_ms = 4;
  oneof result {
    Ok      ok      = 10;
    Error   error   = 11;
  }
}

message Ok    {}
message Error {
  enum Code {
    UNKNOWN = 0;
    UNAUTHENTICATED = 1;
    RATE_LIMITED    = 2;
    INVALID         = 3;
    INTERNAL        = 4;
  }
  Code code           = 1;
  string message      = 2;
  int32  retry_after_ms = 3;   // 0 = do not retry
}
```

### Auth

- **API key in metadata**: `authorization: Bearer <api_key>` (same
  hashing as HTTP sources). Always on.
- **mTLS**: optional, same opt-in semantics as HTTP sources.
- **Per-source `type='grpc'`** row carries the rate limit and HMAC
  secret if any. gRPC does **not** use HMAC — TLS + API key is enough
  inside our infra.

### Server-side behavior

- One stream = one source. Stream-level rate limit
  (`sources.rate_limit_per_sec`) applied as a sliding window; per-message
  backpressure by sending `Error.RATE_LIMITED` with `retry_after_ms`.
- Per-message flow is identical to HTTP: validate → dedupe (Redis) →
  publish to NATS `alerts.<company_id>`. The `Ack` carries the same
  `dedupe_count` contract.
- Max in-flight messages per stream: 256 (server-side). Source gets
  flow-controlled when the buffer fills — it should stop sending until
  it receives Acks.
- Stream keepalive: 30s ping, 10s ack timeout. Dead streams are closed.
- All existing observability (Prometheus metrics, OTel traces,
  structured logs) applies; add `source.transport="grpc"` label.

### Repo additions for gRPC

- `proto/broadannounce/v1/ingest.proto` — checked in
- `internal/grpcserver/` — server skeleton on `ingestd`
- `internal/grpcclient/` — reusable Go client lib for other services
- `gen/go/broadannounce/v1/` — generated stubs (via `buf generate`)
- `Makefile` target `proto` running `buf generate`
- `docs/sources/grpc.md` — quickstart for an internal service

### Milestone

Landed as **M11**, after M10 (load test). M11 exit criterion:
internal Go service can publish ≥ 10k alerts/sec on one stream
sustained 10 min, p99 server-side `Ack` latency ≤ 50ms.

## 20. Repo layout (planned)
```
broad-announce/
├── SPEC.md                ← this file
├── ARCHITECTURE.md        ← diagrams, sequence flows, SLOs
├── PROMPT.md              ← build log, decisions, lessons
├── README.md
├── docker-compose.yml     ← one-shot local stack
├── .env.example
├── migrations/            ← sql migrations (golang-migrate)
├── ingestd/               ← binary: HTTP/WS/MQTT ingest
├── routerd/               ← binary: recipient expansion
├── deliverd/              ← binary: per-channel delivery workers
├── admind/                ← binary: admin HTTP API + (later) UI host
├── loadgen/               ← (coming) traffic generators per protocol
├── internal/
│   ├── alert/             ← Alert v1 type, validation
│   ├── broker/            ← NATS wrapper
│   ├── store/             ← Postgres + Timescale + ClickHouse
│   ├── dedupe/            ← Redis dedupe
│   ├── ratelimit/         ← Redis token bucket
│   ├── fcm/               ← FCM HTTP v1 client + topic mgmt
│   ├── telegram/          ← Bot client, command parser
│   ├── retry/             ← exponential backoff + DLQ
│   ├── observability/     ← slog + OTel + Prometheus
│   └── config/            ← env-driven config
└── deploy/
    ├── prometheus/
    ├── grafana/
    └── loki/
```

## 21. Traffic generators (loadgen)

Four small Go binaries in `loadgen/` — one per source protocol — that
behave as **reference producers** of alerts. Same wire format, same
auth, same client libs an internal service would use. Copy the
binaries to as many hosts as you want, point them at a target, and
they generate load.

### Goals

- **Realistic traffic** ("normal"): matches the shape of real-world
  sources — bursty, dedupe-heavy, mix of severities, mix of
  company_ids.
- **Stress traffic** ("stress"): maximum sustained rate, no dedupe,
  fat payloads, with fault-injection knobs (drop %, latency spikes,
  random disconnects).
- **Distributed by design**: any number of `loadgen` instances on
  any number of hosts coordinate via a tiny shared state in
  NATS JetStream KV (`loadgen.coordination`). No single host is
  in charge.
- **Self-contained**: no separate deploy, no extra observability
  stack. Emit Prometheus metrics on `:9091`, structured logs to
  stdout.

### Binaries

| binary | protocol | client lib | what it generates |
|---|---|---|---|
| `loadgen-http` | HTTP POST | `internal/httpclient/` | signed webhooks, configurable concurrency, N concurrent connections |
| `loadgen-ws` | WebSocket | `internal/wsclient/` | one long-lived socket per instance, alert frames every X ms |
| `loadgen-mqtt` | MQTT | `internal/mqttclient/` | QoS 1, per-company topic, configurable in-flight |
| `loadgen-grpc` | gRPC bidi | `internal/grpcclient/` | one stream per instance, batched send with bounded in-flight |

### CLI surface (same shape across all four)

```bash
loadgen-http \
  --target https://broad-announce:8800 \
  --api-key $SOURCE_API_KEY \
  --mode normal | stress \
  --rate 1000               # alerts/sec target per instance
  --duration 10m \
  --company-prefix acme-    # company_ids are "acme-001" .. "acme-100" cycled
  --severities info:70,critical:5,inminent_colapse:1 \
  --dedupe-ratio 0.30       # 30% of alerts share a dedupe_key within 60s
  --payload-bytes 512       # avg data{} size
  --concurrency 64          # concurrent HTTP connections
  --drop-pct 0.0            # stress only: drop this % of sends
  --latency-spike-ms 0      # stress only: inject X ms of extra sleep
  --reconnect-every 0       # stress only: drop+reconnect every X seconds
  --metrics :8891
  --coordinator nats://nats:4222  # for distributed mode
```

### Traffic profiles (built-in)

| profile | rate per instance | dedupe | severity mix | payload | use case |
|---|---|---|---|---|---|
| `normal` | 500/s | 30% | 70/25/4/1 | 512B | soak tests, capacity planning |
| `burst` | peak 5k/s for 10s, idle 50s | 50% | 50/30/15/5 | 1KB | alert-storm resilience |
| `chatty-iot` | 50/s × 1000 sources | 0% | 95/5/0/0 | 128B | IoT fleet size test |
| `pager-storm` | 10k/s | 0% | 0/0/50/50 | 256B | inminent_colapse path |
| `stress` | unlimited (CPU bound) | 0% | 25/25/25/25 | 4KB | find the breaking point |

### Distributed mode

Each instance advertises its rate in NATS KV. A coordinator (any one
instance, elected) sums the rates and assigns per-instance targets
to hit a **cluster-wide** rate. Use case: spin up 10 hosts, each
running 4 instances of `loadgen-grpc`, target 50k/s cluster-wide.

```
loadgen-grpc --mode stress --cluster-target 50000 --cluster-id bench-2026-06-13
```

Coordination is **best-effort** (no leader lock, no consensus) — if
the coordinator drops, the next instance takes over. Over- and
under-shoot within ±5% is acceptable for soak tests.

### Metrics

- `loadgen_alerts_sent_total{profile,instance,mode}` counter
- `loadgen_alerts_failed_total{reason}` counter (drop, timeout, 4xx, 5xx)
- `loadgen_alert_send_latency_seconds_bucket{mode}` histogram
- `loadgen_dedupe_hits_total` counter (computed server-side; we just
  read it from `Ack` and increment locally)
- `loadgen_cluster_target_rate` and `loadgen_cluster_actual_rate`
  gauges (only in distributed mode)

### What it is NOT

- Not a Chaos Mesh / Litmus replacement. No pod kills, no network
  partitions. (We can add that layer later with toxiproxy.)
- Not a load test for the **delivery** tier. Stubbing the delivery
  tier in `M10-bench` is the way to test the broker+router ceiling;
  the full delivery path is tested in M10 with the real sinks
  pointed at a fake-FCM / fake-Telegram server.
- Not for **production use**. The bin won't even compile with the
  release tag — it has its own `loadgen` Go module so it can't
  accidentally end up in a service image.

### Layout

```
loadgen/
├── go.mod                       # separate module
├── cmd/
│   ├── http/main.go
│   ├── ws/main.go
│   ├── mqtt/main.go
│   └── grpc/main.go
├── profiles/                    # YAML profiles; --mode picks one
│   ├── normal.yaml
│   ├── burst.yaml
│   ├── chatty-iot.yaml
│   ├── pager-storm.yaml
│   └── stress.yaml
├── internal/
│   ├── client/                  # wrappers around the same clients the services use
│   ├── coord/                   # NATS KV coordination
│   ├── traffic/                 # profile + dedupe + severity mix
│   └── metrics/                 # prom + slog
└── README.md                    # cookbook: "how to soak at 5k/s"
```

### Milestone

A "loadgen usable" check is part of **M10 exit**:
`loadgen-http --mode normal --rate 5000` sustained 10 min against
the running stack with p99 ingest latency ≤ 100ms, then
`loadgen-grpc --mode stress --cluster-target 50000` against
broker+router stubbed for **M10-bench**.

## 22. Source protection (throttling, quotas, circuit breaking)

Multi-tenant ingest dies the moment one runaway source saturates the
shared tier. Seven layers of protection, evaluated in this exact
order per request, so the cheap gates fire first:

```
incoming message
   │
   ▼
1. payload-size cap                  ── reject before parse
   ▼
2. per-IP / per-connection cap       ── reject DoS, no auth needed
   ▼
3. per-source token bucket           ── reject excess from this source
   ▼
4. per-company token bucket          ── reject excess summed across sources
   ▼
5. schema validate                   ── reject malformed
   ▼
6. circuit breaker (downstream)      ── fail fast if broker is sick
   ▼
7. per-source quarantine (cold)      ── hard block repeat offenders
   ▼
accept → dedupe → publish to NATS
```

### Layer 1 — Payload size cap

| transport | cap | reject code |
|---|---|---|
| HTTP | 256 KB (`Content-Length` + net/http `MaxBytesReader`) | `413 Payload Too Large` |
| WebSocket | 256 KB per frame | close `1009 Message Too Big` |
| MQTT | 256 KB per PUBLISH | disconnect with reason code `0x95 Payload format invalid` |
| gRPC | 256 KB (`grpc.MaxRecvMsgSize` per server) | `RESOURCE_EXHAUSTED` |

256 KB is ~1000× a normal alert (a 256 B alert with 1 KB of `data{}`
is the 99th percentile). Anything bigger is almost certainly a bug
or an attack. Configurable per source via `sources.max_payload_bytes`.

### Layer 2 — Per-IP / per-connection concurrency cap

DoS hardening. Default caps:

| transport | cap | rejection |
|---|---|---|
| HTTP | 64 concurrent conns per source IP | `429` + close |
| WebSocket | 32 concurrent sockets per source IP | close `1013 Try Again Later` |
| MQTT | 256 in-flight per `client_id` | broker holds the packet, no PUBACK |
| gRPC | 256 in-flight messages per stream | sender-side backpressure via `Ack{retry_after_ms}` |

Implementation: in-memory `sync.Map[ip]atomic.Int64` on `ingestd`,
with a small janitor that prunes idle entries. No Redis on the hot
path for this gate (would self-DoS).

### Layer 3 — Per-source token bucket (Redis)

Every `sources` row carries `rate_limit_per_sec` (default 100/s,
overridable per source). The bucket is a Redis key:

```
key   = rl:source:{source_id}
cap   = sources.rate_limit_per_sec
fill  = cap tokens/sec
```

Token consumption = one `INCRBY` + (if first in window) `EXPIRE`.
Use the `redis-cell` module if available (atomic, ~1 RTT); fall
back to `INCR`+`EXPIRE` otherwise. On reject:

| transport | response |
|---|---|
| HTTP | `429` + `Retry-After: <seconds>` + `X-RateLimit-*` headers |
| WebSocket | server-sent `{"error":"rate_limited","retry_after_ms":N}`; client may close or pause |
| MQTT | disconnect reason code `0x97 Quota exceeded` (or hold packet — broker config) |
| gRPC | `Ack{error: RATE_LIMITED, retry_after_ms}` — does *not* close the stream |

`429`s are **not** counted as faults by `ingestd`; they are the
documented contract. The `loadgen` `--rate` knob uses these
response codes to drive fault-injection tests.

### Layer 4 — Per-company token bucket (Redis)

Same shape, looser limit. Key `rl:company:{company_id}`, value =
`companies.rate_limit_per_sec` (default 10k/s, sum of all sources
under normal use). Catches the case where a company has 100
sources, each under their per-source cap, but the *sum* saturates
the broker.

### Layer 5 — Schema validate

Strict JSON Schema (or `.proto` for gRPC). Reject:

- Unknown top-level fields (forward-compat + catches misbehaving clients)
- Missing `severity`, `company_id`, `source_id`
- `severity` not in {`info`, `warning`, `critical`, `inminent_colapse`}
- `dedupe_key` longer than 128 chars
- `data` object with > 64 keys or > 8 KB serialized

Cost: ~50µs per message, negligible. Errors return `400` / gRPC
`INVALID_ARGUMENT`. Logged at `warn` level; counted in
`ingestd_schema_rejects_total{reason}`.

### Layer 6 — Circuit breaker on the broker (NATS)

If NATS JetStream's publish path is unhealthy (ack timeout > 250ms
3 times in a row, or stream full), `ingestd` opens a circuit
breaker and immediately returns `503` to all sources. Sources are
expected to retry with backoff. We do **not** queue alerts in
`ingestd` itself — that's what the broker is for.

States: `closed` (normal) → `open` (reject everything) → after
30s cool-down, `half-open` (let one request through) → `closed` on
success, `open` on failure. Standard `sony/gobreaker` config.

### Layer 7 — Per-source quarantine (the nuclear option)

If a source crosses **N=100 rate-limit hits in M=60s** *after* we
already sent the proper `429`s, we **temporarily quarantine** the
source for 5 min:

- All requests answered `403` from a local in-memory map (no Redis
  call). Protects Redis from being hammered by the same client.
- Operator gets a Prometheus alert
  (`ingestd_source_quarantined_total`) and a webhook into
  `admind` for incident review.
- After 5 min, quarantine auto-lifts. Persistent offenders are
  flagged in the admin UI for manual disable.

Quarantine is a **safety valve**, not a primary mechanism. The
token buckets should make it rare.

### Per-source config (additive to schema in §4)

```sql
ALTER TABLE sources
  ADD COLUMN rate_limit_per_sec          INTEGER NOT NULL DEFAULT 100,
  ADD COLUMN max_payload_bytes           INTEGER NOT NULL DEFAULT 262144,
  ADD COLUMN max_concurrent_connections  INTEGER NOT NULL DEFAULT 64,
  ADD COLUMN quarantine_hits_threshold   INTEGER NOT NULL DEFAULT 100,
  ADD COLUMN quarantine_window_seconds  INTEGER NOT NULL DEFAULT 60,
  ADD COLUMN quarantine_duration_seconds INTEGER NOT NULL DEFAULT 300;
```

Same columns on `companies` for the per-company bucket
(`rate_limit_per_sec` only — the rest are source-scoped).

### Prometheus metrics (additive)

```
ingestd_payload_rejected_total{transport,reason}      counter
ingestd_connection_rejected_total{transport}          counter
ingestd_rate_limited_total{transport,scope}           counter
# scope ∈ {source, company, quarantine}
ingestd_schema_rejected_total{reason}                 counter
ingestd_circuit_breaker_state{component}              gauge   # 0=closed 1=half 2=open
ingestd_source_quarantined_total{source_id,company_id} counter
ingestd_source_quarantine_active{source_id}           gauge
ingestd_rejection_latency_seconds_bucket{transport,reason} histogram
```

### Rejection code matrix (the contract)

| reason | HTTP | WS | MQTT | gRPC |
|---|---|---|---|---|
| payload too large | 413 | close 1009 | reason 0x95 | RESOURCE_EXHAUSTED |
| per-IP conn cap | 429 + close | close 1013 | reason 0x97 | n/a (per-stream cap) |
| per-source rate | 429 + Retry-After | `{"error":"rate_limited"}` | reason 0x97 | `Ack{error:RATE_LIMITED,retry_after_ms}` |
| per-company rate | 429 + Retry-After | same as above | same | same |
| schema invalid | 400 | `{"error":"invalid"}` | reason 0x99 | INVALID_ARGUMENT |
| broker down (cb open) | 503 + Retry-After | `{"error":"unavailable"}` | disconnect | `Ack{error:INTERNAL,retry_after_ms:1000}` |
| quarantined | 403 | close 1008 | disconnect | `Ack{error:UNAUTHENTICATED}` + close |

### Milestone rollout

| milestone | layers shipped |
|---|---|
| M0 | layer 3 (per-source bucket, simplest impl) |
| M1 | + layers 1, 4, 5 (size cap, per-company, schema) |
| M5 | + layer 2 (per-IP cap, obvious once WS is in) |
| M6 | + dedupe-aware rate shaping (don't burn a token on a dupe) |
| M9 | + layer 6 (circuit breaker), layer 7 (quarantine), all metrics + dashboards (shipped 2026-06-15) |
| M10 | prove one runaway source can't push p99 above 5s for *other* sources (loadgen fault-injection) |

### What we explicitly do NOT add in v1

- **WAF / OWASP CRS** in front of `ingestd`: alerts don't carry
  SQL/HTML; nginx with CRS is fine if you want it but out of
  spec.
- **Per-payload-cost rate limiting** (fat alerts cost more tokens):
  too clever for v1, easy to bolt on.
- **Auto-tuning of `rate_limit_per_sec`**: per-source is a config
  value, ops sets it.
- **Geo-fencing**: until v2 multi-region.
- **Quota reset emails / billing integration**: separate concern.

## 23. Milestones

| # | milestone | exit criterion |
|---|---|---|
| M0 | Skeleton + docker-compose up | Postgres + NATS + Redis up, all 4 services start, /health green | **✅ shipped 2026-06-13** |
| M1 | HTTP POST ingest end-to-end | send a signed webhook → FCM test push to a fake device; layers 1, 3, 4, 5 in | **✅ shipped 2026-06-13** (live smoke test all 9 steps green; see `M1_SMOKE_LOG.md`) |
| M2 | Recipient resolution | per-source `allowed_targets` honored, subscriptions applied | **✅ shipped 2026-06-13** (live smoke test all 7 steps green; see `M2_SMOKE_LOG.md`) |
| M3 | Telegram delivery + bot commands | user can `/subscribe` and receive an alert via Telegram | **✅ shipped 2026-06-14** (live smoke test all 8 steps green; 14 deliveries, 8 sendMessage calls; see `M3_VERIFICATION.md` + `M3_SMOKE_LOG.md`) |
| M4 | MQTT ingest | EMQX up, QoS 1, per-company topic ACLs | **✅ shipped 2026-06-14** (live smoke test all 5 steps green; 12 deliveries, 0 failures; see `M4_VERIFICATION.md` + `M4_SMOKE_LOG.md`) |
| M5 | WebSocket ingest + live tail | admin UI (or wscat) sees alerts as they arrive; layer 2 in | **✅ shipped 2026-06-14** (live smoke test all 7 steps green; 3 consecutive green runs; 13/13 checks each; +36 deliveries cumulative; see `M5_VERIFICATION.md` + `M5_SMOKE_LOG.md`) |
| M6 | Dedupe + dedupe_count | burst of 100 identical alerts → recipient sees "×100"; dedupe-aware rate shaping in | **✅ shipped 2026-06-14** (live smoke test all 6 steps green; 3 consecutive green runs; 11/11 checks each; sliding-window Lua + 600-alert dupe storm confirms free-for-dupes; see `M6_VERIFICATION.md` + `M6_SMOKE_LOG.md`) |
| M6.5 | Router-level dedupe collapse (closes the M6 "What's NOT" loop) | burst of 100 identical alerts → recipient sees 1 message, not 100; tail still shows the storm; per-source isolation | **✅ shipped 2026-06-14** (live smoke test all 5 steps green; 3 consecutive green runs; 9/9 checks each; 100-alert burst → 1 message, 180-alert continuous → 3 messages; see `M6.5_VERIFICATION.md` + `M6.5_SMOKE_LOG.md`) |
| M7 | Timescale + ClickHouse | 7d retention + archive job | **✅ shipped 2026-06-14** (live smoke test all 4 steps green; 3 consecutive green runs; 9/9 checks each; on remote playground `parres` 192.168.44.94; see `M7_VERIFICATION.md` + `M7_SMOKE_LOG.md`) |
| M8 | DLQ + replay UI | operator can replay a failed delivery | **✅ shipped 2026-06-14** (live smoke test all 4 steps green; 3 consecutive 12/12 runs; m8_smoke.sh: happy-path / DLQ creation (10 attempts, 1 DLQ row) / replay / discard; on local docker-compose stack; see `M8_VERIFICATION.md` + `M8_SMOKE_LOG.md`) |
| M9 | Observability (Prom/Grafana) | 1 dashboard per tier + per-company drilldown; layers 6, 7 in |
| M10 | Load test 5k/s on docker-compose | soak 10 min, p99 ≤ 5s, zero DLQ, run via `loadgen`; runaway-source test passes | **✅ shipped 2026-06-15** (live smoke test: 3 consecutive green runs on local; 1 run on remote `parres`; all 20 soak samples within 268–278/s, p99=0.248s, DLQ=0; runaway-source fault injection: p99 clean for healthy companies throughout 60s rogue load; see `M10_VERIFICATION.md` + `M10_SMOKE_LOG.md`) |
| M10-bench | Broker + router ceiling bench | 50k/s via `loadgen` against broker+router (delivery stubbed); p99 router latency ≤ 50ms; no broker backpressure | **✅ shipped 2026-06-15** (1 green run on remote `parres`; 10 samples over 5 min, router p99=5.0ms throughout, NATS qd informational only; see `M10_BENCH_VERIFICATION.md`; HTTP loadgen RTT ceiling limits realistic rate to ~275/s, target adjusted accordingly) |
| M11 | gRPC bidi-streaming ingest | internal Go service pushes ≥ 10k alerts/sec on one stream, p99 server-side `Ack` ≤ 50ms |