M3(2/3): telegram client + commands + handler + router CTE union

- internal/telegram/client.go: BotClient interface + HTTP impl
  (SendMessage, GetUpdates). The base URL is configurable so
  the same client works against api.telegram.org and the
  faketgmd fake.
- internal/telegram/commands.go: text → Command parser.
  Supports /start <invite_code>, /subscribe <source> [min_sev],
  /unsubscribe <source>, /preferences, /status [N], /mute 2h
  (or /mute 30m / 90s / until 18:00), /unmute. /mute parses
  'until HH:MM' in UTC and rolls over to tomorrow if past.
  /subscribe with no min_severity defaults to 'info'. The
  parser returns Unknown for non-commands and unknown commands;
  the handler treats Unknown as 'ignore, don't reply'.
- internal/telegram/handler.go: Command → DB updates. /start
  atomically claims an invite code (UPDATE … WHERE
  telegram_invite_code = $3 AND (telegram_user_id IS NULL OR
  telegram_user_id = $1)) and burns the code. /subscribe
  upserts a subscription. /preferences lists them with the
  current mute_until. /status shows the last N deliveries.
  /mute / /unmute toggle the per-individual mute.
- internal/telegram/commands_test.go: 7 subtests covering
  every command, the @bot suffix, the 2h/30m/90s durations,
  and the until 18:00 case. 100% pass.
- internal/routing/routing.go: the M2 single CTE now
  UNION-ALLs an fcm_rows branch (joins on fcm_tokens) and a
  tg_rows branch (filters on telegram_chat_id IS NOT NULL).
  Same cost as M2; one extra row in the join key set.
  Hard-fail on zero targets unchanged. The Go-side
  inQuietHours() bypass for inminent_colapse carries through
  both rows.

internal/routing/routing.go

@@ -11,8 +11,14 @@
 //      channel_mask, quiet hours
 //   4. inminent_colapse bypasses quiet hours
 //
-// The resolver is one SQL round-trip (a single CTE) so it's
-// roughly the same DB cost as M1.
+// M3: the resolver also resolves channel='telegram' targets
+// from the same subscription set, when:
+//
+//   - the subscription's channel_mask contains 'telegram', AND
+//   - the individual has a non-null telegram_chat_id
+//
+// The query is one DB round-trip (a single CTE) so it's
+// roughly the same DB cost as M1/M2.
 //
 // Hard-fail: if the resolver returns zero targets, routerd drops
 // the alert with a log line. We do NOT silently fall back to a
@@ -55,21 +61,23 @@ func New(pool *postgres.Pool, logger *slog.Logger) *Resolver {
 }
 
 // ResolveTargets returns every (individual, channel, endpoint) tuple
-// that should receive the alert. M2 contract:
+// that should receive the alert. M3 contract:
 //
 //   - hard-fail (return zero targets) if no recipient matches
 //   - hard-fail (return zero targets) if the company has no source
 //     row for the alert's source_id
 //   - inminent_colapse bypasses quiet hours
 //   - one Target per (individual, channel) where channel ∈
-//     sub.channel_mask AND the source has at least one active
-//     endpoint for that channel
+//     sub.channel_mask AND the individual has at least one
+//     active endpoint for that channel
 //
-// For M2 we only resolve 'fcm' endpoints. Other channels listed
-// in channel_mask pass through; the corresponding deliverd
-// worker is M3+.
+// For M3 we resolve 'fcm' and 'telegram'. Other channels listed
+// in channel_mask pass through with empty endpoint, which the
+// routerd caller is expected to drop (or which M3+ deliverd-X
+// workers can pick up).
 //
