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.
// M8 adds the retry + DLQ chain (SPEC §9).
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"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/dlq"
	"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/retry"
	"git3.techno-world.net/lrosales/broad-announce/internal/telegram"
	"github.com/nats-io/nats.go/jetstream"
)

const telegramChannel = "telegram"

// deliverdMetrics is the package-level metrics instance.
// Wired in main() from observability.NewDeliverdMetrics.
var deliverdMetrics *observability.DeliverdMetrics

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,
		"max_attempts", cfg.DeliverdMaxAttempts,
		"retry_base_ms", cfg.DeliverdRetryBaseMs,
		"retry_max_ms", cfg.DeliverdRetryMaxMs,
		"retry_budget_ms", cfg.DeliverdRetryBudgetMs,
	)

	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},
	}

	// M8: same retry config as deliverd-fcm. See
	// internal/retry for the policy.
	retryCfg := retry.Config{
		MaxAttempts: cfg.DeliverdMaxAttempts,
		BaseDelay:   time.Duration(cfg.DeliverdRetryBaseMs) * time.Millisecond,
		MaxDelay:    time.Duration(cfg.DeliverdRetryMaxMs) * time.Millisecond,
		Budget:      time.Duration(cfg.DeliverdRetryBudgetMs) * time.Millisecond,
	}

	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()

	// M9: deliverd metrics. Separate registry from deliverd-fcm
	// so each gets its own service label in Prometheus.
	reg, _ := observability.NewRegistry("deliverd-telegram")
	deliverdMetrics = observability.NewDeliverdMetrics(reg, "deliverd-telegram")

	go consume(runCtx, logger, consumer, pool, client, botToken, retryCfg)
	srv := httpserver.New(httpserver.Config{
		Addr:          cfg.HTTPAddr,
		ServiceName:   "deliverd-telegram",
		ShutdownGrace: cfg.ShutdownGrace,
	}, logger, observability.MetricsHandler(reg))

	// M13a W5: admin routes (JWT-gated). When BA_AUTHD_JWT_SECRET
	// is unset, wireAdminRoutes is a no-op so the LAN deploy
	// path keeps working unchanged.
	wireAdminRoutes(srv.Mux(), pool, logger)

	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, retryCfg retry.Config) {
	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, retryCfg)
			if batch.Error() != nil {
				logger.Warn("batch error", "err", batch.Error())
				break
			}
		}
	}
}

