broad-announce/cmd/deliverd-telegram/main.go

// 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"`
		// 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.<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, 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