feat(spec): add gRPC bidi-streaming as fourth ingest protocol (M11)

- SPEC: new §19 gRPC ingest, M11 milestone added; .proto sketch
  with StreamAlerts(Ack) and same dedupe_count contract as HTTP
- ARCHITECTURE: gRPC source added to system-context flowchart,
  new §3 gRPC sequence diagram
- README: source list updated to include gRPC
- PROMPT: decision recorded (option a, internal high-volume only,
  HTTP stays for public SaaS, WS for browsers, MQTT for IoT)
- Mermaid: drop '(M11)' and other parens from node labels;
  gogs mermaid renderer chokes on them. mermaid-validator passes
  all 8 diagrams.

ARCHITECTURE.md

@@ -11,6 +11,7 @@ flowchart LR
         WH[Webhook / HTTP]
         WS[WebSocket clients]
         MQ[MQTT publishers]
+        GR[gRPC bidi-stream]
     end
     subgraph BA[Broad-Announce]
         IG[ingestd]
@@ -36,6 +37,7 @@ flowchart LR
     WH --> IG
     WS  --> IG
     MQ  --> IG
+    GR  --> IG
     IG  --> RD
     IG  --> BR
     BR  --> RT
@@ -102,7 +104,47 @@ sequenceDiagram
     end
 ```
 
-## 3. Recipient resolution algorithm
+## 3. gRPC ingest (M11)
+
+```mermaid
+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
 
 ```mermaid
 flowchart TD
@@ -122,7 +164,7 @@ flowchart TD
     J -- no --> K[emit one Delivery per channel in sub.channel_mask]
 ```
 
-## 4. Storage topology
+## 5. Storage topology
 
 ```mermaid
 flowchart LR
@@ -139,7 +181,7 @@ flowchart LR
     CH --- AGG[materialized aggregates<br/>per-company, per-channel]
 ```
 
-## 5. NATS subject layout
+## 6. NATS subject layout
 
 | subject | producer | consumer | retention |
 |---|---|---|---|
@@ -156,7 +198,7 @@ flowchart LR
 Subject-based partitioning keeps a single company's alert stream in
 order (mostly) and lets us add per-company worker affinity in K8s later.
 
-## 6. FCM delivery paths
+## 7. FCM delivery paths
 
 ```mermaid
 flowchart TD
@@ -200,7 +242,7 @@ flowchart TD
 The Android app reads `data.category` and `data.severity` to pick
 the right sound + channel.
 
-## 7. Telegram bot surface
+## 8. Telegram bot surface
 
 ```mermaid
 sequenceDiagram
@@ -211,12 +253,14 @@ sequenceDiagram
     participant RT as routerd
     participant PG as Postgres
 
-    U->>TG: /start <invite_code>
+    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;<br/>admin creates row + invite code,<br/>bot matches and burns 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(chat_id, "Linked to <name>")
+    BOT->>TG: sendMessage "Linked to NAME"
 
     U->>TG: /mute 2h
     TG->>BOT: update
@@ -224,17 +268,17 @@ sequenceDiagram
 
     Note over RT,PG: New alert arrives
     RT->>PG: resolve recipients incl. telegram
-    RT->>BR: deliveries.telegram.<company_id>
+    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>
+    BOT->>BR: ack.ALERT_ID.INDIVIDUAL_ID
     BR->>RT: close alert (or notify source)
 ```
 
-## 8. Failure & retry
+## 9. Failure & retry
 
 | layer | failure mode | behavior |
 |---|---|---|
@@ -246,7 +290,7 @@ sequenceDiagram
 | FCM | token unregistered | mark `fcm_tokens.status='unregistered'`, skip on next send |
 | Telegram | chat not found | mark `individuals.telegram_status='revoked'`, skip |
 
-## 9. SLOs
+## 10. SLOs
 
 | SLI | target | measured at |
 |---|---|---|
@@ -257,7 +301,7 @@ sequenceDiagram
 | DLQ rate | < 0.5% of deliveries | deliverd |
 | Per-company data isolation | 100% (no cross-tenant queries possible) | audit + e2e test |
 
-## 10. Capacity model (back-of-envelope)
+## 11. Capacity model (back-of-envelope)
 
 Assumption: 50k alerts/sec peak, avg fan-out 10 recipients × 2 channels = 20 deliveries per alert.
 
@@ -278,7 +322,7 @@ 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.
 
-## 11. Security model
+## 12. Security model
 
 ```mermaid
 flowchart LR
