feat(internal/controller/pipes/telegram): make basic telegram communication functional

internal/controller/pipes/telegram/client.go

@@ -0,0 +1,287 @@
+package telegram
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"nukumizu-backend/internal/netproxy"
+	"nukumizu-backend/postLog"
+)
+
+const apiBase = "https://api.telegram.org/bot"
+
+// User mirrors a Telegram user.
+type User struct {
+	ID        int64  `json:"id"`
+	IsBot     bool   `json:"is_bot"`
+	FirstName string `json:"first_name"`
+	Username  string `json:"username"`
+}
+
+// Chat mirrors a Telegram chat.
+type Chat struct {
+	ID    int64  `json:"id"`
+	Type  string `json:"type"` // "private", "group", "supergroup", "channel"
+	Title string `json:"title"`
+}
+
+// Message mirrors a Telegram message. Only the fields used for command
+// handling are modeled.
+type Message struct {
+	MessageID int64           `json:"message_id"`
+	From      *User           `json:"from"`
+	Chat      Chat            `json:"chat"`
+	Date      int64           `json:"date"`
+	Text      string          `json:"text"`
+	Entities  []MessageEntity `json:"entities"`
+}
+
+// MessageEntity mirrors a Telegram message entity (bot_command, mention, ...).
+type MessageEntity struct {
+	Type   string `json:"type"`
+	Offset int    `json:"offset"`
+	Length int    `json:"length"`
+}
+
+// Update mirrors a Telegram update.
+type Update struct {
+	UpdateID int64    `json:"update_id"`
+	Message  *Message `json:"message"`
+}
+
+// apiResponse is the Telegram Bot API response envelope.
+type apiResponse struct {
+	OK          bool            `json:"ok"`
+	Result      json.RawMessage `json:"result"`
+	ErrorCode   int             `json:"error_code"`
+	Description string          `json:"description"`
+}
+
+// maxMessageLen is the safe chunk size for outbound messages. Telegram's hard
+// limit is 4096 characters; staying under it leaves headroom for encoding.
+const maxMessageLen = 4000
+
+// pollTimeout and pollLimit are the long-polling parameters passed to
+// getUpdates. The server holds the request open for pollTimeout seconds, so the
+// HTTP client timeout below must exceed it.
+const (
+	pollTimeout = 30
+	pollLimit   = 100
+)
+
+// retryDelay is the pause between getUpdates attempts after an error.
+const retryDelay = 5 * time.Second
+
+// Client is a thin Telegram Bot API client. It both polls for incoming updates
+// (long polling) and issues outbound Bot API calls (sendMessage).
+type Client struct {
+	token      string
+	useProxy   bool
+	httpClient *http.Client
+
+	stopCh   chan struct{}
+	stopOnce sync.Once
+}
+
+// NewClient creates a Telegram Bot API client. When useProxy is set, all API
+// calls are routed through the system-wide network proxy.
+func NewClient(token string, useProxy bool) *Client {
+	return &Client{
+		token:      token,
+		useProxy:   useProxy,
+		httpClient: netproxy.HTTPClient(useProxy, 5*time.Minute),
+		stopCh:     make(chan struct{}),
+	}
+}
+
+func (c *Client) baseURL() string {
+	return apiBase + c.token
+}
+
+// call performs an API call and unmarshals the result field into result (when
+// non-nil). Non-OK responses are wrapped as errors, so the 409 Conflict body
+// raised when another poller uses the same token surfaces to the caller.
+func (c *Client) call(method string, params url.Values, result interface{}) error {
+	endpoint := c.baseURL() + "/" + method
+
+	var req *http.Request
+	var err error
+	if len(params) > 0 {
+		req, err = http.NewRequest(http.MethodPost, endpoint, strings.NewReader(params.Encode()))
+		if err != nil {
+			return fmt.Errorf("failed to create telegram request: %w", err)
+		}
+		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	} else {
+		req, err = http.NewRequest(http.MethodGet, endpoint, nil)
+		if err != nil {
+			return fmt.Errorf("failed to create telegram request: %w", err)
+		}
+	}
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("telegram API %s failed: %w", method, err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to read telegram response: %w", err)
+	}
+
+	var apiResp apiResponse
+	if err := json.Unmarshal(body, &apiResp); err != nil {
+		return fmt.Errorf("failed to unmarshal telegram response (%s): %w\n%s", method, err, string(body))
+	}
+	if !apiResp.OK {
+		return fmt.Errorf("telegram API %s error: %d %s", method, apiResp.ErrorCode, apiResp.Description)
+	}
+
+	if result != nil {
+		if err := json.Unmarshal(apiResp.Result, result); err != nil {
+			return fmt.Errorf("failed to unmarshal telegram %s result: %w", method, err)
+		}
+	}
+
+	return nil
+}
+
+// GetMe verifies the bot token and returns the bot's own user object.
+func (c *Client) GetMe() (*User, error) {
+	var u User
+	if err := c.call("getMe", nil, &u); err != nil {
+		return nil, err
+	}
+	return &u, nil
+}
+
+// getUpdates fetches incoming updates. offset is the first update to return;
+// pass the previous last update_id + 1 to acknowledge the received updates.
+func (c *Client) getUpdates(offset int64, limit, timeout int) ([]Update, error) {
+	params := url.Values{
+		"limit":   {strconv.Itoa(limit)},
+		"timeout": {strconv.Itoa(timeout)},
+	}
+	if offset > 0 {
+		params.Set("offset", strconv.FormatInt(offset, 10))
+	}
+
+	var updates []Update
+	if err := c.call("getUpdates", params, &updates); err != nil {
+		return nil, err
+	}
+	return updates, nil
+}
+
+// SendMessage sends a text message to a chat, splitting it into chunks that
+// fit Telegram's 4096-character limit. The text is sent without a parse mode so
+// template asterisks render literally.
+func (c *Client) SendMessage(chatID int64, text string) error {
+	if strings.TrimSpace(text) == "" {
+		return nil
+	}
+	for _, chunk := range splitMessage(text, maxMessageLen) {
+		if err := c.sendMessageChunk(chatID, chunk); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (c *Client) sendMessageChunk(chatID int64, text string) error {
+	params := url.Values{
+		"chat_id": {strconv.FormatInt(chatID, 10)},
+		"text":    {text},
+	}
+	return c.call("sendMessage", params, nil)
+}
+
+// Listen runs the long-polling loop, calling onUpdate for every incoming
+// update. It blocks until Stop is called, retrying on errors (e.g. the 409
+// Conflict raised when another poller uses the same token). Run it in a
+// goroutine.
+func (c *Client) Listen(onUpdate func(Update)) {
+	var offset int64 // Next getUpdates offset: previous last update_id + 1.
+
+	for {
+		select {
+		case <-c.stopCh:
+			return
+		default:
+		}
+
+		updates, err := c.getUpdates(offset, pollLimit, pollTimeout)
+		if err != nil {
+			if strings.Contains(err.Error(), "409") || strings.Contains(err.Error(), "Conflict") {
+				postLog.Error("Telegram getUpdates conflict (409): another poller is using the same bot token. Ensure only one instance is polling.")
+			} else {
+				postLog.Warning("Telegram getUpdates failed: " + err.Error())
+			}
+
+			select {
+			case <-c.stopCh:
+				return
+			case <-time.After(retryDelay):
+			}
+			continue
+		}
+
+		for _, u := range updates {
+			if u.UpdateID >= offset {
+				offset = u.UpdateID + 1
+			}
+			onUpdate(u)
+		}
+	}
+}
+
+// Stop closes the stop channel and unblocks the Listen loop.
+func (c *Client) Stop() {
+	c.stopOnce.Do(func() { close(c.stopCh) })
+}
+
+// splitMessage splits text into chunks of at most maxLen runes, keeping
+// complete lines when possible.
+func splitMessage(text string, maxLen int) []string {
+	if maxLen <= 0 {
+		return []string{text}
+	}
+
+	remaining := []rune(text)
+	if len(remaining) <= maxLen {
+		return []string{text}
+	}
+
+	var chunks []string
+	for len(remaining) > maxLen {
+		// Prefer the last newline within the window so messages aren't cut
+		// mid-line; fall back to a hard rune cut for overlong lines.
+		cut := maxLen
+		if nl := lastIndexRune(remaining[:maxLen], '\n'); nl > 0 {
+			cut = nl + 1
+		}
+		chunks = append(chunks, string(remaining[:cut]))
+		remaining = remaining[cut:]
+	}
+	if len(remaining) > 0 {
+		chunks = append(chunks, string(remaining))
+	}
+	return chunks
+}
+
+func lastIndexRune(s []rune, r rune) int {
+	for i := len(s) - 1; i >= 0; i-- {
+		if s[i] == r {
+			return i
+		}
+	}
+	return -1
+}

internal/controller/pipes/telegram/telegram.go

@@ -0,0 +1,351 @@
+package telegram
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+
+	"nukumizu-backend/config"
+	"nukumizu-backend/internal/controller"
+	"nukumizu-backend/internal/node"
+	"nukumizu-backend/internal/template"
+	"nukumizu-backend/postLog"
+)
+
+// TelegramController handles Telegram Bot interactions via long polling.
+type TelegramController struct {
+	cfg    config.TelegramConfig
+	client *Client
+	bot    *User // The bot's own user object from getMe.
+
+	mu           sync.Mutex
+	usernameToID map[string]int64 // resolved @username -> numeric user ID
+}
+
+// NewTelegramController creates a new Telegram controller.
+func NewTelegramController(cfg config.TelegramConfig) *TelegramController {
+	t := &TelegramController{
+		cfg:          cfg,
+		usernameToID: make(map[string]int64),
+	}
+	if cfg.Enabled {
+		t.client = NewClient(cfg.BotToken, cfg.NetworkUseProxy)
+	}
+	return t
+}
+
+// Name returns the controller name.
+func (t *TelegramController) Name() string {
+	return "telegram"
+}
+
+// Start verifies the bot token via getMe and begins long polling.
+func (t *TelegramController) Start() error {
+	if !t.cfg.Enabled {
+		return nil
+	}
+	if t.client == nil {
+		postLog.Warning("Telegram controller enabled but client is nil")
+		return nil
+	}
+	if t.cfg.BotToken == "" {
+		postLog.Warning("Telegram controller enabled but no bot token configured")
+		return nil
+	}
+
+	// The tutorial's first step: verify the token with getMe. The returned bot
+	// identity is used for @mention mode and echo prevention.
+	bot, err := t.client.GetMe()
+	if err != nil {
+		return fmt.Errorf("failed to verify Telegram bot token (getMe): %w", err)
+	}
+	t.bot = bot
+	postLog.Info(fmt.Sprintf("Telegram bot authenticated: @%s (id %d)", bot.Username, bot.ID))
+
+	postLog.Info("Telegram controller started (long polling)")
+	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				postLog.Error(fmt.Sprintf("Telegram poll loop panic recovered: %v", r))
+			}
+		}()
+		t.client.Listen(t.handleUpdate)
+	}()
+
+	return nil
+}
+
+// Stop shuts down the Telegram controller and its poll loop.
+func (t *TelegramController) Stop() {
+	if t.client != nil {
+		t.client.Stop()
+	}
+	postLog.Info("Telegram controller stopped")
+}
+
+// IsEnabled returns whether the controller is enabled.
+func (t *TelegramController) IsEnabled() bool {
+	return t.cfg.Enabled
+}
+
+// handleUpdate processes a single Telegram update received via long polling.
+func (t *TelegramController) handleUpdate(update Update) {
+	if update.Message == nil {
+		return
+	}
+	msg := update.Message
+
+	if config.GetConfig().System.DebugMode && config.GetConfig().Debug.ShowTelegramMsg {
+		raw, _ := json.Marshal(update)
+		postLog.Debug("Telegram update received: " + string(raw))
+	}
+
+	// Echo prevention: ignore messages from other bots and from this bot itself.
+	if msg.From == nil || msg.From.IsBot {
+		return
+	}
+	if t.bot != nil && msg.From.ID == t.bot.ID {
+		return
+	}
+
+	chatType := telegramChatType(msg.Chat.Type)
+	if chatType == "" {
+		return // channel or other unsupported chat type.
+	}
+
+	// Record username -> ID so admins configured by username can be reached.
+	t.recordUser(msg.From)
+
+	// Only text messages can carry commands.
+	if strings.TrimSpace(msg.Text) == "" {
+		return
+	}
+
+	cmd := controller.Command{
+		RawText:  msg.Text,
+		ChatID:   msg.Chat.ID,
+		ChatType: chatType,
+		SenderID: msg.From.ID,
+	}
+
+	response := t.processCommand(cmd)
+	if response == "" {
+		return
+	}
+
+	if err := t.client.SendMessage(msg.Chat.ID, response); err != nil {
+		postLog.Warning("Failed to send Telegram reply: " + err.Error())
+	}
+}
+
+// processCommand validates an incoming message as a command and hands the
+// complete command to the unified processor. It returns the response text to
+// reply with; an empty response means the message was discarded.
+func (t *TelegramController) processCommand(cmd controller.Command) string {
+	text := cmd.RawText
+
+	// In "at" listen mode, require a mention of the bot and strip it before
+	// parsing, mirroring the QQ CQ-at behavior.
+	if t.cfg.ListenMethod == "at" && t.bot != nil && t.bot.Username != "" {
+		mention := "@" + t.bot.Username
+		if !strings.Contains(text, mention) {
+			return "" // Not mentioned, ignore.
+		}
+		text = strings.ReplaceAll(text, mention, "")
+	}
+
+	// Telegram appends @botname to commands sent in groups (/list@MyBot); strip
+	// it so the command word parses.
+	text = stripCommandBotSuffix(text)
+
+	// First check whether the received message is a command.
+	parsed, ok := controller.ParseCommand(text)
+	if !ok {
+		return "" // Not a command, discard.
+	}
+
+	parsed.ChatID = cmd.ChatID
+	parsed.ChatType = cmd.ChatType
+	parsed.SenderID = cmd.SenderID
+
+	// Hand the complete command to the unified processor, which checks group
+	// vs private, trusted groups, admin permissions, and executes it.
+	response, err := controller.GetManager().Trigger(parsed, t.cfg.TrustedGroups, t.resolvedAdminList(), t.cfg.ListenMethod)
+	if err != nil {
+		postLog.Error("Telegram command processing failed: " + err.Error())
+		return ""
+	}
+	return response
+}
+
+// SendMessage sends an arbitrary message (e.g. the bot initialization message)
+// to all Telegram trusted groups and admins.
+func (t *TelegramController) SendMessage(message string) error {
+	if !t.cfg.Enabled || t.client == nil {
+		return nil
+	}
+	t.sendToGroupsAndAdmins(message)
+	return nil
+}
+
+// SendStatusChange sends a status change notification via Telegram.
+func (t *TelegramController) SendStatusChange(change node.StatusChange) error {
+	if !t.cfg.Enabled || t.client == nil {
+		return nil
+	}
+
+	cfg := config.GetConfig()
+	params := template.BuildParamsFromStatusChange(change)
+	message := template.Render(cfg.ControllerMessage.ServerStatusChanged, params)
+
+	t.sendToGroupsAndAdmins(message)
+	return nil
+}
+
+// SendServerList sends the server list via Telegram.
+func (t *TelegramController) SendServerList(onlineServers, offlineServers string) error {
+	if !t.cfg.Enabled || t.client == nil {
+		return nil
+	}
+
+	cfg := config.GetConfig()
+	params := template.BuildParamsFromServerList()
+	message := template.Render(cfg.ControllerMessage.ServerList, params)
+
+	t.sendToGroups(message)
+	return nil
+}
+
+// SendExecuteResult sends a command execution result via Telegram.
+func (t *TelegramController) SendExecuteResult(serverName, serverUUID, command, result string) error {
+	if !t.cfg.Enabled || t.client == nil {
+		return nil
+	}
+
+	cfg := config.GetConfig()
+	params := template.BuildParamsFromExecResult(serverName, serverUUID, command, result)
+	message := template.Render(cfg.ControllerMessage.ServerExecuteResult, params)
+
+	t.sendToGroups(message)
+	return nil
+}
+
+// telegramChatType maps a Telegram chat type to the unified ChatType value used
+// by the controller package. Empty means the chat type is unsupported.
+func telegramChatType(chatType string) string {
+	switch chatType {
+	case "private":
+		return "private"
+	case "group", "supergroup":
+		return "group"
+	default:
+		return ""
+	}
+}
+
+// stripCommandBotSuffix removes a "@botname" suffix from the command word,
+// turning "/list@MyBot" into "/list".
+func stripCommandBotSuffix(text string) string {
+	if !strings.HasPrefix(text, "/") {
+		return text
+	}
+
+	// The command word is everything before the first space.
+	spaceIdx := strings.Index(text, " ")
+	if spaceIdx < 0 {
+		spaceIdx = len(text)
+	}
+
+	command, _, found := strings.Cut(text[1:spaceIdx], "@")
+	if !found {
+		return text
+	}
+	return "/" + command + text[spaceIdx:]
+}
+
+// recordUser stores the numeric ID for a sender's username so admins configured
+// by username (rather than numeric ID) can be resolved later.
+func (t *TelegramController) recordUser(u *User) {
+	if u == nil || u.Username == "" {
+		return
+	}
+	t.mu.Lock()
+	t.usernameToID[strings.ToLower(u.Username)] = u.ID
+	t.mu.Unlock()
+}
+
+// resolveUsername resolves a Telegram username to its numeric ID, or returns
+// ok == false when the user has not messaged the bot yet.
+func (t *TelegramController) resolveUsername(username string) (int64, bool) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	id, ok := t.usernameToID[strings.ToLower(strings.TrimPrefix(username, "@"))]
+	return id, ok
+}
+
+// resolvedAdminList maps username-based admin entries to numeric IDs so the
+// unified processor's numeric IsAdmin check works. Unresolvable entries (the
+// user has not messaged the bot yet) pass through unchanged and simply never
+// match.
+func (t *TelegramController) resolvedAdminList() []string {
+	result := make([]string, 0, len(t.cfg.Admins))
+	for _, admin := range t.cfg.Admins {
+		if id, ok := t.resolveUsername(admin); ok {
+			result = append(result, strconv.FormatInt(id, 10))
+		} else {
+			result = append(result, admin)
+		}
+	}
+	return result
+}
+
+// sendToGroupsAndAdmins sends a message to all trusted groups and admins.
+func (t *TelegramController) sendToGroupsAndAdmins(message string) {
+	t.sendToGroups(message)
+	for _, admin := range t.cfg.Admins {
+		t.sendAdminMessage(admin, message)
+	}
+}
+
+// sendToGroups sends a message to all trusted groups.
+func (t *TelegramController) sendToGroups(message string) {
+	for _, groupID := range t.cfg.TrustedGroups {
+		t.sendGroupMessage(groupID, message)
+	}
+}
+
+// sendGroupMessage sends a message to a single group chat.
+func (t *TelegramController) sendGroupMessage(groupID string, message string) {
+	chatID, err := strconv.ParseInt(groupID, 10, 64)
+	if err != nil || chatID == 0 {
+		postLog.Warning(fmt.Sprintf("Invalid Telegram group chat ID: %s", groupID))
+		return
+	}
+	t.sendToChat(chatID, message)
+}
+
+// sendAdminMessage sends a message to an admin, resolving a username-based
+// entry to a numeric ID. Telegram bots cannot start conversations, so an admin
+// configured by username is only reachable after they have messaged the bot.
+func (t *TelegramController) sendAdminMessage(admin string, message string) {
+	chatID, err := strconv.ParseInt(admin, 10, 64)
+	if err != nil || chatID == 0 {
+		// Not a numeric ID: try to resolve a username.
+		if id, ok := t.resolveUsername(admin); ok {
+			t.sendToChat(id, message)
+			return
+		}
+		postLog.Warning(fmt.Sprintf("Cannot reach Telegram admin %q: user has not messaged the bot yet (bots cannot start conversations)", admin))
+		return
+	}
+	t.sendToChat(chatID, message)
+}
+
+// sendToChat sends a message to a chat ID, logging failures.
+func (t *TelegramController) sendToChat(chatID int64, message string) {
+	if err := t.client.SendMessage(chatID, message); err != nil {
+		postLog.Warning(fmt.Sprintf("Failed to send Telegram message to %d: %v", chatID, err))
+	}
+}

