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