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)

-- 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:

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).
Data Processing Agreement (DPA) with each customer: required by GDPR. We are the processor; they are the controller.
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

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

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)

loadgen-http \
  --target https://broad-announce:8080 \
  --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 :9091
  --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

429s 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 429s, 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)

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
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
M1	HTTP POST ingest end-to-end	send a signed webhook → FCM test push to a fake device; layers 1, 3, 4, 5 in
M2	Recipient resolution	per-source `allowed_targets` honored, subscriptions applied
M3	Telegram delivery + bot commands	user can `/subscribe` and receive an alert via Telegram
M4	MQTT ingest	EMQX up, QoS 1, per-company topic ACLs
M5	WebSocket ingest + live tail	admin UI (or wscat) sees alerts as they arrive; layer 2 in
M6	Dedupe + dedupe_count	burst of 100 identical alerts → recipient sees "×100"; dedupe-aware rate shaping in
M7	Timescale + ClickHouse	7d retention + archive job
M8	DLQ + replay UI	operator can replay a failed delivery
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
M10-bench	Broker + router ceiling bench	50k/s via `loadgen` against broker+router (delivery stubbed); p99 router latency ≤ 50ms; no broker backpressure
M11	gRPC bidi-streaming ingest	internal Go service pushes ≥ 10k alerts/sec on one stream, p99 server-side `Ack` ≤ 50ms

SPEC.md 37 KB История Исходник

Broad-Announce — Project Spec (v1)

1. Goals (v1)

2. Non-goals (v1)

3. Multi-tier architecture

Why 4 tiers instead of 1 monolith

4. Entities (Postgres schema sketch)

Severity taxonomy (used everywhere)

5. Dedupe + dedupe_count

6. Recipient resolution (the heart of the router)

7. FCM (shared project model)

10k companies × 1k devices = 10M tokens

8. Telegram (primary channel for ops)

9. Delivery retry & DLQ

10. Observability (operate from day 1)

11. Security

12. Data retention

13. Region / data residency (Q24)

14. Compliance (Q25)

15. mTLS vs HMAC vs API key (Q8)

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

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

18. Stack

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

Why a fourth protocol

What it is not

API surface

Auth

Server-side behavior

Repo additions for gRPC

Milestone

20. Repo layout (planned)

21. Traffic generators (loadgen)

Goals

Binaries

CLI surface (same shape across all four)

Traffic profiles (built-in)

Distributed mode

Metrics

What it is NOT

Layout

Milestone

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

Layer 1 — Payload size cap

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

Layer 3 — Per-source token bucket (Redis)

Layer 4 — Per-company token bucket (Redis)

Layer 5 — Schema validate

Layer 6 — Circuit breaker on the broker (NATS)

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

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

Prometheus metrics (additive)

Rejection code matrix (the contract)

Milestone rollout

What we explicitly do NOT add in v1

23. Milestones

SPEC.md 37 KB

История Исходник