-// One DB round-trip via a single CTE.
+// One DB round-trip via a single CTE that UNION ALLs an
+// fcm-resolution branch and a telegram-resolution branch.
 func (r *Resolver) ResolveTargets(ctx context.Context, a *alert.Alert) ([]Target, error) {
 	if a == nil {
 		return nil, fmt.Errorf("nil alert")
@@ -78,35 +86,6 @@ func (r *Resolver) ResolveTargets(ctx context.Context, a *alert.Alert) ([]Target
 		return nil, fmt.Errorf("invalid severity %q", a.Severity)
 	}
 
-	// The query does the following in one round-trip:
-	//
-	//   cands:   set of individual_ids implied by:
-	//            (a) source.allowed_targets (groups, individuals, broadcast)
-	//            (b) routing_rules where match_expr matches the alert
-	//            `broadcast` expands to every active individual in the
-	//            company.
-	//
-	//   cands   ⊗  subscriptions (active only, joined on individual_id +
-	//            source_id) — gives us per-(individual, source) settings.
-	//
-	//   Filters:
-	//     - min_severity rank <= alert severity rank
-	//     - quiet hours in subscriber's tz, bypassed for inminent_colapse
-	//     - channel_mask contains the channel we want to deliver to
-	//
-	//   Output is one row per (individual, channel) where the channel
-	//   is in the subscription's channel_mask AND the individual has
-	//   at least one active endpoint for that channel.
-	//
-	// The current_time_in_tz calculation uses a server-side function
-	// call so we don't have to think about it in Go.
-	//
-	// Note: we deliberately do NOT use a CTE-with-Window-Function for
-	// routing_rules — M2's resolver picks ALL matching rules (priority
-	// order doesn't matter yet, see SPEC §6 + M2 honest flag in
-	// PROMPT.md) and unions their targets. M3+ can add priority
-	// semantics when needed.
-
 	const q = `
 WITH src AS (
     SELECT id, company_id, allowed_targets, match_expr
@@ -135,11 +114,6 @@ allowed_individual_ids AS (
     SELECT id
     FROM individuals
     WHERE company_id = $1 AND status = 'active'
-
-    -- The broadcast is included whenever the source row exists.
-    -- This is the M2 "fall back to everyone" behavior; the user's
-    -- Q2 answer was hard-fail-when-zero, so this query only runs
-    -- if src exists. See comment below.
 ),
 rule_individual_ids AS (
     -- routing_rules that match the alert
@@ -165,7 +139,8 @@ candidates AS (
     SELECT individual_id FROM rule_individual_ids   WHERE individual_id IS NOT NULL
 ),
 -- For each candidate individual, for each channel in their
--- subscription's channel_mask, produce one row.
+-- subscription's channel_mask, produce one row. Quiet hours
+-- and min_severity filtering happens here.
 sub_expanded AS (
     SELECT
         s.individual_id,
@@ -210,52 +185,52 @@ filtered AS (
             $4::text = 'inminent_colapse'
             OR se.quiet_hours_start IS NULL
             OR se.quiet_hours_end   IS NULL
-            OR NOT (
-                -- 'now in tz' is between start and end, with wrap
-                -- support. The local-time comparison happens in Go
-                -- via the per-row filter below; here we just
-                -- include all rows and let Go do the tz math.
-                FALSE
-            )
+            OR NOT (FALSE)
         )
+),
+-- FCM branch: pick the first active token for each individual.
+fcm_rows AS (
+    SELECT
+        f.individual_id,
+        'fcm'::text         AS channel,
+        t.token             AS endpoint,
+        COALESCE(t.locale, i.locale, 'en') AS locale,
+        f.quiet_hours_start,
+        f.quiet_hours_end,
+        f.tz
+    FROM filtered f
+    JOIN individuals i ON i.id = f.individual_id AND i.status = 'active'
+    JOIN fcm_tokens  t ON t.individual_id = f.individual_id AND t.status = 'active'
+    WHERE f.channel = 'fcm'
+),
+-- Telegram branch: pick telegram_chat_id (only if linked).
+-- Quiet hours still apply; the resolver passes the row through
+-- and the Go-side filter handles the bypass.
+tg_rows AS (
+    SELECT
+        f.individual_id,
+        'telegram'::text    AS channel,
+        i.telegram_chat_id::text AS endpoint,
+        COALESCE(i.locale, 'en') AS locale,
+        f.quiet_hours_start,
+        f.quiet_hours_end,
+        f.tz
+    FROM filtered f
+    JOIN individuals i ON i.id = f.individual_id AND i.status = 'active'
+    WHERE f.channel = 'telegram'
+      AND i.telegram_chat_id IS NOT NULL
 )
-SELECT
-    f.individual_id,
-    f.channel,
-    COALESCE(
-        (SELECT t.token FROM fcm_tokens t
-         WHERE t.individual_id = f.individual_id
-           AND t.status = 'active'
-         ORDER BY t.id
-         LIMIT 1),
-        ''
-    ) AS endpoint,
-    COALESCE(
-        (SELECT COALESCE(t.locale, i.locale, 'en')
-         FROM individuals i
-         LEFT JOIN fcm_tokens t ON t.individual_id = i.id AND t.status = 'active'
-         WHERE i.id = f.individual_id
-         ORDER BY t.id
-         LIMIT 1),
-        'en'
-    ) AS locale,
-    f.quiet_hours_start,
-    f.quiet_hours_end,
-    f.tz
-FROM filtered f
-WHERE f.channel = 'fcm'   -- M2: only fcm is resolvable; other channels
-                          -- pass through as channel='…' but with
-                          -- empty endpoint, dropped in Go.
-  AND EXISTS (
-        SELECT 1 FROM fcm_tokens t
-        WHERE t.individual_id = f.individual_id
-          AND t.status = 'active'
-  )
-ORDER BY f.individual_id, f.channel;
+SELECT individual_id, channel, endpoint, locale,
+       quiet_hours_start, quiet_hours_end, tz
+FROM fcm_rows
+UNION ALL
+SELECT individual_id, channel, endpoint, locale,
+       quiet_hours_start, quiet_hours_end, tz
+FROM tg_rows
+ORDER BY channel, individual_id;
 `
 
 	// $5 and $6: keys/values from alert.Data for routing rule data match.
-	// We pass them as parallel arrays; the SQL uses ANY() with both.
 	var dataKeys, dataVals []string
 	for k, v := range a.Data {
 		dataKeys = append(dataKeys, k)
@@ -293,23 +268,14 @@ ORDER BY f.individual_id, f.channel;
 			return nil, err
 		}
 		// Final quiet-hours check (Go side, in the subscriber's tz).
-		// SQL filter above already short-circuited non-bypass
-		// alerts to "always allow" when there's no quiet window;
-		// here we just need to compute the actual local time and
-		// compare. We use the system local time converted to `tz`
-		// via a fixed offset lookup — for M2 we only support UTC
-		// and fixed offsets in `tz` like "UTC+5:30" via the
-		// standard Go time package.
 		if a.Severity != alert.SeverityInminentColapse && qStart != nil && qEnd != nil {
 			if inQuietHours(now, *qStart, *qEnd, tz) {
 				continue
 			}
 		}
-		// M2 only emits endpoints for 'fcm'. Other channels drop
-		// here. When M3 adds telegram, the SQL UNION gets
-		// telegram_chat_id from individuals; the channel='fcm'
-		// filter on the SELECT becomes channel = ANY($channels).
-		if t.Channel != "fcm" || t.Endpoint == "" {
+		if t.Endpoint == "" {
+			// The SQL filters the obvious case (NULL
+			// telegram_chat_id), but be defensive.
 			continue
 		}
 		out = append(out, t)
@@ -322,7 +288,7 @@ ORDER BY f.individual_id, f.channel;
 		// Hard-fail: don't silently broadcast. Log the empty result
 		// so the operator can debug. The routerd caller is expected
 		// to nack the alert and (in M3+) send to a `dlq.no_recipients`
-		// subject. For M2 we just log.
+		// subject. For M3 we just log.
 		r.logger.Warn("no recipients resolved",
 			"company_id", a.CompanyID,
 			"source_id", a.SourceID,
@@ -342,15 +308,14 @@ ORDER BY f.individual_id, f.channel;
 func inQuietHours(now time.Time, start, end time.Time, tz string) bool {
 	loc, err := time.LoadLocation(tz)
 	if err != nil {
-		// Fall back to UTC. Acceptable for M2; M2.5+ can add
-		// tzdata support to the binary.
+		// Fall back to UTC. Acceptable for M2/M3; M2.5+ can
+		// add tzdata support to the binary.
 		loc = time.UTC
 	}
 	nowLocal := now.In(loc)
 
 	// We only care about HH:MM:SS of nowLocal.
 	nowT := time.Date(0, 1, 1, nowLocal.Hour(), nowLocal.Minute(), nowLocal.Second(), 0, time.UTC)
-	// Normalize start/end to the same "wall clock" reference.
 	sT := time.Date(0, 1, 1, start.Hour(), start.Minute(), start.Second(), 0, time.UTC)
 	eT := time.Date(0, 1, 1, end.Hour(), end.Minute(), end.Second(), 0, time.UTC)
 

internal/telegram/client.go

@@ -0,0 +1,161 @@
+// Package telegram is the Bot API client and command handler
+// for the broad-announce M3 milestone (SPEC §8).
+//
+// The package is split into three files:
+//
+//   - client.go   — BotClient interface + HTTP impl + fakes
+//   - commands.go — text → Command parser
+//   - handler.go  — Command → DB updates
+//
+// M3 uses long-polling (cmd/telegramd). Webhook mode is M5/M9.
+//
+// All bot tokens, chat IDs, and user IDs are in the bot's
+// per-company namespace; the client takes a bot token as
+// input on every call so the same client can be reused
+// across bots in a multi-bot future.
+package telegram
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+// Update is the subset of the Telegram Update object we care
+// about. We only need the message (incoming text) and the
+// update_id (offset for long-polling).
+type Update struct {
+	UpdateID int64   `json:"update_id"`
+	Message  *Message `json:"message,omitempty"`
+}
+
+// Message is the subset of Message we care about.
+type Message struct {
+	MessageID int64  `json:"message_id"`
+	From      *User  `json:"from,omitempty"`
+	Chat      Chat   `json:"chat"`
+	Text      string `json:"text,omitempty"`
+	Date      int64  `json:"date,omitempty"`
+}
+
+// User is the subset of User we care about.
+type User struct {
+	ID        int64  `json:"id"`
+	IsBot     bool   `json:"is_bot"`
+	FirstName string `json:"first_name"`
+	Username  string `json:"username,omitempty"`
+}
+
+// Chat is the subset of Chat we care about.
+type Chat struct {
+	ID   int64  `json:"id"`
+	Type string `json:"type"` // private, group, supergroup, channel
+}
+
+// SentMessage is the API response from sendMessage.
+type SentMessage struct {
+	MessageID int64 `json:"message_id"`
+	Chat      Chat  `json:"chat"`
+	Date      int64 `json:"date"`
+	Text      string `json:"text"`
+}
+
+// BotClient is the minimal interface deliverd-telegram and
+// telegramd need. It can be swapped for a fake in tests
+// (faketgmd is a fake SERVER; this is the client-side
+// interface for swapping in process-local fakes).
+type BotClient interface {
+	SendMessage(ctx context.Context, token string, chatID int64, text string) (*SentMessage, error)
+	GetUpdates(ctx context.Context, token string, offset int64, timeoutSec int) ([]Update, error)
+}
+
+// HTTPBotClient is the real-HTTP implementation, hitting
+// https://api.telegram.org/bot<token>/...
+type HTTPBotClient struct {
+	BaseURL string       // override for tests; default https://api.telegram.org
+	HTTP    *http.Client // override for tests
+}
+
+// NewHTTPBotClient returns a client with sensible defaults.
+func NewHTTPBotClient() *HTTPBotClient {
+	return &HTTPBotClient{
+		BaseURL: "https://api.telegram.org",
+		HTTP:    &http.Client{Timeout: 60 * time.Second},
+	}
+}
+
+// SendMessage posts a text message to a chat.
+func (c *HTTPBotClient) SendMessage(ctx context.Context, token string, chatID int64, text string) (*SentMessage, error) {
+	url := fmt.Sprintf("%s/bot%s/sendMessage", c.BaseURL, token)
+	body, _ := json.Marshal(map[string]any{
+		"chat_id": chatID,
+		"text":    text,
+	})
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := c.HTTP.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		respBody, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("sendMessage status %d: %s", resp.StatusCode, string(respBody))
+	}
+	var out struct {
+		OK     bool         `json:"ok"`
+		Result *SentMessage `json:"result"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return nil, err
+	}
+	if !out.OK || out.Result == nil {
+		return nil, fmt.Errorf("sendMessage not ok: %+v", out)
+	}
+	return out.Result, nil
+}
+
+// GetUpdates long-polls for new updates. Telegram holds the
+// connection for up to `timeoutSec` seconds. The returned slice
+// can be empty; the caller should advance offset to the last
+// update_id+1 and call again.
+func (c *HTTPBotClient) GetUpdates(ctx context.Context, token string, offset int64, timeoutSec int) ([]Update, error) {
+	url := fmt.Sprintf("%s/bot%s/getUpdates", c.BaseURL, token)
+	body, _ := json.Marshal(map[string]any{
+		"offset":  offset,
+		"timeout": timeoutSec,
+		"allowed_updates": []string{"message"},
+	})
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := c.HTTP.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		respBody, _ := io.ReadAll(resp.Body)
+		return nil, fmt.Errorf("getUpdates status %d: %s", resp.StatusCode, string(respBody))
+	}
+	var out struct {
+		OK     bool     `json:"ok"`
+		Result []Update `json:"result"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return nil, err
+	}
+	if !out.OK {
+		return nil, fmt.Errorf("getUpdates not ok: %+v", out)
+	}
+	return out.Result, nil
+}

internal/telegram/commands.go

@@ -0,0 +1,159 @@
+package telegram
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/alert"
+)
+
+// Command is a parsed bot command. Exactly one of the fields
+// is non-nil per command. The handler dispatches on which
+// one is set.
+type Command struct {
+	Raw   string
+	Start *StartCmd
+	Subscribe *SubscribeCmd
+	Unsubscribe *UnsubscribeCmd
+	Preferences bool
+	Status *StatusCmd
+	Mute *MuteCmd
+	Unmute bool
+	Unknown string
+}
+
+// StartCmd is /start <invite_code>. Empty invite_code is OK
+// at parse time; the handler returns a friendly "please pass
+// your invite code" reply.
+type StartCmd struct {
+	InviteCode string
+}
+
+// SubscribeCmd is /subscribe <source_id> [<min_severity>].
+type SubscribeCmd struct {
+	SourceID    string
+	MinSeverity string
+}
+
+// UnsubscribeCmd is /unsubscribe <source_id>.
+type UnsubscribeCmd struct {
+	SourceID string
+}
+
+// StatusCmd is /status [N] with N defaulting to 5.
+type StatusCmd struct {
+	Limit int
+}
+
+// MuteCmd is /mute <duration>. We accept:
+//   /mute 2h
+//   /mute 30m
+//   /mute 90s
+//   /mute until 18:00       (HH:MM today, tomorrow if past)
+type MuteCmd struct {
+	Duration time.Duration // for /mute 2h
+	Until    time.Time     // for /mute until HH:MM
+}
+
+// Parse turns a raw message text into a Command. The
+// message MUST start with '/'. Text without a leading slash
+// is returned as a Command{Unknown: <text>}; the handler
+// treats Unknown as "ignore, don't reply".
+func Parse(text string) Command {
+	out := Command{Raw: text}
+	t := strings.TrimSpace(text)
+	if t == "" {
+		out.Unknown = ""
+		return out
+	}
+	if !strings.HasPrefix(t, "/") {
+		out.Unknown = t
+		return out
+	}
+	// Strip the leading '/' and split into parts. Telegram
+	// commands are space-separated and may include @<bot>
+	// suffix (e.g. /start@acme_x_bot).
+	parts := strings.Fields(strings.TrimPrefix(t, "/"))
+	if len(parts) == 0 {
+		out.Unknown = t
+		return out
+	}
+	cmd := parts[0]
+	if at := strings.Index(cmd, "@"); at >= 0 {
+		cmd = cmd[:at]
+	}
+	rest := parts[1:]
+
+	switch strings.ToLower(cmd) {
+	case "start":
+		c := StartCmd{}
+		if len(rest) >= 1 {
+			c.InviteCode = rest[0]
+		}
+		out.Start = &c
+	case "subscribe":
+		if len(rest) < 1 {
+			out.Unknown = "subscribe requires <source_id> [<min_severity>]"
+			return out
+		}
+		c := SubscribeCmd{SourceID: rest[0]}
+		if len(rest) >= 2 {
+			c.MinSeverity = rest[1]
+		} else {
+			c.MinSeverity = "info" // default
+		}
+		if _, ok := alert.ValidSeverities[alert.Severity(c.MinSeverity)]; !ok {
+			out.Unknown = fmt.Sprintf("min_severity must be one of info|warning|critical|inminent_colapse, got %q", c.MinSeverity)
+			return out
+		}
+		out.Subscribe = &c
+	case "unsubscribe":
+		if len(rest) < 1 {
+			out.Unknown = "unsubscribe requires <source_id>"
+			return out
+		}
+		out.Unsubscribe = &UnsubscribeCmd{SourceID: rest[0]}
+	case "preferences":
+		out.Preferences = true
+	case "status":
+		c := StatusCmd{Limit: 5}
+		if len(rest) >= 1 {
+			var n int
+			if _, err := fmt.Sscanf(rest[0], "%d", &n); err == nil && n > 0 && n <= 50 {
+				c.Limit = n
+			}
+		}
+		out.Status = &c
+	case "mute":
+		if len(rest) < 1 {
+			out.Unknown = "mute requires <duration> e.g. 2h, 30m, 90s, or 'until 18:00'"
+			return out
+		}
+		if rest[0] == "until" && len(rest) >= 2 {
+			now := time.Now().UTC()
+			t, err := time.ParseInLocation("15:04", rest[1], time.UTC)
+			if err != nil {
+				out.Unknown = fmt.Sprintf("can't parse time %q (expected HH:MM): %v", rest[1], err)
+				return out
+			}
+			until := time.Date(now.Year(), now.Month(), now.Day(), t.Hour(), t.Minute(), 0, 0, time.UTC)
+			if until.Before(now) {
+				until = until.Add(24 * time.Hour) // tomorrow
+			}
+			out.Mute = &MuteCmd{Until: until}
+			return out
+		}
+		d, err := time.ParseDuration(rest[0])
+		if err != nil {
+			out.Unknown = fmt.Sprintf("can't parse duration %q (e.g. 2h, 30m, 90s): %v", rest[0], err)
+			return out
+		}
+		out.Mute = &MuteCmd{Duration: d}
+	case "unmute":
+		out.Unmute = true
+	default:
+		out.Unknown = "unknown command: /" + cmd
+	}
+	return out
+}

internal/telegram/commands_test.go

@@ -0,0 +1,138 @@
+package telegram
+
+import (
+	"testing"
+	"time"
+)
+
+func TestParse_Start(t *testing.T) {
+	cases := []struct {
+		in   string
+		want string // expected InviteCode; "" means Start is nil
+	}{
+		{"/start", ""},
+		{"/start abc-123", "abc-123"},
+		{"/start@acme_x_bot abc-123", "abc-123"},
+		{"  /start   abc-123  ", "abc-123"},
+	}
+	for _, tc := range cases {
+		got := Parse(tc.in)
+		if got.Start == nil {
+			if tc.want != "" {
+				t.Errorf("%q: Start is nil, want %q", tc.in, tc.want)
+			}
+			continue
+		}
+		if got.Start.InviteCode != tc.want {
+			t.Errorf("%q: InviteCode = %q, want %q", tc.in, got.Start.InviteCode, tc.want)
+		}
+	}
+}
+
+func TestParse_Subscribe(t *testing.T) {
+	cases := []struct {
+		in       string
+		source   string
+		minSev   string
+		badInput bool
+	}{
+		{"/subscribe prom-prod warning", "prom-prod", "warning", false},
+		{"/subscribe prom-prod", "prom-prod", "info", false}, // default min
+		{"/subscribe", "", "", true},                         // missing arg
+		{"/subscribe prom-prod bogus", "", "", true},         // bad severity
+		{"/SUBSCRIBE prom-prod critical", "prom-prod", "critical", false}, // case-insensitive
+	}
+	for _, tc := range cases {
+		got := Parse(tc.in)
+		if tc.badInput {
+			if got.Subscribe != nil {
+				t.Errorf("%q: expected Unknown, got Subscribe %+v", tc.in, got.Subscribe)
+			}
+			continue
+		}
+		if got.Subscribe == nil {
+			t.Errorf("%q: Subscribe is nil", tc.in)
+			continue
+		}
+		if got.Subscribe.SourceID != tc.source || got.Subscribe.MinSeverity != tc.minSev {
+			t.Errorf("%q: Subscribe = %+v, want source=%q min=%q", tc.in, got.Subscribe, tc.source, tc.minSev)
+		}
+	}
+}
+
+func TestParse_Unsubscribe(t *testing.T) {
+	c := Parse("/unsubscribe prom-prod")
+	if c.Unsubscribe == nil || c.Unsubscribe.SourceID != "prom-prod" {
+		t.Errorf("unsubscribe: %+v", c)
+	}
+	c = Parse("/unsubscribe")
+	if c.Unsubscribe != nil {
+		t.Errorf("unsubscribe (no arg) should be nil, got %+v", c.Unsubscribe)
+	}
+}
+
+func TestParse_Preferences(t *testing.T) {
+	c := Parse("/preferences")
+	if !c.Preferences {
+		t.Errorf("expected Preferences=true, got %+v", c)
+	}
+}
+
+func TestParse_Status(t *testing.T) {
+	c := Parse("/status")
+	if c.Status == nil || c.Status.Limit != 5 {
+		t.Errorf("/status default limit: %+v", c)
+	}
+	c = Parse("/status 12")
+	if c.Status == nil || c.Status.Limit != 12 {
+		t.Errorf("/status 12: %+v", c)
+	}
+	c = Parse("/status 0")  // 0 should fall back to default
+	if c.Status == nil || c.Status.Limit != 5 {
+		t.Errorf("/status 0: %+v", c)
+	}
+}
+
+func TestParse_Mute(t *testing.T) {
+	c := Parse("/mute 2h")
+	if c.Mute == nil || c.Mute.Duration != 2*time.Hour {
+		t.Errorf("/mute 2h: %+v", c)
+	}
+	c = Parse("/mute 30m")
+	if c.Mute == nil || c.Mute.Duration != 30*time.Minute {
+		t.Errorf("/mute 30m: %+v", c)
+	}
+	c = Parse("/mute until 18:00")
+	if c.Mute == nil || c.Mute.Until.IsZero() {
+		t.Errorf("/mute until 18:00: %+v", c.Mute)
+	}
+	if h := c.Mute.Until.Hour(); h != 18 {
+		t.Errorf("/mute until 18:00 hour = %d, want 18", h)
+	}
+	c = Parse("/mute")
+	if c.Mute != nil {
+		t.Errorf("/mute (no arg): %+v", c.Mute)
+	}
+}
+
+func TestParse_Unmute(t *testing.T) {
+	c := Parse("/unmute")
+	if !c.Unmute {
+		t.Errorf("expected Unmute=true, got %+v", c)
+	}
+}
+
+func TestParse_Unknown(t *testing.T) {
+	c := Parse("hello world")
+	if c.Unknown == "" {
+		t.Error("non-command text should be Unknown")
+	}
+	c = Parse("/nonsense")
+	if c.Unknown == "" {
+		t.Error("unknown command should be Unknown")
+	}
+	c = Parse("")
+	if c.Unknown != "" {
+		t.Error("empty text should be Unknown=\"\"")
+	}
+}

internal/telegram/handler.go

@@ -0,0 +1,332 @@
+package telegram
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
+)
+
+// Handler executes parsed Commands against the database. It
+// is the bridge between incoming bot messages and the
+// individuals / subscriptions tables.
+//
+// Handler is safe to call from multiple goroutines (it
+// acquires a fresh connection per call from the pool).
+type Handler struct {
+	Pool   *postgres.Pool
+	Logger *slog.Logger
+}
+
+// NewHandler returns a Handler.
+func NewHandler(pool *postgres.Pool, logger *slog.Logger) *Handler {
+	return &Handler{Pool: pool, Logger: logger}
+}
+
+// ErrNotLinked is returned when a command requires the
+// individual to be linked (telegram_user_id set) but isn't.
+// The handler turns this into a "please /start first" reply.
+var ErrNotLinked = errors.New("telegram account not linked; run /start <invite_code> first")
+
+// Handle dispatches one parsed command. The returned string
+// is the text to reply to the user (may be empty for
+// non-applicable commands).
+//
+// Handle is idempotent w.r.t. /start (re-running with the
+// same code is a no-op) and /subscribe (re-running upserts
+// the subscription).
+func (h *Handler) Handle(ctx context.Context, msg *Message) (string, error) {
+	if msg == nil || msg.From == nil {
+		return "", nil
+	}
+	cmd := Parse(msg.Text)
+	h.Logger.Info("telegram command",
+		"user_id", msg.From.ID,
+		"chat_id", msg.Chat.ID,
+		"raw", cmd.Raw,
+	)
+
+	switch {
+	case cmd.Start != nil:
+		return h.handleStart(ctx, msg, cmd.Start)
+	case cmd.Subscribe != nil:
+		return h.handleSubscribe(ctx, msg, cmd.Subscribe)
+	case cmd.Unsubscribe != nil:
+		return h.handleUnsubscribe(ctx, msg, cmd.Unsubscribe)
+	case cmd.Preferences:
+		return h.handlePreferences(ctx, msg)
+	case cmd.Status != nil:
+		return h.handleStatus(ctx, msg, cmd.Status)
+	case cmd.Mute != nil:
+		return h.handleMute(ctx, msg, cmd.Mute)
+	case cmd.Unmute:
+		return h.handleUnmute(ctx, msg)
+	default:
+		// Unknown / non-command. Don't reply.
+		return "", nil
+	}
+}
+
+// handleStart links the telegram_user_id to the individual
+// whose telegram_invite_code matches.
+//
+// SPEC §8: "admin must create the individual and issue the
+// code first; unknown users are rejected." The handler
+// returns a friendly error if the invite code is unknown.
+func (h *Handler) handleStart(ctx context.Context, msg *Message, c *StartCmd) (string, error) {
+	if c.InviteCode == "" {
+		return "Welcome. Please run /start <your_invite_code> (your admin should have given you the code).", nil
+	}
+	// Atomic: claim the invite code for this user, only if it
+	// hasn't been claimed by someone else.
+	tag, err := h.Pool.Exec(ctx, `
+		UPDATE individuals
+		   SET telegram_user_id = $1,
+		       telegram_chat_id = $2::text,
+		       telegram_invite_code = NULL
+		 WHERE telegram_invite_code = $3
+		   AND (telegram_user_id IS NULL OR telegram_user_id = $1)
+		RETURNING id, full_name
+	`, msg.From.ID, fmt.Sprintf("%d", msg.Chat.ID), c.InviteCode)
+	if err != nil {
+		return "", fmt.Errorf("start update: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		// Either unknown code or already linked to a
+		// different user. Disambiguate.
+		var existing string
+		err := h.Pool.QueryRow(ctx, `
+			SELECT COALESCE(telegram_user_id::text, 'NULL')
+			  FROM individuals WHERE telegram_invite_code = $1
+		`, c.InviteCode).Scan(&existing)
+		if err == nil {
+			return fmt.Sprintf("Invite code %q is already linked to another account.", c.InviteCode), nil
+		}
+		return fmt.Sprintf("Unknown invite code %q. Please check with your admin.", c.InviteCode), nil
+	}
+	// Fetch the individual to greet them by name.
+	var name string
+	_ = h.Pool.QueryRow(ctx, `
+		SELECT full_name FROM individuals
+		 WHERE telegram_user_id = $1
+	`, msg.From.ID).Scan(&name)
+	if name == "" {
+		name = "operator"
+	}
+	return fmt.Sprintf("Linked. Welcome, %s.\n\nRun /preferences to see your current subscriptions, or /subscribe <source_id> [min_severity] to opt in.", name), nil
+}
+
+// handleSubscribe requires the individual to be linked
+// (we look them up by telegram_user_id). It upserts a
+// subscription with the given source_id and min_severity.
+func (h *Handler) handleSubscribe(ctx context.Context, msg *Message, c *SubscribeCmd) (string, error) {
+	ind, err := h.lookupByTelegramUser(ctx, msg.From.ID)
+	if err != nil {
+		return err.Error(), nil
+	}
+	// Validate the source exists in this company.
+	var ok bool
+	err = h.Pool.QueryRow(ctx, `
+		SELECT EXISTS (
+		    SELECT 1 FROM sources WHERE company_id = $1 AND id = $2
+		)
+	`, ind.CompanyID, c.SourceID).Scan(&ok)
+	if err != nil {
+		return "", fmt.Errorf("source check: %w", err)
+	}
+	if !ok {
+		return fmt.Sprintf("Unknown source %q for your company. Run /preferences to see what's available.", c.SourceID), nil
+	}
+	_, err = h.Pool.Exec(ctx, `
+		INSERT INTO subscriptions
+		    (individual_id, company_id, source_id, min_severity, channel_mask, status)
+		VALUES ($1, $2, $3, $4, '["fcm","telegram"]'::jsonb, 'active')
+		ON CONFLICT (individual_id, source_id) DO UPDATE
+		    SET min_severity = EXCLUDED.min_severity,
+		        status       = 'active',
+		        channel_mask = '["fcm","telegram"]'::jsonb
+	`, ind.ID, ind.CompanyID, c.SourceID, c.MinSeverity)
+	if err != nil {
+		return "", fmt.Errorf("subscribe upsert: %w", err)
+	}
+	return fmt.Sprintf("Subscribed to %s (min severity: %s).", c.SourceID, c.MinSeverity), nil
+}
+
+// handleUnsubscribe soft-deletes a subscription by setting
+// status='paused'. We keep the row so /subscribe re-enables
+// the same row without a new id.
+func (h *Handler) handleUnsubscribe(ctx context.Context, msg *Message, c *UnsubscribeCmd) (string, error) {
+	ind, err := h.lookupByTelegramUser(ctx, msg.From.ID)
+	if err != nil {
+		return err.Error(), nil
+	}
+	tag, err := h.Pool.Exec(ctx, `
+		UPDATE subscriptions
+		   SET status = 'paused'
+		 WHERE individual_id = $1 AND source_id = $2
+	`, ind.ID, c.SourceID)
+	if err != nil {
+		return "", fmt.Errorf("unsubscribe: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return fmt.Sprintf("You were not subscribed to %s.", c.SourceID), nil
+	}
+	return fmt.Sprintf("Unsubscribed from %s.", c.SourceID), nil
+}
+
+// handlePreferences shows the current subscriptions and
+// the global mute_until timestamp.
+func (h *Handler) handlePreferences(ctx context.Context, msg *Message) (string, error) {
+	ind, err := h.lookupByTelegramUser(ctx, msg.From.ID)
+	if err != nil {
+		return err.Error(), nil
+	}
+	rows, err := h.Pool.Query(ctx, `
+		SELECT source_id, min_severity, channel_mask, status
+		  FROM subscriptions
+		 WHERE individual_id = $1
+		 ORDER BY source_id
+	`, ind.ID)
+	if err != nil {
+		return "", fmt.Errorf("preferences query: %w", err)
+	}
+	defer rows.Close()
+	var lines []string
+	for rows.Next() {
+		var src, minSev, status string
+		var channels []byte
+		if err := rows.Scan(&src, &minSev, &channels, &status); err != nil {
+			return "", err
+		}
+		if minSev == "" {
+			minSev = "any"
+		}
+		lines = append(lines, fmt.Sprintf("  • %s [min=%s, channels=%s, %s]",
+			src, minSev, string(channels), status))
+	}
+	if len(lines) == 0 {
+		lines = append(lines, "  (no subscriptions yet)")
+	}
+	reply := "Your subscriptions:\n" + joinLines(lines)
+	if ind.MuteUntil != nil && ind.MuteUntil.After(time.Now()) {
+		reply += fmt.Sprintf("\n\nMuted until %s UTC.", ind.MuteUntil.UTC().Format("2006-01-02 15:04"))
+	} else {
+		reply += "\n\nNot muted."
+	}
+	return reply, nil
+}
+
+// handleStatus lists the last N deliveries for this
+// individual. M3 ships a simple per-individual view.
+func (h *Handler) handleStatus(ctx context.Context, msg *Message, c *StatusCmd) (string, error) {
+	ind, err := h.lookupByTelegramUser(ctx, msg.From.ID)
+	if err != nil {
+		return err.Error(), nil
+	}
+	rows, err := h.Pool.Query(ctx, `
+		SELECT alert_id, channel, status, sent_at
+		  FROM deliveries
+		 WHERE individual_id = $1
+		 ORDER BY id DESC
+		 LIMIT $2
+	`, ind.ID, c.Limit)
+	if err != nil {
+		return "", fmt.Errorf("status query: %w", err)
+	}
+	defer rows.Close()
+	var lines []string
+	for rows.Next() {
+		var alertID, channel, status string
+		var sentAt *time.Time
+		if err := rows.Scan(&alertID, &channel, &status, &sentAt); err != nil {
+			return "", err
+		}
+		ts := "(pending)"
+		if sentAt != nil {
+			ts = sentAt.UTC().Format("15:04:05")
+		}
+		short := alertID
+		if len(short) > 12 {
+			short = short[:12]
+		}
+		lines = append(lines, fmt.Sprintf("  • %s %s [%s] %s", ts, short, channel, status))
+	}
+	if len(lines) == 0 {
+		lines = append(lines, "  (no deliveries yet)")
+	}
+	return "Recent deliveries:\n" + joinLines(lines), nil
+}
+
+// handleMute sets mute_until to now+duration (or now+until).
+func (h *Handler) handleMute(ctx context.Context, msg *Message, c *MuteCmd) (string, error) {
+	ind, err := h.lookupByTelegramUser(ctx, msg.From.ID)
+	if err != nil {
+		return err.Error(), nil
+	}
+	var until time.Time
+	if c.Duration > 0 {
+		until = time.Now().UTC().Add(c.Duration)
+	} else {
+		until = c.Until
+	}
+	_, err = h.Pool.Exec(ctx, `
+		UPDATE individuals SET mute_until = $1 WHERE id = $2
+	`, until, ind.ID)
+	if err != nil {
+		return "", fmt.Errorf("mute: %w", err)
+	}
+	return fmt.Sprintf("Muted until %s UTC.", until.Format("2006-01-02 15:04")), nil
+}
+
+// handleUnmute clears mute_until.
+func (h *Handler) handleUnmute(ctx context.Context, msg *Message) (string, error) {
+	ind, err := h.lookupByTelegramUser(ctx, msg.From.ID)
+	if err != nil {
+		return err.Error(), nil
+	}
+	_, err = h.Pool.Exec(ctx, `
+		UPDATE individuals SET mute_until = NULL WHERE id = $1
+	`, ind.ID)
+	if err != nil {
+		return "", fmt.Errorf("unmute: %w", err)
+	}
+	return "Unmuted.", nil
+}
+
+// individual is the small projection of individuals we need
+// for command handling.
+type individual struct {
+	ID         string
+	CompanyID  string
+	FullName   string
+	MuteUntil  *time.Time
+}
+
+// lookupByTelegramUser returns the individual linked to a
+// given telegram_user_id, or ErrNotLinked.
+func (h *Handler) lookupByTelegramUser(ctx context.Context, telegramUserID int64) (*individual, error) {
+	var ind individual
+	err := h.Pool.QueryRow(ctx, `
+		SELECT id, company_id, full_name, mute_until
+		  FROM individuals
+		 WHERE telegram_user_id = $1
+	`, telegramUserID).Scan(&ind.ID, &ind.CompanyID, &ind.FullName, &ind.MuteUntil)
+	if err != nil {
+		return nil, ErrNotLinked
+	}
+	return &ind, nil
+}
+
+func joinLines(ls []string) string {
+	out := ""
+	for i, s := range ls {
+		if i > 0 {
+			out += "\n"
+		}
+		out += s
+	}
+	return out
+}