1package telegram
2
3import (
4 "context"
5 "fmt"
6 "strings"
7 "sync"
8 "time"
9
10 "github.com/patriceckhart/zot/internal/core"
11 "github.com/patriceckhart/zot/internal/provider"
12)
13
14// Bot owns the Telegram polling loop and dispatches inbound DMs to
15// the agent. It is a long-running goroutine; Run blocks until ctx
16// cancels.
17type Bot struct {
18 Client *Client
19 Agent *core.Agent
20 Config Config
21 ZotHome string
22 // Save persists cfg to bot.json. Called whenever the bot pairs
23 // with a new allowed user or advances LastUpdateID.
24 Save func(Config) error
25 // RefreshCreds is called before every turn to pick up newly
26 // refreshed OAuth tokens. Optional; when nil, the bot uses the
27 // credential it was built with. Implementations typically call
28 // agent.ResolveCredentialFull which auto-refreshes expired tokens.
29 RefreshCreds func() error
30
31 mu sync.Mutex
32 busy bool
33 activeCtx context.CancelFunc
34 queue []queuedTurn
35}
36
37// queuedTurn is an inbound DM waiting to become a prompt.
38type queuedTurn struct {
39 chatID int64
40 messageID int
41 prompt string
42 images []provider.ImageBlock
43}
44
45// Run drives the bot. Returns when ctx is cancelled or GetMe fails.
46func (b *Bot) Run(ctx context.Context) error {
47 if b.Config.BotToken == "" {
48 return fmt.Errorf("no bot token configured; run `zot bot setup` first")
49 }
50 me, err := b.Client.GetMe(ctx)
51 if err != nil {
52 return fmt.Errorf("getMe: %w", err)
53 }
54 // Keep the stored username/id in sync with the actual bot.
55 if b.Config.BotID != me.ID || b.Config.BotUsername != me.Username {
56 b.Config.BotID = me.ID
57 b.Config.BotUsername = me.Username
58 _ = b.Save(b.Config)
59 }
60
61 fmt.Printf("telegram bridge online as @%s (id=%d)\n", me.Username, me.ID)
62 if b.Config.AllowedUserID == 0 {
63 fmt.Println("no user paired yet — send /start to the bot from Telegram to claim it")
64 } else {
65 fmt.Printf("paired with telegram user id %d\n", b.Config.AllowedUserID)
66 }
67
68 return b.pollLoop(ctx)
69}
70
71// pollLoop long-polls Telegram for updates and dispatches them.
72func (b *Bot) pollLoop(ctx context.Context) error {
73 backoff := time.Second
74 for {
75 if err := ctx.Err(); err != nil {
76 return err
77 }
78 updates, err := b.Client.GetUpdates(ctx, b.Config.LastUpdateID+1, 30)
79 if err != nil {
80 if ctx.Err() != nil {
81 return ctx.Err()
82 }
83 fmt.Fprintln(stderr(), "telegram: getUpdates error:", err)
84 select {
85 case <-ctx.Done():
86 return ctx.Err()
87 case <-time.After(backoff):
88 }
89 if backoff < 30*time.Second {
90 backoff *= 2
91 }
92 continue
93 }
94 backoff = time.Second
95 for _, u := range updates {
96 if err := b.handleUpdate(ctx, u); err != nil {
97 fmt.Fprintln(stderr(), "telegram: handleUpdate:", err)
98 }
99 b.Config.LastUpdateID = u.UpdateID
100 _ = b.Save(b.Config)
101 }
102 }
103}
104
105// handleUpdate processes a single Telegram update.
106func (b *Bot) handleUpdate(ctx context.Context, u Update) error {
107 msg := u.Message
108 if msg == nil {
109 msg = u.Edited
110 }
111 if msg == nil || msg.From == nil || msg.From.IsBot {
112 return nil
113 }
114 if msg.Chat.Type != "private" {
115 return nil
116 }
117
118 // Pairing: first user who sends /start claims the bridge.
119 text := strings.TrimSpace(msg.Text)
120 if b.Config.AllowedUserID == 0 {
121 if strings.HasPrefix(text, "/start") {
122 b.Config.AllowedUserID = msg.From.ID
123 _ = b.Save(b.Config)
124 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
125 fmt.Sprintf("paired with @%s. send any message and i'll forward it to zot.", msg.From.Username),
126 msg.MessageID)
127 return nil
128 }
129 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
130 "this bot isn't paired yet. send /start to claim it.",
131 msg.MessageID)
132 return nil
133 }
134
135 // Enforce allowed user.
136 if msg.From.ID != b.Config.AllowedUserID {
137 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
138 "this bot is paired with a different user.",
139 msg.MessageID)
140 return nil
141 }
142
143 // Built-in commands that bypass the agent.
144 switch text {
145 case "/start", "/help":
146 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
147 "send me any message and i'll forward it to zot. attach an image and i'll pass it to the model. commands: /status, /stop.",
148 msg.MessageID)
149 return nil
150 case "/status":
151 return b.sendStatus(ctx, msg.Chat.ID, msg.MessageID)
152 case "/stop":
153 b.mu.Lock()
154 cancel := b.activeCtx
155 b.mu.Unlock()
156 if cancel != nil {
157 cancel()
158 _ = b.Client.SendMessage(ctx, msg.Chat.ID, "cancelled the current turn.", msg.MessageID)
159 } else {
160 _ = b.Client.SendMessage(ctx, msg.Chat.ID, "nothing running.", msg.MessageID)
161 }
162 return nil
163 }
164
165 // Build the prompt: combine text + caption; download image attachments.
166 prompt := strings.TrimSpace(msg.Text)
167 if msg.Caption != "" {
168 if prompt != "" {
169 prompt += "\n"
170 }
171 prompt += msg.Caption
172 }
173
174 var images []provider.ImageBlock
175 if len(msg.Photo) > 0 {
176 // Photos arrive in multiple sizes; take the largest (last in the slice).
177 largest := msg.Photo[len(msg.Photo)-1]
178 if data, mime, err := b.download(ctx, largest.FileID, ""); err == nil {
179 images = append(images, provider.ImageBlock{MimeType: mime, Data: data})
180 } else {
181 fmt.Fprintln(stderr(), "telegram: download photo:", err)
182 }
183 }
184 if msg.Document != nil && isImageMIME(msg.Document.MimeType) {
185 if data, mime, err := b.download(ctx, msg.Document.FileID, msg.Document.MimeType); err == nil {
186 images = append(images, provider.ImageBlock{MimeType: mime, Data: data})
187 }
188 }
189
190 if prompt == "" && len(images) == 0 {
191 return nil
192 }
193
194 b.mu.Lock()
195 b.queue = append(b.queue, queuedTurn{
196 chatID: msg.Chat.ID,
197 messageID: msg.MessageID,
198 prompt: prompt,
199 images: images,
200 })
201 idle := !b.busy
202 b.mu.Unlock()
203
204 if idle {
205 go b.drainQueue(ctx)
206 }
207 return nil
208}
209
210// drainQueue runs queued turns one at a time until the queue is empty.
211func (b *Bot) drainQueue(parent context.Context) {
212 for {
213 b.mu.Lock()
214 if len(b.queue) == 0 {
215 b.busy = false
216 b.activeCtx = nil
217 b.mu.Unlock()
218 return
219 }
220 t := b.queue[0]
221 b.queue = b.queue[1:]
222 b.busy = true
223 turnCtx, cancel := context.WithCancel(parent)
224 b.activeCtx = cancel
225 b.mu.Unlock()
226
227 if b.RefreshCreds != nil {
228 if err := b.RefreshCreds(); err != nil {
229 fmt.Fprintln(stderr(), "telegram: refresh creds:", err)
230 }
231 }
232 b.runTurn(turnCtx, t)
233 cancel()
234 }
235}
236
237// runTurn sends the queued prompt to the agent and streams the reply.
238func (b *Bot) runTurn(ctx context.Context, t queuedTurn) {
239 stopTyping := b.startTyping(ctx, t.chatID)
240 defer stopTyping()
241
242 var replyBuilder strings.Builder
243 var lastAssistantText string
244 var turnErr error
245
246 sink := func(ev core.AgentEvent) {
247 switch e := ev.(type) {
248 case core.EvTextDelta:
249 replyBuilder.WriteString(e.Delta)
250 case core.EvAssistantMessage:
251 var sb strings.Builder
252 for _, c := range e.Message.Content {
253 if tb, ok := c.(provider.TextBlock); ok {
254 if sb.Len() > 0 {
255 sb.WriteString("\n")
256 }
257 sb.WriteString(tb.Text)
258 }
259 }
260 if sb.Len() > 0 {
261 lastAssistantText = sb.String()
262 }
263 replyBuilder.Reset()
264 case core.EvTurnEnd:
265 if e.Err != nil {
266 turnErr = e.Err
267 }
268 }
269 }
270
271 if err := b.Agent.Prompt(ctx, t.prompt, t.images, sink); err != nil {
272 turnErr = err
273 }
274
275 reply := strings.TrimSpace(lastAssistantText)
276 if reply == "" {
277 reply = strings.TrimSpace(replyBuilder.String())
278 }
279 if turnErr != nil && ctx.Err() == nil {
280 reply = "error: " + turnErr.Error()
281 }
282 if reply == "" {
283 reply = "(no reply)"
284 }
285 // Telegram caps messages at 4096 chars. Chunk to be safe.
286 for _, chunk := range chunkMessage(reply, 4000) {
287 if err := b.Client.SendMessage(context.Background(), t.chatID, chunk, 0); err != nil {
288 fmt.Fprintln(stderr(), "telegram: sendMessage:", err)
289 break
290 }
291 }
292}
293
294// startTyping keeps Telegram's "typing..." indicator alive until the
295// returned stop function is called.
296func (b *Bot) startTyping(ctx context.Context, chatID int64) func() {
297 tctx, cancel := context.WithCancel(ctx)
298 go func() {
299 for {
300 _ = b.Client.SendChatAction(tctx, chatID, "typing")
301 select {
302 case <-tctx.Done():
303 return
304 case <-time.After(4 * time.Second):
305 }
306 }
307 }()
308 return cancel
309}
310
311// sendStatus describes agent state to the Telegram user.
312func (b *Bot) sendStatus(ctx context.Context, chatID int64, replyTo int) error {
313 b.mu.Lock()
314 busy := b.busy
315 queued := len(b.queue)
316 b.mu.Unlock()
317
318 state := "idle"
319 if busy {
320 state = "working"
321 }
322 lines := []string{
323 fmt.Sprintf("state: %s", state),
324 fmt.Sprintf("queued: %d", queued),
325 fmt.Sprintf("model: %s", b.Agent.Model),
326 }
327 cost := b.Agent.Cost()
328 lines = append(lines, fmt.Sprintf("cost: $%.4f (%d in / %d out)",
329 cost.CostUSD, cost.InputTokens, cost.OutputTokens))
330 return b.Client.SendMessage(ctx, chatID, strings.Join(lines, "\n"), replyTo)
331}
332
333// download fetches a file from Telegram and returns bytes + mime.
334func (b *Bot) download(ctx context.Context, fileID, mime string) ([]byte, string, error) {
335 f, err := b.Client.GetFile(ctx, fileID)
336 if err != nil {
337 return nil, "", err
338 }
339 data, err := b.Client.DownloadFile(ctx, f.FilePath)
340 if err != nil {
341 return nil, "", err
342 }
343 if mime == "" {
344 mime = guessImageMIME(f.FilePath)
345 }
346 return data, mime, nil
347}
348
349// chunkMessage splits s into chunks no larger than limit runes, on line
350// boundaries when possible.
351func chunkMessage(s string, limit int) []string {
352 if len(s) <= limit {
353 return []string{s}
354 }
355 var out []string
356 lines := strings.Split(s, "\n")
357 var cur strings.Builder
358 for _, l := range lines {
359 if cur.Len()+len(l)+1 > limit && cur.Len() > 0 {
360 out = append(out, cur.String())
361 cur.Reset()
362 }
363 if len(l) > limit {
364 // Line itself too long; hard-split.
365 for len(l) > limit {
366 out = append(out, l[:limit])
367 l = l[limit:]
368 }
369 }
370 if cur.Len() > 0 {
371 cur.WriteString("\n")
372 }
373 cur.WriteString(l)
374 }
375 if cur.Len() > 0 {
376 out = append(out, cur.String())
377 }
378 return out
379}
380
381// isImageMIME returns true for MIME types the model can probably ingest
382// as a vision input.
383func isImageMIME(m string) bool {
384 switch strings.ToLower(m) {
385 case "image/png", "image/jpeg", "image/jpg", "image/gif", "image/webp":
386 return true
387 }
388 return false
389}
390
391// guessImageMIME infers a mime type from a filename suffix. Falls back
392// to image/png because telegram photos are always re-encoded to jpeg
393// but getFile's file_path may omit the extension.
394func guessImageMIME(path string) string {
395 lower := strings.ToLower(path)
396 switch {
397 case strings.HasSuffix(lower, ".png"):
398 return "image/png"
399 case strings.HasSuffix(lower, ".jpg"), strings.HasSuffix(lower, ".jpeg"):
400 return "image/jpeg"
401 case strings.HasSuffix(lower, ".gif"):
402 return "image/gif"
403 case strings.HasSuffix(lower, ".webp"):
404 return "image/webp"
405 }
406 return "image/jpeg"
407}