M3(1/3): telegram tables + seed_m3 + per-channel binary split + compose/Dockerfile

- migrations/004_telegram.up.sql: telegram_bots table (one row per
  (company, bot)) and 4 new columns on individuals
  (telegram_chat_id, telegram_user_id, telegram_invite_code,
  mute_until) with two partial unique indexes for fast /start
  lookup and link-already-set check.
- migrations/seed_m3.sql: idempotent seed for M3. Inserts the
  per-company telegram_bots row, sets telegram_invite_code on
  all 3 individuals, and pre-links Alice (chat_id=1001,
  user_id=900001) so the very first POST exercises both fcm and
  telegram. Expands every subscription's channel_mask to
  ['fcm','telegram'] (M2 was ['fcm']).
- cmd/seed/main.go: runner now applies seed.sql → seed_m2.sql →
  seed_m3.sql. All three are idempotent.
- cmd/deliverd → cmd/deliverd-fcm: rename. M3 ships two
  per-channel binaries per the user's Q2 answer (independent
  deploys, no cross-channel blast radius). fcm delivery is
  unchanged.
- cmd/deliverd-telegram: new binary. Subscribes to
  deliveries.telegram.<company_id>, builds a
  severity-prefixed text message (🟧🟥🟥🟥⬛), posts to the
  Bot API sendMessage, writes a deliveries row. Single-attempt
  (M9 adds retry+DLQ).
- cmd/telegramd: new binary. Long-polling bot loop. Loads
  telegram_bots at startup, polls getUpdates per bot,
  dispatches commands via internal/telegram.Handler, replies
  via sendMessage. One process per deployment; per-bot
  sharding later if needed.
- testfakes/faketgmd: ~350 LoC fake Bot API server. /admin/queue
  to enqueue an incoming update, /admin/sent to read every
  sendMessage call, /admin/reset to clear state. Used purely
  for the M3 smoke test; never deployed.
- docker-compose.yml: adds deliverd-telegram (port 8821),
  telegramd (port 8822), faketgmd (port 8830). Grafana is now
  3001:3000 because :3000 is held by gogs on this host
  (one-line exception; canonical ports stay elsewhere).
- Dockerfile: builds 7 binaries (4 originals + 2 M3 + 1 fake).

What stays out of M3 (and not supposed to be): real FCM auth,
SMS/email/Slack/Teams/voice, webhook mode, bot token encryption
at rest, retry+DLQ for telegram, per-company bot token resolution
in deliverd-telegram.

Dockerfile

@@ -18,7 +18,11 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
       -o /out/routerd  ./cmd/routerd && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
-      -o /out/deliverd ./cmd/deliverd && \
+      -o /out/deliverd-fcm ./cmd/deliverd-fcm && \
+    CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
+      -o /out/deliverd-telegram ./cmd/deliverd-telegram && \
+    CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
+      -o /out/telegramd ./cmd/telegramd && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
       -o /out/admind   ./cmd/admind && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
@@ -28,7 +32,9 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
       -o /out/loadgen-http ./cmd/http && \
     cd .. && \
     CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
-      -o /out/fakefcmd ./testfakes/fakefcmd
+      -o /out/fakefcmd ./testfakes/fakefcmd && \
+    CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" \
+      -o /out/faketgmd ./testfakes/faketgmd
 
 FROM alpine:3.20
 RUN apk add --no-cache ca-certificates

cmd/deliverd/main.go → cmd/deliverd-fcm/main.go

@@ -1,10 +1,16 @@
-// Command deliverd consumes deliveries.<channel>.<company_id> subjects
-// and pushes the alert to the appropriate third-party sink
-// (FCM, Telegram, SMS, email, Slack, Teams, webhook).
+// Command deliverd-fcm consumes deliveries.fcm.<company_id> subjects
+// and posts each alert to the FCM HTTP v1 endpoint. M3 splits
+// this out of the original `cmd/deliverd` (which is now two
+// per-channel binaries: deliverd-fcm and deliverd-telegram).
+// That gives us the option to scale, restart, and deploy each
+// channel independently — your Q2 answer.
 //
 // M0: per-channel worker binary that connects to NATS, /health, /metrics.
 // M1: deliverd-fcm — consumes deliveries.fcm.*, posts to fakefcmd,
 //     records a row in Postgres per attempt.
