broad-announce/internal/telegram/handler.go

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
}