internal/controller/pipes/telegram/telegram_test.go

@@ -0,0 +1,85 @@
+package telegram
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestSplitMessageWithinLimit(t *testing.T) {
+	got := splitMessage("short message", maxMessageLen)
+	if len(got) != 1 || got[0] != "short message" {
+		t.Fatalf("expected single chunk, got %q", got)
+	}
+}
+
+func TestSplitMessageChunks(t *testing.T) {
+	// A message longer than the limit must be split, and the rune content must
+	// be preserved exactly when the chunks are re-joined.
+	text := strings.Repeat("a", maxMessageLen+100)
+	got := splitMessage(text, maxMessageLen)
+	if len(got) < 2 {
+		t.Fatalf("expected multiple chunks, got %d", len(got))
+	}
+	for _, c := range got {
+		if len([]rune(c)) > maxMessageLen {
+			t.Fatalf("chunk exceeds %d runes: %d", maxMessageLen, len([]rune(c)))
+		}
+	}
+	if strings.Join(got, "") != text {
+		t.Fatalf("chunks do not reconstruct the original text")
+	}
+}
+
+func TestSplitMessageKeepsLines(t *testing.T) {
+	// Splitting must prefer a newline boundary over cutting mid-line.
+	text := strings.Repeat("x", maxMessageLen-50) + "\n" + strings.Repeat("y", 200)
+	got := splitMessage(text, maxMessageLen)
+	if len(got) != 2 {
+		t.Fatalf("expected 2 chunks, got %d", len(got))
+	}
+	if strings.HasSuffix(got[0], "\n") == false {
+		t.Fatalf("first chunk should end at the newline, got %q...", got[0][len(got[0])-10:])
+	}
+	if !strings.HasPrefix(got[1], "y") {
+		t.Fatalf("second chunk should start after the newline, got %q...", got[1][:10])
+	}
+}
+
+func TestSplitMessageOverlongLine(t *testing.T) {
+	// A single line longer than the limit falls back to a hard rune cut.
+	text := strings.Repeat("x", maxMessageLen*2+10)
+	got := splitMessage(text, maxMessageLen)
+	if strings.Join(got, "") != text {
+		t.Fatalf("chunks do not reconstruct the original text")
+	}
+}
+
+func TestStripCommandBotSuffix(t *testing.T) {
+	cases := map[string]string{
+		"/list":            "/list",
+		"/list@MyBot":      "/list",
+		"/list@MyBot arg1": "/list arg1",
+		"/status@Bot u1":   "/status u1",
+		"not a command":    "not a command",
+	}
+	for in, want := range cases {
+		if got := stripCommandBotSuffix(in); got != want {
+			t.Errorf("stripCommandBotSuffix(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+func TestTelegramChatType(t *testing.T) {
+	cases := map[string]string{
+		"private":   "private",
+		"group":     "group",
+		"supergroup": "group",
+		"channel":   "",
+		"":          "",
+	}
+	for in, want := range cases {
+		if got := telegramChatType(in); got != want {
+			t.Errorf("telegramChatType(%q) = %q, want %q", in, got, want)
+		}
+	}
+}