+// M3: rename only. No behavior change. The HTTP target URL is
+//     BA_FAKECMD_URL (dev) or BA_FCM_BASE_URL (prod); M11+ will
+//     add real FCM auth.
 package main
 
 import (

cmd/deliverd-telegram/main.go

@@ -0,0 +1,296 @@
+// Command deliverd-telegram consumes deliveries.telegram.<company_id>
+// subjects and posts each alert to the Telegram Bot API
+// (sendMessage). It is the M3 counterpart of deliverd-fcm;
+// both consume from the same alerts flow but are deployed,
+// scaled, and restarted independently.
+//
+// Per the user's M3 Q2 answer: two binaries, not one with a
+// per-channel registry. Pros: independent deploys, no
+// cross-channel blast radius, simpler per-channel config.
+// Cons: more containers to operate. For v1 with a small
+// channel set (fcm, telegram) this is the right call.
+//
+// Configuration:
+//   BA_TELEGRAM_BOT_TOKEN       — default bot token if envelope
+//                                 doesn't carry a per-company one
+//   BA_TELEGRAM_BASE_URL        — override for tests (default
+//                                 https://api.telegram.org).
+//                                 M3 dev: point at faketgmd.
+//   BA_TELEGRAM_FAKE_URL        — alias kept for the smoke
+//                                 script; if set, takes priority.
+//
+// M3 ships single-attempt sends (no retry, no DLQ). That
+// follows the same pattern as deliverd-fcm in M1. M9 adds
+// the retry + DLQ chain.
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
+	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
+	"git3.techno-world.net/lrosales/broad-announce/internal/config"
+	"git3.techno-world.net/lrosales/broad-announce/internal/httpserver"
+	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
+	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
+	"git3.techno-world.net/lrosales/broad-announce/internal/telegram"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const telegramChannel = "telegram"
+
+type deliveryEnvelope struct {
+	Alert        json.RawMessage `json:"alert"`
+	IndividualID string          `json:"individual_id"`
+	Channel      string          `json:"channel"`
+	Endpoint     string          `json:"endpoint"`
+	Locale       string          `json:"locale,omitempty"`
+}
+
+func main() {
+	cfg, err := config.LoadCommon("deliverd-telegram")
+	if err != nil {
+		os.Stderr.WriteString("config: " + err.Error() + "\n")
+		os.Exit(1)
+	}
+	logger := observability.Init(cfg.Env, cfg.LogLevel, "deliverd-telegram")
+	logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr)
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	br, err := broker.Connect(ctx, cfg.NATSURL)
+	if err != nil {
+		logger.Error("nats connect", "err", err)
+		os.Exit(1)
+	}
+	defer br.Close()
+
+	pool, err := postgres.Connect(ctx, cfg.PostgresDSN)
+	if err != nil {
+		logger.Error("postgres connect", "err", err)
+		os.Exit(1)
+	}
+	defer pool.Close()
+
+	// Bot token resolution. M3: read from env. M3+:
+	// look up the per-company bot via the company_id encoded
+	// in the alert or the subject. For now the env-supplied
+	// default works because the M3 seed has one bot per
+	// company.
+	botToken := os.Getenv("BA_TELEGRAM_BOT_TOKEN")
+	if botToken == "" {
+		botToken = "fake-tg-bot-token-acme-001" // dev convenience
+	}
+
+	baseURL := os.Getenv("BA_TELEGRAM_FAKE_URL")
+	if baseURL == "" {
+		baseURL = os.Getenv("BA_TELEGRAM_BASE_URL")
+	}
+	if baseURL == "" {
+		baseURL = "http://faketgmd:8830" // M3 docker-compose default
+	}
+	logger.Info("telegram target", "base_url", baseURL)
+
+	client := &telegram.HTTPBotClient{
+		BaseURL: baseURL,
+		HTTP:    &http.Client{Timeout: 15 * time.Second},
+	}
+
+	js := br.JS()
+	stream, err := js.Stream(ctx, "DELIVERIES")
+	if err != nil {
+		logger.Error("nats stream DELIVERIES", "err", err)
+		os.Exit(1)
+	}
+	consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
+		Name:          "deliverd-telegram",
+		Durable:       "deliverd-telegram",
+		FilterSubjects: []string{"deliveries.telegram.>"},
+		AckPolicy:     jetstream.AckExplicitPolicy,
+	})
+	if err != nil {
+		logger.Error("nats consumer", "err", err)
+		os.Exit(1)
+	}
+
+	runCtx, runCancel := context.WithCancel(ctx)
+	defer runCancel()
+
+	go consume(runCtx, logger, consumer, pool, client, botToken)
+
+	reg, _ := observability.NewRegistry("deliverd-telegram")
+	srv := httpserver.New(httpserver.Config{
+		Addr:          cfg.HTTPAddr,
+		ServiceName:   "deliverd-telegram",
+		ShutdownGrace: cfg.ShutdownGrace,
+	}, logger, observability.MetricsHandler(reg))
+
+	errCh := make(chan error, 1)
+	go func() { errCh <- srv.Start() }()
+	select {
+	case <-ctx.Done():
+		logger.Info("shutdown signal received")
+	case err := <-errCh:
+		if err != nil {
+			logger.Error("http server", "err", err)
+			os.Exit(1)
+		}
+	}
+	runCancel()
+	time.Sleep(500 * time.Millisecond)
+	if err := srv.Shutdown(ctx); err != nil {
+		logger.Warn("graceful shutdown", "err", err)
+	}
+	logger.Info("bye")
+}
+
+func consume(ctx context.Context, logger *slog.Logger, c jetstream.Consumer, pool *postgres.Pool, client telegram.BotClient, botToken string) {
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		batch, err := c.Fetch(16, jetstream.FetchMaxWait(2*time.Second))
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			logger.Warn("nats fetch", "err", err)
+			time.Sleep(500 * time.Millisecond)
+			continue
+		}
+		for m := range batch.Messages() {
+			handleOne(ctx, logger, m, pool, client, botToken)
+			if batch.Error() != nil {
+				logger.Warn("batch error", "err", batch.Error())
+				break
+			}
+		}
+	}
+}
+
+func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *postgres.Pool, client telegram.BotClient, botToken string) {
+	var env deliveryEnvelope
+	if err := json.Unmarshal(m.Data(), &env); err != nil {
+		logger.Warn("malformed delivery envelope", "err", err, "subject", m.Subject())
+		_ = m.Ack()
+		return
+	}
+
+	// Pull the alert fields we need.
+	var ah struct {
+		ID        string `json:"id"`
+		CompanyID string `json:"company_id"`
+		Title     string `json:"title"`
+		Body      string `json:"body"`
+		Severity  alert.Severity `json:"severity"`
+		Category  string `json:"category"`
+		SourceID  string `json:"source_id"`
+	}
+	_ = json.Unmarshal(env.Alert, &ah)
+
+	companyID := ah.CompanyID
+	if companyID == "" {
+		// subject is "deliveries.telegram.<company_id>"
+		parts := strings.SplitN(m.Subject(), ".", 3)
+		if len(parts) == 3 {
+			companyID = parts[2]
+		}
+	}
+	if companyID == "" || env.Endpoint == "" || ah.ID == "" {
+		logger.Warn("delivery envelope missing fields", "subject", m.Subject(), "company", companyID, "endpoint_present", env.Endpoint != "", "alert_id", ah.ID)
+		_ = m.Ack()
+		return
+	}
+
+	chatID, err := parseInt64(env.Endpoint)
+	if err != nil {
+		logger.Warn("endpoint is not a telegram chat id", "endpoint", env.Endpoint)
+		_ = m.Ack()
+		return
+	}
+
+	// Build the text. SPEC §8 says the Android app does
+	// localization; for Telegram we use the source's
+	// pre-localized title + body verbatim, with a severity
+	// prefix so a glance at the chat shows priority.
+	text := formatMessage(ah.Severity, ah.Title, ah.Body, ah.ID)
+
+	status := "failed"
+	lastErr := ""
+	if _, err := client.SendMessage(ctx, botToken, chatID, text); err != nil {
+		lastErr = err.Error()
+		logger.Warn("sendMessage failed", "err", err, "chat_id", chatID, "alert_id", ah.ID)
+	} else {
+		status = "sent"
+	}
+
+	_, dbErr := pool.Exec(ctx, `
+		INSERT INTO deliveries
+		    (alert_id, company_id, individual_id, channel, target, status, attempts, last_error, payload, sent_at)
+		VALUES ($1, $2, $3, $4, $5, $6, 1, NULLIF($7, ''), $8,
+		    CASE WHEN $6 = 'sent' THEN now() ELSE NULL END)
+	`, ah.ID, companyID, env.IndividualID, telegramChannel, env.Endpoint, status, lastErr, json.RawMessage(m.Data()))
+	if dbErr != nil {
+		logger.Warn("delivery row insert", "err", dbErr)
+	}
+
+	logger.Info("delivery",
+		"alert_id", ah.ID,
+		"company", companyID,
+		"individual", env.IndividualID,
+		"channel", telegramChannel,
+		"status", status,
+		"err", lastErr,
+	)
+	_ = m.Ack()
+}
+
+// formatMessage produces a Telegram-friendly rendering of
+// the alert. The format is intentionally plain (Telegram
+// supports Markdown/HTML but they vary across clients);
+// M3.5+ can add formatting once the Android-side payload
+// shape is locked.
+func formatMessage(sev alert.Severity, title, body, alertID string) string {
+	var prefix string
+	switch sev {
+	case alert.SeverityInminentColapse:
+		prefix = "🟥🟥🟥 IMMINENT"
+	case alert.SeverityCritical:
+		prefix = "🟥 CRITICAL"
+	case alert.SeverityWarning:
+		prefix = "🟧 WARNING"
+	default:
+		prefix = "🟦 INFO"
+	}
+	out := fmt.Sprintf("%s: %s", prefix, title)
+	if body != "" {
+		out += "\n" + body
+	}
+	short := alertID
+	if len(short) > 12 {
+		short = short[:12]
+	}
+	out += fmt.Sprintf("\n— alert %s", short)
+	return out
+}
+
+func parseInt64(s string) (int64, error) {
+	var n int64
+	_, err := fmt.Sscanf(s, "%d", &n)
+	return n, err
+}
+
+var _ = bytes.NewReader
+var _ = io.ReadAll

