main.go
Rama main · réplica de lectura del repositorio original.
// Perepost forwards channel posts between Telegram and MAX.
package main
import (
"cmp"
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"log/slog"
"maps"
"net"
"net/http"
"net/url"
"os"
"os/signal"
"path/filepath"
"runtime/debug"
"slices"
"strings"
"sync"
"syscall"
"time"
"github.com/go-telegram/bot"
"github.com/go-telegram/bot/models"
)
// version is set at build time (-ldflags "-X main.version=1.2.3"). A binary
// built from the module (go install …@v1.2.3) takes it from the build info.
var version = "dev"
const (
// albumWait is how long to wait for the next part of an album before sending it.
albumWait = 2 * time.Second
// shutdownGrace is how long to keep sending already received posts on shutdown.
shutdownGrace = 50 * time.Second
// setupTimeout bounds resolving every channel at startup.
setupTimeout = 5 * time.Minute
// maxPostFailures is how many crashes in a row a post gets before it is skipped.
maxPostFailures = 3
)
func main() {
if info, ok := debug.ReadBuildInfo(); ok && version == "dev" && strings.HasPrefix(info.Main.Version, "v") {
version = strings.TrimPrefix(info.Main.Version, "v")
}
showVersion := flag.Bool("version", false, "print the version and exit")
resolve := flag.Bool("resolve-delivery", false, "finish manual recovery of a partial or uncertain delivery: SOURCE POST DESTINATION none|ID [ID...]")
resolveUncertain := flag.Bool("resolve-uncertain", false, "alias for -resolve-delivery")
flag.Parse()
if *showVersion {
fmt.Println(version)
return
}
slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel()})))
start := run
if *resolve || *resolveUncertain {
start = func() error { return resolveDelivery(flag.Args()) }
}
if err := start(); err != nil {
slog.Error(err.Error())
os.Exit(1)
}
}
func run() error {
cfg, err := loadConfig()
if err != nil {
return err
}
slog.Info("perepost", "version", version)
signalContext, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
ctx, cancel := context.WithCancel(signalContext)
defer cancel()
db, err := openStore(ctx, cfg.stateFile)
if err != nil {
return fmt.Errorf("не удалось открыть STATE_FILE: %w", err)
}
defer db.close()
if err := db.unresolved(ctx); err != nil {
return err
}
var options []bot.Option
tgTransport := pacedTransport(http.DefaultTransport, time.Second)
options = append(options, bot.WithHTTPClient(time.Minute, &http.Client{Transport: tgTransport, Timeout: time.Minute}))
fileLimit := int64(tgDownloadLimit)
apiURL, maxFileLimit, maxPostLimit := "https://api.telegram.org", int64(50<<20), int64(200<<20)
if cfg.telegramAPI != "" {
options = append(options, bot.WithServerURL(cfg.telegramAPI))
apiURL = cfg.telegramAPI
fileLimit = 0
maxFileLimit, maxPostLimit = 2_000_000_000, 2_000_000_000
}
tg, err := bot.New(cfg.telegramToken, options...)
if err != nil {
return fmt.Errorf("не удалось подключиться к Telegram: %w", err)
}
maxc, err := newMaxClient(cfg.maxToken)
if err != nil {
return fmt.Errorf("не удалось подключиться к MAX: %w", err)
}
setup, finishSetup := context.WithTimeout(ctx, setupTimeout)
defer finishSetup()
channels, err := connect(setup, tg, maxc, cfg.routes)
if err != nil {
return err
}
list := slices.SortedFunc(maps.Values(channels.telegram), func(a, b *source) int { return cmp.Compare(a.channel.Title, b.channel.Title) })
maxList := slices.SortedFunc(maps.Values(channels.max), func(a, b *maxSource) int { return cmp.Compare(a.channel.title, b.channel.title) })
tgEvents, maxEvents := webhookEvents(cfg.routes)
if len(list) > 0 && cfg.tgWebhookURL == "" {
if err := configureTelegramWebhook(setup, tg, "", "", nil, false); err != nil {
return err
}
}
r := &relay{
tg: tg, max: maxc, http: &http.Client{Timeout: 5 * time.Minute}, adminID: cfg.adminID, fileLimit: fileLimit,
tgAPI: telegramAPI{baseURL: apiURL, token: cfg.telegramToken, http: &http.Client{Transport: tgTransport, Timeout: 15 * time.Minute, CheckRedirect: noRedirect}},
maxFileLimit: maxFileLimit, maxPostLimit: maxPostLimit, mediaSlot: make(chan struct{}, 1),
}
var server *http.Server
var listener net.Listener
handlers := map[string]http.Handler{}
if len(maxList) > 0 {
r.mediaDir = cfg.stateFile + ".media"
if err := prepareMediaDir(r.mediaDir); err != nil {
return err
}
r.maxMediaHTTP, err = maxMediaClient()
if err != nil {
return err
}
u, _ := url.Parse(cfg.maxWebhookURL) // validated by loadConfig
handlers[cmp.Or(u.EscapedPath(), "/")] = maxWebhook(cfg.maxSecret, db, channels.max)
}
if len(list) > 0 && cfg.tgWebhookURL != "" {
u, _ := url.Parse(cfg.tgWebhookURL)
handlers[u.EscapedPath()] = telegramWebhook(cfg.tgSecret, db, channels.telegram)
}
if len(handlers) > 0 {
server = &http.Server{
Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
handler := handlers[req.URL.EscapedPath()]
if handler == nil {
http.NotFound(w, req)
return
}
handler.ServeHTTP(w, req)
}),
ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second,
WriteTimeout: 20 * time.Second, IdleTimeout: 30 * time.Second,
}
listener, err = (&net.ListenConfig{}).Listen(setup, "tcp", cfg.maxListen)
if err != nil {
return fmt.Errorf("не удалось открыть WEBHOOK_LISTEN: %w", err)
}
defer func() { _ = server.Close() }()
}
var failure error
var failOnce sync.Once
fail := func(err error) {
if err != nil {
failOnce.Do(func() {
failure = err
cancel()
})
}
}
var inputs sync.WaitGroup
if server != nil {
inputs.Go(func() {
if err := server.Serve(listener); !errors.Is(err, http.ErrServerClosed) {
fail(err)
}
})
if len(maxList) > 0 {
if err := maxc.webhook(setup, cfg.maxWebhookURL, cfg.maxSecret, maxEvents, true); err != nil {
return err
}
}
if len(list) > 0 && cfg.tgWebhookURL != "" {
if err := configureTelegramWebhook(setup, tg, cfg.tgWebhookURL, cfg.tgSecret, tgEvents, true); err != nil {
return err
}
}
}
finishSetup()
// Sending outlives the stop signal: posts already received get delivered
sending, abort := context.WithCancel(context.Background())
defer abort()
intakeDone := make(chan struct{})
var workers sync.WaitGroup
for _, src := range list {
workers.Go(func() {
key := channelKey{telegramPlatform, src.channel.ID}
fail(process(sending, db, key, intakeDone, r.alert, func(p pendingPost, book *deliveryBook) {
r.forwardTelegramPost(sending, src, p, book)
}))
})
}
for _, src := range maxList {
workers.Go(func() {
key := channelKey{maxPlatform, src.channel.id}
fail(process(sending, db, key, intakeDone, r.alert, func(p pendingPost, book *deliveryBook) {
r.forwardMaxPost(sending, src, p, book)
}))
})
}
summary := describe(list, maxList)
slog.Info("пересылаю посты", "routes", summary)
if err := r.alert(ctx, fmt.Sprintf("запущен, версия %s:\n%s", version, summary)); err != nil {
slog.Warn("написать на TELEGRAM_ADMIN_ID не получилось: откройте бота в Telegram и нажмите «Старт»")
}
health := &monitor{
relay: r, sources: list, to: destinations(list), maxSources: maxList, heartbeat: cfg.heartbeat,
client: &http.Client{Timeout: 30 * time.Second}, db: db, maxWebhookURL: cfg.maxWebhookURL,
tgWebhookURL: cfg.tgWebhookURL, tgEvents: tgEvents, maxEvents: maxEvents,
}
health.lastPoll.Store(time.Now().UnixNano())
var monitoring sync.WaitGroup
monitoring.Go(func() { health.run(ctx) })
if len(list) > 0 && cfg.tgWebhookURL == "" {
inputs.Go(func() {
if err := r.tgAPI.poll(ctx, db, channels.telegram, tgEvents, &health.lastPoll); ctx.Err() == nil {
fail(err)
}
})
}
<-ctx.Done()
stop() // A second interrupt must stop the process even while it drains.
slog.Info("останавливаюсь, досылаю полученные посты")
done := make(chan struct{})
go func() {
if server != nil {
shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := server.Shutdown(shutdown); err != nil {
_ = server.Close()
}
}
inputs.Wait()
close(intakeDone)
workers.Wait()
monitoring.Wait()
close(done)
}()
select {
case <-done:
case <-time.After(shutdownGrace):
abort()
<-done
slog.Info("оставшиеся посты сохранены в очереди")
}
return failure
}
func (r *relay) forwardTelegramPost(ctx context.Context, src *source, p pendingPost, book *deliveryBook) {
if strings.HasPrefix(p.key, "change:") {
r.editFromTelegram(ctx, src, p, book)
return
}
if strings.HasPrefix(p.key, "album:") && len(p.payloads) > 0 {
if book.err = book.store.keepAlbum(ctx, book.source, p.key, p.payloads); book.err != nil {
return
}
// The first attempt freezes the album. A resumed delivery must use the
// same parts even if more updates arrived before the next process started.
p.payloads, book.err = book.store.albumPayloads(ctx, book.source, p.key)
if book.err != nil {
return
}
}
messages, err := telegramMessages(p.payloads)
if err != nil {
slog.Warn("пост Telegram разобран не полностью", "channel", book.source, "post", p.key, "err", err)
action := "переносим доступную часть"
if len(messages) == 0 {
action = "пост пропущен"
}
_ = r.alert(ctx, fmt.Sprintf("пост %s из %s разобран не полностью, %s: %v", p.key, book.source, action, err))
}
if len(messages) > 0 {
r.forward(ctx, src, &post{messages}, book)
}
}
func (r *relay) forwardMaxPost(ctx context.Context, src *maxSource, p pendingPost, book *deliveryBook) {
if strings.HasPrefix(p.key, "change:") {
r.changeFromMax(ctx, src, p, book)
return
}
if len(p.payloads) == 0 {
return
}
var m maxMessage
if err := json.Unmarshal(p.payloads[0], &m); err != nil {
slog.Warn("пост MAX разобран не полностью", "channel", book.source, "post", p.key, "err", err)
_, partial := errors.AsType[*json.UnmarshalTypeError](err)
action := "переносим доступные поля"
if !partial {
action = "пост пропущен"
}
_ = r.alert(ctx, fmt.Sprintf("пост %s из %s разобран не полностью, %s: %v", p.key, book.source, action, err))
if !partial {
return
}
}
r.forwardMax(ctx, src, m, book)
}
// process delivers the posts of one source in order. A crash while handling a
// post stops the service, so the restart turns its in-flight publications into
// uncertain ones; after maxPostFailures crashes in a row the post is skipped.
func process(ctx context.Context, db *store, source channelKey, intakeDone <-chan struct{}, alert func(context.Context, string) error, forward func(pendingPost, *deliveryBook)) (err error) {
defer func() {
if ctx.Err() != nil && errors.Is(err, ctx.Err()) {
err = nil
}
}()
tick := time.NewTicker(250 * time.Millisecond)
defer tick.Stop()
for ctx.Err() == nil {
now := time.Now()
draining := false
select {
case <-intakeDone:
draining = true
default:
}
post, err := db.next(ctx, source, now)
if err != nil {
return err
}
if post.row != 0 {
if source.platform == telegramPlatform && strings.HasPrefix(post.key, "album:") {
moved, err := db.splitLateAlbum(ctx, source, post)
if err != nil {
return err
}
if moved {
continue
}
}
failures, err := db.failures(ctx, source, post.key)
if err != nil {
return err
}
if failures >= maxPostFailures {
slog.Error("пост пропущен после повторных сбоев", "channel", source, "post", post.key, "failures", failures)
_ = alert(ctx, fmt.Sprintf("пост %s из %s пропущен: обработка падала %d раза подряд", post.key, source, failures))
if err := db.complete(ctx, source, post); err != nil {
return err
}
continue
}
book := &deliveryBook{store: db, source: source, post: post.key}
if crash := handle(forward, post, book); crash != nil {
return crashed(ctx, db, source, post.key, crash, alert)
}
if book.err != nil {
return book.err
}
if ctx.Err() != nil {
break
}
if err := db.complete(ctx, source, post); err != nil {
return err
}
continue
}
if draining {
return nil
}
select {
case <-ctx.Done():
case <-tick.C:
case <-intakeDone:
}
}
return nil
}
// crash is a panic caught while handling a post, with the stack it came from.
type crash struct {
value any
stack []byte
}
// handle runs forward and turns a panic into a crash: every handler, in both
// directions, gets the same treatment.
func handle(forward func(pendingPost, *deliveryBook), p pendingPost, book *deliveryBook) (c *crash) {
defer func() {
if v := recover(); v != nil {
c = &crash{v, debug.Stack()}
}
}()
forward(p, book)
return nil
}
// crashed counts the crash, tells the admin and returns the error that stops the service.
func crashed(ctx context.Context, db *store, source channelKey, post string, c *crash, alert func(context.Context, string) error) error {
slog.Error("сбой при обработке поста", "channel", source, "post", post, "panic", c.value, "stack", string(c.stack))
// Shutdown may already be cancelling ctx; the count must still be saved.
saving, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel()
count, err := db.recordFailure(saving, source, post)
if err != nil {
slog.Error("не удалось записать сбой поста", "channel", source, "post", post, "err", err)
}
_ = alert(ctx, fmt.Sprintf("сбой при обработке поста %s из %s (%d из %d, затем пост будет пропущен): %v",
post, source, count, maxPostFailures, c.value))
return fmt.Errorf("обработка поста %s из %s прервана: %v", post, source, c.value)
}
func resolveDelivery(args []string) error {
if len(args) < 4 || (args[3] == "none" && len(args) != 4) {
return errors.New("укажите SOURCE POST DESTINATION и none или id опубликованных сообщений")
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
db, err := openStore(ctx, cmp.Or(os.Getenv("STATE_FILE"), "perepost.sqlite"))
if err != nil {
return err
}
defer db.close()
ids := args[3:]
if ids[0] == "none" {
ids = nil
}
return db.resolveDelivery(ctx, args[0], args[1], args[2], ids)
}
func prepareMediaDir(path string) error {
if err := os.MkdirAll(path, 0o700); err != nil {
return err
}
entries, err := os.ReadDir(path)
if err != nil {
return err
}
for _, entry := range entries {
if !entry.IsDir() && strings.HasPrefix(entry.Name(), "download-") {
if err := os.Remove(filepath.Join(path, entry.Name())); err != nil {
return err
}
}
}
return nil
}
type sources struct {
telegram map[int64]*source
max map[int64]*maxSource
}
// connect resolves every channel of the routes and checks the bot's rights in
// each. A channel met in several routes is resolved once, and a source gets
// the destinations of every route it appears in.
func connect(ctx context.Context, tg *bot.Bot, maxc *maxClient, routes []route) (sources, error) {
telegram := map[string]*models.ChatFullInfo{}
maximum := map[string]maxChat{}
getTelegram := func(ref string) (*models.ChatFullInfo, error) {
if chat := telegram[ref]; chat != nil {
return chat, nil
}
chat, err := useChannel(ctx, tg, ref)
telegram[ref] = chat
return chat, err
}
getMax := func(ref string) (maxChat, error) {
if chat, ok := maximum[ref]; ok {
return chat, nil
}
chat, err := maxc.resolve(ctx, ref)
maximum[ref] = chat
return chat, err
}
result := sources{telegram: map[int64]*source{}, max: map[int64]*maxSource{}}
canPost := map[int64]bool{}
canPostMax := map[int64]bool{}
for _, rt := range routes {
if rt.reverse {
var to []*models.ChatFullInfo
for _, ref := range rt.To {
chat, err := getTelegram(ref)
if err != nil {
return sources{}, err
}
if !canPost[chat.ID] {
if err := checkTelegramPost(ctx, tg, chat); err != nil {
return sources{}, err
}
canPost[chat.ID] = true
}
to = append(to, chat)
}
for _, ref := range rt.From {
chat, err := getMax(ref)
if err != nil {
return sources{}, err
}
src := result.max[chat.id]
if src == nil {
if err := maxc.checkRead(ctx, chat); err != nil {
return sources{}, err
}
src = &maxSource{channel: chat, changes: map[int64]syncOptions{}}
result.max[chat.id] = src
}
for _, dst := range to {
addSyncOptions(src.changes, dst.ID, rt)
if !slices.ContainsFunc(src.to, func(c *models.ChatFullInfo) bool { return c.ID == dst.ID }) {
src.to = append(src.to, dst)
}
}
}
continue
}
var to []maxChat
for _, ref := range rt.To {
chat, err := getMax(ref)
if err != nil {
return sources{}, err
}
if !canPostMax[chat.id] {
if err := maxc.checkAdmin(ctx, chat); err != nil {
return sources{}, err
}
canPostMax[chat.id] = true
}
to = append(to, chat)
}
for _, ref := range rt.From {
channel, err := getTelegram(ref)
if err != nil {
return sources{}, err
}
src := result.telegram[channel.ID]
if src == nil {
src = &source{channel: channel, changes: map[int64]syncOptions{}}
result.telegram[channel.ID] = src
}
for _, chat := range to {
addSyncOptions(src.changes, chat.id, rt)
if !slices.ContainsFunc(src.to, func(c maxChat) bool { return c.id == chat.id }) {
src.to = append(src.to, chat)
}
}
}
}
return result, nil
}
func webhookEvents(routes []route) (telegram, max []string) {
telegram = []string{"channel_post"}
max = []string{maxEvent}
if slices.ContainsFunc(routes, func(r route) bool { return !r.reverse && r.syncEdits }) {
telegram = append(telegram, telegramEdit)
}
if slices.ContainsFunc(routes, func(r route) bool { return r.reverse && r.syncEdits }) {
max = append(max, maxEdit)
}
if slices.ContainsFunc(routes, func(r route) bool { return r.reverse && r.syncDeletes }) {
max = append(max, maxDelete)
}
return telegram, max
}
// destinations lists every MAX channel once.
func destinations(sources []*source) []maxChat {
var all []maxChat
for _, src := range sources {
for _, chat := range src.to {
if !slices.ContainsFunc(all, func(c maxChat) bool { return c.id == chat.id }) {
all = append(all, chat)
}
}
}
return all
}
// describe renders the routes for the log and the startup message.
func describe(sources []*source, maxSources []*maxSource) string {
lines := make([]string, 0, len(sources)+len(maxSources))
for _, src := range sources {
names := make([]string, 0, len(src.to))
for _, chat := range src.to {
names = append(names, "«"+chat.title+"»")
}
lines = append(lines, fmt.Sprintf("Telegram «%s» → MAX %s", src.channel.Title, strings.Join(names, ", ")))
}
for _, src := range maxSources {
names := make([]string, 0, len(src.to))
for _, dst := range src.to {
names = append(names, "«"+dst.Title+"»")
}
lines = append(lines, fmt.Sprintf("MAX «%s» → Telegram %s", src.channel.title, strings.Join(names, ", ")))
}
return strings.Join(lines, "\n")
}
func logLevel() slog.Level {
if strings.EqualFold(os.Getenv("LOG_LEVEL"), "debug") {
return slog.LevelDebug
}
return slog.LevelInfo
}