diff --git a/.env.example b/.env.example index a8886bb..8d3b842 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/.gitignore b/.gitignore index 5f60ac5..efe9b82 100644 --- a/.gitignore +++ b/.gitignore @@ -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 \ No newline at end of file +*.session-journal +plans/ + +# go +/bin/ +*.exe +*.test +*.out +coverage.* +go.work +go.work.sum + +# env +.env + +# tooling +.idea/ +.vscode/ +.claude/ diff --git a/README.md b/README.md index 7ebd231..ccddf5c 100644 --- a/README.md +++ b/README.md @@ -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` for supergroups and channels, +`-` 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 diff --git a/cmd/forward/main.go b/cmd/forward/main.go new file mode 100644 index 0000000..8c8ba6d --- /dev/null +++ b/cmd/forward/main.go @@ -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) + } +} diff --git a/config.py b/config.py deleted file mode 100644 index 40b1d97..0000000 --- a/config.py +++ /dev/null @@ -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')) diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..44d6fca --- /dev/null +++ b/go.mod @@ -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 +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..21d93c9 --- /dev/null +++ b/go.sum @@ -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= diff --git a/internal/commands/forward/command.go b/internal/commands/forward/command.go new file mode 100644 index 0000000..09cfddd --- /dev/null +++ b/internal/commands/forward/command.go @@ -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): + } + } +} diff --git a/internal/commands/forward/command_test.go b/internal/commands/forward/command_test.go new file mode 100644 index 0000000..e174535 --- /dev/null +++ b/internal/commands/forward/command_test.go @@ -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()) + } +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..5f843cc --- /dev/null +++ b/internal/config/config.go @@ -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, - for basic groups and -100 for supergroups +// and channels. +func validChatID(id int64) bool { + marked := constant.TDLibPeerID(id) + return marked.IsUser() || marked.IsChat() || marked.IsChannel() +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..47b4894 --- /dev/null +++ b/internal/config/config_test.go @@ -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) + } + } +} diff --git a/internal/prompt/prompt.go b/internal/prompt/prompt.go new file mode 100644 index 0000000..52c45ad --- /dev/null +++ b/internal/prompt/prompt.go @@ -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 +} diff --git a/internal/telegram/chat-id.go b/internal/telegram/chat-id.go new file mode 100644 index 0000000..d609386 --- /dev/null +++ b/internal/telegram/chat-id.go @@ -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, - for basic groups and -100 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 +} diff --git a/internal/telegram/chat-id_test.go b/internal/telegram/chat-id_test.go new file mode 100644 index 0000000..9f54677 --- /dev/null +++ b/internal/telegram/chat-id_test.go @@ -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) + } + } +} diff --git a/internal/telegram/client.go b/internal/telegram/client.go new file mode 100644 index 0000000..6acadcf --- /dev/null +++ b/internal/telegram/client.go @@ -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 +} diff --git a/internal/telegram/client_test.go b/internal/telegram/client_test.go new file mode 100644 index 0000000..5aecd9b --- /dev/null +++ b/internal/telegram/client_test.go @@ -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 +} diff --git a/internal/telegram/forwarder.go b/internal/telegram/forwarder.go new file mode 100644 index 0000000..10a63f0 --- /dev/null +++ b/internal/telegram/forwarder.go @@ -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): + } + } +} diff --git a/internal/telegram/forwarder_test.go b/internal/telegram/forwarder_test.go new file mode 100644 index 0000000..1bcba7a --- /dev/null +++ b/internal/telegram/forwarder_test.go @@ -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) + } +} diff --git a/internal/telegram/queue.go b/internal/telegram/queue.go new file mode 100644 index 0000000..894be96 --- /dev/null +++ b/internal/telegram/queue.go @@ -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() +} diff --git a/main.py b/main.py deleted file mode 100644 index a28aa76..0000000 --- a/main.py +++ /dev/null @@ -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()) diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 0ede79f..0000000 Binary files a/requirements.txt and /dev/null differ