// Command deliverd-telegram consumes deliveries.telegram. // 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"` // M6: dedupe_count. 0 or 1 means "not a duplicate", // ≥2 means the dedupe window has seen this alert N // times; we surface that to the recipient via the // inline `(×N)` suffix on the title. DedupeCount uint32 `json:"dedupe_count"` } _ = json.Unmarshal(env.Alert, &ah) companyID := ah.CompanyID if companyID == "" { // subject is "deliveries.telegram." 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, ah.DedupeCount) 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. // // M6: when DedupeCount > 1, we append ` (×N)` to the title // so a glance at the chat shows how many times the source // has fired this alert inside the sliding dedupe window. // For DedupeCount == 0 (not set) or == 1 (first arrival), // no suffix is added. func formatMessage(sev alert.Severity, title, body, alertID string, dedupeCount uint32) string { var prefix string switch sev { case alert.SeverityInminentColapse: prefix = "🟥🟥🟥 IMMINENT" case alert.SeverityCritical: prefix = "🟥 CRITICAL" case alert.SeverityWarning: prefix = "🟧 WARNING" default: prefix = "🟦 INFO" } // M6: inline `(×N)` suffix on the title when this alert // was suppressed N-1 times by the dedupe gate. We render // the count as a small visual hint that doesn't // interfere with the severity prefix or the body. displayedTitle := title if dedupeCount > 1 { displayedTitle = fmt.Sprintf("%s (×%d)", title, dedupeCount) } out := fmt.Sprintf("%s: %s", prefix, displayedTitle) 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