package transcoder import ( "bytes" "encoding/json" "fmt" "io" "log/slog" "mime/multipart" "net/http" "os" "path/filepath" "slices" "strconv" "strings" "time" "github.com/SayaAndy/saya-today-article-metadata-add/config" "github.com/SayaAndy/saya-today-article-metadata-add/internal/draft" ) const ( telegramMediaGroupLimit = 10 // telegramCaptionLimit is the max length (runes) of a media caption. telegramCaptionLimit = 1024 // telegramMaxAttempts caps retries when the API returns 429. telegramMaxAttempts = 5 ) type TelegramTranscoder struct { botToken string channelID int groupID int lastUpdateID int imageBaseURL string previewBaseURL string previewExtension string stateDir string http *http.Client } func NewTelegramTranscoder(cfg any) (Transcoder, error) { tgCfg, ok := cfg.(*config.TelegramConfig) if !ok { return nil, fmt.Errorf("invalid config for telegram transcoder") } transcoder := &TelegramTranscoder{ botToken: tgCfg.BotToken, channelID: tgCfg.ChannelID, groupID: tgCfg.GroupID, lastUpdateID: -1, imageBaseURL: strings.TrimRight(tgCfg.ImageBaseURL, "/"), previewBaseURL: strings.TrimRight(tgCfg.PreviewBaseURL, "/"), previewExtension: tgCfg.PreviewExtension, stateDir: tgCfg.StateDir, http: &http.Client{Timeout: time.Minute}, } if _, err := transcoder.getUpdates(); err != nil { return nil, fmt.Errorf("failed to get last update id for checkpoint: %w", err) } return transcoder, nil } func (t *TelegramTranscoder) Name() string { return "telegram" } func (t *TelegramTranscoder) Finalize() error { return nil } // telegramState records, per draft, which section indices have already been // posted and the message ids they produced. Posting is append-only: a section // already present here is never re-sent. type telegramState struct { ChannelID string `json:"channelID"` Posted map[string][]int `json:"posted"` } func (t *TelegramTranscoder) Transcode(doc *draft.Document) error { statePath := filepath.Join(t.stateDir, doc.Codename+".telegram-state.json") state, err := loadTelegramState(statePath) if err != nil { return fmt.Errorf("load telegram state %q: %w", statePath, err) } if state.ChannelID != "" && state.ChannelID != strconv.Itoa(t.channelID) { slog.Warn("telegram channel changed, re-posting all sections to the new channel", slog.String("codename", doc.Codename), slog.String("old", state.ChannelID), slog.Int("new", t.channelID)) state.Posted = map[string][]int{} } state.ChannelID = strconv.Itoa(t.channelID) for _, section := range doc.Sections { idx := strconv.Itoa(section.Index) if _, done := state.Posted[idx]; done { continue } var msgIDs []int pendingText := "" // flushText sends any buffered text as its own message. flushText := func() error { if pendingText == "" { return nil } id, err := t.sendMessage(pendingText) if err != nil { return err } msgIDs = append(msgIDs, id) pendingText = "" return nil } for _, block := range section.Blocks { switch block.Kind { case draft.TextBlock: text := stripRules(block.Text) if text == "" { continue } // flush any earlier text before buffering this one if err := flushText(); err != nil { return fmt.Errorf("send section %d text: %w", section.Index, err) } pendingText = text case draft.GalleryBlock: // Attach the immediately preceding text as the album caption when // it fits; otherwise send it as its own message first. caption := "" if pendingText != "" { if len([]rune(pendingText)) <= telegramCaptionLimit { caption = pendingText pendingText = "" } else if err := flushText(); err != nil { return fmt.Errorf("send section %d text: %w", section.Index, err) } } for i, group := range splitPhotos(block.Gallery.Photos, telegramMediaGroupLimit) { groupCaption := "" if i == 0 { groupCaption = caption } ids, err := t.sendMediaGroup(group, groupCaption, "photo", t.channelID, 0) if err != nil { return fmt.Errorf("send section %d gallery: %w", section.Index, err) } msgIDs = append(msgIDs, ids...) if err := flushText(); err != nil { return fmt.Errorf("send section %d text: %w", section.Index, err) } if len(msgIDs) > 0 { time.Sleep(5 * time.Second) forwardedID, err := t.getForwardedID(msgIDs) if err != nil { return fmt.Errorf("send full documents for section %d gallery: %w", section.Index, err) } if forwardedID == 0 { return fmt.Errorf("fail to get forwarded id of section %d gallery", section.Index) } _, err = t.sendMediaGroup(group, "", "document", t.groupID, forwardedID) if err != nil { return fmt.Errorf("send section %d documents: %w", section.Index, err) } } } } } state.Posted[idx] = msgIDs if err := saveTelegramState(statePath, state); err != nil { return fmt.Errorf("save telegram state %q: %w", statePath, err) } slog.Info("telegram posted section", slog.String("codename", doc.Codename), slog.Int("section", section.Index), slog.Int("messages", len(msgIDs))) } return nil } func loadTelegramState(path string) (*telegramState, error) { raw, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { return &telegramState{Posted: map[string][]int{}}, nil } return nil, err } var state telegramState if err := json.Unmarshal(raw, &state); err != nil { return nil, err } if state.Posted == nil { state.Posted = map[string][]int{} } return &state, nil } func saveTelegramState(path string, state *telegramState) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return err } raw, err := json.MarshalIndent(state, "", " ") if err != nil { return err } return os.WriteFile(path, raw, 0o644) } // stripRules drops standalone "---" horizontal-rule lines (meaningless in a // Telegram message) and trims the remaining text. func stripRules(text string) string { lines := strings.Split(text, "\n") kept := lines[:0] for _, line := range lines { if strings.TrimSpace(line) == "---" { continue } kept = append(kept, line) } return strings.TrimSpace(strings.Join(kept, "\n")) } // splitPhotos divides photos into near-equal groups each no larger than limit. func splitPhotos(photos []draft.Photo, limit int) [][]draft.Photo { n := len(photos) if n == 0 { return nil } groupCount := (n + limit - 1) / limit base := n / groupCount rem := n % groupCount var groups [][]draft.Photo start := 0 for g := range groupCount { size := base if g < rem { size++ } groups = append(groups, photos[start:start+size]) start += size } return groups } type inputMediaPhoto struct { Type string `json:"type"` Media string `json:"media"` Caption string `json:"caption,omitempty"` ParseMode string `json:"parse_mode,omitempty"` } func (t *TelegramTranscoder) photoURL(key string) string { return t.imageBaseURL + "/" + key } func (t *TelegramTranscoder) previewURL(key string) string { base := t.previewBaseURL if base == "" { base = t.imageBaseURL } if t.previewExtension != "" { key = strings.TrimSuffix(key, filepath.Ext(key)) + t.previewExtension } return base + "/" + key } func (t *TelegramTranscoder) sendMessage(text string) (int, error) { payload := map[string]any{ "chat_id": strconv.Itoa(t.channelID), "text": markdownToTelegramHTML(text), "parse_mode": "HTML", "disable_web_page_preview": true, } var result struct { MessageID int `json:"message_id"` } if err := t.call("sendMessage", payload, &result); err != nil { return 0, err } return result.MessageID, nil } func (t *TelegramTranscoder) sendMediaGroup(photos []draft.Photo, caption string, sendAsType string, id int, replyTo int) ([]int, error) { media := make([]inputMediaPhoto, 0, len(photos)) files := map[string]uploadFile{} for i, p := range photos { item := inputMediaPhoto{Type: sendAsType} if sendAsType == "document" { // Upload the original file directly rather than passing Telegram a // URL to fetch: full-resolution originals routinely fail Telegram's // server-side downloader with WEBPAGE_CURL_FAILED. data, err := t.downloadPhoto(p.Key) if err != nil { return nil, fmt.Errorf("download %q for upload: %w", p.Key, err) } field := fmt.Sprintf("file%d", i) files[field] = uploadFile{name: filepath.Base(p.Key), data: data} item.Media = "attach://" + field } else { item.Media = t.previewURL(p.Key) } // Telegram shows an album's caption in the feed only when exactly one // item is captioned. So the section text goes on the first photo and // every other photo is left uncaptioned (per-photo captions dropped). if i == 0 && caption != "" { item.Caption = markdownToTelegramHTML(caption) item.ParseMode = "HTML" } media = append(media, item) } var result []struct { MessageID int `json:"message_id"` } if len(files) > 0 { mediaJSON, err := json.Marshal(media) if err != nil { return nil, err } fields := map[string]string{ "chat_id": strconv.Itoa(id), "media": string(mediaJSON), } if replyTo != 0 { replyJSON, err := json.Marshal(map[string]any{"message_id": replyTo}) if err != nil { return nil, err } fields["reply_parameters"] = string(replyJSON) } if err := t.callMultipart("sendMediaGroup", fields, files, &result); err != nil { return nil, err } } else { payload := map[string]any{ "chat_id": strconv.Itoa(id), "media": media, } if replyTo != 0 { payload["reply_parameters"] = map[string]any{ "message_id": replyTo, } } if err := t.call("sendMediaGroup", payload, &result); err != nil { return nil, err } } ids := make([]int, 0, len(result)) for _, r := range result { ids = append(ids, r.MessageID) } return ids, nil } type uploadFile struct { name string data []byte } // downloadPhoto fetches the original bytes for a photo key so they can be // uploaded directly to Telegram instead of fetched server-side by Telegram. func (t *TelegramTranscoder) downloadPhoto(key string) ([]byte, error) { url := t.photoURL(key) resp, err := t.http.Get(url) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("unexpected status %d fetching %q", resp.StatusCode, url) } return io.ReadAll(resp.Body) } func (t *TelegramTranscoder) getForwardedID(channelMessageIDs []int) (int, error) { var payloads []getUpdatesPayload var err error slog.Debug("search for forwarded message id", slog.Any("original_id", channelMessageIDs), slog.Int("group_id", t.groupID), slog.Int("channel_id", t.channelID)) for range 3 { for payloads, err = t.getUpdates(); len(payloads) > 0 && err == nil; payloads, err = t.getUpdates() { for _, payload := range payloads { if payload.Message.MessageOrigin.Type == "channel" && payload.Message.MessageOrigin.Chat.ID == t.channelID && slices.Contains(channelMessageIDs, payload.Message.MessageOrigin.MessageID) && payload.Message.Chat.ID == t.groupID { return payload.Message.MessageID, nil } } } time.Sleep(3 * time.Second) } return 0, err } type getUpdatesPayload struct { UpdateID int `json:"update_id"` Message struct { MessageID int `json:"message_id"` Chat struct { ID int `json:"id"` Title string `json:"title"` Type string `json:"type"` } `json:"chat"` MessageOrigin struct { Type string `json:"type"` Date int `json:"date"` Chat struct { ID int `json:"id"` Title string `json:"title"` Type string `json:"type"` } `json:"chat"` MessageID int `json:"message_id"` } `json:"forward_origin"` Date int `json:"date"` Text string `json:"text"` } `json:"message"` } func (t *TelegramTranscoder) getUpdates() ([]getUpdatesPayload, error) { var out []getUpdatesPayload payload := struct { Offset int `json:"offset"` }{ Offset: t.lastUpdateID, } if err := t.call("getUpdates", payload, &out); err != nil { return nil, fmt.Errorf("failed to transcode received message: %w", err) } for _, p := range out { pjson, _ := json.Marshal(p) slog.Debug("parsed new update", slog.String("payload", string(pjson))) if p.UpdateID > t.lastUpdateID { t.lastUpdateID = p.UpdateID } } t.lastUpdateID++ return out, nil } // handleResponse decodes a Telegram API response envelope. When the API asks us // to back off (HTTP 429) it returns a positive wait duration and a nil error so // the caller can retry; otherwise it unmarshals the result into out (if any) or // returns the API error verbatim. func (t *TelegramTranscoder) handleResponse(method string, resp *http.Response, attempt int, out any) (time.Duration, error) { var envelope struct { OK bool `json:"ok"` ErrorCode int `json:"error_code"` Description string `json:"description"` Parameters struct { RetryAfter int `json:"retry_after"` } `json:"parameters"` Result json.RawMessage `json:"result"` } decodeErr := json.NewDecoder(resp.Body).Decode(&envelope) resp.Body.Close() if decodeErr != nil { return 0, fmt.Errorf("decode telegram response (%s): %w", method, decodeErr) } envelopeBytes, _ := json.Marshal(envelope) slog.Debug("telegram response", slog.String("method", method), slog.String("response", string(envelopeBytes))) if envelope.OK { if out == nil { return 0, nil } return 0, json.Unmarshal(envelope.Result, out) } // 429 Too Many Requests: Telegram tells us how long to wait in // parameters.retry_after. Back off and retry instead of failing. if envelope.ErrorCode == 429 && attempt < telegramMaxAttempts { wait := time.Duration(envelope.Parameters.RetryAfter) * time.Second if wait <= 0 { wait = time.Second } slog.Warn("telegram rate limited, backing off", slog.String("method", method), slog.Int("attempt", attempt), slog.Duration("retry_after", wait)) return wait, nil } return 0, fmt.Errorf("telegram %s failed: %s", method, envelope.Description) } // call invokes a Telegram Bot API method with a JSON body and decodes the // "result" field into out. The API's error description is surfaced verbatim. func (t *TelegramTranscoder) call(method string, payload any, out any) error { body, err := json.Marshal(payload) if err != nil { return err } url := fmt.Sprintf("https://api.telegram.org/bot%s/%s", t.botToken, method) for attempt := 1; ; attempt++ { req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Content-Type", "application/json") slog.Debug("send message", slog.String("method", method), slog.String("request", string(body))) resp, err := t.http.Do(req) if err != nil { return err } wait, err := t.handleResponse(method, resp, attempt, out) if wait > 0 { time.Sleep(wait) continue } return err } } // callMultipart invokes a Telegram Bot API method using multipart/form-data, // uploading the given files directly instead of handing Telegram URLs to fetch // server-side. fields carries the non-file form values (chat_id, media JSON, // reply_parameters, ...); files maps a form field name to its bytes and upload // filename, referenced from the media JSON via attach://. func (t *TelegramTranscoder) callMultipart(method string, fields map[string]string, files map[string]uploadFile, out any) error { url := fmt.Sprintf("https://api.telegram.org/bot%s/%s", t.botToken, method) for attempt := 1; ; attempt++ { var body bytes.Buffer w := multipart.NewWriter(&body) for k, v := range fields { if err := w.WriteField(k, v); err != nil { return err } } for field, f := range files { part, err := w.CreateFormFile(field, f.name) if err != nil { return err } if _, err := part.Write(f.data); err != nil { return err } } if err := w.Close(); err != nil { return err } req, err := http.NewRequest(http.MethodPost, url, &body) if err != nil { return err } req.Header.Set("Content-Type", w.FormDataContentType()) slog.Debug("send media group upload", slog.String("method", method), slog.Int("files", len(files))) resp, err := t.http.Do(req) if err != nil { return err } wait, err := t.handleResponse(method, resp, attempt, out) if wait > 0 { time.Sleep(wait) continue } return err } }