@@ -304,7 +348,7 @@ flowchart LR
     APP --> RD
 ```
 
-## 12. What changes when we move to K8s
+## 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`

PROMPT.md

@@ -51,4 +51,23 @@
   that compounds is the **subject layout in the broker**.
   `alerts.<company_id>` is fine; `alerts.<company_id>.<source_id>`
   would let us scale router consumer groups per source. Locked
-  in §5 of ARCHITECTURE.md.
+  in §6 of ARCHITECTURE.md.
+
+**2026-06-13 — added gRPC ingest (option a)**
+- Decision: ship gRPC bidi-streaming as the **fourth** ingest
+  protocol, scoped to **internal high-volume sources only**.
+  Public SaaS webhooks stay on HTTP POST, browsers stay on
+  WebSocket, IoT stays on MQTT.
+- Rationale: typed schemas, HTTP/2 + protobuf, native backpressure,
+  no reconnect-loop code, generated Go/Java/Python/Node clients.
+  Cost: one more protocol to operate + a `buf generate` build step.
+- Gated on **M11**, after M10 (load test). M10 now caps at
+  **5k/s on docker-compose** (not 50k/s) — 50k/s is the design
+  ceiling that requires K8s + multi-broker (M12 in v2).
+- `StreamAlerts(Alert) → Ack` carries the same `dedupe_count`
+  contract as HTTP/WS/MQTT.
+- Proto lives at `proto/broadannounce/v1/ingest.proto`; server
+  stub in `internal/grpcserver/`, reusable client in
+  `internal/grpcclient/`. Auth = API key in metadata + optional
+  mTLS. Per-stream rate limit + 256 in-flight cap = natural
+  backpressure.

README.md

@@ -1,10 +1,11 @@
 # broad-announce
 
 Multi-tenant notification router. Receives alerts from many source
-systems (HTTP/JSON webhooks, WebSockets, MQTT), normalizes them,
-resolves recipients via `companies` → `groups` → `individuals` +
-`subscriptions`, and delivers to FCM (Android), Telegram, SMS, email,
-voice, Slack, MS Teams, and arbitrary outbound webhooks.
+systems (HTTP/JSON webhooks, WebSockets, MQTT, gRPC bidi-streaming),
+normalizes them, resolves recipients via `companies` → `groups` →
+`individuals` + `subscriptions`, and delivers to FCM (Android),
+Telegram, SMS, email, voice, Slack, MS Teams, and arbitrary outbound
+webhooks.
 
 > **Status**: spec + architecture, no code yet. See `SPEC.md` for
 > requirements and `ARCHITECTURE.md` for diagrams / sequence flows /

SPEC.md

@@ -7,8 +7,8 @@
 
 ## 1. Goals (v1)
 
-- Accept alerts via **HTTP POST (JSON)**, **WebSockets (client→server)**, and
-  **MQTT**.
+- 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).
@@ -35,6 +35,8 @@
    │  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)
@@ -389,7 +391,119 @@ us per-device delivery reports.
 - **Migrations**: `golang-migrate`
 - **No-code admin UI** (later): React + Vite, served by `admind`
 
-## 19. Repo layout (planned)
+## 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/
@@ -421,7 +535,7 @@ broad-announce/
     └── loki/
 ```
 
-## 20. Milestones
+## 21. Milestones
 
 | # | milestone | exit criterion |
 |---|---|---|
@@ -435,4 +549,5 @@ broad-announce/
 | 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 |
-| M10 | Load test 50k/s | soak 10 min, p99 ≤ 5s, zero DLQ |
+| M10 | Load test 5k/s on docker-compose | soak 10 min, p99 ≤ 5s, zero DLQ |
+| M11 | gRPC bidi-streaming ingest | internal Go service pushes ≥ 10k alerts/sec on one stream, p99 server-side `Ack` ≤ 50ms |