// handleOne mirrors deliverd-fcm's structure:
//   1. Parse the envelope (drop poison messages).
//   2. Retry loop: each attempt calls SendMessage and
//      writes a `deliveries` audit row. On success,
//      break out. On exhaustion, write a `deliveries_dlq`
//      row + a final 'dlq' audit row.
//   3. Ack the NATS message either way.
func handleOne(ctx context.Context, logger *slog.Logger, m jetstream.Msg, pool *postgres.Pool, client telegram.BotClient, botToken string, retryCfg retry.Config) {
	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
	}

	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"`
		DedupeCount uint32         `json:"dedupe_count"`
	}
	_ = json.Unmarshal(env.Alert, &ah)

	companyID := ah.CompanyID
	if companyID == "" {
		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 once, retry the same text. Format
	// unchanged from M3.
	text := formatMessage(ah.Severity, ah.Title, ah.Body, ah.ID, ah.DedupeCount)

	// M9: track attempt start time for DLQ latency metric.
	firstAttemptTime := time.Now()

	// M8: retry loop. Telegram returns errors on:
	//   - 4xx (chat not found, bot blocked, etc.) — these
	//     are permanent; we use PermanentError to skip
	//     further retries.
	//   - 5xx, network timeouts — these are transient;
	//     retry with backoff.
	//   - 429 (rate limited) — transient; retry.
	var lastErr error
	var sent bool
	res := retry.Run(ctx, retryCfg, func(attemptCtx context.Context, attempt int) error {
		attemptCtx, cancel := context.WithTimeout(attemptCtx, 15*time.Second)
		defer cancel()
		_, sendErr := client.SendMessage(attemptCtx, botToken, chatID, text)
		status := "failed"
		lastErrStr := ""
		if sendErr == nil {
			status = "sent"
			sent = true
		} else {
			lastErrStr = sendErr.Error()
		}

		_, dbErr := pool.Exec(attemptCtx, `
			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, $7, NULLIF($8, ''), $9,
			    CASE WHEN $6 = 'sent' THEN now() ELSE NULL END)
		`, ah.ID, companyID, env.IndividualID, telegramChannel, env.Endpoint, status, attempt, lastErrStr, json.RawMessage(m.Data()))
		if dbErr != nil {
			logger.Warn("delivery row insert", "err", dbErr, "attempt", attempt)
		}

		// M9: record per-channel delivery attempt metric.
		if deliverdMetrics != nil {
			deliverdMetrics.DeliveryAttempts.WithLabelValues(telegramChannel, status).Inc()
		}

		if status == "sent" {
			logger.Info("delivery sent",
				"alert_id", ah.ID,
				"company", companyID,
				"individual", env.IndividualID,
				"channel", telegramChannel,
				"attempt", attempt,
			)
			return nil
		}

		// Permanent-error detection. The Telegram Bot API
		// returns errors of the form "telegram api <code>:
		// <description>". 4xx codes (400, 401, 403, 404,
		// 409) are not retryable; 429 (Too Many Requests) is
		// transient and stays in the retry path. The
		// faketgmd test double mirrors this convention.
		if isTelegramPermanent(lastErrStr) {
			logger.Warn("permanent telegram error",
				"alert_id", ah.ID,
				"err", lastErrStr,
				"attempt", attempt,
			)
			lastErr = &retry.PermanentError{Err: fmt.Errorf("telegram permanent %s", lastErrStr)}
			return lastErr
		}
		logger.Warn("delivery attempt failed",
			"alert_id", ah.ID,
			"company", companyID,
			"channel", telegramChannel,
			"attempt", attempt,
			"err", lastErrStr,
		)
		lastErr = fmt.Errorf("%s", lastErrStr)
		return lastErr
	})

	if sent {
		_ = m.Ack()
		return
	}

	// All attempts exhausted (or permanent error). DLQ
	// insert for the operator UI; replay re-INSERTs the
	// original payload into the same NATS subject.
	dlqID, dlqErr := dlq.Write(ctx, pool, dlq.Entry{
		AlertID:         ah.ID,
		CompanyID:       companyID,
		IndividualID:    env.IndividualID,
		Channel:         telegramChannel,
		Target:          env.Endpoint,
		OriginalSubject: m.Subject(),
		Attempts:        res.Attempts,
		LastError:       errString(lastErr),
		Payload:         json.RawMessage(m.Data()),
	})
	if dlqErr != nil {
		logger.Error("dlq insert FAILED",
			"err", dlqErr,
			"alert_id", ah.ID,
			"company", companyID,
		)
	} else {
		logger.Warn("delivery → DLQ",
			"dlq_id", dlqID,
			"alert_id", ah.ID,
			"company", companyID,
			"channel", telegramChannel,
			"attempts", res.Attempts,
			"err", errString(lastErr),
		)
		// M9: record DLQ metric and latency.
		if deliverdMetrics != nil {
			deliverdMetrics.DLQTotal.WithLabelValues(telegramChannel).Inc()
			deliverdMetrics.DLQLatency.Observe(time.Since(firstAttemptTime).Seconds())
			deliverdMetrics.RetryAttempts.WithLabelValues(telegramChannel).Add(float64(res.Attempts))
		}
	}
	_ = m.Ack()
}

// isTelegramPermanent classifies a Telegram Bot API
// error string as permanent (no point retrying). The
// faketgmd test double prefixes errors with "telegram
// api <code>:" so we can match on the code.
//
// 429 (rate limited) is intentionally NOT permanent —
// the retry helper will space it out via backoff.
func isTelegramPermanent(errStr string) bool {
	if errStr == "" {
		return false
	}
	// Look for "telegram api NNN" or "status NNN" in the
	// error string. 4xx (except 429) is permanent.
	for _, prefix := range []string{"telegram api ", "status "} {
		i := strings.Index(errStr, prefix)
		if i < 0 {
			continue
		}
		rest := errStr[i+len(prefix):]
		// Parse a 3-digit code.
		if len(rest) < 3 {
			continue
		}
		codeStr := rest[:3]
		var code int
		if _, err := fmt.Sscanf(codeStr, "%d", &code); err != nil {
			continue
		}
		if code >= 400 && code < 500 && code != 429 && code != 408 {
			return true
		}
	}
	return false
}

// 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"
	}
	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
}

func errString(err error) string {
	if err == nil {
		return ""
	}
	return err.Error()
}