path: root/internal/transcoder/telegram.go

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://<field>.
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
	}
}