git.giacomolli.uk - onechat/blob - telegram.go

   1 package main
   2
   3 import (
   4         "bytes"
   5         "context"
   6         "encoding/json"
   7         "errors"
   8         "fmt"
   9         "io"
  10         "log"
  11         "os"
  12         "path/filepath"
  13         "strings"
  14         "sync"
  15         "time"
  16 )
  17
  18 const (
  19         // telegramAPI is the Telegram Bot API service.
  20         telegramAPI = "https://api.telegram.org"
  21
  22         // botConfigFileName is the bot config in the data directory. The daemon
  23         // runs the bot only when the file exists.
  24         botConfigFileName = "telegram.json"
  25
  26         // offsetFileName is the update offset of the bot in the agent directory.
  27         offsetFileName = "telegram-offset.json"
  28
  29         // telegramSource is the source of the user messages from Telegram.
  30         telegramSource = "telegram"
  31
  32         // messageUnits is the most UTF-16 code units of one Telegram message.
  33         messageUnits = 4096
  34
  35         // typingInterval is how often the bot shows the typing action while a
  36         // turn runs. Telegram shows the action for about five seconds.
  37         typingInterval = 4 * time.Second
  38
  39         // textOnly is the answer to a message that is not text.
  40         textOnly = "Only text messages are supported."
  41 )
  42
  43 // botConfig holds the settings of telegram.json: the bot token and the user ID
  44 // of the owner. The token is a secret, so the file must have mode 0600.
  45 type botConfig struct {
  46         Token string `json:"token"`
  47         Owner int64  `json:"owner"`
  48 }
  49
  50 // telegramBot connects the private chat of the owner with the agent. It polls
  51 // the Bot API for new messages and sends each text message of the owner to the
  52 // agent. It ignores every other update. Its sender sends the replies, the
  53 // errors, and the user messages of the other clients to the chat as plain text,
  54 // and shows the typing action while a turn runs. Tool calls do not go to the
  55 // chat.
  56 //
  57 // The offset file holds the ID of the next update. It moves past an update
  58 // only after the message of the update is durable in the log, so a restart
  59 // does not lose a message. A crash between the flush and the offset write
  60 // makes Telegram send the update again, and the message repeats once.
  61 //
  62 // A failed Bot API call waits and tries again, from minWait up to maxWait, or
  63 // as long as the Bot API asks. It never stops the agent.
  64 type telegramBot struct {
  65         api        *botAPI
  66         owner      int64
  67         offsetPath string
  68         logger     *log.Logger
  69         minWait    time.Duration
  70         maxWait    time.Duration
  71
  72         mutex sync.Mutex
  73         queue []Event
  74         ready chan struct{}
  75 }
  76
  77 // offsetFile is the content of the offset file.
  78 type offsetFile struct {
  79         Offset int64 `json:"offset"`
  80 }
  81
  82 // loadBotConfig reads telegram.json in dir. It returns false when the file
  83 // does not exist. An error never holds the token.
  84 func loadBotConfig(dir string) (botConfig, bool, error) {
  85         path := filepath.Join(dir, botConfigFileName)
  86         info, err := os.Stat(path)
  87         if errors.Is(err, os.ErrNotExist) {
  88                 return botConfig{}, false, nil
  89         }
  90
  91         if err != nil {
  92                 return botConfig{}, false, fmt.Errorf("read bot config: %w", err)
  93         }
  94
  95         if info.Mode().Perm()&0o077 != 0 {
  96                 return botConfig{}, false, fmt.Errorf("%s holds the bot token, so its mode must be 0600", botConfigFileName)
  97         }
  98
  99         data, err := os.ReadFile(path)
 100         if err != nil {
 101                 return botConfig{}, false, fmt.Errorf("read bot config: %w", err)
 102         }
 103
 104         var config botConfig
 105         decoder := json.NewDecoder(bytes.NewReader(data))
 106         decoder.DisallowUnknownFields()
 107         err = decoder.Decode(&config)
 108         if err == nil && !errors.Is(decoder.Decode(&struct{}{}), io.EOF) {
 109                 err = errors.New("data after the JSON object")
 110         }
 111
 112         if err != nil {
 113                 return botConfig{}, false, fmt.Errorf("%s is not valid JSON with the fields token and owner", botConfigFileName)
 114         }
 115
 116         err = config.validate()
 117         if err != nil {
 118                 return botConfig{}, false, fmt.Errorf("%s: %w", botConfigFileName, err)
 119         }
 120
 121         return config, true, nil
 122 }
 123
 124 // validate makes sure that the token has the form of a bot token and that the
 125 // owner is a user ID. The token goes into a URL path, so it can only hold the
 126 // characters of a bot token.
 127 func (c botConfig) validate() error {
 128         id, secret, found := strings.Cut(c.Token, ":")
 129         if !found || id == "" || secret == "" || strings.Trim(id, "0123456789") != "" || strings.Trim(secret, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-") != "" {
 130                 return errors.New("token does not have the form of a bot token from BotFather")
 131         }
 132
 133         if c.Owner <= 0 {
 134                 return errors.New("owner must be the numeric Telegram user ID of the owner")
 135         }
 136
 137         return nil
 138 }
 139
 140 // newTelegramBot returns the bot of owner on api. The offset file is at
 141 // offsetPath, and logger gets the failures. No log line holds the text of a
 142 // message.
 143 func newTelegramBot(api *botAPI, owner int64, offsetPath string, logger *log.Logger) *telegramBot {
 144         return &telegramBot{
 145                 api:        api,
 146                 owner:      owner,
 147                 offsetPath: offsetPath,
 148                 logger:     logger,
 149                 minWait:    time.Second,
 150                 maxWait:    time.Minute,
 151                 ready:      make(chan struct{}, 1),
 152         }
 153 }
 154
 155 // Notify queues event for the chat. It never blocks, so a slow network does not
 156 // stop the agent.
 157 func (b *telegramBot) Notify(event Event) {
 158         switch event.Kind {
 159         case "busy", "idle", "reply", "error":
 160         case "user":
 161                 if event.Source == telegramSource {
 162                         return
 163                 }
 164         default:
 165                 return
 166         }
 167
 168         b.mutex.Lock()
 169         b.queue = append(b.queue, event)
 170         b.mutex.Unlock()
 171
 172         select {
 173         case b.ready <- struct{}{}:
 174         default:
 175         }
 176 }
 177
 178 // Run polls for the messages of the owner and sends the queued events to the
 179 // chat until ctx ends. It returns an error only when a message or the offset
 180 // cannot be written, because then the bot would lose or repeat messages.
 181 func (b *telegramBot) Run(ctx context.Context, l *loop) error {
 182         ctx, cancel := context.WithCancel(ctx)
 183         defer cancel()
 184
 185         var sender sync.WaitGroup
 186         sender.Add(1)
 187         go func() {
 188                 defer sender.Done()
 189
 190                 b.send(ctx)
 191         }()
 192
 193         err := b.poll(ctx, l)
 194         cancel()
 195         sender.Wait()
 196
 197         return err
 198 }
 199
 200 // send sends each queued event to the chat in order until ctx ends. While a
 201 // turn runs, it shows the typing action every typingInterval and after each
 202 // message, because a message ends the action.
 203 func (b *telegramBot) send(ctx context.Context) {
 204         ticker := time.NewTicker(typingInterval)
 205         defer ticker.Stop()
 206
 207         busy := false
 208         for ctx.Err() == nil {
 209                 event, found := b.next()
 210                 if !found {
 211                         select {
 212                         case <-b.ready:
 213                         case <-ticker.C:
 214                                 if busy {
 215                                         b.typing(ctx)
 216                                 }
 217                         case <-ctx.Done():
 218                         }
 219
 220                         continue
 221                 }
 222
 223                 switch event.Kind {
 224                 case "busy":
 225                         busy = true
 226                         b.typing(ctx)
 227                         ticker.Reset(typingInterval)
 228                 case "idle":
 229                         busy = false
 230                 default:
 231                         for _, part := range splitMessage(botText(event), messageUnits) {
 232                                 b.deliver(ctx, part)
 233                         }
 234
 235                         if busy {
 236                                 b.typing(ctx)
 237                         }
 238                 }
 239         }
 240 }
 241
 242 // next removes the first queued event and returns it.
 243 func (b *telegramBot) next() (Event, bool) {
 244         b.mutex.Lock()
 245         defer b.mutex.Unlock()
 246
 247         if len(b.queue) == 0 {
 248                 return Event{}, false
 249         }
 250
 251         event := b.queue[0]
 252         b.queue = b.queue[1:]
 253
 254         return event, true
 255 }
 256
 257 // typing shows the typing action once. A failure is not logged, because the
 258 // next message or poll logs the state of the network.
 259 func (b *telegramBot) typing(ctx context.Context) {
 260         _ = b.api.SendTyping(ctx, b.owner)
 261 }
 262
 263 // deliver sends text to the chat. It tries again after each failure that a
 264 // later try can fix, until ctx ends. It drops the text after any other failure.
 265 func (b *telegramBot) deliver(ctx context.Context, text string) {
 266         wait := b.minWait
 267         for {
 268                 err := b.api.SendMessage(ctx, b.owner, text)
 269                 if err == nil || ctx.Err() != nil {
 270                         return
 271                 }
 272
 273                 retry, asked := retryable(err)
 274                 if !retry {
 275                         b.logger.Printf("message dropped: %v", err)
 276                         return
 277                 }
 278
 279                 wait = b.pause(ctx, err, wait, asked)
 280         }
 281 }
 282
 283 // pause logs err and waits the longer of wait and asked, or until ctx ends. It
 284 // returns the wait of the next failure: twice wait, up to maxWait.
 285 func (b *telegramBot) pause(ctx context.Context, err error, wait, asked time.Duration) time.Duration {
 286         delay := max(wait, asked)
 287         b.logger.Printf("%v (next try in %s)", err, delay)
 288
 289         timer := time.NewTimer(delay)
 290         defer timer.Stop()
 291
 292         select {
 293         case <-timer.C:
 294         case <-ctx.Done():
 295         }
 296
 297         return min(2*wait, b.maxWait)
 298 }
 299
 300 // poll gets the updates from the stored offset on and handles each one, until
 301 // ctx ends. Each update is handled to its end also when ctx ends during it, so
 302 // a stop does not repeat its message.
 303 func (b *telegramBot) poll(ctx context.Context, l *loop) error {
 304         offset, err := loadOffset(b.offsetPath)
 305         if err != nil {
 306                 return err
 307         }
 308
 309         wait := b.minWait
 310         for ctx.Err() == nil {
 311                 updates, err := b.api.GetUpdates(ctx, offset)
 312                 if ctx.Err() != nil {
 313                         return nil
 314                 }
 315
 316                 if err != nil {
 317                         _, asked := retryable(err)
 318                         wait = b.pause(ctx, err, wait, asked)
 319
 320                         continue
 321                 }
 322
 323                 wait = b.minWait
 324                 for _, update := range updates {
 325                         if ctx.Err() != nil {
 326                                 return nil
 327                         }
 328
 329                         err = b.handle(context.WithoutCancel(ctx), l, update)
 330                         if err != nil {
 331                                 return err
 332                         }
 333
 334                         offset = update.UpdateID + 1
 335                         err = saveOffset(b.offsetPath, offset)
 336                         if err != nil {
 337                                 return err
 338                         }
 339                 }
 340         }
 341
 342         return nil
 343 }
 344
 345 // handle runs one update. A text message of the owner in the private chat goes
 346 // to the agent, and handle returns when it is durable in the log. The line
 347 // /stop interrupts the turn. Any other message of the owner gets the answer
 348 // that only text is supported. Every other update is ignored, and its content
 349 // is not logged.
 350 func (b *telegramBot) handle(ctx context.Context, l *loop, update botUpdate) error {
 351         message := update.Message
 352         if message == nil || message.Chat.Type != "private" || message.Chat.ID != b.owner || message.From == nil || message.From.ID != b.owner {
 353                 b.logger.Print("update ignored: not a message of the owner in the private chat")
 354                 return nil
 355         }
 356
 357         switch {
 358         case message.Text == "":
 359                 b.notice(textOnly)
 360         case strings.TrimSpace(message.Text) == stopCommand:
 361                 if !l.Interrupt() {
 362                         b.notice("No turn runs.")
 363                 }
 364         default:
 365                 _, err := l.Submit(telegramSource, message.Text)
 366                 if err != nil {
 367                         return err
 368                 }
 369
 370                 err = l.Flush(ctx)
 371                 if err != nil {
 372                         return fmt.Errorf("flush the message from Telegram: %w", err)
 373                 }
 374         }
 375
 376         return nil
 377 }
 378
 379 // notice queues a short answer of the bot for the chat.
 380 func (b *telegramBot) notice(text string) {
 381         b.Notify(Event{
 382                 Kind: "reply",
 383                 Text: text,
 384         })
 385 }
 386
 387 // loadOffset reads the update offset at path. A missing file gives 0, which
 388 // asks for every update that Telegram keeps.
 389 func loadOffset(path string) (int64, error) {
 390         data, err := os.ReadFile(path)
 391         if errors.Is(err, os.ErrNotExist) {
 392                 return 0, nil
 393         }
 394
 395         if err != nil {
 396                 return 0, fmt.Errorf("read update offset: %w", err)
 397         }
 398
 399         var file offsetFile
 400         err = json.Unmarshal(data, &file)
 401         if err != nil {
 402                 return 0, fmt.Errorf("%s: %w", path, err)
 403         }
 404
 405         return file.Offset, nil
 406 }
 407
 408 // saveOffset replaces the update offset at path.
 409 func saveOffset(path string, offset int64) error {
 410         data, err := json.Marshal(offsetFile{
 411                 Offset: offset,
 412         })
 413         if err != nil {
 414                 return fmt.Errorf("encode update offset: %w", err)
 415         }
 416
 417         err = writeFileAtomic(path, append(data, '\n'))
 418         if err != nil {
 419                 return fmt.Errorf("write update offset: %w", err)
 420         }
 421
 422         return nil
 423 }
 424
 425 // botText returns the text of event for the chat. A user message of another
 426 // client has its source as a mark.
 427 func botText(event Event) string {
 428         switch event.Kind {
 429         case "user":
 430                 return "[" + event.Source + "] " + event.Text
 431         case "error":
 432                 return "Error: " + event.Text
 433         default:
 434                 return event.Text
 435         }
 436 }
 437
 438 // splitMessage cuts text into parts of at most limit UTF-16 code units, the
 439 // unit of the Telegram message limit. It cuts after the last line break or
 440 // space of a part when one is in its second half, and else at the limit. It
 441 // never cuts a character.
 442 func splitMessage(text string, limit int) []string {
 443         var parts []string
 444         for {
 445                 end := prefixEnd(text, limit)
 446                 if end == len(text) {
 447                         return append(parts, text)
 448                 }
 449
 450                 cut := strings.LastIndexByte(text[:end], '\n') + 1
 451                 if cut <= end/2 {
 452                         cut = strings.LastIndexByte(text[:end], ' ') + 1
 453                 }
 454
 455                 if cut <= end/2 {
 456                         cut = end
 457                 }
 458
 459                 parts = append(parts, text[:cut])
 460                 text = text[cut:]
 461         }
 462 }
 463
 464 // prefixEnd returns the byte length of the longest prefix of text with at most
 465 // limit UTF-16 code units.
 466 func prefixEnd(text string, limit int) int {
 467         units := 0
 468         for index, char := range text {
 469                 size := 1
 470                 if char > 0xFFFF {
 471                         size = 2
 472                 }
 473
 474                 if units+size > limit {
 475                         return index
 476                 }
 477
 478                 units += size
 479         }
 480
 481         return len(text)
 482 }