feat: rewrite the forwarder in Go with gotd

Replace the Telethon script with a Go port built on gotd/td. It keeps the
PHONE, API_ID, API_HASH, SOURCE_GROUP_ID and TARGET_GROUP_ID contract,
stores the session in a local session.json, queues forwards outside the
updates loop, and catches up on updates missed across reconnects. The
Telethon session file is not reused, so the first run logs in once.
This commit is contained in:
tiennm99 committed 2026-10-15 16:53:05 +07:00
1 parent 1d723da967
commit eba7f835a9
21 files changed
+1158 -226

No files matched your search

+2 -2
View File
@@ -1,5 +1,5 @@
PHONE=+1234567890
API_ID=123456789
API_HASH=0123456789abcdef0123456789abcdef
SOURCE_GROUP_ID=-123456789
TARGET_GROUP_ID=-123456789
SOURCE_GROUP_ID=-1001234567890
TARGET_GROUP_ID=-1009876543210
+21 -163
View File
@@ -1,164 +1,22 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
.idea/
# project
session.json
*.session
*.session-journal
*.session-journal
plans/
# go
/bin/
*.exe
*.test
*.out
coverage.*
go.work
go.work.sum
# env
.env
# tooling
.idea/
.vscode/
.claude/
+74 -19
View File
@@ -1,34 +1,89 @@
# forward-telegram-message
Mirror Telegram messages between two chats using Telethon — runs as a user-account, not a bot.
Forward every new message from one Telegram chat to another with Go and
[gotd/td](https://github.com/gotd/td). It runs as a **user account**, not a
bot, so it can read any chat the account is in. Text and media (photos,
documents, videos) are forwarded with Telegram's "Forwarded from" header.
## Quick start
## How to use
1. Clone this repository and install [Go](https://go.dev/dl/):
```bash
cp .env.example .env
# Fill in PHONE, API_ID, API_HASH, SOURCE_GROUP_ID, TARGET_GROUP_ID
pip install -r requirements.txt
python main.py
git clone https://github.com/tiennm99/forward-telegram-message.git
```
Get `API_ID` and `API_HASH` from [my.telegram.org](https://my.telegram.org).
2. Create a new Telegram app at [https://my.telegram.org](https://my.telegram.org) and get the `api_id` and `api_hash`.
3. Copy `.env.example` to `.env` and fill in the values.
4. Run the forwarder:
```bash
go run ./cmd/forward
```
The first run asks for the Telegram login code (and the two-step
verification password, if the account has one), then stores the session in
`session.json`. Later runs reuse it and **do not** prompt again. If the login
fails (no answer on stdin, a wrong code or password, or Telegram rejecting the
phone), the forwarder exits instead of retrying, so it does not keep
requesting new codes. Run the first login in an interactive terminal.
The forwarder runs until interrupted (Ctrl+C). When the connection drops or
Telegram returns an error, it reconnects after 5 seconds. A message that fails
to forward is reported and skipped.
Console output keeps the Python version's wording (`Starting...`,
`Bot is running.`, `Exception on forward_message: ...`,
`Exception on main: ... Restarting...`) on stdout. Errors that stop the
forwarder (bad config, failed login, unknown chat) go to stderr with exit
status 1.
## Configuration
| Variable | Required | Description |
|----------|----------|-------------|
| `PHONE` | Yes | Your Telegram phone number (e.g. `+84912345678`) |
| `API_ID` | Yes | App API ID from my.telegram.org |
| `API_HASH` | Yes | App API hash from my.telegram.org |
| `SOURCE_GROUP_ID` | Yes | ID of the source chat/group to mirror from |
| `TARGET_GROUP_ID` | Yes | ID of the destination chat/group to forward into |
Set these in `.env` or in the environment:
## Notes
| Variable | Description |
|----------|-------------|
| `PHONE` | Your Telegram phone number in international format (e.g. `+84912345678`) |
| `API_ID` | App API ID from my.telegram.org |
| `API_HASH` | App API hash from my.telegram.org |
| `SOURCE_GROUP_ID` | Chat to forward from |
| `TARGET_GROUP_ID` | Chat to forward into |
- Runs as a **user account** (Telethon), not a bot — requires first-time interactive login to generate a session file.
- Forwards both text messages and media (photos, documents, videos).
- Re-uses the session file on subsequent runs; no re-auth needed.
- `SOURCE_GROUP_ID` and `TARGET_GROUP_ID` can be negative (supergroup IDs) or positive (private chat IDs).
Chat IDs use the usual marked form: `-100<id>` for supergroups and channels,
`-<id>` for basic groups, and the other person's user ID (a positive number)
for private chats. The account must already be in both chats: the forwarder
looks them up in its chat list at startup and stops if either is missing.
Messages sent while the forwarder is running are forwarded in order,
including ones the account itself sends to the source chat. Messages sent
during a reconnect are caught up once the connection is back; messages sent
while the process is not running are not. Service messages such as joins and
pins are skipped.
## Session file
`session.json` grants full access to the Telegram account. It is written with
`0600` permissions and ignored by Git; delete it to force a clean re-login.
Sessions from the earlier Python version (`*.session`) are not reused, so the
first Go run asks for a login code once.
## Build a binary
The code is pure Go, so a static binary builds without cgo:
```bash
CGO_ENABLED=0 go build -o bin/ ./cmd/forward
```
## Development
```bash
go vet ./...
go test ./...
```
Tests run offline; Telegram calls are not exercised.
## License
+21
View File
@@ -0,0 +1,21 @@
package main
import (
"context"
"fmt"
"os"
"os/signal"
"syscall"
"github.com/tiennm99/forward-telegram-message/internal/commands/forward"
)
func main() {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
if err := forward.Run(ctx, os.Stdin, os.Stdout); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}
-11
View File
@@ -1,11 +0,0 @@
import os
from dotenv import load_dotenv
load_dotenv()
phone = os.getenv('PHONE')
api_id = os.getenv('API_ID')
api_hash = os.getenv('API_HASH')
source_group_id = int(os.getenv('SOURCE_GROUP_ID'))
target_group_id = int(os.getenv('TARGET_GROUP_ID'))
+51
View File
@@ -0,0 +1,51 @@
module github.com/tiennm99/forward-telegram-message
go 1.27.1
require (
github.com/gotd/td v0.162.0
github.com/joho/godotenv v1.5.1
golang.org/x/term v0.47.0
)
require (
github.com/andybalholm/brotli v1.2.1 // indirect
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/coder/websocket v1.8.15 // indirect
github.com/dlclark/regexp2 v1.12.0 // indirect
github.com/fatih/color v1.19.0 // indirect
github.com/ghodss/yaml v1.0.0 // indirect
github.com/go-faster/errors v0.8.0 // indirect
github.com/go-faster/jx v1.2.0 // indirect
github.com/go-faster/xor v1.0.0 // indirect
github.com/go-faster/yaml v0.4.6 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gotd/ige v0.3.0 // indirect
github.com/gotd/log v0.1.0 // indirect
github.com/gotd/neo v0.1.5 // indirect
github.com/klauspost/compress v1.19.1 // indirect
github.com/mattn/go-colorable v0.1.15 // indirect
github.com/mattn/go-isatty v0.0.24 // indirect
github.com/ogen-go/ogen v1.23.0 // indirect
github.com/refraction-networking/utls v1.8.2 // indirect
github.com/segmentio/asm v1.2.1 // indirect
github.com/shopspring/decimal v1.4.0 // indirect
github.com/yuin/goldmark v1.8.5 // indirect
go.opentelemetry.io/otel v1.44.0 // indirect
go.opentelemetry.io/otel/metric v1.44.0 // indirect
go.opentelemetry.io/otel/trace v1.44.0 // indirect
go.uber.org/atomic v1.12.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.28.0 // indirect
golang.org/x/crypto v0.54.0 // indirect
golang.org/x/exp v0.0.0-20230725093048-515e97ebf090 // indirect
golang.org/x/mod v0.38.0 // indirect
golang.org/x/net v0.57.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.49.0 // indirect
golang.org/x/text v0.40.0 // indirect
golang.org/x/tools v0.48.0 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
rsc.io/qr v0.2.0 // indirect
)
+114
View File
@@ -0,0 +1,114 @@
github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dlclark/regexp2 v1.12.0 h1:0j4c5qQmnC6XOWNjP3PIXURXN2gWx76rd3KvgdPkCz8=
github.com/dlclark/regexp2 v1.12.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8=
github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w=
github.com/fatih/color v1.19.0/go.mod h1:zNk67I0ZUT1bEGsSGyCZYZNrHuTkJJB+r6Q9VuMi0LE=
github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
github.com/go-faster/errors v0.8.0 h1:9T9eJrM+72dFk7n4DfhuaDDe6cyuFCSW2oNUkN77Yqc=
github.com/go-faster/errors v0.8.0/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo=
github.com/go-faster/jx v1.2.0 h1:T2YHJPrFaYu21fJtUxC9GzmluKu8rVIFDwwGBKTDseI=
github.com/go-faster/jx v1.2.0/go.mod h1:UWLOVDmMG597a5tBFPLIWJdUxz5/2emOpfsj9Neg0PE=
github.com/go-faster/xor v1.0.0 h1:2o8vTOgErSGHP3/7XwA5ib1FTtUsNtwCoLLBjl31X38=
github.com/go-faster/xor v1.0.0/go.mod h1:x5CaDY9UKErKzqfRfFZdfu+OSTfoZny3w5Ak7UxcipQ=
github.com/go-faster/yaml v0.4.6 h1:lOK/EhI04gCpPgPhgt0bChS6bvw7G3WwI8xxVe0sw9I=
github.com/go-faster/yaml v0.4.6/go.mod h1:390dRIvV4zbnO7qC9FGo6YYutc+wyyUSHBgbXL52eXk=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gotd/ige v0.3.0 h1:4f6LEHWsVDLBG0bT9wWG2/9TZb5aWm265G8ZlTXmRRU=
github.com/gotd/ige v0.3.0/go.mod h1:FE9bTaQtvfArizAcZuI4sS6gXaEUBmixdUufVHoCKac=
github.com/gotd/log v0.1.0 h1:4LJUEvafD1xtBwx2QkrlzFnRgbYXTlWqJPDi8BvrLbU=
github.com/gotd/log v0.1.0/go.mod h1:5ilhdu1Ux0QvDY/FF3Ojfw24Ws3SlCtyLwOpXy8KYXs=
github.com/gotd/log/logzap v0.1.1 h1:O6l7d8HUbODe+UMcrM47eXYDwdJ6RNmpQejLjrlcEIQ=
github.com/gotd/log/logzap v0.1.1/go.mod h1:5ObZkITbfhbsBOLzBkzmMk9QxXc0eNQpimau7zRL+Y8=
github.com/gotd/neo v0.1.5 h1:oj0iQfMbGClP8xI59x7fE/uHoTJD7NZH9oV1WNuPukQ=
github.com/gotd/neo v0.1.5/go.mod h1:9A2a4bn9zL6FADufBdt7tZt+WMhvZoc5gWXihOPoiBQ=
github.com/gotd/td v0.162.0 h1:3pUzBsEbb0I5QquvfSkXihkVBXI+QTtdd/upijjUbkw=
github.com/gotd/td v0.162.0/go.mod h1:ZsGbErIos7XHF7HL0VlsjL6mmX4b1JpinHfDb7gOdvs=
github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy0/jY=
github.com/mattn/go-colorable v0.1.15/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI=
github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
github.com/ogen-go/ogen v1.23.0 h1:QaWeKm2KZ2zy7NkqqO1Vdl5idNqlG+svxdgwVAX+zbo=
github.com/ogen-go/ogen v1.23.0/go.mod h1:bwwvC3AmCV+LrL5lazyQwwof90402mdcSyI0FOzzpfM=
github.com/refraction-networking/utls v1.8.2 h1:j4Q1gJj0xngdeH+Ox/qND11aEfhpgoEvV+S9iJ2IdQo=
github.com/refraction-networking/utls v1.8.2/go.mod h1:jkSOEkLqn+S/jtpEHPOsVv/4V4EVnelwbMQl4vCWXAM=
github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0=
github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
github.com/yuin/goldmark v1.8.5 h1:r6N5afV5qj/5S4UTch8agZHJ8UxNCMwX7WjkkJam2NA=
github.com/yuin/goldmark v1.8.5/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg=
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
go.uber.org/atomic v1.12.0 h1:BvcXdFKuviU4fTL/f+SxdQ5qJX/Jix8pAkgdUcb3XOE=
go.uber.org/atomic v1.12.0/go.mod h1:I6c4cg+6HCxRjfjSsYtApoFILnpc0CGUdGkXVqbYVNk=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo=
go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q=
go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
golang.org/x/exp v0.0.0-20230725093048-515e97ebf090 h1:Di6/M8l0O2lCLc6VVRWhgCiApHV8MnQurBnFSHsQtNY=
golang.org/x/exp v0.0.0-20230725093048-515e97ebf090/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc=
golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk=
golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40=
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA=
golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/term v0.47.0 h1:cpyOwQkr2Nzxe2p6KnrTG7Ei2wtzjsuSjhpXXepZ/BM=
golang.org/x/term v0.47.0/go.mod h1:efAfAuG6RKABvg1ArGebRkvX6LXul14SM67o2e4mUCY=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE=
golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y=
nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=
rsc.io/qr v0.2.0 h1:6vBLea5/NRMVTz8V66gipeLycZMl/+UlFmk8DvqQ6WY=
rsc.io/qr v0.2.0/go.mod h1:IF+uZjkb9fqyeF/4tlBoynqmQxUoPfWEKh921coOuXs=
+55
View File
@@ -0,0 +1,55 @@
// Package forward runs the forwarder and reconnects when it fails.
package forward
import (
"context"
"errors"
"fmt"
"io"
"time"
"github.com/gotd/td/session"
"github.com/tiennm99/forward-telegram-message/internal/config"
"github.com/tiennm99/forward-telegram-message/internal/prompt"
"github.com/tiennm99/forward-telegram-message/internal/telegram"
)
// sessionFile holds the Telegram login session. It grants full account
// access, so it is written with 0600 permissions and ignored by Git.
const sessionFile = "session.json"
// restartDelay is how long to wait before reconnecting after a failure.
const restartDelay = 5 * time.Second
// Run forwards messages until ctx ends. Connection failures restart the
// forwarder; a bad config, a failed login, or an unknown chat stops it.
func Run(ctx context.Context, in io.Reader, out io.Writer) error {
cfg, err := config.FromEnv()
if err != nil {
return err
}
storage := &session.FileStorage{Path: sessionFile}
ask := prompt.New(in, out)
listener := telegram.NewListener(cfg, out)
for {
fmt.Fprintln(out, "Starting...")
err := listener.Listen(ctx, storage, ask)
if ctx.Err() != nil {
return nil
}
switch {
case errors.Is(err, telegram.ErrChatNotFound):
return fmt.Errorf("%w; the account must be a member of both chats", err)
case errors.Is(err, telegram.ErrLogin):
return err
}
fmt.Fprintf(out, "Exception on main: %v. Restarting...\n", err)
select {
case <-ctx.Done():
return nil
case <-time.After(restartDelay):
}
}
}
+23
View File
@@ -0,0 +1,23 @@
package forward
import (
"bytes"
"context"
"strings"
"testing"
)
func TestRunRejectsMissingConfigBeforeTelegram(t *testing.T) {
for _, name := range []string{"PHONE", "API_ID", "API_HASH", "SOURCE_GROUP_ID", "TARGET_GROUP_ID"} {
t.Setenv(name, "")
}
var out bytes.Buffer
err := Run(context.Background(), strings.NewReader(""), &out)
if err == nil || !strings.Contains(err.Error(), "missing PHONE") {
t.Fatalf("Run error = %v", err)
}
if out.Len() != 0 {
t.Fatalf("Run wrote %q before failing", out.String())
}
}
+76
View File
@@ -0,0 +1,76 @@
// Package config reads the forwarder settings from the environment.
package config
import (
"errors"
"fmt"
"os"
"strconv"
"strings"
"github.com/gotd/td/constant"
"github.com/joho/godotenv"
)
// Config is what the forwarder needs to log in and route messages.
type Config struct {
Phone string
APIID int
APIHash string
SourceID int64
TargetID int64
}
// FromEnv loads .env when present, then reads the process environment.
func FromEnv() (Config, error) {
_ = godotenv.Load()
return Parse(os.Getenv)
}
// Parse reads every setting through getenv and reports all missing ones at
// once.
func Parse(getenv func(string) string) (Config, error) {
value := func(name string) string { return strings.TrimSpace(getenv(name)) }
var missing []string
for _, name := range []string{"PHONE", "API_ID", "API_HASH", "SOURCE_GROUP_ID", "TARGET_GROUP_ID"} {
if value(name) == "" {
missing = append(missing, name)
}
}
if len(missing) > 0 {
return Config{}, fmt.Errorf("missing %s in environment or .env", strings.Join(missing, ", "))
}
cfg := Config{Phone: value("PHONE"), APIHash: value("API_HASH")}
var err error
if cfg.APIID, err = strconv.Atoi(value("API_ID")); err != nil || cfg.APIID <= 0 {
return Config{}, fmt.Errorf("invalid API_ID %q: must be a positive integer", value("API_ID"))
}
if cfg.SourceID, err = parseChatID("SOURCE_GROUP_ID", value("SOURCE_GROUP_ID")); err != nil {
return Config{}, err
}
if cfg.TargetID, err = parseChatID("TARGET_GROUP_ID", value("TARGET_GROUP_ID")); err != nil {
return Config{}, err
}
if cfg.SourceID == cfg.TargetID {
return Config{}, errors.New("SOURCE_GROUP_ID and TARGET_GROUP_ID must differ")
}
return cfg, nil
}
func parseChatID(name, raw string) (int64, error) {
id, err := strconv.ParseInt(raw, 10, 64)
if err != nil || !validChatID(id) {
return 0, fmt.Errorf("invalid %s %q: use a marked chat ID such as -1001234567890", name, raw)
}
return id, nil
}
// validChatID reports whether id is a marked chat ID: the other person's user
// ID for private chats, -<id> for basic groups and -100<id> for supergroups
// and channels.
func validChatID(id int64) bool {
marked := constant.TDLibPeerID(id)
return marked.IsUser() || marked.IsChat() || marked.IsChannel()
}
+76
View File
@@ -0,0 +1,76 @@
package config
import (
"strings"
"testing"
)
func env(values map[string]string) func(string) string {
return func(name string) string { return values[name] }
}
func validEnv() map[string]string {
return map[string]string{
"PHONE": "+84912345678",
"API_ID": "123456",
"API_HASH": "0123456789abcdef",
"SOURCE_GROUP_ID": "-1001234567890",
"TARGET_GROUP_ID": " -42 ",
}
}
func TestParse(t *testing.T) {
got, err := Parse(env(validEnv()))
if err != nil {
t.Fatalf("Parse: %v", err)
}
want := Config{
Phone: "+84912345678",
APIID: 123456,
APIHash: "0123456789abcdef",
SourceID: -1001234567890,
TargetID: -42,
}
if got != want {
t.Fatalf("Parse = %+v, want %+v", got, want)
}
}
func TestParseReportsAllMissing(t *testing.T) {
_, err := Parse(env(map[string]string{"PHONE": "+1"}))
if err == nil || !strings.Contains(err.Error(), "API_ID, API_HASH, SOURCE_GROUP_ID, TARGET_GROUP_ID") {
t.Fatalf("Parse error = %v", err)
}
}
func TestParseRejectsBadValues(t *testing.T) {
tests := []struct {
name, value, want string
}{
{"API_ID", "abc", "invalid API_ID"},
{"API_ID", "-5", "invalid API_ID"},
{"SOURCE_GROUP_ID", "group", "invalid SOURCE_GROUP_ID"},
{"SOURCE_GROUP_ID", "-1000000000000", "invalid SOURCE_GROUP_ID"},
{"TARGET_GROUP_ID", "-1001234567890", "must differ"},
}
for _, tt := range tests {
values := validEnv()
values[tt.name] = tt.value
if _, err := Parse(env(values)); err == nil || !strings.Contains(err.Error(), tt.want) {
t.Errorf("%s=%q: error = %v, want %q", tt.name, tt.value, err, tt.want)
}
}
}
func TestValidChatID(t *testing.T) {
for _, id := range []int64{777000, -42, -999999999999, -1000000000001, -1001234567890} {
if !validChatID(id) {
t.Errorf("validChatID(%d) = false", id)
}
}
for _, id := range []int64{0, -1000000000000} {
if validChatID(id) {
t.Errorf("validChatID(%d) = true", id)
}
}
}
+55
View File
@@ -0,0 +1,55 @@
// Package prompt reads answers from the terminal.
package prompt
import (
"bufio"
"errors"
"fmt"
"io"
"os"
"strings"
"golang.org/x/term"
)
// Prompter asks questions on out and reads answers from in.
type Prompter struct {
in *bufio.Reader
fd int
tty bool
out io.Writer
}
// New returns a Prompter. Secrets are read without echo when in is a
// terminal.
func New(in io.Reader, out io.Writer) *Prompter {
p := &Prompter{in: bufio.NewReader(in), out: out}
if f, ok := in.(*os.File); ok && term.IsTerminal(int(f.Fd())) {
p.fd, p.tty = int(f.Fd()), true
}
return p
}
// Line asks for one line of text and trims it.
func (p *Prompter) Line(label string) (string, error) {
fmt.Fprintf(p.out, "%s: ", label)
line, err := p.in.ReadString('\n')
if err != nil && !(errors.Is(err, io.EOF) && line != "") {
return "", fmt.Errorf("read %s: %w", strings.ToLower(label), err)
}
return strings.TrimSpace(line), nil
}
// Secret asks for a value without echoing it on a terminal.
func (p *Prompter) Secret(label string) (string, error) {
if !p.tty {
return p.Line(label)
}
fmt.Fprintf(p.out, "%s: ", label)
raw, err := term.ReadPassword(p.fd)
fmt.Fprintln(p.out)
if err != nil {
return "", fmt.Errorf("read %s: %w", strings.ToLower(label), err)
}
return strings.TrimSpace(string(raw)), nil
}
+41
View File
@@ -0,0 +1,41 @@
package telegram
import (
"github.com/gotd/td/constant"
"github.com/gotd/td/tg"
)
// Chat IDs use the usual "marked" form: the other person's user ID for private
// chats, -<id> for basic groups and -100<id> for supergroups and channels.
// peerChatID returns the marked ID of a message's chat.
func peerChatID(p tg.PeerClass) (int64, bool) {
var id constant.TDLibPeerID
switch p := p.(type) {
case *tg.PeerUser:
id.User(p.UserID)
case *tg.PeerChat:
id.Chat(p.ChatID)
case *tg.PeerChannel:
id.Channel(p.ChannelID)
default:
return 0, false
}
return int64(id), true
}
// inputPeerChatID returns the marked ID of a dialog's input peer.
func inputPeerChatID(p tg.InputPeerClass) (int64, bool) {
var id constant.TDLibPeerID
switch p := p.(type) {
case *tg.InputPeerUser:
id.User(p.UserID)
case *tg.InputPeerChat:
id.Chat(p.ChatID)
case *tg.InputPeerChannel:
id.Channel(p.ChannelID)
default:
return 0, false
}
return int64(id), true
}
+46
View File
@@ -0,0 +1,46 @@
package telegram
import (
"testing"
"github.com/gotd/td/tg"
)
func TestPeerChatID(t *testing.T) {
tests := []struct {
peer tg.PeerClass
want int64
}{
{&tg.PeerUser{UserID: 777000}, 777000},
{&tg.PeerChat{ChatID: 42}, -42},
{&tg.PeerChannel{ChannelID: 1234567890}, -1001234567890},
}
for _, tt := range tests {
got, ok := peerChatID(tt.peer)
if !ok || got != tt.want {
t.Errorf("peerChatID(%v) = %d, %v; want %d", tt.peer, got, ok, tt.want)
}
}
if _, ok := peerChatID(nil); ok {
t.Error("peerChatID(nil) ok")
}
}
func TestInputPeerChatID(t *testing.T) {
tests := []struct {
peer tg.InputPeerClass
want int64
ok bool
}{
{&tg.InputPeerUser{UserID: 5, AccessHash: 1}, 5, true},
{&tg.InputPeerChat{ChatID: 42}, -42, true},
{&tg.InputPeerChannel{ChannelID: 1, AccessHash: 1}, -1000000000001, true},
{&tg.InputPeerEmpty{}, 0, false},
}
for _, tt := range tests {
got, ok := inputPeerChatID(tt.peer)
if got != tt.want || ok != tt.ok {
t.Errorf("inputPeerChatID(%v) = %d, %v; want %d, %v", tt.peer, got, ok, tt.want, tt.ok)
}
}
}
+150
View File
@@ -0,0 +1,150 @@
// Package telegram logs in to Telegram and forwards new messages.
package telegram
import (
"context"
"errors"
"fmt"
"io"
"github.com/gotd/td/session"
"github.com/gotd/td/telegram"
"github.com/gotd/td/telegram/auth"
"github.com/gotd/td/telegram/updates"
"github.com/gotd/td/telegram/updates/hook"
"github.com/gotd/td/tg"
"github.com/gotd/td/tgerr"
"github.com/tiennm99/forward-telegram-message/internal/config"
"github.com/tiennm99/forward-telegram-message/internal/prompt"
)
// ErrLogin means signing in failed in a way reconnecting cannot fix, such as
// no answer on the terminal or Telegram rejecting the phone, code, or
// password. Retrying would only request more login codes.
var ErrLogin = errors.New("telegram login failed")
var errNoAccount = fmt.Errorf("%w: this phone has no Telegram account; sign up in an official app first", ErrLogin)
// Listener forwards new source messages to the target across reconnects.
// It keeps the update state and the pending forwards between Listen calls,
// so messages sent while reconnecting are caught up.
type Listener struct {
cfg config.Config
out io.Writer
fwd *forwarder
gaps *updates.Manager
}
// NewListener returns a Listener for cfg that reports on out.
func NewListener(cfg config.Config, out io.Writer) *Listener {
fwd := &forwarder{sourceID: cfg.SourceID, queue: newQueue(), out: out}
dispatcher := tg.NewUpdateDispatcher()
dispatcher.OnNewMessage(func(_ context.Context, _ tg.Entities, u *tg.UpdateNewMessage) error {
fwd.handle(u.Message)
return nil
})
dispatcher.OnNewChannelMessage(func(_ context.Context, _ tg.Entities, u *tg.UpdateNewChannelMessage) error {
fwd.handle(u.Message)
return nil
})
return &Listener{
cfg: cfg,
out: out,
fwd: fwd,
gaps: updates.New(updates.Config{Handler: dispatcher}),
}
}
// Listen connects with the stored session, signs in only when that session is
// missing or expired, and forwards every new source message to the target
// until ctx ends or the connection fails.
func (l *Listener) Listen(ctx context.Context, storage session.Storage, ask *prompt.Prompter) error {
defer l.gaps.Reset()
client := telegram.NewClient(l.cfg.APIID, l.cfg.APIHash, telegram.Options{
SessionStorage: storage,
UpdateHandler: l.gaps,
Middlewares: []telegram.Middleware{hook.UpdateHook(l.gaps.Handle)},
})
return client.Run(ctx, func(ctx context.Context) error {
flow := auth.NewFlow(terminalAuth{phone: l.cfg.Phone, ask: ask}, auth.SendCodeOptions{})
if err := client.Auth().IfNecessary(ctx, flow); err != nil {
if loginFatal(err) {
return fmt.Errorf("%w: %w", ErrLogin, err)
}
return fmt.Errorf("telegram login: %w", err)
}
self, err := client.Self(ctx)
if err != nil {
return fmt.Errorf("read own account: %w", err)
}
r, err := resolveRoute(ctx, client.API(), l.cfg.SourceID, l.cfg.TargetID)
if err != nil {
return err
}
l.fwd.route.Store(&r)
ctx, cancel := context.WithCancel(ctx)
done := make(chan struct{})
go func() {
defer close(done)
l.fwd.work(ctx, client.API())
}()
defer func() {
cancel()
<-done
}()
return l.gaps.Run(ctx, client.API(), self.ID, updates.AuthOptions{
IsBot: self.Bot,
OnStart: func(context.Context) {
fmt.Fprintln(l.out, "Bot is running.")
},
})
})
}
// loginFatal reports whether a login error is one reconnecting cannot fix:
// a failed terminal prompt, a wrong password, or Telegram rejecting the
// request (4xx errors, flood waits included). Network and server errors stay
// retryable.
func loginFatal(err error) bool {
if errors.Is(err, ErrLogin) || errors.Is(err, auth.ErrPasswordInvalid) {
return true
}
rpc, ok := tgerr.As(err)
return ok && rpc.Code < 500
}
// terminalAuth answers the login flow from the terminal.
type terminalAuth struct {
phone string
ask *prompt.Prompter
}
func (a terminalAuth) Phone(context.Context) (string, error) { return a.phone, nil }
func (a terminalAuth) Password(context.Context) (string, error) {
return answer(a.ask.Secret("Password"))
}
func (a terminalAuth) Code(context.Context, *tg.AuthSentCode) (string, error) {
return answer(a.ask.Line("Enter code"))
}
func (a terminalAuth) AcceptTermsOfService(context.Context, tg.HelpTermsOfService) error {
return errNoAccount
}
func (a terminalAuth) SignUp(context.Context) (auth.UserInfo, error) {
return auth.UserInfo{}, errNoAccount
}
// answer marks a failed terminal prompt as a login failure.
func answer(s string, err error) (string, error) {
if err != nil {
return "", fmt.Errorf("%w: %w", ErrLogin, err)
}
return s, nil
}
+37
View File
@@ -0,0 +1,37 @@
package telegram
import (
"errors"
"fmt"
"io"
"testing"
"github.com/gotd/td/telegram/auth"
"github.com/gotd/td/tgerr"
)
func TestLoginFatal(t *testing.T) {
tests := []struct {
name string
err error
want bool
}{
{"no terminal answer", fmt.Errorf("callback: %w", mustAnswerErr(io.EOF)), true},
{"no account", errNoAccount, true},
{"wrong password", fmt.Errorf("sign in: %w", auth.ErrPasswordInvalid), true},
{"wrong code", tgerr.New(400, "PHONE_CODE_INVALID"), true},
{"flood wait", tgerr.New(420, "FLOOD_WAIT_60"), true},
{"server error", tgerr.New(500, "INTERNAL"), false},
{"network", errors.New("connection reset"), false},
}
for _, tt := range tests {
if got := loginFatal(tt.err); got != tt.want {
t.Errorf("%s: loginFatal = %v, want %v", tt.name, got, tt.want)
}
}
}
func mustAnswerErr(err error) error {
_, err = answer("", err)
return err
}
+141
View File
@@ -0,0 +1,141 @@
package telegram
import (
"context"
"errors"
"fmt"
"io"
"math/rand/v2"
"sync/atomic"
"time"
"github.com/gotd/td/telegram/query/dialogs"
"github.com/gotd/td/tg"
"github.com/gotd/td/tgerr"
)
// ErrChatNotFound means a configured chat is not in the account's dialogs.
// Retrying cannot fix it, so callers should stop instead of reconnecting.
var ErrChatNotFound = errors.New("chat not found in this account's chats")
// dialogsPageSize is how many dialogs each lookup request fetches.
const dialogsPageSize = 100
// maxFloodWait caps how long one call may sleep on FLOOD_WAIT before giving up.
const maxFloodWait = 5 * time.Minute
// route holds the resolved source and target chats.
type route struct {
source tg.InputPeerClass
target tg.InputPeerClass
}
// resolveRoute finds the source and target chats in the account's dialogs.
//
// A marked chat ID alone cannot address a supergroup, channel, or user:
// Telegram also needs its access hash. The dialog list carries both, so the
// account must already be in both chats.
func resolveRoute(ctx context.Context, api *tg.Client, sourceID, targetID int64) (route, error) {
var r route
iter := dialogs.NewQueryBuilder(api).GetDialogs().BatchSize(dialogsPageSize).Iter()
for (r.source == nil || r.target == nil) && iter.Next(ctx) {
peer := iter.Value().Peer
switch id, _ := inputPeerChatID(peer); id {
case sourceID:
r.source = peer
case targetID:
r.target = peer
}
}
if err := iter.Err(); err != nil {
return route{}, fmt.Errorf("list chats: %w", err)
}
if r.source == nil {
return route{}, fmt.Errorf("source %d: %w", sourceID, ErrChatNotFound)
}
if r.target == nil {
return route{}, fmt.Errorf("target %d: %w", targetID, ErrChatNotFound)
}
return r, nil
}
// forwarder queues new source messages from the updates loop and forwards
// them in order from a separate worker. Updates can arrive before the route
// is resolved, so route is read atomically.
type forwarder struct {
sourceID int64
route atomic.Pointer[route]
queue *queue
out io.Writer
}
// handle queues m when it is a source message. It never calls Telegram:
// it runs inside the updates loop, which must keep draining while a forward
// waits on the network or a flood wait.
func (f *forwarder) handle(m tg.MessageClass) {
if f.route.Load() == nil {
return
}
if msg, ok := sourceMessage(m, f.sourceID); ok {
f.queue.push(msg.ID)
}
}
// work forwards queued messages until ctx ends. A failed forward is reported
// and skipped so later messages still go through; one cut short by ctx stays
// queued for the next connection.
func (f *forwarder) work(ctx context.Context, api *tg.Client) {
for {
id, err := f.queue.front(ctx)
if err != nil {
return
}
r := f.route.Load()
req := &tg.MessagesForwardMessagesRequest{
FromPeer: r.source,
ToPeer: r.target,
ID: []int{id},
RandomID: []int64{rand.Int64()},
}
_, err = withFloodWait(ctx, func() (tg.UpdatesClass, error) {
return api.MessagesForwardMessages(ctx, req)
})
if ctx.Err() != nil {
return
}
if err != nil {
fmt.Fprintf(f.out, "Exception on forward_message: %v\n", err)
}
f.queue.pop()
}
}
// sourceMessage returns m as a regular message when it belongs to the source
// chat. Service messages such as joins and pins are skipped.
func sourceMessage(m tg.MessageClass, sourceID int64) (*tg.Message, bool) {
msg, ok := m.(*tg.Message)
if !ok {
return nil, false
}
if id, ok := peerChatID(msg.PeerID); !ok || id != sourceID {
return nil, false
}
return msg, true
}
// withFloodWait retries call after Telegram's requested FLOOD_WAIT delay, so
// a burst of source messages does not drop forwards on rate limits.
func withFloodWait[T any](ctx context.Context, call func() (T, error)) (T, error) {
for {
result, err := call()
wait, ok := tgerr.AsFloodWait(err)
if !ok || wait > maxFloodWait {
return result, err
}
select {
case <-ctx.Done():
return result, ctx.Err()
case <-time.After(wait + time.Second):
}
}
}
+121
View File
@@ -0,0 +1,121 @@
package telegram
import (
"bytes"
"context"
"errors"
"strings"
"testing"
"time"
"github.com/gotd/td/bin"
"github.com/gotd/td/tg"
)
func TestSourceMessage(t *testing.T) {
const source = -1001234567890
tests := []struct {
name string
msg tg.MessageClass
ok bool
}{
{"source channel", &tg.Message{ID: 1, PeerID: &tg.PeerChannel{ChannelID: 1234567890}}, true},
{"outgoing in source", &tg.Message{ID: 2, Out: true, PeerID: &tg.PeerChannel{ChannelID: 1234567890}}, true},
{"other channel", &tg.Message{ID: 3, PeerID: &tg.PeerChannel{ChannelID: 99}}, false},
{"basic group with same raw id", &tg.Message{ID: 4, PeerID: &tg.PeerChat{ChatID: 1234567890}}, false},
{"service message", &tg.MessageService{ID: 5, PeerID: &tg.PeerChannel{ChannelID: 1234567890}}, false},
{"empty message", &tg.MessageEmpty{ID: 6}, false},
}
for _, tt := range tests {
msg, ok := sourceMessage(tt.msg, source)
if ok != tt.ok || (ok && msg != tt.msg) {
t.Errorf("%s: got %v, %v; want ok %v", tt.name, msg, ok, tt.ok)
}
}
private := &tg.Message{ID: 7, PeerID: &tg.PeerUser{UserID: 555}}
if _, ok := sourceMessage(private, 555); !ok {
t.Error("private chat message from source user skipped")
}
}
// fakeInvoker records forwarded message IDs and fails or blocks on demand.
type fakeInvoker struct {
forwarded chan int
fail bool
block bool
}
func (f *fakeInvoker) Invoke(ctx context.Context, input bin.Encoder, _ bin.Decoder) error {
if f.block {
<-ctx.Done()
return ctx.Err()
}
req := input.(*tg.MessagesForwardMessagesRequest)
f.forwarded <- req.ID[0]
if f.fail {
return errors.New("forward failed")
}
return nil
}
func newTestForwarder(out *bytes.Buffer) *forwarder {
f := &forwarder{sourceID: 555, queue: newQueue(), out: out}
f.route.Store(&route{source: &tg.InputPeerUser{UserID: 555}, target: &tg.InputPeerChat{ChatID: 42}})
return f
}
func TestHandleQueuesWithoutTelegram(t *testing.T) {
f := newTestForwarder(&bytes.Buffer{})
for i := 1; i <= 100; i++ {
f.handle(&tg.Message{ID: i, PeerID: &tg.PeerUser{UserID: 555}})
}
f.handle(&tg.Message{ID: 999, PeerID: &tg.PeerUser{UserID: 1}})
if got := len(f.queue.ids); got != 100 {
t.Fatalf("queued %d messages, want 100", got)
}
unrouted := &forwarder{sourceID: 555, queue: newQueue()}
unrouted.handle(&tg.Message{ID: 1, PeerID: &tg.PeerUser{UserID: 555}})
if len(unrouted.queue.ids) != 0 {
t.Fatal("queued a message before the route was resolved")
}
}
func TestWorkForwardsInOrderAndSkipsFailures(t *testing.T) {
var out bytes.Buffer
f := newTestForwarder(&out)
inv := &fakeInvoker{forwarded: make(chan int), fail: true}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
defer close(done)
f.work(ctx, tg.NewClient(inv))
}()
for _, id := range []int{1, 2, 3} {
f.queue.push(id)
}
for _, want := range []int{1, 2, 3} {
if got := <-inv.forwarded; got != want {
t.Fatalf("forwarded %d, want %d", got, want)
}
}
cancel()
<-done
if !strings.Contains(out.String(), "Exception on forward_message: ") {
t.Fatalf("failure not reported: %q", out.String())
}
}
func TestWorkKeepsInterruptedForwardQueued(t *testing.T) {
f := newTestForwarder(&bytes.Buffer{})
f.queue.push(7)
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
f.work(ctx, tg.NewClient(&fakeInvoker{block: true}))
if len(f.queue.ids) != 1 || f.queue.ids[0] != 7 {
t.Fatalf("queue = %v, want [7]", f.queue.ids)
}
}
+54
View File
@@ -0,0 +1,54 @@
package telegram
import (
"context"
"sync"
)
// queue is an unbounded FIFO of message IDs. push never blocks, so the
// updates loop that feeds it never waits on a forward in progress.
type queue struct {
mu sync.Mutex
ids []int
ready chan struct{}
}
func newQueue() *queue {
return &queue{ready: make(chan struct{}, 1)}
}
func (q *queue) push(id int) {
q.mu.Lock()
q.ids = append(q.ids, id)
q.mu.Unlock()
select {
case q.ready <- struct{}{}:
default:
}
}
// front waits for the oldest ID without removing it, so an ID whose forward
// is interrupted by a restart is retried on the next connection.
func (q *queue) front(ctx context.Context) (int, error) {
for {
q.mu.Lock()
if len(q.ids) > 0 {
id := q.ids[0]
q.mu.Unlock()
return id, nil
}
q.mu.Unlock()
select {
case <-ctx.Done():
return 0, ctx.Err()
case <-q.ready:
}
}
}
// pop removes the oldest ID.
func (q *queue) pop() {
q.mu.Lock()
q.ids = q.ids[1:]
q.mu.Unlock()
}
-31
View File
@@ -1,31 +0,0 @@
import asyncio
from telethon import TelegramClient
from telethon import events
from config import *
client = TelegramClient(phone, api_id, api_hash)
@client.on(events.NewMessage(chats=[source_group_id]))
async def forward_message(event):
try:
await client.forward_messages(target_group_id, event.message)
except Exception as e:
print(f"Exception on forward_message: {str(e)}")
async def main():
while True:
try:
print("Starting...")
await client.start(phone)
print("Bot is running.")
await client.run_until_disconnected()
except Exception as e:
print(f"Exception on main: {str(e)}. Restarting...")
if __name__ == '__main__':
asyncio.run(main())
BIN
View File
Binary file not shown.