1package telegram
2
3import (
4 "context"
5 "fmt"
6 "strings"
7 "sync"
8 "time"
9
10 "github.com/patriceckhart/zot/packages/core"
11 "github.com/patriceckhart/zot/packages/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 Provider string
23 AuthMethod string
24 CWD string
25 // Save persists cfg to bot.json. Called whenever the bot pairs
26 // with a new allowed user or advances LastUpdateID.
27 Save func(Config) error
28 // RefreshCreds is called before every turn to pick up newly
29 // refreshed OAuth tokens. Optional; when nil, the bot uses the
30 // credential it was built with. Implementations typically call
31 // agent.ResolveCredentialFull which auto-refreshes expired tokens.
32 RefreshCreds func() error
33
34 mu sync.Mutex
35 busy bool
36 activeCtx context.CancelFunc
37 queue []queuedTurn
38 lastCtxInput int
39}
40
41// queuedTurn is an inbound DM waiting to become a prompt.
42type queuedTurn struct {
43 chatID int64
44 messageID int
45 prompt string
46 images []provider.ImageBlock
47}
48
49// Run drives the bot. Returns when ctx is cancelled or GetMe fails.
50func (b *Bot) Run(ctx context.Context) error {
51 if b.Config.BotToken == "" {
52 return fmt.Errorf("no bot token configured; run `zot bot setup` first")
53 }
54 me, err := b.Client.GetMe(ctx)
55 if err != nil {
56 return fmt.Errorf("getMe: %w", err)
57 }
58 // Keep the stored username/id in sync with the actual bot.
59 if b.Config.BotID != me.ID || b.Config.BotUsername != me.Username {
60 b.Config.BotID = me.ID
61 b.Config.BotUsername = me.Username
62 _ = b.Save(b.Config)
63 }
64
65 fmt.Printf("telegram bridge online as @%s (id=%d)\n", me.Username, me.ID)
66 if b.Config.AllowedUserID == 0 {
67 fmt.Println("no user paired yet — send /start to the bot from Telegram to claim it")
68 } else {
69 fmt.Printf("paired with telegram user id %d\n", b.Config.AllowedUserID)
70 }
71
72 return b.pollLoop(ctx)
73}
74
75// pollLoop long-polls Telegram for updates and dispatches them.
76func (b *Bot) pollLoop(ctx context.Context) error {
77 backoff := time.Second
78 for {
79 if err := ctx.Err(); err != nil {
80 return err
81 }
82 updates, err := b.Client.GetUpdates(ctx, b.Config.LastUpdateID+1, 30)
83 if err != nil {
84 if ctx.Err() != nil {
85 return ctx.Err()
86 }
87 fmt.Fprintln(stderr(), "telegram: getUpdates error:", err)
88 select {
89 case <-ctx.Done():
90 return ctx.Err()
91 case <-time.After(backoff):
92 }
93 if backoff < 30*time.Second {
94 backoff *= 2
95 }
96 continue
97 }
98 backoff = time.Second
99 for _, u := range updates {
100 if err := b.handleUpdate(ctx, u); err != nil {
101 fmt.Fprintln(stderr(), "telegram: handleUpdate:", err)
102 }
103 b.Config.LastUpdateID = u.UpdateID
104 _ = b.Save(b.Config)
105 }
106 }
107}
108
109// handleUpdate processes a single Telegram update.
110func (b *Bot) handleUpdate(ctx context.Context, u Update) error {
111 msg := u.Message
112 if msg == nil {
113 msg = u.Edited
114 }
115 if msg == nil || msg.From == nil || msg.From.IsBot {
116 return nil
117 }
118 if msg.Chat.Type != "private" {
119 return nil
120 }
121
122 // Pairing: first user who sends /start claims the bridge.
123 text := strings.TrimSpace(msg.Text)
124 if b.Config.AllowedUserID == 0 {
125 if strings.HasPrefix(text, "/start") {
126 b.Config.AllowedUserID = msg.From.ID
127 _ = b.Save(b.Config)
128 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
129 fmt.Sprintf("paired with @%s. send any message and i'll forward it to zot.", msg.From.Username),
130 msg.MessageID)
131 return nil
132 }
133 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
134 "this bot isn't paired yet. send /start to claim it.",
135 msg.MessageID)
136 return nil
137 }
138
139 // Enforce allowed user.
140 if msg.From.ID != b.Config.AllowedUserID {
141 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
142 "this bot is paired with a different user.",
143 msg.MessageID)
144 return nil
145 }
146
147 // Built-in commands that bypass the agent.
148 switch text {
149 case "/start", "/help":
150 _ = b.Client.SendMessage(ctx, msg.Chat.ID,
151 "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, or plain stop.",
152 msg.MessageID)
153 return nil
154 case "/status":
155 return b.sendStatus(ctx, msg.Chat.ID, msg.MessageID)
156 case "/stop":
157 b.cancelActiveTurn(ctx, msg.Chat.ID, msg.MessageID)
158 return nil
159 }
160 if isStopCommand(text) {
161 b.cancelActiveTurn(ctx, msg.Chat.ID, msg.MessageID)
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.EvUsage:
251 b.mu.Lock()
252 if e.Usage.InputTokens > 0 {
253 b.lastCtxInput = e.Usage.InputTokens + e.Usage.CacheReadTokens + e.Usage.CacheWriteTokens
254 }
255 b.mu.Unlock()
256 case core.EvAssistantMessage:
257 var sb strings.Builder
258 for _, c := range e.Message.Content {
259 if tb, ok := c.(provider.TextBlock); ok {
260 if sb.Len() > 0 {
261 sb.WriteString("\n")
262 }
263 sb.WriteString(tb.Text)
264 }
265 }
266 if sb.Len() > 0 {
267 lastAssistantText = sb.String()
268 }
269 replyBuilder.Reset()
270 case core.EvTurnEnd:
271 if e.Err != nil {
272 turnErr = e.Err
273 }
274 }
275 }
276
277 if err := b.Agent.Prompt(ctx, t.prompt, t.images, sink); err != nil {
278 turnErr = err
279 }
280
281 reply := strings.TrimSpace(lastAssistantText)
282 if reply == "" {
283 reply = strings.TrimSpace(replyBuilder.String())
284 }
285 if turnErr != nil && ctx.Err() == nil {
286 reply = "error: " + turnErr.Error()
287 }
288 if reply == "" {
289 reply = "(no reply)"
290 }
291 // Telegram caps messages at 4096 chars. Chunk to be safe.
292 for _, chunk := range chunkMessage(reply, 4000) {
293 if err := b.Client.SendMessage(context.Background(), t.chatID, chunk, 0); err != nil {
294 fmt.Fprintln(stderr(), "telegram: sendMessage:", err)
295 break
296 }
297 }
298}
299
300// startTyping keeps Telegram's "typing..." indicator alive until the
301// returned stop function is called.
302func (b *Bot) startTyping(ctx context.Context, chatID int64) func() {
303 tctx, cancel := context.WithCancel(ctx)
304 go func() {
305 for {
306 _ = b.Client.SendChatAction(tctx, chatID, "typing")
307 select {
308 case <-tctx.Done():
309 return
310 case <-time.After(4 * time.Second):
311 }
312 }
313 }()
314 return cancel
315}
316
317func (b *Bot) cancelActiveTurn(ctx context.Context, chatID int64, replyTo int) {
318 b.mu.Lock()
319 cancel := b.activeCtx
320 b.mu.Unlock()
321 if cancel != nil {
322 cancel()
323 _ = b.Client.SendMessage(ctx, chatID, "cancelled the current turn.", replyTo)
324 } else {
325 _ = b.Client.SendMessage(ctx, chatID, "nothing running.", replyTo)
326 }
327}
328
329// sendStatus describes agent state to the Telegram user.
330func (b *Bot) sendStatus(ctx context.Context, chatID int64, replyTo int) error {
331 b.mu.Lock()
332 busy := b.busy
333 queued := len(b.queue)
334 ctxUsed := b.lastCtxInput
335 providerName := b.Provider
336 authMethod := b.AuthMethod
337 cwd := b.CWD
338 b.mu.Unlock()
339
340 model := b.Agent.Model
341 ctxMax := 0
342 if m, err := provider.FindModel(providerName, model); err == nil {
343 ctxMax = m.ContextWindow
344 }
345 return b.Client.SendMessage(ctx, chatID, FormatStatus(StatusSnapshot{
346 Provider: providerName,
347 Model: model,
348 CWD: cwd,
349 Usage: b.Agent.Cost(),
350 Subscription: authMethod == "oauth",
351 ContextUsed: ctxUsed,
352 ContextMax: ctxMax,
353 Busy: busy,
354 Queued: queued,
355 }), replyTo)
356}
357
358// download fetches a file from Telegram and returns bytes + mime.
359func (b *Bot) download(ctx context.Context, fileID, mime string) ([]byte, string, error) {
360 f, err := b.Client.GetFile(ctx, fileID)
361 if err != nil {
362 return nil, "", err
363 }
364 data, err := b.Client.DownloadFile(ctx, f.FilePath)
365 if err != nil {
366 return nil, "", err
367 }
368 if mime == "" {
369 mime = guessImageMIME(f.FilePath)
370 }
371 return data, mime, nil
372}
373
374// chunkMessage splits s into chunks no larger than limit runes, on line
375// boundaries when possible.
376func chunkMessage(s string, limit int) []string {
377 if len(s) <= limit {
378 return []string{s}
379 }
380 var out []string
381 lines := strings.Split(s, "\n")
382 var cur strings.Builder
383 for _, l := range lines {
384 if cur.Len()+len(l)+1 > limit && cur.Len() > 0 {
385 out = append(out, cur.String())
386 cur.Reset()
387 }
388 if len(l) > limit {
389 // Line itself too long; hard-split.
390 for len(l) > limit {
391 out = append(out, l[:limit])
392 l = l[limit:]
393 }
394 }
395 if cur.Len() > 0 {
396 cur.WriteString("\n")
397 }
398 cur.WriteString(l)
399 }
400 if cur.Len() > 0 {
401 out = append(out, cur.String())
402 }
403 return out
404}
405
406// isImageMIME returns true for MIME types the model can probably ingest
407// as a vision input.
408func isImageMIME(m string) bool {
409 switch strings.ToLower(m) {
410 case "image/png", "image/jpeg", "image/jpg", "image/gif", "image/webp":
411 return true
412 }
413 return false
414}
415
416// guessImageMIME infers a mime type from a filename suffix. Falls back
417// to image/png because telegram photos are always re-encoded to jpeg
418// but getFile's file_path may omit the extension.
419func guessImageMIME(path string) string {
420 lower := strings.ToLower(path)
421 switch {
422 case strings.HasSuffix(lower, ".png"):
423 return "image/png"
424 case strings.HasSuffix(lower, ".jpg"), strings.HasSuffix(lower, ".jpeg"):
425 return "image/jpeg"
426 case strings.HasSuffix(lower, ".gif"):
427 return "image/gif"
428 case strings.HasSuffix(lower, ".webp"):
429 return "image/webp"
430 }
431 return "image/jpeg"
432}