cmd/seed/main.go

@@ -45,9 +45,9 @@ func main() {
 		die("apply migrations: " + err.Error())
 	}
 	// Seed files run in lexical order, so seed.sql → seed_m2.sql
-	// → seed_m3.sql → … Each is idempotent (ON CONFLICT DO NOTHING)
-	// and safe to re-run.
-	for _, name := range []string{"seed.sql", "seed_m2.sql"} {
+	// → seed_m3.sql → … Each is idempotent (ON CONFLICT DO NOTHING
+	// or ON CONFLICT DO UPDATE) and safe to re-run.
+	for _, name := range []string{"seed.sql", "seed_m2.sql", "seed_m3.sql"} {
 		p := filepath.Join(dir, name)
 		if _, err := os.Stat(p); err != nil {
 			fmt.Fprintln(os.Stderr, "no", name, "in", dir, "(skipped)")

cmd/telegramd/main.go

@@ -0,0 +1,195 @@
+// Command telegramd is the long-polling bot loop for
+// broad-announce M3 (SPEC §8). It pulls the list of
+// active telegram_bots from Postgres, then for each bot
+// long-polls getUpdates, parses commands, applies them
+// via internal/telegram.Handler, and replies via
+// SendMessage.
+//
+// M3 ships long-polling only. Webhook mode is M5/M9.
+//
+// Configuration:
+//   BA_TELEGRAM_BOT_TOKEN        — fallback if DB has no row
+//   BA_TELEGRAM_BASE_URL         — override for tests (default
+//                                  https://api.telegram.org).
+//                                  M3 dev: point at faketgmd.
+//
+// One process polls all bots. With M3's one-bot-per-company
+// this is a single update loop; M3+ can shard by company
+// hash if the volume warrants.
+package main
+
+import (
+	"context"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/broker"
+	"git3.techno-world.net/lrosales/broad-announce/internal/config"
+	"git3.techno-world.net/lrosales/broad-announce/internal/httpserver"
+	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
+	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
+	"git3.techno-world.net/lrosales/broad-announce/internal/telegram"
+)
+
+type botConfig struct {
+	BotID     string
+	CompanyID string
+	BotToken  string
+}
+
+func main() {
+	cfg, err := config.LoadCommon("telegramd")
+	if err != nil {
+		os.Stderr.WriteString("config: " + err.Error() + "\n")
+		os.Exit(1)
+	}
+	logger := observability.Init(cfg.Env, cfg.LogLevel, "telegramd")
+	logger.Info("starting", "env", cfg.Env, "addr", cfg.HTTPAddr)
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	br, err := broker.Connect(ctx, cfg.NATSURL)
+	if err != nil {
+		logger.Error("nats connect", "err", err)
+		os.Exit(1)
+	}
+	defer br.Close()
+	_ = br // M3: long-polling only; NATS used later for ack fan-out
+
+	pool, err := postgres.Connect(ctx, cfg.PostgresDSN)
+	if err != nil {
+		logger.Error("postgres connect", "err", err)
+		os.Exit(1)
+	}
+	defer pool.Close()
+
+	baseURL := os.Getenv("BA_TELEGRAM_FAKE_URL")
+	if baseURL == "" {
+		baseURL = os.Getenv("BA_TELEGRAM_BASE_URL")
+	}
+	if baseURL == "" {
+		baseURL = "http://faketgmd:8830" // M3 docker-compose default
+	}
+	logger.Info("telegram target", "base_url", baseURL)
+
+	client := &telegram.HTTPBotClient{
+		BaseURL: baseURL,
+		HTTP:    &http.Client{Timeout: 60 * time.Second},
+	}
+	handler := telegram.NewHandler(pool, logger.With("subsystem", "telegram"))
+
+	runCtx, runCancel := context.WithCancel(ctx)
+	defer runCancel()
+	go pollAllBots(runCtx, logger, pool, client, handler)
+
+	reg, _ := observability.NewRegistry("telegramd")
+	srv := httpserver.New(httpserver.Config{
+		Addr:          cfg.HTTPAddr,
+		ServiceName:   "telegramd",
+		ShutdownGrace: cfg.ShutdownGrace,
+	}, logger, observability.MetricsHandler(reg))
+
+	errCh := make(chan error, 1)
+	go func() { errCh <- srv.Start() }()
+	select {
+	case <-ctx.Done():
+		logger.Info("shutdown signal received")
+	case err := <-errCh:
+		if err != nil {
+			logger.Error("http server", "err", err)
+			os.Exit(1)
+		}
+	}
+	runCancel()
+	time.Sleep(500 * time.Millisecond)
+	if err := srv.Shutdown(ctx); err != nil {
+		logger.Warn("graceful shutdown", "err", err)
+	}
+	logger.Info("bye")
+}
+
+// pollAllBots loads the bot list once at startup, then
+// long-polls each in a goroutine. M3: a single reload on
+// SIGUSR1 is overkill; restarting the binary picks up new
+// bots. M5 can add a watch on the table.
+func pollAllBots(ctx context.Context, logger *slog.Logger, pool *postgres.Pool, client telegram.BotClient, h *telegram.Handler) {
+	bots, err := loadBots(ctx, pool)
+	if err != nil {
+		logger.Error("load bots", "err", err)
+		return
+	}
+	logger.Info("loaded bots", "count", len(bots))
+	if len(bots) == 0 {
+		// No bots yet. Block on ctx so the process stays up.
+		<-ctx.Done()
+		return
+	}
+	for _, b := range bots {
+		go pollOneBot(ctx, logger.With("bot", b.BotID, "company", b.CompanyID), client, h, b)
+	}
+	<-ctx.Done()
+}
+
+func loadBots(ctx context.Context, pool *postgres.Pool) ([]botConfig, error) {
+	rows, err := pool.Query(ctx, `
+		SELECT bot_id, company_id, bot_token
+		  FROM telegram_bots
+		 WHERE status = 'active'
+		 ORDER BY company_id, bot_id
+	`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []botConfig
+	for rows.Next() {
+		var b botConfig
+		if err := rows.Scan(&b.BotID, &b.CompanyID, &b.BotToken); err != nil {
+			return nil, err
+		}
+		out = append(out, b)
+	}
+	return out, rows.Err()
+}
+
+func pollOneBot(ctx context.Context, logger *slog.Logger, client telegram.BotClient, h *telegram.Handler, b botConfig) {
+	var offset int64
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		updates, err := client.GetUpdates(ctx, b.BotToken, offset, 25)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			logger.Warn("getUpdates", "err", err)
+			time.Sleep(2 * time.Second)
+			continue
+		}
+		for _, u := range updates {
+			if u.UpdateID >= offset {
+				offset = u.UpdateID + 1
+			}
+			if u.Message == nil {
+				continue
+			}
+			reply, err := h.Handle(ctx, u.Message)
+			if err != nil {
+				logger.Warn("command handle", "err", err, "update_id", u.UpdateID)
+				continue
+			}
+			if reply == "" {
+				continue
+			}
+			if _, err := client.SendMessage(ctx, b.BotToken, u.Message.Chat.ID, reply); err != nil {
+				logger.Warn("sendMessage reply", "err", err, "chat_id", u.Message.Chat.ID)
+			}
+		}
+	}
+}

docker-compose.yml

@@ -3,7 +3,7 @@
 # Then: see M0_VERIFICATION.md for the loadgen smoke test.
 #
 # Port conventions (project rule):
-#   - app HTTP services:  8800–8899  (ingestd/routerd/deliverd/admind/loadgen)
+#   - app HTTP services:  8800–8899  (ingestd/routerd/deliverd-fcm/deliverd-telegram/telegramd/admind/loadgen/faketgmd)
 #   - canonical ports stay (5432 postgres, 4222 nats, 6379 redis,
 #     1883 mqtt, 9090 prometheus, 3000 grafana, etc.)
 
@@ -100,9 +100,9 @@ services:
       nats:     { condition: service_healthy }
       postgres: { condition: service_healthy }
 
-  deliverd:
+  deliverd-fcm:
     build: .
-    command: ["/app/deliverd"]
+    command: ["/app/deliverd-fcm"]
     environment:
       BA_ENV: dev
       BA_HTTP_ADDR: ":8802"
@@ -115,6 +115,42 @@ services:
       postgres:  { condition: service_healthy }
       fakefcmd:  { condition: service_started }
 
+  deliverd-telegram:
+    build: .
+    command: ["/app/deliverd-telegram"]
+    environment:
+      BA_ENV: dev
+      BA_HTTP_ADDR: ":8821"
+      BA_NATS_URL: nats://nats:4222
+      BA_POSTGRES_DSN: postgres://ba:ba@postgres:5432/ba?sslmode=disable
+      BA_TELEGRAM_BOT_TOKEN: "fake-tg-bot-token-acme-001"
+      BA_TELEGRAM_FAKE_URL: "http://faketgmd:8830"
+    ports: ["8821:8821"]
+    depends_on:
+      nats:      { condition: service_healthy }
+      postgres:  { condition: service_healthy }
+      faketgmd:  { condition: service_started }
+
+  telegramd:
+    build: .
+    command: ["/app/telegramd"]
+    environment:
+      BA_ENV: dev
+      BA_HTTP_ADDR: ":8822"
+      BA_NATS_URL: nats://nats:4222
+      BA_POSTGRES_DSN: postgres://ba:ba@postgres:5432/ba?sslmode=disable
+      BA_TELEGRAM_FAKE_URL: "http://faketgmd:8830"
+    ports: ["8822:8822"]
+    depends_on:
+      nats:      { condition: service_healthy }
+      postgres:  { condition: service_healthy }
+      faketgmd:  { condition: service_started }
+
+  faketgmd:
+    build: .
+    command: ["/app/faketgmd", "--addr", ":8830", "--timeout", "5"]
+    ports: ["8830:8830"]
+
   fakefcmd:
     build: .
     command: ["/app/fakefcmd", "--addr", ":8820"]
@@ -157,11 +193,15 @@ services:
     volumes:
       - ./deploy/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro
     ports: ["9090:9090"]
-    depends_on: [ingestd, routerd, deliverd, admind]
+    depends_on: [ingestd, routerd, deliverd-fcm, deliverd-telegram, telegramd, admind]
 
   grafana:
     image: grafana/grafana:latest
-    ports: ["3000:3000"]
+    # :3000 is held by gogs (git3) on this host. Map to :3001
+    # on the host, keep :3000 internal. M0 ports convention
+    # says canonical ports stay; the host-port override is a
+    # one-line exception, fully isolated to grafana.
+    ports: ["3001:3000"]
     environment:
       GF_SECURITY_ADMIN_USER: admin
       GF_SECURITY_ADMIN_PASSWORD: admin

migrations/004_telegram.down.sql

@@ -0,0 +1,13 @@
+-- 004_telegram.down.sql
+-- Symmetric drop. Children first.
+
+DROP INDEX IF EXISTS idx_individuals_telegram_user;
+DROP INDEX IF EXISTS idx_individuals_invite_code;
+
+ALTER TABLE individuals
+    DROP COLUMN IF EXISTS mute_until,
+    DROP COLUMN IF EXISTS telegram_invite_code,
+    DROP COLUMN IF EXISTS telegram_user_id,
+    DROP COLUMN IF EXISTS telegram_chat_id;
+
+DROP TABLE IF EXISTS telegram_bots;

migrations/004_telegram.up.sql

@@ -0,0 +1,61 @@
+-- 004_telegram.up.sql
+-- M3 Telegram delivery + bot commands. See SPEC §8.
+--
+-- What lands in M3:
+--   telegram_bots    — one row per (company, bot). For M3 we
+--                      support one bot per company; the table
+--                      keys on (company_id, bot_id) so adding
+--                      more later is a one-line change.
+--   individuals:     — adds telegram_chat_id, telegram_user_id,
+--                      telegram_invite_code (set on individual
+--                      creation by the admin; admin hands the
+--                      code to the user out-of-band), and
+--                      mute_until (per-individual global mute
+--                      for the /mute command).
+--
+-- What stays out of M3:
+--   Bot token encryption at rest. M3 stores the token in
+--   plaintext in telegram_bots.bot_token with a dev-only
+--   annotation. AES-256-GCM is a security milestone (§11).
+--   Real Telegram API. M3 ships faketgmd and a long-poll
+--   bot loop. Webhook mode is M5/M9.
+--   Retry + DLQ for failed sends. The single-attempt pattern
+--   from M1's fcm path is repeated. M9 adds the retry chain.
+--
+-- Naming: snake_case to match the rest of the schema.
+
+-- ── telegram_bots ────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS telegram_bots (
+    bot_id            TEXT NOT NULL,                       -- short id; e.g. "primary"
+    company_id        TEXT NOT NULL REFERENCES companies(id) ON DELETE CASCADE,
+    name              TEXT NOT NULL,                       -- human label; e.g. "Acme Ops"
+    bot_token         TEXT NOT NULL,                       -- DEV ONLY; encrypt in M11 (security milestone)
+    status            TEXT NOT NULL DEFAULT 'active',      -- active | paused
+    last_seen_at      TIMESTAMPTZ,                         -- last getUpdates / webhook hit
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (company_id, bot_id)
+);
+CREATE INDEX IF NOT EXISTS idx_telegram_bots_company ON telegram_bots(company_id) WHERE status = 'active';
+
+-- ── extend individuals ───────────────────────────────────────────
+-- All four columns are nullable: an individual that has not yet
+-- linked their Telegram account has no chat_id / user_id. The
+-- invite_code is set when the admin creates the individual.
+ALTER TABLE individuals
+    ADD COLUMN IF NOT EXISTS telegram_chat_id     TEXT,
+    ADD COLUMN IF NOT EXISTS telegram_user_id     BIGINT,
+    ADD COLUMN IF NOT EXISTS telegram_invite_code TEXT,
+    ADD COLUMN IF NOT EXISTS mute_until           TIMESTAMPTZ;
+
+-- Unique index for fast "/start <code>" lookup. Partial index
+-- so historical / legacy individuals without a code don't bloat
+-- the index.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_individuals_invite_code
+    ON individuals(telegram_invite_code)
+    WHERE telegram_invite_code IS NOT NULL;
+
+-- Unique index for fast "is this telegram_user_id already linked?"
+-- check. Partial: most individuals won't be linked.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_individuals_telegram_user
+    ON individuals(company_id, telegram_user_id)
+    WHERE telegram_user_id IS NOT NULL;

migrations/seed_m3.sql

@@ -0,0 +1,53 @@
+-- seed_m3.sql
+-- M3 seed. Adds the per-company telegram_bots row, sets
+-- telegram_invite_code on every individual, and **pre-links**
+-- Alice to a fake Telegram account so the smoke test can
+-- exercise the full M3 path (FCM + Telegram delivery) on
+-- the very first POST.
+--
+-- Bob and Carol get invite codes but are NOT linked. The
+-- smoke test simulates `/start <code>` via faketgmd's
+-- /admin/queue endpoint, watches the link happen, then
+-- continues.
+--
+-- Idempotent (ON CONFLICT DO UPDATE for the link, ON
+-- CONFLICT DO NOTHING for the rest). Safe to re-run.
+
+-- ── telegram_bots ────────────────────────────────────────────────
+-- One bot for acme-001. The bot_token is the faketgmd sentinel
+-- "fake-tg-bot-token-acme-001". In production this is a real
+-- token from @BotFather; see M3_VERIFICATION §Step 8.
+INSERT INTO telegram_bots (bot_id, company_id, name, bot_token) VALUES
+    ('primary', 'acme-001', 'Acme Ops', 'fake-tg-bot-token-acme-001')
+ON CONFLICT (company_id, bot_id) DO NOTHING;
+
+-- ── individuals: invite codes + Alice's pre-link ───────────────
+-- All three individuals get an invite code. Re-running the
+-- seed is safe: the code is overwritten on conflict.
+UPDATE individuals SET telegram_invite_code = 'acme-alice-001' WHERE id = 'ind-acme-001';
+UPDATE individuals SET telegram_invite_code = 'acme-bob-002'   WHERE id = 'ind-acme-002';
+UPDATE individuals SET telegram_invite_code = 'acme-carol-003' WHERE id = 'ind-acme-003';
+
+-- Alice is already linked. The faketgmd admin endpoint can
+-- override these later; for M3 verification step 2 we want
+-- her to be linked so the first POST already exercises
+-- both fcm and telegram delivery paths.
+UPDATE individuals
+   SET telegram_chat_id = 1001,
+       telegram_user_id = 900001
+ WHERE id = 'ind-acme-001';
+
+-- ── channel_mask expansion ─────────────────────────────────────
+-- Bob's subscription was ['fcm'] in M2. M3 needs it to be
+-- ['fcm','telegram'] so that, once he's linked via /start,
+-- the resolver starts emitting telegram endpoints for him.
+-- For step 2 the existing fcm path still fires for him; the
+-- telegram path is no-op until step 5 (the /start sim).
+UPDATE subscriptions
+   SET channel_mask = '["fcm","telegram"]'::jsonb
+ WHERE company_id = 'acme-001'
+   AND individual_id IN ('ind-acme-001','ind-acme-002','ind-acme-003');
+
+-- Carol stays with channel_mask = ['fcm','telegram'] too, so
+-- the same channel expansion is exercised. Her quiet_hours
+-- keep her out of any non-inminent_colapse alert regardless.

testfakes/faketgmd/main.go

@@ -0,0 +1,353 @@
+// Command faketgmd is a fake Telegram Bot API server for the
+// M3 smoke test. It mimics the parts of api.telegram.org
+// that broad-announce uses:
+//
+//   POST /bot<token>/sendMessage   — accept the message, log it
+//   POST /bot<token>/getUpdates    — return queued updates, or
+//                                    block for `timeout` seconds
+//
+// Plus admin endpoints for the smoke test:
+//
+//   POST /admin/queue              — queue a fake incoming
+//                                    update (user text) for a
+//                                    given bot token. The
+//                                    next getUpdates call
+//                                    will return it.
+//   GET  /admin/sent               — JSON list of every
+//                                    sendMessage call received,
+//                                    with timestamp. Used by
+//                                    the smoke test to confirm
+//                                    a delivery happened.
+//   POST /admin/reset              — clear sent + queue.
+//   GET  /health                   — liveness probe.
+//
+// M3 uses this purely for tests. The production switch is
+// one env var (BA_TELEGRAM_BASE_URL on the deliverd /
+// telegramd side); faketgmd is never deployed.
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"io"
+	"log/slog"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// sentMessage and queuedUpdate are the records faketgmd
+// keeps in memory. M3 lives in dev; persistence is
+// explicitly out of scope.
+type sentMessage struct {
+	BotToken string    `json:"bot_token"`
+	ChatID   int64     `json:"chat_id"`
+	Text     string    `json:"text"`
+	SentAt   time.Time `json:"sent_at"`
+}
+
+type queuedUpdate struct {
+	BotToken  string
+	UpdateID  int64
+	UserID    int64
+	ChatID    int64
+	Text      string
+	FirstName string
+}
+
+type server struct {
+	mu              sync.Mutex
+	sent            []sentMessage
+	queue           []queuedUpdate
+	offsetByBot     map[string]int64
+	defaultTimeout  int
+	failRatePercent int
+	logger          *slog.Logger
+}
+
+func main() {
+	addr := flag.String("addr", ":8830", "listen address")
+	timeout := flag.Int("timeout", 25, "default long-poll timeout in seconds")
+	failRate := flag.Int("fail-rate", 0, "percent of sendMessage calls to fail (0-100)")
+	flag.Parse()
+
+	logger := slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
+
+	s := &server{
+		offsetByBot:     map[string]int64{},
+		defaultTimeout:  *timeout,
+		failRatePercent: *failRate,
+		logger:          logger,
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/health", s.handleHealth)
+	mux.HandleFunc("/admin/queue", s.handleAdminQueue)
+	mux.HandleFunc("/admin/sent", s.handleAdminSent)
+	mux.HandleFunc("/admin/reset", s.handleAdminReset)
+	// Telegram-shaped endpoints. ServeMux's `/bot` (no
+	// trailing slash) is exact-match only. We register
+	// `/` (root subtree) so we can read the path inside
+	// the handler. /admin/* and /health are still matched
+	// by the more specific patterns above because ServeMux
+	// prefers the longest match.
+	mux.HandleFunc("/", s.routeAny)
+
+	logger.Info("faketgmd listening", "addr", *addr, "timeout", *timeout, "fail_rate", *failRate)
+	if err := http.ListenAndServe(*addr, mux); err != nil {
+		logger.Error("listen", "err", err)
+		os.Exit(1)
+	}
+}
+
+func (s *server) handleHealth(w http.ResponseWriter, r *http.Request) {
+	s.mu.Lock()
+	nSent := len(s.sent)
+	s.mu.Unlock()
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]any{
+		"status":   "ok",
+		"service":  "faketgmd",
+		"received": nSent,
+		"failed":   0,
+	})
+}
+
+// handleBot dispatches /bot<token>/<method> to the right
+// handler. ServeMux only matches exact paths unless the
+// registered pattern ends in "/", and Telegram's URL
+// (`/bot<token>/<method>`) doesn't have a slash after
+// "bot", so we register `/` and dispatch on the path
+// inside routeAny.
+func (s *server) routeAny(w http.ResponseWriter, r *http.Request) {
+	path := r.URL.Path
+	switch {
+	case strings.HasPrefix(path, "/bot"):
+		rest := strings.TrimPrefix(path, "/bot")
+		if rest == "" {
+			http.Error(w, "bad path; expected /bot<token>/<method>", http.StatusBadRequest)
+			return
+		}
+		parts := strings.SplitN(rest, "/", 2)
+		if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
+			http.Error(w, "bad path; expected /bot<token>/<method>", http.StatusBadRequest)
+			return
+		}
+		token, method := parts[0], parts[1]
+		switch method {
+		case "sendMessage":
+			s.handleSendMessage(w, r, token)
+		case "getUpdates":
+			s.handleGetUpdates(w, r, token)
+		default:
+			http.Error(w, "unknown method "+method, http.StatusNotFound)
+		}
+	default:
+		http.NotFound(w, r)
+	}
+}
+
+// handleBot kept for backward-compat with earlier callers; routes
+// through routeAny.
+func (s *server) handleBot(w http.ResponseWriter, r *http.Request) {
+	s.routeAny(w, r)
+}
+
+// handleSendMessage accepts the JSON body, optionally
+// fails, and logs to the sent[] ring.
+func (s *server) handleSendMessage(w http.ResponseWriter, r *http.Request, token string) {
+	body, _ := io.ReadAll(r.Body)
+	defer r.Body.Close()
+
+	var req struct {
+		ChatID int64  `json:"chat_id"`
+		Text   string `json:"text"`
+	}
+	if err := json.Unmarshal(body, &req); err != nil {
+		http.Error(w, "bad json: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	// Roll the failure dice.
+	if s.failRatePercent > 0 {
+		if n := time.Now().UnixNano() % 100; int(n) < s.failRatePercent {
+			s.logger.Warn("faketgmd failing this sendMessage", "fail_rate", s.failRatePercent)
+			http.Error(w, `{"ok":false,"description":"intentional fake fail"}`, http.StatusInternalServerError)
+			return
+		}
+	}
+
+	s.mu.Lock()
+	s.sent = append(s.sent, sentMessage{
+		BotToken: token,
+		ChatID:   req.ChatID,
+		Text:     req.Text,
+		SentAt:   time.Now().UTC(),
+	})
+	s.mu.Unlock()
+
+	resp := map[string]any{
+		"ok": true,
+		"result": map[string]any{
+			"message_id": time.Now().UnixNano() % 1_000_000,
+			"chat":       map[string]any{"id": req.ChatID, "type": "private"},
+			"date":       time.Now().Unix(),
+			"text":       req.Text,
+		},
+	}
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(resp)
+}
+
+// handleGetUpdates long-polls: waits up to defaultTimeout
+// seconds for at least one queued update for this bot,
+// returns everything queued, advances the offset.
+func (s *server) handleGetUpdates(w http.ResponseWriter, r *http.Request, token string) {
+	body, _ := io.ReadAll(r.Body)
+	defer r.Body.Close()
+
+	var req struct {
+		Offset  int64 `json:"offset"`
+		Timeout int   `json:"timeout"`
+	}
+	if len(body) > 0 {
+		_ = json.Unmarshal(body, &req)
+	}
+	if req.Timeout == 0 {
+		req.Timeout = s.defaultTimeout
+	}
+
+	deadline := time.Now().Add(time.Duration(req.Timeout) * time.Second)
+	for time.Now().Before(deadline) {
+		s.mu.Lock()
+		var updates []map[string]any
+		for _, u := range s.queue {
+			if u.BotToken != token {
+				continue
+			}
+			if u.UpdateID < req.Offset {
+				continue
+			}
+			updates = append(updates, map[string]any{
+				"update_id": u.UpdateID,
+				"message": map[string]any{
+					"message_id": u.UpdateID,
+					"from": map[string]any{
+						"id":         u.UserID,
+						"is_bot":     false,
+						"first_name": u.FirstName,
+					},
+					"chat": map[string]any{
+						"id":   u.ChatID,
+						"type": "private",
+					},
+					"text": u.Text,
+					"date": time.Now().Unix(),
+				},
+			})
+		}
+		// Advance offset to last+1.
+		var maxID int64
+		for _, u := range s.queue {
+			if u.BotToken == token && u.UpdateID > maxID {
+				maxID = u.UpdateID
+			}
+		}
+		if maxID >= req.Offset {
+			s.offsetByBot[token] = maxID + 1
+		}
+		s.mu.Unlock()
+
+		if len(updates) > 0 {
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(map[string]any{
+				"ok":     true,
+				"result": updates,
+			})
+			return
+		}
+		// Sleep a bit, then poll again. Long-poll in the
+		// spec keeps the connection open; we simulate by
+		// re-checking every 500ms.
+		time.Sleep(500 * time.Millisecond)
+	}
+
+	// Timeout with no updates.
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]any{
+		"ok":     true,
+		"result": []map[string]any{},
+	})
+}
+
+// handleAdminQueue accepts a JSON body to enqueue a fake
+// incoming update for a bot. Used by the smoke test to
+// simulate a user typing /start <code> or /subscribe ...
+func (s *server) handleAdminQueue(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "POST only", http.StatusMethodNotAllowed)
+		return
+	}
+	body, _ := io.ReadAll(r.Body)
+	defer r.Body.Close()
+
+	var req struct {
+		BotToken  string `json:"bot_token"`
+		UserID    int64  `json:"user_id"`
+		ChatID    int64  `json:"chat_id"`
+		Text      string `json:"text"`
+		FirstName string `json:"first_name"`
+	}
+	if err := json.Unmarshal(body, &req); err != nil {
+		http.Error(w, "bad json: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+	if req.BotToken == "" || req.UserID == 0 || req.ChatID == 0 {
+		http.Error(w, "bot_token, user_id, chat_id required", http.StatusBadRequest)
+		return
+	}
+	if req.FirstName == "" {
+		req.FirstName = "FakeUser"
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.queue = append(s.queue, queuedUpdate{
+		BotToken:  req.BotToken,
+		UpdateID:  int64(len(s.queue) + 1),
+		UserID:    req.UserID,
+		ChatID:    req.ChatID,
+		Text:      req.Text,
+		FirstName: req.FirstName,
+	})
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]any{
+		"queued":    true,
+		"update_id": len(s.queue),
+	})
+}
+
+func (s *server) handleAdminSent(w http.ResponseWriter, r *http.Request) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]any{
+		"count": len(s.sent),
+		"items": s.sent,
+	})
+}
+
+func (s *server) handleAdminReset(w http.ResponseWriter, r *http.Request) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.sent = nil
+	s.queue = nil
+	s.offsetByBot = map[string]int64{}
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]any{"ok": true})
+}
+
+var _ = strconv.Itoa // keep import