diff --git a/plugins/channels/telegram/adapter.go b/plugins/channels/telegram/adapter.go new file mode 100644 index 0000000..63e0d4b --- /dev/null +++ b/plugins/channels/telegram/adapter.go @@ -0,0 +1,244 @@ +// Package telegram implements the Telegram Bot API as a channel adapter for +// Cyrene v2. It conforms to internal/channel/adapter.PlatformAdapter, translating +// Telegram updates (getUpdates long polling, or a registered webhook) into the +// normalized UnifiedMessage model and delivering outbound replies through the +// Bot API send* methods. +// +// Media is carried by Telegram file_id values: inbound attachments populate +// UnifiedMessage.Media with URL set to the file_id (plus size/name/mime when the +// update provides them), and outbound image/voice/documents are sent back by +// file_id via sendPhoto/sendVoice/sendDocument. The design authority is +// DevDocs/design/08-渠道接入.md section 4; the contract is defined in sections +// 0-2. Proactive push is enabled (design: Telegram is a strong push channel). +// +// The adapter is self-contained: it depends only on the standard library plus +// the project's internal adapter, config-independent logger, and channel/actor +// model. Tokens are never logged. +package telegram + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "sort" + "sync" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// Compile-time assertion that Adapter satisfies the shared contract. +var _ adapter.PlatformAdapter = (*Adapter)(nil) + +// Adapter is a Telegram Bot API channel adapter. It is safe for concurrent use. +// The zero use is via New; do not copy it after first use. +type Adapter struct { + mu sync.RWMutex + cfg Config + client *http.Client + baseURL string + log *slog.Logger + + connected bool + closed bool + webhook bool + handler adapter.MessageHandler + pollCancel context.CancelFunc + runID uint64 + + offset int64 + chats map[string]adapter.ChannelInfo +} + +// Name returns the platform identifier. +func (a *Adapter) Name() string { return "telegram" } + +// Config returns a snapshot of the adapter's configuration. The BotToken is the +// sensitive field; callers that persist this map must treat it as a secret. +func (a *Adapter) Config() map[string]any { + a.mu.RLock() + defer a.mu.RUnlock() + return map[string]any{ + "bot_token": a.cfg.BotToken, + "webhook_url": a.cfg.WebhookURL, + } +} + +// Connect establishes the session. It is idempotent. When WebhookURL is +// configured the webhook is registered with the Bot API and long polling is +// disabled; otherwise the adapter prepares for getUpdates polling. +func (a *Adapter) Connect(ctx context.Context) error { + a.mu.Lock() + if a.connected { + a.mu.Unlock() + return nil + } + if a.closed { + a.mu.Unlock() + return fmt.Errorf("telegram: %w", adapter.ErrClosed) + } + if a.cfg.BotToken == "" { + a.mu.Unlock() + return fmt.Errorf("telegram: %w", ErrMissingToken) + } + token := a.cfg.BotToken + webhookURL := a.cfg.WebhookURL + a.mu.Unlock() + + webhookMode := false + if webhookURL != "" { + if err := a.setWebhook(ctx, webhookURL); err != nil { + return fmt.Errorf("telegram: register webhook: %w", err) + } + webhookMode = true + } + a.log.Info("connected", "webhook", webhookMode, "token_len", len(token)) + + a.mu.Lock() + a.connected = true + a.webhook = webhookMode + a.offset = 0 + a.mu.Unlock() + return nil +} + +// Disconnect tears down the session and stops an in-flight poll loop. It is +// idempotent. +func (a *Adapter) Disconnect(ctx context.Context) error { + a.mu.Lock() + if !a.connected { + a.mu.Unlock() + return nil + } + a.connected = false + cancel := a.pollCancel + a.mu.Unlock() + if cancel != nil { + cancel() + } + a.log.Info("disconnected") + return nil +} + +// Close permanently shuts the adapter down, stopping any polling loop and +// marking it closed. It is idempotent; subsequent operations fail with +// adapter.ErrClosed. +func (a *Adapter) Close() error { + a.mu.Lock() + if a.closed { + a.mu.Unlock() + return nil + } + a.closed = true + a.connected = false + cancel := a.pollCancel + a.mu.Unlock() + if cancel != nil { + cancel() + } + a.log.Info("closed") + return nil +} + +// IsConnected reports whether the session is currently live. +func (a *Adapter) IsConnected() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.connected +} + +// HealthCheck verifies the session is usable by calling getMe against the Bot +// API. It returns an error wrapping adapter.ErrNotConnected when not connected. +func (a *Adapter) HealthCheck(ctx context.Context) error { + if err := a.requireConnected(); err != nil { + return err + } + bot, err := a.getMe(ctx) + if err != nil { + return fmt.Errorf("telegram: health check: %w", err) + } + a.log.Debug("health check ok", "bot", bot) + return nil +} + +// Capabilities declares what the Telegram platform supports. Proactive push is +// enabled because Telegram permits arbitrary outbound messages. +func (a *Adapter) Capabilities() adapter.Capabilities { + return adapter.Capabilities{ + MaxMessageLength: maxMessageLen, + Markdown: true, + Image: true, + Voice: true, + Reaction: false, + TypingHint: false, + RecommendBurstMax: 1, + Proactive: true, + }.Normalize() +} + +// SetMessageHandler installs the push-style inbound callback used by the webhook +// delivery path (see ProcessUpdate). A nil fn clears the handler. +func (a *Adapter) SetMessageHandler(fn adapter.MessageHandler) { + a.mu.Lock() + defer a.mu.Unlock() + a.handler = fn +} + +// ListChannels returns the conversations observed so far, ordered by ChannelID. +func (a *Adapter) ListChannels(ctx context.Context) ([]adapter.ChannelInfo, error) { + a.mu.RLock() + defer a.mu.RUnlock() + out := make([]adapter.ChannelInfo, 0, len(a.chats)) + for _, c := range a.chats { + out = append(out, c) + } + sort.Slice(out, func(i, j int) bool { return out[i].ChannelID < out[j].ChannelID }) + return out, nil +} + +// ProcessUpdate decodes a raw webhook payload (a single Telegram Update) and +// delivers the normalized message through the installed message handler. It is +// the push-side counterpart to ReadMessages and returns the number of messages +// delivered. Non-message updates (for example callback queries) are ignored. +func (a *Adapter) ProcessUpdate(ctx context.Context, raw []byte) (int, error) { + var update telegramUpdate + if err := json.Unmarshal(raw, &update); err != nil { + return 0, fmt.Errorf("telegram: decode webhook update: %w", err) + } + msg, ok := a.toUnifiedMessage(update) + if !ok { + return 0, nil + } + if err := msg.Validate(); err != nil { + return 0, fmt.Errorf("telegram: validate webhook update: %w", err) + } + a.mu.RLock() + handler := a.handler + a.mu.RUnlock() + if handler != nil { + handler(ctx, msg) + } + return 1, nil +} + +// requireConnected guards operations that need a live session. +func (a *Adapter) requireConnected() error { + a.mu.RLock() + defer a.mu.RUnlock() + if a.closed { + return fmt.Errorf("telegram: %w", adapter.ErrClosed) + } + if !a.connected { + return fmt.Errorf("telegram: %w", adapter.ErrNotConnected) + } + return nil +} + +// isWebhookMode reports whether the adapter is serving via webhook rather than +// polling. +func (a *Adapter) isWebhookMode() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.webhook +} diff --git a/plugins/channels/telegram/botapi.go b/plugins/channels/telegram/botapi.go new file mode 100644 index 0000000..6bf13d1 --- /dev/null +++ b/plugins/channels/telegram/botapi.go @@ -0,0 +1,205 @@ +package telegram + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" +) + +// botResponse is the shared envelope Telegram returns for every method: ok plus +// either result or the error description/code. +type botResponse struct { + OK bool `json:"ok"` + Result json.RawMessage `json:"result"` + Description string `json:"description"` + ErrorCode int `json:"error_code"` +} + +// doJSON posts a JSON payload to a Bot API method and decodes the envelope. The +// caller is responsible for decoding Result; a non-OK reply is returned as +// ErrBadResponse wrapped with the API description. +func (a *Adapter) doJSON(ctx context.Context, method string, payload any) (*botResponse, error) { + var buf bytes.Buffer + if payload != nil { + if err := json.NewEncoder(&buf).Encode(payload); err != nil { + return nil, fmt.Errorf("telegram: encode %s request: %w", method, err) + } + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.endpoint(method), &buf) + if err != nil { + return nil, fmt.Errorf("telegram: build %s request: %w", method, err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + + resp, err := a.client.Do(req) + if err != nil { + return nil, fmt.Errorf("telegram: %s call: %w", method, err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + desc := strings.TrimSpace(string(body)) + if desc != "" { + return nil, fmt.Errorf("telegram: %s: status %d: %s: %w", method, resp.StatusCode, desc, ErrBadResponse) + } + return nil, fmt.Errorf("telegram: %s: status %d: %w", method, resp.StatusCode, ErrBadResponse) + } + + var out botResponse + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, fmt.Errorf("telegram: decode %s response: %w", method, err) + } + if !out.OK { + return nil, fmt.Errorf("telegram: %s: %w", method, ErrBadResponse) + } + return &out, nil +} + +// getMe verifies the token and returns the bot's username. +func (a *Adapter) getMe(ctx context.Context) (string, error) { + out, err := a.doJSON(ctx, "getMe", nil) + if err != nil { + return "", err + } + var me struct { + Username string `json:"username"` + } + if err := json.Unmarshal(out.Result, &me); err != nil { + return "", fmt.Errorf("telegram: getMe result: %w", err) + } + return me.Username, nil +} + +// setWebhook registers the webhook URL and disables getUpdates polling. +func (a *Adapter) setWebhook(ctx context.Context, url string) error { + _, err := a.doJSON(ctx, "setWebhook", map[string]any{"url": url}) + return err +} + +// getUpdates performs a long-poll for updates at the current offset. It returns +// a list of updates; the caller advances a.offset after acknowledging them. +func (a *Adapter) getUpdates(ctx context.Context) ([]telegramUpdate, error) { + a.mu.RLock() + offset := a.offset + a.mu.RUnlock() + + payload := map[string]any{ + "offset": offset, + "limit": maxUpdatesPerPoll, + "timeout": longPollTimeoutSec, + } + out, err := a.doJSON(ctx, "getUpdates", payload) + if err != nil { + return nil, err + } + var updates []telegramUpdate + if err := json.Unmarshal(out.Result, &updates); err != nil { + return nil, fmt.Errorf("telegram: getUpdates result: %w", err) + } + if updates == nil { + updates = []telegramUpdate{} + } + return updates, nil +} + +// sendMessage sends a text message and returns the new message id. +func (a *Adapter) sendMessage(ctx context.Context, chatID, text string, replyTo int64, parseMode string) (int64, error) { + payload := map[string]any{ + "chat_id": chatID, + "text": text, + } + if replyTo != 0 { + payload["reply_to_message_id"] = replyTo + } + if parseMode != "" { + payload["parse_mode"] = parseMode + } + return a.sendPayload(ctx, "sendMessage", payload) +} + +// sendPhoto sends an existing photo by file_id and returns the new message id. +func (a *Adapter) sendPhoto(ctx context.Context, chatID, fileID string, replyTo int64, caption string) (int64, error) { + payload := map[string]any{ + "chat_id": chatID, + "photo": fileID, + } + if caption != "" { + payload["caption"] = caption + payload["parse_mode"] = "Markdown" + } + if replyTo != 0 { + payload["reply_to_message_id"] = replyTo + } + return a.sendPayload(ctx, "sendPhoto", payload) +} + +// sendVoice sends an existing voice clip by file_id and returns the new id. +func (a *Adapter) sendVoice(ctx context.Context, chatID, fileID string, replyTo int64, caption string) (int64, error) { + payload := map[string]any{ + "chat_id": chatID, + "voice": fileID, + } + if caption != "" { + payload["caption"] = caption + payload["parse_mode"] = "Markdown" + } + if replyTo != 0 { + payload["reply_to_message_id"] = replyTo + } + return a.sendPayload(ctx, "sendVoice", payload) +} + +// sendDocument sends an existing document by file_id and returns the new id. +func (a *Adapter) sendDocument(ctx context.Context, chatID, fileID, fileName string, replyTo int64, caption string) (int64, error) { + payload := map[string]any{ + "chat_id": chatID, + "document": fileID, + } + if fileName != "" { + payload["file_name"] = fileName + } + if caption != "" { + payload["caption"] = caption + payload["parse_mode"] = "Markdown" + } + if replyTo != 0 { + payload["reply_to_message_id"] = replyTo + } + return a.sendPayload(ctx, "sendDocument", payload) +} + +// sendPayload posts a media/message payload and returns the acknowledged +// message id from the Bot API result. +func (a *Adapter) sendPayload(ctx context.Context, method string, payload map[string]any) (int64, error) { + out, err := a.doJSON(ctx, method, payload) + if err != nil { + return 0, err + } + var res struct { + MessageID int64 `json:"message_id"` + } + if err := json.Unmarshal(out.Result, &res); err != nil { + return 0, fmt.Errorf("telegram: %s result: %w", method, err) + } + return res.MessageID, nil +} + +// parseReplyToID converts a reply context identifier to a Telegram message id. +// A non-numeric or empty value yields 0 (no reply). +func parseReplyToID(s string) int64 { + if s == "" { + return 0 + } + n, err := strconv.ParseInt(s, 10, 64) + if err != nil { + return 0 + } + return n +} diff --git a/plugins/channels/telegram/config.go b/plugins/channels/telegram/config.go new file mode 100644 index 0000000..006a73e --- /dev/null +++ b/plugins/channels/telegram/config.go @@ -0,0 +1,90 @@ +package telegram + +import ( + "io" + "log/slog" + "net/http" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// Defaults and knob values for the Telegram Bot API integration. +const ( + // defaultBaseURL is the Telegram API root for unconfigured adapters. + defaultBaseURL = "https://api.telegram.org" + // httpTimeout bounds a single Bot API request. + httpTimeout = 180 * time.Second + // longPollTimeoutSec is the getUpdates long-poll timeout in seconds. + longPollTimeoutSec = 50 + // pollInterval is how long to sleep between getUpdates batches. + pollInterval = 1 * time.Second + // maxUpdatesPerPoll caps the number of updates fetched in one getUpdates call. + maxUpdatesPerPoll = 100 + // maxMessageLen is Telegram's per-message text limit. + maxMessageLen = 4096 +) + +// Config holds the settings required to reach the Telegram Bot API. +type Config struct { + // BotToken is the secret token issued by BotFather. It is stored and + // surfaced through Config() but never written to logs. + BotToken string `json:"bot_token"` + // WebhookURL is the public HTTPS endpoint Telegram delivers updates to. + // When empty the adapter uses getUpdates long polling instead. + WebhookURL string `json:"webhook_url"` +} + +// Option customizes an Adapter at construction time. +type Option func(*Adapter) + +// WithHTTPClient injects the *http.Client used for Bot API requests. Tests pass +// the client of an httptest.Server so the mock endpoint answers calls. +func WithHTTPClient(c *http.Client) Option { + return func(a *Adapter) { a.client = c } +} + +// WithBaseURL overrides the Telegram API root (default https://api.telegram.org). +// Tests point it at an httptest.Server URL. +func WithBaseURL(base string) Option { + return func(a *Adapter) { + if base != "" { + a.baseURL = base + } + } +} + +// WithLogger injects the logger used by the adapter. When nil the adapter logs +// through a discard logger so a bare adapter stays silent; callers wire their +// production logger via this option. +func WithLogger(l *slog.Logger) Option { + return func(a *Adapter) { + if l == nil { + l = logger.New(logger.Options{Out: io.Discard}) + } + a.log = logger.WithSource(l, "plugins/channels/telegram") + } +} + +// New builds a Telegram adapter. It performs no network I/O; call Connect to +// establish the session. +func New(cfg Config, opts ...Option) *Adapter { + a := &Adapter{ + cfg: cfg, + client: &http.Client{Timeout: httpTimeout}, + baseURL: defaultBaseURL, + log: logger.WithSource(logger.New(logger.Options{Out: io.Discard}), "plugins/channels/telegram"), + chats: map[string]adapter.ChannelInfo{}, + } + for _, opt := range opts { + opt(a) + } + return a +} + +// endpoint returns the absolute Bot API URL for the given method, bundling the +// token into the path (never into query parameters or logs). +func (a *Adapter) endpoint(method string) string { + return a.baseURL + "/bot" + a.cfg.BotToken + "/" + method +} diff --git a/plugins/channels/telegram/config_test.go b/plugins/channels/telegram/config_test.go new file mode 100644 index 0000000..db4ac6f --- /dev/null +++ b/plugins/channels/telegram/config_test.go @@ -0,0 +1,217 @@ +package telegram + +import ( + "context" + "errors" + "net/http" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestName(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + if got := a.Name(); got != "telegram" { + t.Fatalf("Name() = %q, want telegram", got) + } +} + +func TestConfig(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "abc123", WebhookURL: "https://example.com/hook"}) + got := a.Config() + if got["bot_token"] != "abc123" { + t.Fatalf("bot_token = %v, want abc123", got["bot_token"]) + } + if got["webhook_url"] != "https://example.com/hook" { + t.Fatalf("webhook_url = %v, want https://example.com/hook", got["webhook_url"]) + } + // Round-trip through the shared decoder to prove the map is decodable. + var round Config + if err := adapter.DecodeConfig(got, &round); err != nil { + t.Fatalf("DecodeConfig: %v", err) + } + if round.BotToken != "abc123" || round.WebhookURL != "https://example.com/hook" { + t.Fatalf("round trip mismatch: %+v", round) + } +} + +func TestCapabilities(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + caps := a.Capabilities() + if !caps.Proactive { + t.Error("Proactive should be true for telegram") + } + if !caps.Markdown || !caps.Image || !caps.Voice { + t.Errorf("expected Markdown/Image/Voice supported, got %+v", caps) + } + if caps.Reaction || caps.TypingHint { + t.Errorf("expected Reaction/TypingHint unsupported, got %+v", caps) + } + if caps.MaxMessageLength != 4096 { + t.Errorf("MaxMessageLength = %d, want 4096", caps.MaxMessageLength) + } + if caps.RecommendBurstMax <= 0 { + t.Errorf("RecommendBurstMax should be positive, got %d", caps.RecommendBurstMax) + } +} + +func TestListChannels_Empty(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + got, err := a.ListChannels(context.Background()) + if err != nil { + t.Fatalf("ListChannels: %v", err) + } + if got == nil || len(got) != 0 { + t.Fatalf("expected empty non-nil slice, got %#v", got) + } +} + +func TestConnect_MissingToken(t *testing.T) { + t.Parallel() + a := New(Config{}) + err := a.Connect(context.Background()) + if !errors.Is(err, ErrMissingToken) { + t.Fatalf("err = %v, want ErrMissingToken", err) + } + if a.IsConnected() { + t.Fatal("should not be connected") + } +} + +func TestConnect_WebhookIdempotent(t *testing.T) { + t.Parallel() + var mu sync.Mutex + calls := 0 + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "setWebhook" { + http.Error(w, "unexpected method", http.StatusNotFound) + return + } + mu.Lock() + calls++ + mu.Unlock() + okResponse(w, map[string]any{"webhook_url": "https://example.com/hook"}) + })) + a.cfg.WebhookURL = "https://example.com/hook" + + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if !a.IsConnected() { + t.Fatal("expected connected") + } + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("second Connect should be idempotent: %v", err) + } + mu.Lock() + if calls != 1 { + t.Fatalf("expected 1 setWebhook call, got %d", calls) + } + mu.Unlock() +} + +func TestConnect_PollingMode(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "no api call expected in poll mode", http.StatusNotFound) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if !a.IsConnected() { + t.Fatal("expected connected") + } + if a.isWebhookMode() { + t.Fatal("expected polling mode") + } +} + +func TestDisconnect_Idempotent(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "no api call expected", http.StatusNotFound) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if err := a.Disconnect(context.Background()); err != nil { + t.Fatalf("Disconnect: %v", err) + } + if a.IsConnected() { + t.Fatal("expected disconnected") + } + if err := a.Disconnect(context.Background()); err != nil { + t.Fatalf("second Disconnect should be idempotent: %v", err) + } +} + +func TestClose_Terminal(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "no api call expected", http.StatusNotFound) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if err := a.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if a.IsConnected() { + t.Fatal("expected disconnected after close") + } + if err := a.Connect(context.Background()); !errors.Is(err, adapter.ErrClosed) { + t.Fatalf("Connect after close err = %v, want ErrClosed", err) + } + if err := a.HealthCheck(context.Background()); !errors.Is(err, adapter.ErrClosed) { + t.Fatalf("HealthCheck after close err = %v, want ErrClosed", err) + } +} + +func TestHealthCheck_OK(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "getMe" { + http.Error(w, "unexpected method", http.StatusNotFound) + return + } + okResponse(w, map[string]any{"id": 123, "username": "cyrene_bot"}) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + if err := a.HealthCheck(context.Background()); err != nil { + t.Fatalf("HealthCheck: %v", err) + } +} + +func TestHealthCheck_NotConnected(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + err := a.HealthCheck(context.Background()) + if !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("err = %v, want ErrNotConnected", err) + } +} + +func TestHealthCheck_APIError(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "getMe" { + http.Error(w, "unexpected method", http.StatusNotFound) + return + } + failResponse(w, "Unauthorized", 401) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + err := a.HealthCheck(context.Background()) + if !errors.Is(err, ErrBadResponse) { + t.Fatalf("err = %v, want ErrBadResponse", err) + } +} diff --git a/plugins/channels/telegram/errors.go b/plugins/channels/telegram/errors.go new file mode 100644 index 0000000..0728abd --- /dev/null +++ b/plugins/channels/telegram/errors.go @@ -0,0 +1,17 @@ +package telegram + +import "errors" + +// Sentinel errors returned by the Telegram adapter. Consumers test for them +// with errors.Is; failures that mean "not connected" wrap adapter.ErrNotConnected +// so the shared channel sentinel still matches. +var ( + // ErrMissingToken is returned when no BotToken is configured. + ErrMissingToken = errors.New("telegram: missing bot token") + // ErrBadResponse is returned when the Bot API replies with OK=false or an + // unexpected payload shape. + ErrBadResponse = errors.New("telegram: bad bot api response") + // ErrNoChatTarget is returned when an outbound message carries no usable + // conversation/group identifier. + ErrNoChatTarget = errors.New("telegram: message has no target chat") +) diff --git a/plugins/channels/telegram/helpers_test.go b/plugins/channels/telegram/helpers_test.go new file mode 100644 index 0000000..f1f0c02 --- /dev/null +++ b/plugins/channels/telegram/helpers_test.go @@ -0,0 +1,61 @@ +package telegram + +import ( + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// discardLogger returns a silent, source-tagged logger for tests. +func discardLogger() *slog.Logger { + return logger.WithSource(logger.New(logger.Options{Out: io.Discard}), "plugins/channels/telegram") +} + +// newTestAdapter wires a Telegram adapter to an httptest server whose handler +// answers the Bot API methods. The server is closed on test cleanup. +func newTestAdapter(t *testing.T, h http.Handler) (*Adapter, *httptest.Server) { + t.Helper() + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + a := New( + Config{BotToken: "123456:TESTTOKEN"}, + WithBaseURL(srv.URL), + WithHTTPClient(srv.Client()), + WithLogger(discardLogger()), + ) + return a, srv +} + +// methodOf extracts the Bot API method name from the request path, which takes +// the form /bot/. +func methodOf(r *http.Request) string { + parts := strings.Split(r.URL.Path, "/") + return parts[len(parts)-1] +} + +// decodeBody unmarshals an incoming JSON request body into a map for assertions. +func decodeBody(r *http.Request) map[string]any { + var m map[string]any + if err := json.NewDecoder(r.Body).Decode(&m); err != nil { + return nil + } + return m +} + +// okResponse writes a generic Bot API success envelope with the given result. +func okResponse(w http.ResponseWriter, result any) { + body := map[string]any{"ok": true, "result": result} + _ = json.NewEncoder(w).Encode(body) +} + +// failResponse writes a Bot API error envelope. +func failResponse(w http.ResponseWriter, desc string, code int) { + body := map[string]any{"ok": false, "description": desc, "error_code": code} + _ = json.NewEncoder(w).Encode(body) +} diff --git a/plugins/channels/telegram/poll.go b/plugins/channels/telegram/poll.go new file mode 100644 index 0000000..bc27d2e --- /dev/null +++ b/plugins/channels/telegram/poll.go @@ -0,0 +1,93 @@ +package telegram + +import ( + "context" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// ReadMessages runs a blocking getUpdates long-poll loop, emitting each +// normalized inbound message to emit until ctx is canceled or the adapter is +// disconnected. When running in webhook mode it does not poll; it simply waits +// for ctx to be canceled, since messages are delivered through the message +// handler via ProcessUpdate instead. +// +// The update offset is advanced only after a batch has been fully processed, so +// a canceled read does not lose the batch; acknowledgment is batched so the +// getUpdates call does not re-deliver the same update_id flood. +func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) { + if emit == nil { + return + } + if a.isWebhookMode() { + <-ctx.Done() + return + } + if err := a.requireConnected(); err != nil { + a.log.Warn("read not started", "err", err) + return + } + + pollCtx, cancel := context.WithCancel(ctx) + a.mu.Lock() + a.runID++ + a.pollCancel = cancel + runID := a.runID + a.mu.Unlock() + defer func() { + a.mu.Lock() + if a.runID == runID { + a.pollCancel = nil + } + a.mu.Unlock() + cancel() + }() + + for { + if err := pollCtx.Err(); err != nil { + return + } + updates, err := a.getUpdates(pollCtx) + if err != nil { + a.log.Warn("long poll failed", "err", err) + if !sleepWait(pollCtx, pollInterval) { + return + } + continue + } + if len(updates) == 0 { + if !sleepWait(pollCtx, pollInterval) { + return + } + continue + } + for _, u := range updates { + msg, ok := a.toUnifiedMessage(u) + if !ok { + continue + } + if err := msg.Validate(); err != nil { + a.log.Warn("skip invalid message", "err", err) + continue + } + emit(msg) + } + a.mu.Lock() + a.offset = updates[len(updates)-1].UpdateID + 1 + a.mu.Unlock() + } +} + +// sleepWait blocks for d or until ctx is canceled, returning false when the +// context fires first. +func sleepWait(ctx context.Context, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} diff --git a/plugins/channels/telegram/poll_test.go b/plugins/channels/telegram/poll_test.go new file mode 100644 index 0000000..9383fe8 --- /dev/null +++ b/plugins/channels/telegram/poll_test.go @@ -0,0 +1,171 @@ +package telegram + +import ( + "context" + "net/http" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// fakeUpdate builds a minimal telegramUpdate JSON object. +func fakeUpdate(id int, text string) map[string]any { + return map[string]any{ + "update_id": id, + "message": map[string]any{ + "message_id": id, + "from": map[string]any{"id": 1, "first_name": "Tester"}, + "chat": map[string]any{"id": 10, "type": "private"}, + "date": 1700000000, + "text": text, + }, + } +} + +func TestReadMessages_LongPollOffset(t *testing.T) { + t.Parallel() + var mu sync.Mutex + first := true + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "getUpdates" { + http.Error(w, "unexpected method", http.StatusNotFound) + return + } + mu.Lock() + firstCall := first + first = false + mu.Unlock() + if firstCall { + okResponse(w, []map[string]any{fakeUpdate(1, "one"), fakeUpdate(2, "two")}) + return + } + okResponse(w, []map[string]any{}) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + emitted := make(chan adapter.UnifiedMessage, 4) + done := make(chan struct{}) + go func() { + a.ReadMessages(ctx, func(m adapter.UnifiedMessage) { emitted <- m }) + close(done) + }() + + got := make([]adapter.UnifiedMessage, 0, 2) + for i := 0; i < 2; i++ { + select { + case m := <-emitted: + got = append(got, m) + case <-time.After(3 * time.Second): + t.Fatal("timeout waiting for emitted message") + } + } + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("ReadMessages did not return after ctx cancel") + } + + if len(got) != 2 { + t.Fatalf("emitted %d messages, want 2", len(got)) + } + if got[0].Content != "one" || got[1].Content != "two" { + t.Errorf("contents = %q/%q, want one/two", got[0].Content, got[1].Content) + } + a.mu.RLock() + offset := a.offset + a.mu.RUnlock() + if offset != 3 { + t.Errorf("offset = %d, want 3", offset) + } +} + +func TestReadMessages_CtxPreCanceled(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "no api call expected", http.StatusNotFound) + })) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + done := make(chan struct{}) + go func() { + a.ReadMessages(ctx, func(m adapter.UnifiedMessage) {}) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("ReadMessages should return immediately on canceled ctx") + } +} + +func TestReadMessages_WebhookWaits(t *testing.T) { + t.Parallel() + var calls int + var mu sync.Mutex + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) == "setWebhook" { + okResponse(w, map[string]any{"webhook_url": "https://example.com/hook"}) + return + } + mu.Lock() + calls++ + mu.Unlock() + http.Error(w, "no getUpdates expected in webhook mode", http.StatusNotFound) + })) + a.cfg.WebhookURL = "https://example.com/hook" + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + a.ReadMessages(ctx, func(m adapter.UnifiedMessage) {}) + close(done) + }() + time.Sleep(100 * time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("ReadMessages should return on cancel in webhook mode") + } + mu.Lock() + if calls != 0 { + t.Errorf("getUpdates called %d times in webhook mode", calls) + } + mu.Unlock() +} + +func TestReadMessages_ErrorRetriesThenCancel(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "getUpdates" { + http.Error(w, "unexpected method", http.StatusNotFound) + return + } + failResponse(w, "Conflict", 409) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + a.ReadMessages(ctx, func(m adapter.UnifiedMessage) {}) + close(done) + }() + time.Sleep(150 * time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("ReadMessages should return on cancel despite getUpdates errors") + } +} diff --git a/plugins/channels/telegram/send.go b/plugins/channels/telegram/send.go new file mode 100644 index 0000000..dfbff5d --- /dev/null +++ b/plugins/channels/telegram/send.go @@ -0,0 +1,94 @@ +package telegram + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// markdownParseMode is the parse_mode sent with outgoing text/captions so the +// Markdown capability is actually exercised. +const markdownParseMode = "Markdown" + +// SendMessage delivers the reply messages contained in resp to their target +// chats. Each message goes to its own ChannelID (or GroupID) conversation. +func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error { + if err := a.requireConnected(); err != nil { + return err + } + for _, m := range resp.Messages { + if err := a.sendOne(ctx, m, resp.ReplyTo); err != nil { + return fmt.Errorf("telegram: send message: %w", err) + } + } + return nil +} + +// SendProactive pushes a message not triggered by a recent inbound message. +// Telegram permits arbitrary proactive push, so this is only gated by the +// connection state. +func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error { + if err := a.requireConnected(); err != nil { + return err + } + return a.sendOne(ctx, msg, "") +} + +// sendOne delivers a single message to its conversation, choosing the Bot API +// method that matches the content type and the presence of a file_id payload. +func (a *Adapter) sendOne(ctx context.Context, m adapter.UnifiedMessage, replyTo string) error { + chatID := m.ChannelID + if chatID == "" { + chatID = m.GroupID + } + if chatID == "" { + return fmt.Errorf("telegram: %w", ErrNoChatTarget) + } + replyToID := parseReplyToID(replyTo) + fileID := mediaFileID(m.Media) + + switch m.ContentType { + case adapter.ContentImage: + if fileID != "" { + _, err := a.sendPhoto(ctx, chatID, fileID, replyToID, m.Content) + return err + } + case adapter.ContentVoice: + if fileID != "" { + _, err := a.sendVoice(ctx, chatID, fileID, replyToID, m.Content) + return err + } + case adapter.ContentFile: + if fileID != "" { + _, err := a.sendDocument(ctx, chatID, fileID, mediaFileName(m.Media), replyToID, m.Content) + return err + } + } + + if m.Content == "" { + return fmt.Errorf("telegram: %w: empty message content", ErrBadResponse) + } + _, err := a.sendMessage(ctx, chatID, m.Content, replyToID, markdownParseMode) + return err +} + +// mediaFileID returns the first non-empty file_id carried by the media list. +func mediaFileID(media []adapter.Media) string { + for _, md := range media { + if md.URL != "" { + return md.URL + } + } + return "" +} + +// mediaFileName returns the first non-empty file name carried by the media list. +func mediaFileName(media []adapter.Media) string { + for _, md := range media { + if md.Name != "" { + return md.Name + } + } + return "" +} diff --git a/plugins/channels/telegram/send_test.go b/plugins/channels/telegram/send_test.go new file mode 100644 index 0000000..19a6782 --- /dev/null +++ b/plugins/channels/telegram/send_test.go @@ -0,0 +1,324 @@ +package telegram + +import ( + "context" + "errors" + "net/http" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestSendMessage_TextConstruction(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "sendMessage" { + http.Error(w, "unexpected method: "+methodOf(r), http.StatusNotFound) + return + } + mu.Lock() + body = decodeBody(r) + mu.Unlock() + okResponse(w, map[string]any{"message_id": 100}) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + + resp := adapter.UnifiedResponse{ + ReplyTo: "42", + Messages: []adapter.UnifiedMessage{ + {ChannelID: "200", ChannelType: adapter.ChannelDirect, Content: "hello", ContentType: adapter.ContentText}, + }, + } + if err := a.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + mu.Lock() + defer mu.Unlock() + if body["chat_id"] != "200" { + t.Errorf("chat_id = %v, want 200", body["chat_id"]) + } + if body["text"] != "hello" { + t.Errorf("text = %v, want hello", body["text"]) + } + if body["parse_mode"] != "Markdown" { + t.Errorf("parse_mode = %v, want Markdown", body["parse_mode"]) + } + if body["reply_to_message_id"] != float64(42) { + t.Errorf("reply_to_message_id = %v, want 42", body["reply_to_message_id"]) + } +} + +func TestSendMessage_ImageByFileID(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + var method string + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + m := methodOf(r) + if m != "sendPhoto" { + http.Error(w, "unexpected method: "+m, http.StatusNotFound) + return + } + mu.Lock() + method = m + body = decodeBody(r) + mu.Unlock() + okResponse(w, map[string]any{"message_id": 101}) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + resp := adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{ + { + ChannelID: "200", + ChannelType: adapter.ChannelDirect, + Content: "a caption", + ContentType: adapter.ContentImage, + Media: []adapter.Media{{Kind: adapter.MediaImage, URL: "file-abc"}}, + }, + }, + } + if err := a.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + mu.Lock() + defer mu.Unlock() + if method != "sendPhoto" { + t.Errorf("method = %q, want sendPhoto", method) + } + if body["chat_id"] != "200" || body["photo"] != "file-abc" { + t.Errorf("body = %v, want chat_id 200 and photo file-abc", body) + } + if body["caption"] != "a caption" || body["parse_mode"] != "Markdown" { + t.Errorf("caption/parse_mode = %v/%v", body["caption"], body["parse_mode"]) + } +} + +func TestSendMessage_ImageNoFileIDFallsBackToText(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var method string + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + m := methodOf(r) + mu.Lock() + method = m + mu.Unlock() + if m != "sendMessage" { + http.Error(w, "unexpected method: "+m, http.StatusNotFound) + return + } + okResponse(w, map[string]any{"message_id": 102}) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + resp := adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{ + {ChannelID: "200", ContentType: adapter.ContentImage, Content: "no file_id so text"}, + }, + } + if err := a.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + mu.Lock() + defer mu.Unlock() + if method != "sendMessage" { + t.Errorf("method = %q, want fallback sendMessage", method) + } +} + +func TestSendMessage_VoiceAndDocument(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + ct adapter.ContentType + media adapter.Media + wantMethod string + wantField string + }{ + {name: "voice", ct: adapter.ContentVoice, media: adapter.Media{Kind: adapter.MediaVoice, URL: "voice-1"}, wantMethod: "sendVoice", wantField: "voice"}, + {name: "document", ct: adapter.ContentFile, media: adapter.Media{Kind: adapter.MediaFile, URL: "doc-1", Name: "x.pdf"}, wantMethod: "sendDocument", wantField: "document"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + var method string + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + m := methodOf(r) + mu.Lock() + method = m + body = decodeBody(r) + mu.Unlock() + if m != tc.wantMethod { + http.Error(w, "unexpected method: "+m, http.StatusNotFound) + return + } + okResponse(w, map[string]any{"message_id": 103}) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + resp := adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{ + {ChannelID: "200", ChannelType: adapter.ChannelDirect, ContentType: tc.ct, Media: []adapter.Media{tc.media}}, + }, + } + if err := a.SendMessage(context.Background(), resp); err != nil { + t.Fatalf("SendMessage: %v", err) + } + mu.Lock() + defer mu.Unlock() + if method != tc.wantMethod { + t.Errorf("method = %q, want %q", method, tc.wantMethod) + } + field := tc.media.URL + if body[tc.wantField] != field { + t.Errorf("%s = %v, want %v", tc.wantField, body[tc.wantField], field) + } + }) + } +} + +func TestSendProactive(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var body map[string]any + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "sendMessage" { + http.Error(w, "unexpected method", http.StatusNotFound) + return + } + mu.Lock() + body = decodeBody(r) + mu.Unlock() + okResponse(w, map[string]any{"message_id": 104}) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + msg := adapter.UnifiedMessage{ + ChannelID: "300", ChannelType: adapter.ChannelDirect, + Content: "reminder", ContentType: adapter.ContentText, + } + if err := a.SendProactive(context.Background(), msg); err != nil { + t.Fatalf("SendProactive: %v", err) + } + mu.Lock() + defer mu.Unlock() + if body["chat_id"] != "300" || body["text"] != "reminder" { + t.Errorf("body = %v, want chat_id 300 text reminder", body) + } + if _, ok := body["reply_to_message_id"]; ok { + t.Errorf("proactive message should not carry reply_to_message_id, got %v", body) + } +} + +func TestSendMessage_NotConnected(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{}) + if !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("err = %v, want ErrNotConnected", err) + } + if err := a.SendProactive(context.Background(), adapter.UnifiedMessage{}); !errors.Is(err, adapter.ErrNotConnected) { + t.Fatalf("SendProactive err = %v, want ErrNotConnected", err) + } +} + +func TestSendMessage_NoTargetChat(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "unexpected method", http.StatusNotFound) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ContentType: adapter.ContentText, Content: "hi"}}, + }) + if !errors.Is(err, ErrNoChatTarget) { + t.Fatalf("err = %v, want ErrNoChatTarget", err) + } +} + +func TestSendMessage_APIError(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if methodOf(r) != "sendMessage" { + http.Error(w, "unexpected method", http.StatusNotFound) + return + } + failResponse(w, "Bad Request: chat not found", 400) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + err := a.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{ChannelID: "999", ContentType: adapter.ContentText, Content: "hi"}}, + }) + if !errors.Is(err, ErrBadResponse) { + t.Fatalf("err = %v, want ErrBadResponse", err) + } +} + +func TestProcessUpdate_DeliversToHandler(t *testing.T) { + t.Parallel() + a, _ := newTestAdapter(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "unexpected method", http.StatusNotFound) + })) + if err := a.Connect(context.Background()); err != nil { + t.Fatalf("Connect: %v", err) + } + + var mu sync.Mutex + var got *adapter.UnifiedMessage + a.SetMessageHandler(func(ctx context.Context, m adapter.UnifiedMessage) { + mu.Lock() + got = &m + mu.Unlock() + }) + raw := []byte(`{"update_id":1,"message":{"message_id":5,"from":{"id":1,"first_name":"Ada"},"chat":{"id":10,"type":"private"},"date":1700000000,"text":"webhook hi"}}`) + n, err := a.ProcessUpdate(context.Background(), raw) + if err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + if n != 1 { + t.Errorf("delivered %d messages, want 1", n) + } + mu.Lock() + defer mu.Unlock() + if got == nil { + t.Fatal("handler was not called") + } + if got.Content != "webhook hi" || got.Platform != "telegram" { + t.Errorf("message = %+v", got) + } +} + +func TestProcessUpdate_NonMessageIgnored(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + raw := []byte(`{"update_id":2,"callback_query":{"id":"cq","from":{"id":1},"data":"btn"}}`) + n, err := a.ProcessUpdate(context.Background(), raw) + if err != nil { + t.Fatalf("ProcessUpdate: %v", err) + } + if n != 0 { + t.Errorf("delivered %d messages, want 0 for callback query", n) + } +} + +func TestProcessUpdate_InvalidJSON(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + if _, err := a.ProcessUpdate(context.Background(), []byte("{not json")); err == nil { + t.Fatal("expected error decoding invalid JSON") + } +} diff --git a/plugins/channels/telegram/updates.go b/plugins/channels/telegram/updates.go new file mode 100644 index 0000000..7540f72 --- /dev/null +++ b/plugins/channels/telegram/updates.go @@ -0,0 +1,223 @@ +package telegram + +import ( + "strconv" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// telegramUpdate is the top-level object Telegram sends for every event. +type telegramUpdate struct { + UpdateID int64 `json:"update_id"` + Message *telegramMessage `json:"message"` + EditedMessage *telegramMessage `json:"edited_message"` + ChannelPost *telegramMessage `json:"channel_post"` + EditedChannelPost *telegramMessage `json:"edited_channel_post"` +} + +// message returns the concrete chat message carried by the update, or nil. +func (u telegramUpdate) message() *telegramMessage { + switch { + case u.Message != nil: + return u.Message + case u.EditedMessage != nil: + return u.EditedMessage + case u.ChannelPost != nil: + return u.ChannelPost + case u.EditedChannelPost != nil: + return u.EditedChannelPost + default: + return nil + } +} + +// telegramMessage is a chat message as delivered by the Bot API. +type telegramMessage struct { + MessageID int64 `json:"message_id"` + From *telegramUser `json:"from"` + Chat telegramChat `json:"chat"` + Date int64 `json:"date"` + Text string `json:"text"` + Caption string `json:"caption"` + Photo []telegramPhotoSize `json:"photo"` + Voice *telegramAudio `json:"voice"` + Audio *telegramAudio `json:"audio"` + Document *telegramDocument `json:"document"` + Video *telegramVideo `json:"video"` + ReplyToMessage *telegramMessage `json:"reply_to_message"` + MessageThreadID int64 `json:"message_thread_id"` +} + +// telegramUser is the sender of a message. +type telegramUser struct { + ID int64 `json:"id"` + FirstName string `json:"first_name"` + LastName string `json:"last_name"` + Username string `json:"username"` +} + +// telegramChat is the conversation a message belongs to. +type telegramChat struct { + ID int64 `json:"id"` + Type string `json:"type"` // private | group | supergroup | channel + Title string `json:"title"` +} + +// telegramPhotoSize is one entry of a photo attachment. +type telegramPhotoSize struct { + FileID string `json:"file_id"` + FileSize int64 `json:"file_size"` +} + +// telegramAudio describes a voice or audio attachment. +type telegramAudio struct { + FileID string `json:"file_id"` + FileSize int64 `json:"file_size"` + MIMEType string `json:"mime_type"` +} + +// telegramDocument describes a document attachment. +type telegramDocument struct { + FileID string `json:"file_id"` + FileSize int64 `json:"file_size"` + FileName string `json:"file_name"` + MIMEType string `json:"mime_type"` +} + +// telegramVideo describes a video attachment. +type telegramVideo struct { + FileID string `json:"file_id"` + FileSize int64 `json:"file_size"` + MIMEType string `json:"mime_type"` +} + +// toUnifiedMessage converts an update to the normalized model. It returns ok +// false for updates that carry no deliverable chat message. +func (a *Adapter) toUnifiedMessage(update telegramUpdate) (adapter.UnifiedMessage, bool) { + m := update.message() + if m == nil { + return adapter.UnifiedMessage{}, false + } + msg := adapter.UnifiedMessage{ + Platform: a.Name(), + ChannelID: strconv.FormatInt(m.Chat.ID, 10), + ChannelType: chatType(m.Chat.Type), + MessageID: strconv.FormatInt(m.MessageID, 10), + Timestamp: time.Unix(m.Date, 0), + } + + if m.Chat.Type == "group" || m.Chat.Type == "supergroup" || m.Chat.Type == "channel" { + msg.GroupID = strconv.FormatInt(m.Chat.ID, 10) + msg.GroupName = m.Chat.Title + } + if m.MessageThreadID != 0 { + msg.ThreadID = strconv.FormatInt(m.MessageThreadID, 10) + } + + content, contentType, media := extractPayload(m) + msg.Content = content + msg.ContentType = contentType + msg.Media = media + + if m.From != nil { + msg.SenderID = strconv.FormatInt(m.From.ID, 10) + msg.SenderName = displayName(m.From) + msg.Sender = adapter.Actor{ + Role: adapter.RoleUnknown, + UserID: msg.SenderID, + DisplayName: msg.SenderName, + GroupID: msg.GroupID, + GroupName: msg.GroupName, + }.Resolve() + } + + if m.ReplyToMessage != nil { + msg.ReplyToID = strconv.FormatInt(m.ReplyToMessage.MessageID, 10) + msg.ReplyToText = m.ReplyToMessage.Text + if msg.ReplyToText == "" { + msg.ReplyToText = m.ReplyToMessage.Caption + } + } + + a.recordChat(m.Chat, m.From) + return msg, true +} + +// extractPayload picks the primary content and attachment list from a message. +func extractPayload(m *telegramMessage) (string, adapter.ContentType, []adapter.Media) { + media := make([]adapter.Media, 0, 2) + content := m.Caption + if m.Text != "" { + content = m.Text + } + ct := adapter.ContentText + + if len(m.Photo) > 0 { + p := m.Photo[len(m.Photo)-1] + media = append(media, adapter.Media{Kind: adapter.MediaImage, URL: p.FileID, Size: p.FileSize, MIMEType: "image/jpeg"}) + ct = adapter.ContentImage + } + if m.Voice != nil { + media = append(media, adapter.Media{Kind: adapter.MediaVoice, URL: m.Voice.FileID, Size: m.Voice.FileSize, MIMEType: m.Voice.MIMEType}) + ct = adapter.ContentVoice + } else if m.Audio != nil { + media = append(media, adapter.Media{Kind: adapter.MediaVoice, URL: m.Audio.FileID, Size: m.Audio.FileSize, MIMEType: m.Audio.MIMEType}) + ct = adapter.ContentVoice + } + if m.Document != nil { + media = append(media, adapter.Media{Kind: adapter.MediaFile, URL: m.Document.FileID, Size: m.Document.FileSize, Name: m.Document.FileName, MIMEType: m.Document.MIMEType}) + ct = adapter.ContentFile + } + if m.Video != nil { + media = append(media, adapter.Media{Kind: adapter.MediaVideo, URL: m.Video.FileID, Size: m.Video.FileSize, MIMEType: m.Video.MIMEType}) + ct = adapter.ContentFile + } + + return content, ct, media +} + +// chatType maps a Telegram chat type to the adapter's ChannelType. +func chatType(t string) adapter.ChannelType { + if t == "group" || t == "supergroup" || t == "channel" { + return adapter.ChannelGroup + } + return adapter.ChannelDirect +} + +// displayName renders a sender's best human-readable name without leaking +// anything sensitive. +func displayName(u *telegramUser) string { + if u.FirstName != "" && u.LastName != "" { + return u.FirstName + " " + u.LastName + } + if u.FirstName != "" { + return u.FirstName + } + if u.Username != "" { + return u.Username + } + return strconv.FormatInt(u.ID, 10) +} + +// recordChat caches a conversation so ListChannels can report it. +func (a *Adapter) recordChat(c telegramChat, from *telegramUser) { + info := adapter.ChannelInfo{ + ChannelID: strconv.FormatInt(c.ID, 10), + ChannelType: chatType(c.Type), + ChannelName: c.Title, + } + if info.ChannelType == adapter.ChannelGroup { + info.GroupID = info.ChannelID + info.GroupName = c.Title + } else if from != nil { + info.ChannelName = displayName(from) + } + if info.ChannelName == "" { + info.ChannelName = info.ChannelID + } + + a.mu.Lock() + a.chats[info.ChannelID] = info + a.mu.Unlock() +} diff --git a/plugins/channels/telegram/updates_test.go b/plugins/channels/telegram/updates_test.go new file mode 100644 index 0000000..b421c66 --- /dev/null +++ b/plugins/channels/telegram/updates_test.go @@ -0,0 +1,244 @@ +package telegram + +import ( + "context" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestToUnifiedMessage_PrivateText(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + u := telegramUpdate{ + UpdateID: 1, + Message: &telegramMessage{ + MessageID: 42, + From: &telegramUser{ID: 100, FirstName: "Alice", Username: "alice"}, + Chat: telegramChat{ID: 200, Type: "private"}, + Date: 1700000000, + Text: "hello world", + }, + } + msg, ok := a.toUnifiedMessage(u) + if !ok { + t.Fatal("expected ok") + } + if msg.Platform != "telegram" { + t.Errorf("Platform = %q, want telegram", msg.Platform) + } + if msg.ChannelType != adapter.ChannelDirect { + t.Errorf("ChannelType = %q, want direct", msg.ChannelType) + } + if msg.ChannelID != "200" { + t.Errorf("ChannelID = %q, want 200", msg.ChannelID) + } + if msg.SenderID != "100" { + t.Errorf("SenderID = %q, want 100", msg.SenderID) + } + if msg.SenderName != "Alice" { + t.Errorf("SenderName = %q, want Alice", msg.SenderName) + } + if msg.Content != "hello world" || msg.ContentType != adapter.ContentText { + t.Errorf("content = %q/%q, want hello world/text", msg.Content, msg.ContentType) + } + if msg.Sender.UserID != "100" || msg.Sender.DisplayName != "Alice" { + t.Errorf("Sender actor = %+v", msg.Sender) + } + if msg.Sender.IsGroupChat() { + t.Error("private message should not be a group actor") + } + wantTS := time.Unix(1700000000, 0) + if !msg.Timestamp.Equal(wantTS) { + t.Errorf("Timestamp = %v, want %v", msg.Timestamp, wantTS) + } +} + +func TestToUnifiedMessage_Group(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + u := telegramUpdate{ + UpdateID: 2, + Message: &telegramMessage{ + MessageID: 7, + From: &telegramUser{ID: 300, FirstName: "Bob"}, + Chat: telegramChat{ID: -1000, Type: "supergroup", Title: "Team Chat"}, + Date: 1700000001, + Text: "ping", + }, + } + msg, ok := a.toUnifiedMessage(u) + if !ok { + t.Fatal("expected ok") + } + if msg.ChannelType != adapter.ChannelGroup { + t.Errorf("ChannelType = %q, want group", msg.ChannelType) + } + if msg.GroupID != "-1000" || msg.GroupName != "Team Chat" { + t.Errorf("group = %q/%q, want -1000/Team Chat", msg.GroupID, msg.GroupName) + } + if !msg.Sender.IsGroupChat() { + t.Error("Sender should be a group actor") + } + if msg.Sender.GroupID != "-1000" { + t.Errorf("Sender.GroupID = %q, want -1000", msg.Sender.GroupID) + } +} + +func TestToUnifiedMessage_ImageFileID(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + u := telegramUpdate{ + UpdateID: 3, + Message: &telegramMessage{ + MessageID: 9, + From: &telegramUser{ID: 400, FirstName: "Cara"}, + Chat: telegramChat{ID: 500, Type: "private"}, + Date: 1700000002, + Caption: "look at this", + Photo: []telegramPhotoSize{{FileID: "photo-small", FileSize: 100}, {FileID: "photo-big", FileSize: 2000}}, + }, + } + msg, ok := a.toUnifiedMessage(u) + if !ok { + t.Fatal("expected ok") + } + if msg.ContentType != adapter.ContentImage { + t.Errorf("ContentType = %q, want image", msg.ContentType) + } + if msg.Content != "look at this" { + t.Errorf("Content = %q, want caption", msg.Content) + } + if len(msg.Media) != 1 { + t.Fatalf("media length = %d, want 1", len(msg.Media)) + } + m := msg.Media[0] + if m.Kind != adapter.MediaImage || m.URL != "photo-big" || m.Size != 2000 { + t.Errorf("media = %+v, want image/photo-big/2000", m) + } +} + +func TestToUnifiedMessage_VoiceFileID(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + u := telegramUpdate{ + UpdateID: 4, + Message: &telegramMessage{ + MessageID: 10, + From: &telegramUser{ID: 600, FirstName: "Dan"}, + Chat: telegramChat{ID: 700, Type: "private"}, + Date: 1700000003, + Voice: &telegramAudio{FileID: "voice-id", FileSize: 5000, MIMEType: "audio/ogg"}, + }, + } + msg, ok := a.toUnifiedMessage(u) + if !ok { + t.Fatal("expected ok") + } + if msg.ContentType != adapter.ContentVoice { + t.Errorf("ContentType = %q, want voice", msg.ContentType) + } + if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaVoice || msg.Media[0].URL != "voice-id" { + t.Errorf("media = %+v, want voice/voice-id", msg.Media) + } +} + +func TestToUnifiedMessage_DocumentFileID(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + u := telegramUpdate{ + UpdateID: 5, + Message: &telegramMessage{ + MessageID: 11, + From: &telegramUser{ID: 800, FirstName: "Eve"}, + Chat: telegramChat{ID: 900, Type: "private"}, + Date: 1700000004, + Document: &telegramDocument{FileID: "doc-id", FileSize: 9000, FileName: "report.pdf", MIMEType: "application/pdf"}, + }, + } + msg, ok := a.toUnifiedMessage(u) + if !ok { + t.Fatal("expected ok") + } + if msg.ContentType != adapter.ContentFile { + t.Errorf("ContentType = %q, want file", msg.ContentType) + } + m := msg.Media[0] + if m.Kind != adapter.MediaFile || m.URL != "doc-id" || m.Name != "report.pdf" { + t.Errorf("media = %+v, want file/doc-id/report.pdf", m) + } +} + +func TestToUnifiedMessage_ReplyAndThread(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + u := telegramUpdate{ + UpdateID: 6, + Message: &telegramMessage{ + MessageID: 20, + From: &telegramUser{ID: 1000, FirstName: "Finn"}, + Chat: telegramChat{ID: -2000, Type: "group", Title: "Thread Group"}, + Date: 1700000005, + Text: "a reply", + MessageThreadID: 33, + ReplyToMessage: &telegramMessage{MessageID: 19, Text: "original"}, + }, + } + msg, ok := a.toUnifiedMessage(u) + if !ok { + t.Fatal("expected ok") + } + if msg.ThreadID != "33" { + t.Errorf("ThreadID = %q, want 33", msg.ThreadID) + } + if msg.ReplyToID != "19" || msg.ReplyToText != "original" { + t.Errorf("reply = %q/%q, want 19/original", msg.ReplyToID, msg.ReplyToText) + } +} + +func TestToUnifiedMessage_NoMessage(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + u := telegramUpdate{UpdateID: 7} + if _, ok := a.toUnifiedMessage(u); ok { + t.Fatal("expected ok=false for non-message update") + } +} + +func TestListChannels_FromConversations(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + a.toUnifiedMessage(telegramUpdate{ + UpdateID: 8, + Message: &telegramMessage{ + MessageID: 1, + From: &telegramUser{ID: 1, FirstName: "Ada"}, + Chat: telegramChat{ID: 10, Type: "private"}, + Text: "hi", + }, + }) + a.toUnifiedMessage(telegramUpdate{ + UpdateID: 9, + Message: &telegramMessage{ + MessageID: 2, + From: &telegramUser{ID: 2, FirstName: "Bob"}, + Chat: telegramChat{ID: -20, Type: "group", Title: "Gang"}, + Text: "yo", + }, + }) + got, err := a.ListChannels(context.Background()) + if err != nil { + t.Fatalf("ListChannels: %v", err) + } + if len(got) != 2 { + t.Fatalf("channels length = %d, want 2: %+v", len(got), got) + } + // Sort puts -20 (group) before 10 (private) as strings. + if got[0].ChannelID != "-20" || got[0].ChannelType != adapter.ChannelGroup || got[0].GroupName != "Gang" { + t.Errorf("channels[0] = %+v, want group -20", got[0]) + } + if got[1].ChannelID != "10" || got[1].ChannelType != adapter.ChannelDirect || got[1].ChannelName != "Ada" { + t.Errorf("channels[1] = %+v, want direct 10", got[1]) + } +}