Compare commits

..

3 commits

25 changed files with 2649 additions and 1272 deletions

23
.gitignore vendored
View file

@ -1,12 +1,29 @@
/dist/
/downloads/
# Binaries
/ztrr
/ztorrent*
/torrent-client*
*.log
*.exe
*.dll
*.so
*.dylib
*.out
bin/
cmd/ztrr/ztrr
# Logs and databases
*.log
/logs/
/downloads/
.DS_Store
# Temporary and build files
/dist/
/tmp/
*.test
*.prof
.ztorrent-parts/
profile.txt
testlip.go
# JSON config files
*.json

155
TODO.md
View file

@ -1,126 +1,47 @@
# ZTorrent — TODO & Контекст улучшений
# ZTORRENT — ROADMAP & TODO
> Цель: довести ztorrent до уровня qBittorrent по функциональности и скорости.
> Ведётся автоматически — отражает актуальное состояние реализации.
> **Глобальная цель:** Довести Ztorrent до продакшен-уровня (стандарты qBittorrent и выше), сохранив консольную эстетику.
---
## Архитектура проекта
## 🚀 Глобальный план развития (Roadmap)
```
internal/
torrent/
engine.go — основной движок: загрузка торрента, воркеры, координация
peerwire.go — BitTorrent wire protocol: handshake, download piece, messages
piecescheduler.go — выбор кусков (rarest-first), endgame mode
debug.go — debugf helper
tracker/
tracker.go — HTTP + UDP announce
torrentfile/
torrentfile.go — парсинг .torrent (bencode)
app/
controller.go — тонкая обёртка над Engine для UI
ui/
ui.go — Bubble Tea TUI (welcome, loading, dashboard)
styles.go — цветовая палитра
```
### 1. Движок уровня qBittorrent (Core Engine)
- [ ] **Оптимизация I/O:** Асинхронная запись на диск (mmap / AIO), чтобы не было фризов UI при скачивании на гигабитных скоростях.
- [ ] **Продвинутый кэш:** Умное кэширование кусков в ОЗУ перед сбросом на диск.
- [ ] **Smart Ban / Anti-Leech:** Автоматический бан плохих пиров, защита от спам-хэшей и некорректных кусков.
- [ ] **Приоритизация файлов:** Возможность выбирать, какие файлы из раздачи качать первыми.
### 2. Идеальный DHT
- [ ] **BEP 44 (Arbitrary Data):** Хранение произвольных данных в DHT.
- [ ] **Умный роутинг (Kademlia 2.0):** Динамическое поддержание "живых" нод в bucket'ах, пинги в фоне.
- [ ] **IPv6 DHT:** Поддержка поиска пиров в IPv6 сетях.
- [ ] **Локальный поиск (LSD):** Поиск пиров в локальной сети (Local Service Discovery).
### 3. Умное логирование (Smart Logging)
- [ ] **Уровни логов (Levels):** Переход на структурированные логи (DEBUG, INFO, WARN, ERROR).
- [ ] **Фильтрация логов в UI:** Возможность прямо в TUI отфильтровать логи по "DHT", "Peers", "Disk".
- [ ] **Аналитика сессии:** Summary по итогам скачивания (сколько отброшено кусков, средняя скорость, топовые пиры).
### 4. Эргономика UI (UX Refactoring)
- [ ] **Интерактивные таблицы:** Возможность сортировки по столбцам (по скорости, проценту скачивания).
- [ ] **Просмотр файлов внутри торрента:** Древовидная структура (`tree view`) для многофайловых торрентов.
- [ ] **Настройки "на лету":** Вызов окна настроек (`Config`) прямо из UI (лимиты скорости, порты, пути).
- [ ] **Визуализация Swarm'а:** Красивая карта пиров или график скорости (ASCII Sparklines).
### 5. Killer-Features (Убийцы qBit)
- [ ] **Стриминг в плеер (VLC/mpv):** Скачивание торрента "на лету" (куски по порядку) с пробросом HTTP-стрима прямо в медиаплеер (нажал кнопку — фильм сразу открылся).
- [ ] **Интеграция с Telegram / Discord:** Уведомления об окончании загрузки прямо на телефон.
- [ ] **CLI Remote Control:** Возможность поднять Ztorrent как демона на сервере и управлять им через SSH/TUI удалённо по RPC.
- [ ] **Web-виджет (Read-Only):** Поднятие легковесного веб-сервера для просмотра статуса загрузок с телефона.
---
## P0 — Критичные ✅ ГОТОВО
## 🛠 Выполнено (Архив)
- [x] **Resume download**`createOrOpenPartFile()` без O_TRUNC + SHA-1 верификация
- [x] **Keep-alive**`keepaliveLoop()` горутина, 90с, `msgCancel=8`
- [x] **Endgame mode**`pieceDone` broadcast + `SendCancel()` при дублировании
## P1 — Производительность ✅ ГОТОВО
- [x] **Adaptive pipeline** — depth 464, пересчёт каждые 4 блока по RTT+BW
- [x] **IPv6 compact peers**`parsePeers6()` в tracker.go
- [x] **Rate limiting**`rateLimitedConn`, `SetDownloadLimit`/`SetUploadLimit`
- [x] **Bitmap resume data**`.part.bitmap` JSON, `asyncSaveBitmap()` в heartbeat
## Баги ✅ ПОФИКШЕНО
- [x] **`announce` пустой** — BEP 12 fallback: берём первый URL из `announce-list`
> `torrentfile.go``primaryAnnounce` fallback в `Open()`
- [x] **`peerBackoff` O(n)** → O(1) через `peerIdx map[string]int`
> Добавлен `Engine.peerIdx`, `peerByKey()` helper
> Все peer-методы: `setPeerState`, `setPeerError`, `peerBackoff`, `recordPieceComplete`
- [x] **Скорость без EWMA** → сглаживание α=0.3 в `updateSpeedLocked()`
> `downloadSpeed = 0.3*sample + 0.7*prev`
- [x] **Reannounce блокирует heartbeat** → запускается в отдельной горутине
> `heartbeatTicker.C``go func() { refreshPeersFromTrackers() }()`
> Таймаут уменьшен с 20с до 15с
## Фича ✅ РЕАЛИЗОВАНО
- [x] **Incoming connections** — TCP listener на :6881
> `listenIncoming()` + `handleIncoming()` в engine.go
> Handshake верификация (info_hash), регистрация в `peerIdx`
> Входящие пиры подхватываются `startWorkers` на следующем heartbeat тике
---
## P2 — Долгосрочные (не начато)
- [x] **DHT (BEP 5)** — Kademlia DHT для поиска пиров без трекера
> Новый пакет `internal/dht/` — реализовано!
- [x] **Магнет-ссылки (BEP 9)**`magnet:?xt=...`, ut_metadata extension
> Новый пакет `internal/magnet/` — средняя сложность
- [x] **Seeding / upload (tit-for-tat)** — раздача после завершения
> `peerwire.go` → обработка `msgRequest`, отправка `msgPiece`
- [x] **PEX (BEP 11)** — Peer Exchange между пирами
> Extension handshake + `ut_pex` message
---
## Технические решения
### peerIdx — O(1) peer lookup
```
Engine.peers []PeerStatus // слайс со всеми пирами
Engine.peerIdx map[string]int // "host:port" → индекс в слайсе
buildPeerIdx(peers) // строится при LoadTorrent и reannounce
peerByKey(key) *PeerStatus // O(1) под e.mu
```
### EWMA сглаживание скорости
```
updateSpeedLocked():
sample = delta_bytes / elapsed_sec
speed = 0.3*sample + 0.7*prev_speed // α=0.3
```
### Incoming connections flow
```
runDownload()
└── go listenIncoming(ctx, tf)
└── net.Listen(":6881")
└── for { conn = Accept() → go handleIncoming(conn) }
├── ReadFull → verify wireProtocol + infoHash
├── Write reply handshake
└── register peer in peerIdx
→ подхватывается startWorkers на следующем heartbeat
```
### Resume flow
```
createOrOpenPartFile():
├── если .part существует + .bitmap → verifyPieces() (быстро)
├── если .part существует, нет bitmap → verifyAllPieces() (SHA-1 scan)
└── если .part нет → создаём новый, Truncate(size)
heartbeat:
asyncSaveBitmap() → .part.bitmap
завершение:
os.Remove(.part) + os.Remove(.part.bitmap)
```
- ✅ **Magnet-ссылки (BEP 9)** и метаданные.
- ✅ **Базовый DHT (BEP 5)** с непрерывным поиском.
- ✅ **Мультитрекерность** (HTTP + UDP).
- ✅ **Resume Data / Endgame Mode / Keep-alive.**
- ✅ **Журнал (History) и Логи (Session Logs).**
- ✅ **TUI интерфейс** с глобальными оверлеями (О программе, Журнал, Логи).

View file

@ -1,14 +1,39 @@
package main
import (
"log"
"net/http"
_ "net/http/pprof"
"os"
"github.com/veggiedefender/torrent-client/internal/app"
"github.com/veggiedefender/torrent-client/internal/config"
"github.com/veggiedefender/torrent-client/internal/tgbot"
"github.com/veggiedefender/torrent-client/ui"
)
func main() {
go func() {
log.Println(http.ListenAndServe("localhost:6060", nil))
}()
controller := app.NewController()
ui.Start(controller)
cfg, _ := config.Load()
var token string
if cfg != nil {
token = cfg.TelegramToken
}
if envTok := os.Getenv("TG_BOT_TOKEN"); envTok != "" {
token = envTok
}
bot, err := tgbot.NewBot(controller, token)
if err != nil {
// Log but don't fail, bot is optional
} else if bot != nil {
bot.Start()
}
ui.Start(controller)
}

3
go.mod
View file

@ -6,6 +6,8 @@ require (
github.com/charmbracelet/bubbles v1.0.0
github.com/charmbracelet/bubbletea v1.3.10
github.com/charmbracelet/lipgloss v1.1.0
github.com/go-telegram-bot-api/telegram-bot-api/v5 v5.5.1
github.com/huin/goupnp v1.3.0
github.com/jackpal/bencode-go v1.0.2
golang.org/x/time v0.15.0
)
@ -33,6 +35,7 @@ require (
)
require (
golang.org/x/sync v0.11.0 // indirect
golang.org/x/sys v0.38.0 // indirect
golang.org/x/text v0.22.0 // indirect
)

7
go.sum
View file

@ -28,6 +28,10 @@ github.com/clipperhouse/uax29/v2 v2.5.0 h1:x7T0T4eTHDONxFJsL94uKNKPHrclyFI0lm7+w
github.com/clipperhouse/uax29/v2 v2.5.0/go.mod h1:Wn1g7MK6OoeDT0vL+Q0SQLDz/KpfsVRgg6W7ihQeh4g=
github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f h1:Y/CXytFA4m6baUTXGLOoWe4PQhGxaX0KpnayAqC48p4=
github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f/go.mod h1:vw97MGsxSvLiUE2X8qFplwetxpGLQrlU1Q9AUEIzCaM=
github.com/go-telegram-bot-api/telegram-bot-api/v5 v5.5.1 h1:wG8n/XJQ07TmjbITcGiUaOtXxdrINDz1b0J1w0SzqDc=
github.com/go-telegram-bot-api/telegram-bot-api/v5 v5.5.1/go.mod h1:A2S0CWkNylc2phvKXWBBdD3K0iGnDBGbzRpISP2zBl8=
github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc=
github.com/huin/goupnp v1.3.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8=
github.com/jackpal/bencode-go v1.0.2 h1:LcCNfZ344u0LpBPOZNjpCLps/wUOuN4r87Fy9+5yU8g=
github.com/jackpal/bencode-go v1.0.2/go.mod h1:6jI9mUjO3GQbZti3JizEfxTzRfWOM8oBBcwbwlTfceI=
github.com/lucasb-eyer/go-colorful v1.3.0 h1:2/yBRLdWBZKrf7gB40FoiKfAWYQ0lqNcbuQwVHXptag=
@ -50,6 +54,9 @@ github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavM
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI=
golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=

View file

@ -3,13 +3,17 @@ package app
import (
"context"
"encoding/hex"
"sync"
"time"
"github.com/veggiedefender/torrent-client/internal/config"
"github.com/veggiedefender/torrent-client/internal/history"
"github.com/veggiedefender/torrent-client/internal/tgbot"
"github.com/veggiedefender/torrent-client/internal/torrent"
)
type Controller struct {
mu sync.Mutex
engine *torrent.Engine
ctx context.Context
cancel context.CancelFunc
@ -29,13 +33,14 @@ func NewController() *Controller {
}
func (c *Controller) StartTorrent(path, outputRoot string) error {
// Cancel any previous history updater
c.mu.Lock()
if c.cancel != nil {
c.cancel()
}
c.ctx, c.cancel = context.WithCancel(context.Background())
c.currentPath = path
c.currentOut = outputRoot
c.mu.Unlock()
err := c.engine.LoadTorrent(path, outputRoot)
if err != nil {
@ -44,26 +49,51 @@ func (c *Controller) StartTorrent(path, outputRoot string) error {
c.updateHistory()
go func() {
go func(ctx context.Context) {
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-c.ctx.Done():
case <-ctx.Done():
return
case <-ticker.C:
c.updateHistory()
}
}
}()
}(c.ctx)
return nil
}
func (c *Controller) DownloadMagnet(magnetURI string) error {
cfg, _ := config.Load()
outDir := "."
if cfg != nil && cfg.DefaultDir != "" {
outDir = cfg.DefaultDir
}
return c.StartTorrent(magnetURI, outDir)
}
func (c *Controller) InitTelegramBot(token string) error {
if token == "" {
return nil
}
bot, err := tgbot.NewBot(c, token)
if err != nil {
return err
}
if bot != nil {
bot.Start()
}
return nil
}
func (c *Controller) StopTorrent() {
c.mu.Lock()
if c.cancel != nil {
c.cancel()
}
c.mu.Unlock()
c.updateHistory()
c.engine.Stop()
}
@ -79,11 +109,16 @@ func (c *Controller) updateHistory() {
hashHex = hex.EncodeToString(status.InfoHash[:])
}
c.mu.Lock()
path := c.currentPath
out := c.currentOut
c.mu.Unlock()
hItem := history.Item{
InfoHash: hashHex,
Name: status.Name,
TorrentPath: c.currentPath,
OutputDir: c.currentOut,
TorrentPath: path,
OutputDir: out,
Status: status.Phase,
Progress: c.engine.Progress() * 100, // store as percentage 0-100
Size: status.TotalBytes,
@ -96,5 +131,14 @@ func (c *Controller) Progress() float64 {
}
func (c *Controller) Status() torrent.Status {
if c.engine == nil {
return torrent.Status{}
}
return c.engine.Status()
}
func (c *Controller) ToggleFilePriority(fileIdx int) {
if c.engine != nil {
c.engine.ToggleFilePriority(fileIdx)
}
}

79
internal/config/config.go Normal file
View file

@ -0,0 +1,79 @@
package config
import (
"encoding/json"
"os"
"path/filepath"
"sync"
)
type Config struct {
TelegramToken string `json:"telegram_token"`
TelegramOwnerID int64 `json:"telegram_owner_id"`
DefaultDir string `json:"default_dir"`
DownloadLimit int `json:"download_limit_kbs"` // 0 = no limit
UploadLimit int `json:"upload_limit_kbs"` // 0 = no limit
DisableAnimations bool `json:"disable_animations"`
}
var (
mu sync.Mutex
)
func getConfigPath() (string, error) {
home, err := os.UserHomeDir()
if err != nil {
return "", err
}
return filepath.Join(home, ".ztrr", "config.json"), nil
}
func Load() (*Config, error) {
mu.Lock()
defer mu.Unlock()
path, err := getConfigPath()
if err != nil {
return nil, err
}
data, err := os.ReadFile(path)
if err != nil {
if os.IsNotExist(err) {
return &Config{}, nil
}
return nil, err
}
var cfg Config
if err := json.Unmarshal(data, &cfg); err != nil {
return nil, err
}
return &cfg, nil
}
func Save(cfg *Config) error {
mu.Lock()
defer mu.Unlock()
path, err := getConfigPath()
if err != nil {
return err
}
if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
return err
}
data, err := json.MarshalIndent(cfg, "", " ")
if err != nil {
return err
}
tmpPath := path + ".tmp"
if err := os.WriteFile(tmpPath, data, 0644); err != nil {
return err
}
return os.Rename(tmpPath, path)
}

View file

@ -6,12 +6,12 @@ import (
"crypto/rand"
"encoding/hex"
"fmt"
"log"
"net"
"strings"
"sync"
"time"
"github.com/veggiedefender/torrent-client/internal/logger"
"github.com/veggiedefender/torrent-client/internal/tracker"
)
@ -69,7 +69,7 @@ func (s *Server) Start(ctx context.Context, port int) error {
go s.readLoop(ctx)
go s.bootstrap(ctx)
log.Printf("DHT Server listening on %s with ID %x", conn.LocalAddr(), s.ID[:8])
logger.Info("DHT", "DHT Server listening on %s with ID %x", conn.LocalAddr(), s.ID[:8])
return nil
}
@ -170,9 +170,19 @@ func (s *Server) handleMsg(msg Msg, from *net.UDPAddr) {
}
func (s *Server) sendQuery(ctx context.Context, addr *net.UDPAddr, q string, a map[string]interface{}) (Msg, error) {
var tid string
s.transactionsMu.Lock()
for {
tidBytes := make([]byte, 2)
rand.Read(tidBytes)
tid := string(tidBytes)
tid = string(tidBytes)
if _, exists := s.transactions[tid]; !exists {
break
}
}
ch := make(chan Msg, 1)
s.transactions[tid] = ch
s.transactionsMu.Unlock()
a["id"] = string(s.ID[:])
msg := NewQuery(tid, q, a)
@ -181,11 +191,6 @@ func (s *Server) sendQuery(ctx context.Context, addr *net.UDPAddr, q string, a m
return Msg{}, err
}
ch := make(chan Msg, 1)
s.transactionsMu.Lock()
s.transactions[tid] = ch
s.transactionsMu.Unlock()
defer func() {
s.transactionsMu.Lock()
delete(s.transactions, tid)
@ -233,20 +238,20 @@ func (s *Server) bootstrap(ctx context.Context) {
"target": string(s.ID[:]),
})
if err != nil {
log.Printf("DHT bootstrap %s failed: %v", name, err)
logger.Info("DHT", "DHT bootstrap %s failed: %v", name, err)
return
}
if resp.R != nil {
if nodesStr, ok := resp.R["nodes"].(string); ok {
count := len(nodesStr) / 26
log.Printf("DHT bootstrap %s: got %d nodes", name, count)
logger.Info("DHT", "DHT bootstrap %s: got %d nodes", name, count)
s.parseAndAddNodes(nodesStr)
}
}
}(addr, addrStr)
}
wg.Wait()
log.Printf("DHT bootstrap done, routing table: %d nodes", s.routingTable.Len())
logger.Info("DHT", "DHT bootstrap done, routing table: %d nodes", s.routingTable.Len())
}
// SearchForPeers непрерывно ищет пиров для данного info_hash.
@ -259,7 +264,7 @@ func (s *Server) SearchForPeers(ctx context.Context, infoHash [20]byte) {
}
time.Sleep(300 * time.Millisecond)
}
log.Printf("DHT SearchForPeers starting, routing table: %d nodes", s.routingTable.Len())
logger.Info("DHT", "DHT SearchForPeers starting, routing table: %d nodes", s.routingTable.Len())
targetID := NodeID(infoHash)
queried := make(map[string]bool) // ключ = IP:port строка
@ -285,27 +290,21 @@ func (s *Server) SearchForPeers(ctx context.Context, infoHash [20]byte) {
}
if len(toQuery) == 0 {
// Нет новых нод — сбрасываем карту уже запрошенных и пробуем снова
// (новые ноды могли добавиться в routing table)
log.Printf("DHT: no new nodes to query (%d total queried), resetting and retrying", len(queried))
logger.Info("DHT", "DHT: no new nodes to query (%d total queried), waiting before retry", len(queried))
queried = make(map[string]bool)
// Если routing table совсем пуста — делаем повторный bootstrap
if s.routingTable.Len() == 0 {
go s.bootstrap(ctx)
}
select {
case <-ctx.Done():
return
case <-time.After(5 * time.Second):
case <-time.After(30 * time.Second):
}
continue
}
var wg sync.WaitGroup
for _, n := range toQuery {
wg.Add(1)
go func(node Node) {
defer wg.Done()
qctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
resp, err := s.sendQuery(qctx, node.Addr, "get_peers", map[string]interface{}{
@ -330,17 +329,23 @@ func (s *Server) SearchForPeers(ctx context.Context, infoHash [20]byte) {
}
}
if peers, err := tracker.ParsePeers(allPeerData); err == nil && len(peers) > 0 {
log.Printf("DHT: found %d peers from %s", len(peers), node.Addr)
logger.Info("DHT", "DHT: found %d peers from %s", len(peers), node.Addr)
select {
case s.PeersFound <- peers:
default:
// канал полный — пытаемся без блокировки
case <-ctx.Done():
return
case <-time.After(2 * time.Second):
}
}
}
}(n)
}
wg.Wait()
select {
case <-ctx.Done():
return
case <-time.After(500 * time.Millisecond):
}
}
}

View file

@ -33,48 +33,76 @@ func Distance(a, b NodeID) *big.Int {
return new(big.Int).SetBytes(xor[:])
}
// RoutingTable manages known nodes.
// Bucket represents a Kademlia k-bucket.
type Bucket struct {
nodes []Node
}
// RoutingTable manages known nodes using Kademlia k-buckets.
type RoutingTable struct {
mu sync.RWMutex
ownID NodeID
nodes []Node
buckets [160]*Bucket
}
// NewRoutingTable creates a new routing table.
func NewRoutingTable(ownID NodeID) *RoutingTable {
return &RoutingTable{
rt := &RoutingTable{
ownID: ownID,
nodes: make([]Node, 0),
}
for i := 0; i < 160; i++ {
rt.buckets[i] = &Bucket{nodes: make([]Node, 0, MaxNodes)}
}
return rt
}
// bucketIndex calculates the appropriate bucket index for a given node ID.
// Returns an index from 0 to 159, or -1 if the ID is our own.
func (rt *RoutingTable) bucketIndex(target NodeID) int {
for i := 0; i < 20; i++ {
xor := rt.ownID[i] ^ target[i]
if xor != 0 {
// Find the most significant bit set in the byte
for j := 7; j >= 0; j-- {
if (xor & (1 << j)) != 0 {
return 159 - (i*8 + (7 - j))
}
}
}
}
return -1
}
// AddNode adds a node to the routing table or updates it.
func (rt *RoutingTable) AddNode(n Node) {
idx := rt.bucketIndex(n.ID)
if idx == -1 {
return // Do not add ourselves
}
rt.mu.Lock()
defer rt.mu.Unlock()
if n.ID == rt.ownID {
return
}
bucket := rt.buckets[idx]
for i, existing := range rt.nodes {
// Check if already exists and update
for i, existing := range bucket.nodes {
if existing.ID == n.ID {
rt.nodes[i].Addr = n.Addr
bucket.nodes[i].Addr = n.Addr
// Move to end (most recently seen)
node := bucket.nodes[i]
bucket.nodes = append(bucket.nodes[:i], bucket.nodes[i+1:]...)
bucket.nodes = append(bucket.nodes, node)
return
}
}
rt.nodes = append(rt.nodes, n)
// Sort by distance to our own ID and keep top 1000 nodes for simplicity.
sort.Slice(rt.nodes, func(i, j int) bool {
distI := Distance(rt.nodes[i].ID, rt.ownID)
distJ := Distance(rt.nodes[j].ID, rt.ownID)
return distI.Cmp(distJ) < 0
})
if len(rt.nodes) > 1000 {
rt.nodes = rt.nodes[:1000]
// Add new node if bucket is not full
if len(bucket.nodes) < MaxNodes {
bucket.nodes = append(bucket.nodes, n)
} else {
// In a full Kademlia implementation, we would ping the oldest node
// and replace it if it doesn't respond. For now, just drop the new one.
}
}
@ -83,26 +111,43 @@ func (rt *RoutingTable) ClosestNodes(target NodeID, count int) []Node {
rt.mu.RLock()
defer rt.mu.RUnlock()
// Since we don't have true k-buckets, we sort the entire list.
// This is O(n log n) but fine for a small simplified table of 1000 nodes.
sortedNodes := make([]Node, len(rt.nodes))
copy(sortedNodes, rt.nodes)
var allNodes []Node
sort.Slice(sortedNodes, func(i, j int) bool {
distI := Distance(sortedNodes[i].ID, target)
distJ := Distance(sortedNodes[j].ID, target)
// Fast path: find the target's bucket
idx := rt.bucketIndex(target)
if idx != -1 {
allNodes = append(allNodes, rt.buckets[idx].nodes...)
}
// Collect nodes from neighboring buckets if we don't have enough
if len(allNodes) < count {
// We just collect all nodes and sort them for simplicity.
// Optimizing this is possible but not strictly necessary for < 1280 nodes total.
allNodes = nil
for i := 0; i < 160; i++ {
allNodes = append(allNodes, rt.buckets[i].nodes...)
}
}
sort.Slice(allNodes, func(i, j int) bool {
distI := Distance(allNodes[i].ID, target)
distJ := Distance(allNodes[j].ID, target)
return distI.Cmp(distJ) < 0
})
if len(sortedNodes) > count {
return sortedNodes[:count]
if len(allNodes) > count {
return allNodes[:count]
}
return sortedNodes
return allNodes
}
// Len returns the number of nodes in the routing table.
// Len returns the total number of nodes in the routing table.
func (rt *RoutingTable) Len() int {
rt.mu.RLock()
defer rt.mu.RUnlock()
return len(rt.nodes)
count := 0
for i := 0; i < 160; i++ {
count += len(rt.buckets[i].nodes)
}
return count
}

107
internal/logger/logger.go Normal file
View file

@ -0,0 +1,107 @@
package logger
import (
"fmt"
"sync"
"time"
)
type Level int
const (
LevelDebug Level = iota
LevelInfo
LevelWarn
LevelError
)
func (l Level) String() string {
switch l {
case LevelDebug:
return "DEBUG"
case LevelInfo:
return "INFO"
case LevelWarn:
return "WARN"
case LevelError:
return "ERROR"
default:
return "UNKNOWN"
}
}
type Entry struct {
Time time.Time
Level Level
Category string
Message string
}
type RingBuffer struct {
mu sync.RWMutex
entries []Entry
head int
count int
size int
}
func NewRingBuffer(size int) *RingBuffer {
return &RingBuffer{
entries: make([]Entry, size),
size: size,
}
}
func (b *RingBuffer) Add(e Entry) {
b.mu.Lock()
defer b.mu.Unlock()
b.entries[b.head] = e
b.head = (b.head + 1) % b.size
if b.count < b.size {
b.count++
}
}
func (b *RingBuffer) Snapshot() []Entry {
b.mu.RLock()
defer b.mu.RUnlock()
result := make([]Entry, b.count)
for i := 0; i < b.count; i++ {
idx := (b.head - b.count + i + b.size) % b.size
result[i] = b.entries[idx]
}
return result
}
// Global logger instance
var (
GlobalBuffer = NewRingBuffer(1000)
)
func logf(level Level, category string, format string, v ...interface{}) {
msg := fmt.Sprintf(format, v...)
GlobalBuffer.Add(Entry{
Time: time.Now(),
Level: level,
Category: category,
Message: msg,
})
}
func Debug(category string, format string, v ...interface{}) {
logf(LevelDebug, category, format, v...)
}
func Info(category string, format string, v ...interface{}) {
logf(LevelInfo, category, format, v...)
}
func Warn(category string, format string, v ...interface{}) {
logf(LevelWarn, category, format, v...)
}
func Error(category string, format string, v ...interface{}) {
logf(LevelError, category, format, v...)
}

View file

@ -0,0 +1,115 @@
package portforward
import (
"context"
"net"
"sync"
"time"
"github.com/huin/goupnp/dcps/internetgateway1"
"github.com/veggiedefender/torrent-client/internal/logger"
)
type PortManager struct {
mu sync.Mutex
clients []*internetgateway1.WANIPConnection1
port uint16
opened bool
}
func NewPortManager() *PortManager {
return &PortManager{}
}
// OpenPort attempts to open the specified port on all discovered UPnP routers.
func (pm *PortManager) OpenPort(ctx context.Context, port uint16, desc string) error {
pm.mu.Lock()
defer pm.mu.Unlock()
if pm.opened && pm.port == port {
return nil
}
logger.Info("SYS", "Discovering UPnP routers for port %d...", port)
discoverCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
discoverDone := make(chan []*internetgateway1.WANIPConnection1, 1)
go func() {
clients, _, _ := internetgateway1.NewWANIPConnection1Clients()
discoverDone <- clients
}()
var clients []*internetgateway1.WANIPConnection1
select {
case <-discoverCtx.Done():
logger.Warn("SYS", "UPnP discovery timed out")
return nil
case clients = <-discoverDone:
}
if len(clients) == 0 {
logger.Warn("SYS", "No UPnP routers discovered")
return nil
}
localIP, err := getLocalIP()
if err != nil {
logger.Warn("SYS", "Could not get local IP for UPnP")
return nil
}
success := false
for _, client := range clients {
// TCP Mapping
errTCP := client.AddPortMapping("", port, "TCP", port, localIP, true, desc, 0)
// UDP Mapping (for DHT)
errUDP := client.AddPortMapping("", port, "UDP", port, localIP, true, desc, 0)
if errTCP == nil || errUDP == nil {
success = true
}
}
if success {
pm.clients = clients
pm.port = port
pm.opened = true
logger.Info("SYS", "UPnP successfully forwarded port %d", port)
} else {
logger.Warn("SYS", "Failed to add UPnP port mapping")
}
return nil
}
// ClosePort closes the currently opened port.
func (pm *PortManager) ClosePort() error {
pm.mu.Lock()
defer pm.mu.Unlock()
if !pm.opened || len(pm.clients) == 0 {
return nil
}
for _, client := range pm.clients {
_ = client.DeletePortMapping("", pm.port, "TCP")
_ = client.DeletePortMapping("", pm.port, "UDP")
}
logger.Info("SYS", "UPnP successfully cleared port %d", pm.port)
pm.opened = false
return nil
}
// getLocalIP returns the preferred outbound IP of this machine
func getLocalIP() (string, error) {
conn, err := net.Dial("udp", "8.8.8.8:80")
if err != nil {
return "", err
}
defer conn.Close()
localAddr := conn.LocalAddr().(*net.UDPAddr)
return localAddr.IP.String(), nil
}

View file

@ -1,185 +0,0 @@
package sessionlog
import (
"fmt"
"io"
"log"
"os"
"path/filepath"
"sort"
"strings"
"time"
)
// Entry представляет один лог-файл сессии.
type Entry struct {
Name string // имя файла без расширения
Path string // полный путь
ModTime time.Time // время изменения
}
// Dir возвращает директорию логов (~/.ztorrent/logs).
func Dir() (string, error) {
home, err := os.UserHomeDir()
if err != nil {
return "", err
}
dir := filepath.Join(home, ".ztorrent", "logs")
if err := os.MkdirAll(dir, 0755); err != nil {
return "", err
}
return dir, nil
}
// NewWriter создаёт новый md-файл для текущей сессии и возвращает:
// - путь к файлу
// - io.WriteCloser для записи
// - ошибку
func NewWriter(torrentName string) (string, io.WriteCloser, error) {
dir, err := Dir()
if err != nil {
return "", nil, err
}
ts := time.Now().Format("2006-01-02_15-04-05")
safe := sanitize(torrentName)
if safe == "" {
safe = "session"
}
fname := fmt.Sprintf("%s_%s.md", ts, safe)
path := filepath.Join(dir, fname)
f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644)
if err != nil {
return "", nil, err
}
// Записываем заголовок
header := fmt.Sprintf("# Сессия: %s\n\n**Начало:** %s \n**Торрент:** %s\n\n---\n\n```\n",
fname, time.Now().Format("02.01.2006 15:04:05"), torrentName)
_, _ = f.WriteString(header)
return path, f, nil
}
// sanitize убирает из имени файла недопустимые символы.
func sanitize(s string) string {
replacer := strings.NewReplacer(
"/", "_", "\\", "_", ":", "_", "*", "_",
"?", "_", "\"", "_", "<", "_", ">", "_", "|", "_",
" ", "_",
)
s = replacer.Replace(s)
if len(s) > 40 {
s = s[:40]
}
return s
}
// List возвращает список лог-файлов, отсортированных от новых к старым.
func List() ([]Entry, error) {
dir, err := Dir()
if err != nil {
return nil, err
}
entries, err := os.ReadDir(dir)
if err != nil {
return nil, err
}
var result []Entry
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".md") {
continue
}
info, err := e.Info()
if err != nil {
continue
}
result = append(result, Entry{
Name: strings.TrimSuffix(e.Name(), ".md"),
Path: filepath.Join(dir, e.Name()),
ModTime: info.ModTime(),
})
}
// Сортировка от новых к старым
sort.Slice(result, func(i, j int) bool {
return result[i].ModTime.After(result[j].ModTime)
})
return result, nil
}
// Read читает содержимое лог-файла.
func Read(path string) (string, error) {
data, err := os.ReadFile(path)
if err != nil {
return "", err
}
return string(data), nil
}
// Delete удаляет лог-файл.
func Delete(path string) error {
return os.Remove(path)
}
// LogWriter оборачивает os.File и при записи:
// - пишет в файл
// - пишет в стандартный log (если enable)
type LogWriter struct {
file *os.File
path string
closed bool
}
// NewLogWriter создаёт LogWriter, настраивает стандартный log на запись в файл.
func NewLogWriter(torrentName string) (*LogWriter, error) {
dir, err := Dir()
if err != nil {
return nil, err
}
ts := time.Now().Format("2006-01-02_15-04-05")
safe := sanitize(torrentName)
if safe == "" {
safe = "session"
}
fname := fmt.Sprintf("%s_%s.md", ts, safe)
path := filepath.Join(dir, fname)
f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644)
if err != nil {
return nil, err
}
header := fmt.Sprintf("# Сессия: %s\n\n**Начало:** %s \n**Торрент:** %s\n\n---\n\n```\n",
fname, time.Now().Format("02.01.2006 15:04:05"), torrentName)
_, _ = f.WriteString(header)
lw := &LogWriter{file: f, path: path}
log.SetOutput(lw)
return lw, nil
}
// Write реализует io.Writer.
func (lw *LogWriter) Write(p []byte) (n int, err error) {
if lw.file == nil {
return len(p), nil
}
return lw.file.Write(p)
}
// Path возвращает путь к файлу лога.
func (lw *LogWriter) Path() string {
return lw.path
}
// Close закрывает файл и дописывает завершающий блок.
func (lw *LogWriter) Close() error {
if lw.closed || lw.file == nil {
return nil
}
lw.closed = true
_, _ = lw.file.WriteString("```\n\n---\n\n*Сессия завершена: " + time.Now().Format("02.01.2006 15:04:05") + "*\n")
return lw.file.Close()
}

67
internal/sorter/sorter.go Normal file
View file

@ -0,0 +1,67 @@
package sorter
import (
"fmt"
"os"
"path/filepath"
"regexp"
"github.com/veggiedefender/torrent-client/internal/logger"
)
var (
tvShowRegex = regexp.MustCompile(`(?i)(s\d{2}e\d{2}|season\s*\d+|сезон\s*\d+)`)
movieRegex = regexp.MustCompile(`(?i)(1080p|720p|2160p|4k|bdrip|web-dl|bluray|camrip)`)
gameRegex = regexp.MustCompile(`(?i)(repack|fitgirl|dodi|skidrow|codex|crack|iso)`)
musicRegex = regexp.MustCompile(`(?i)(discography|flac|mp3\s*320)`)
)
// Categorize analyzes the torrent name and determines its category.
func Categorize(name string) string {
if tvShowRegex.MatchString(name) {
return "TV Shows"
}
if movieRegex.MatchString(name) {
return "Movies"
}
if gameRegex.MatchString(name) {
return "Games"
}
if musicRegex.MatchString(name) {
return "Music"
}
return "Other"
}
// SortAndMove moves the downloaded files (or folder) to a sub-folder based on its category.
func SortAndMove(outputRoot, torrentName string) (string, error) {
category := Categorize(torrentName)
if category == "Other" || category == "" {
return "", nil // Do not move if category is unknown
}
sourcePath := filepath.Join(outputRoot, torrentName)
if _, err := os.Stat(sourcePath); os.IsNotExist(err) {
return "", fmt.Errorf("source path does not exist: %s", sourcePath)
}
destDir := filepath.Join(outputRoot, category)
if err := os.MkdirAll(destDir, 0o755); err != nil {
return "", fmt.Errorf("failed to create category directory: %w", err)
}
destPath := filepath.Join(destDir, torrentName)
// If it already exists in the destination, maybe we resume/overwrite
if _, err := os.Stat(destPath); err == nil {
logger.Warn("SYS", "Destination path already exists, skipping move: %s", destPath)
return "", nil
}
if err := os.Rename(sourcePath, destPath); err != nil {
return "", fmt.Errorf("failed to move files to category folder: %w", err)
}
logger.Info("SYS", "Smart Sorter moved '%s' to '%s'", torrentName, category)
return destDir, nil
}

View file

@ -5,6 +5,7 @@ import (
"io"
"os"
"path/filepath"
"sync"
"github.com/veggiedefender/torrent-client/internal/torrentfile"
)
@ -15,6 +16,9 @@ type PieceReader struct {
outputRoot string
pieceLength int
totalLength int
mu sync.Mutex
fdMap map[string]*os.File
}
// NewPieceReader creates a new PieceReader.
@ -24,9 +28,36 @@ func NewPieceReader(files []torrentfile.File, pieceLength, totalLength int, outp
outputRoot: outputRoot,
pieceLength: pieceLength,
totalLength: totalLength,
fdMap: make(map[string]*os.File),
}
}
// Close closes all open file descriptors
func (pr *PieceReader) Close() {
pr.mu.Lock()
defer pr.mu.Unlock()
for _, f := range pr.fdMap {
f.Close()
}
pr.fdMap = make(map[string]*os.File)
}
func (pr *PieceReader) getFile(path string) (*os.File, error) {
pr.mu.Lock()
defer pr.mu.Unlock()
if f, ok := pr.fdMap[path]; ok {
return f, nil
}
f, err := os.Open(path)
if err != nil {
return nil, err
}
pr.fdMap[path] = f
return f, nil
}
// ReadBlock reads a specific block of data from the files.
func (pr *PieceReader) ReadBlock(pieceIndex, begin, length int) ([]byte, error) {
absoluteOffset := pieceIndex*pr.pieceLength + begin
@ -59,18 +90,21 @@ func (pr *PieceReader) ReadBlock(pieceIndex, begin, length int) ([]byte, error)
}
filePath := filepath.Join(pr.outputRoot, file.Path)
f, err := os.Open(filePath)
f, err := pr.getFile(filePath)
if err != nil {
return nil, err
}
_, err = f.ReadAt(buf[bytesRead:bytesRead+readLength], int64(fileOffset))
f.Close()
n, err := f.ReadAt(buf[bytesRead:bytesRead+readLength], int64(fileOffset))
if err != nil && err != io.EOF {
return nil, err
}
bytesRead += readLength
bytesRead += n
if n < readLength {
break // EOF reached prematurely
}
currentOffset += file.Length
}

165
internal/tgbot/bot.go Normal file
View file

@ -0,0 +1,165 @@
package tgbot
import (
"fmt"
"io"
"net/http"
"os"
"strings"
tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
"github.com/veggiedefender/torrent-client/internal/config"
"github.com/veggiedefender/torrent-client/internal/logger"
)
type Controller interface {
DownloadMagnet(magnetURI string) error
// You can add more methods if you want to support downloading .torrent files or getting status.
}
type Bot struct {
api *tgbotapi.BotAPI
controller Controller
}
func NewBot(ctrl Controller, token string) (*Bot, error) {
if token == "" {
// Bot is disabled if no token is provided.
logger.Info("TGBOT", "TG_BOT_TOKEN not set, bot integration is disabled")
return nil, nil
}
api, err := tgbotapi.NewBotAPI(token)
if err != nil {
return nil, fmt.Errorf("failed to create bot: %w", err)
}
logger.Info("TGBOT", "Authorized on account %s", api.Self.UserName)
return &Bot{
api: api,
controller: ctrl,
}, nil
}
func (b *Bot) Start() {
if b == nil || b.api == nil {
return
}
u := tgbotapi.NewUpdate(0)
u.Timeout = 60
updates := b.api.GetUpdatesChan(u)
go func() {
cfg, _ := config.Load()
var ownerID int64
if cfg != nil {
ownerID = cfg.TelegramOwnerID
}
for update := range updates {
if update.Message == nil {
continue
}
// Authorization Check
if ownerID == 0 {
if update.Message.Text == "/start" {
ownerID = update.Message.From.ID
if cfg == nil {
cfg = &config.Config{}
}
cfg.TelegramOwnerID = ownerID
_ = config.Save(cfg)
b.api.Send(tgbotapi.NewMessage(update.Message.Chat.ID, "You are now registered as the owner of this Ztorrent instance."))
} else {
b.api.Send(tgbotapi.NewMessage(update.Message.Chat.ID, "Ztorrent is waiting for the owner to send /start."))
}
continue
} else if update.Message.From.ID != ownerID {
logger.Warn("TGBOT", "Unauthorized access attempt from %s", update.Message.From.UserName)
continue
}
// Check for documents (.torrent files)
if update.Message.Document != nil {
doc := update.Message.Document
if strings.HasSuffix(strings.ToLower(doc.FileName), ".torrent") {
logger.Info("TGBOT", "Received .torrent file from %s: %s", update.Message.From.UserName, doc.FileName)
// Get file URL from Telegram
fileURL, err := b.api.GetFileDirectURL(doc.FileID)
if err != nil {
b.api.Send(tgbotapi.NewMessage(update.Message.Chat.ID, fmt.Sprintf("Error getting file: %v", err)))
continue
}
b.api.Send(tgbotapi.NewMessage(update.Message.Chat.ID, "Torrent file received, starting download..."))
// Download file contents async
go func(chatId int64, url, name string) {
err := b.downloadAndStartTorrent(chatId, url, name)
if err != nil {
b.api.Send(tgbotapi.NewMessage(chatId, fmt.Sprintf("Error starting torrent: %v", err)))
} else {
b.api.Send(tgbotapi.NewMessage(chatId, "Download started successfully!"))
}
}(update.Message.Chat.ID, fileURL, doc.FileName)
continue
}
}
text := strings.TrimSpace(update.Message.Text)
if strings.HasPrefix(text, "magnet:?") {
logger.Info("TGBOT", "Received magnet link from %s", update.Message.From.UserName)
msg := tgbotapi.NewMessage(update.Message.Chat.ID, "Magnet link received, starting download...")
b.api.Send(msg)
err := b.controller.DownloadMagnet(text)
if err != nil {
errMsg := tgbotapi.NewMessage(update.Message.Chat.ID, fmt.Sprintf("Error starting download: %v", err))
b.api.Send(errMsg)
} else {
succMsg := tgbotapi.NewMessage(update.Message.Chat.ID, "Download started successfully!")
b.api.Send(succMsg)
}
continue
}
if text == "/start" {
b.api.Send(tgbotapi.NewMessage(update.Message.Chat.ID, "Welcome back, Owner! Send me a magnet link or a .torrent file."))
}
}
}()
}
func (b *Bot) downloadAndStartTorrent(chatID int64, fileURL, fileName string) error {
resp, err := http.Get(fileURL)
if err != nil {
return fmt.Errorf("failed to download file from telegram: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("bad status code: %d", resp.StatusCode)
}
out, err := os.CreateTemp("", "*_"+fileName)
if err != nil {
return fmt.Errorf("failed to create temp file: %w", err)
}
tmpPath := out.Name()
_, err = io.Copy(out, resp.Body)
out.Close()
if err != nil {
return fmt.Errorf("failed to write temp file: %w", err)
}
// We pass the local path to controller
return b.controller.DownloadMagnet(tmpPath)
}

View file

@ -1,6 +1,6 @@
package torrent
import "log"
import "github.com/veggiedefender/torrent-client/internal/logger"
// debug enables verbose protocol and scheduler logs.
var debug = true
@ -9,7 +9,7 @@ func debugf(format string, args ...any) {
if !debug {
return
}
log.Printf(format, args...)
logger.Debug("ENGINE", format, args...)
}
func SetDebug(enabled bool) {

View file

@ -10,7 +10,6 @@ import (
"errors"
"fmt"
"io"
"log"
mathrand "math/rand"
"net"
"os"
@ -26,7 +25,10 @@ import (
"golang.org/x/time/rate"
"github.com/veggiedefender/torrent-client/internal/dht"
"github.com/veggiedefender/torrent-client/internal/logger"
"github.com/veggiedefender/torrent-client/internal/magnet"
"github.com/veggiedefender/torrent-client/internal/portforward"
"github.com/veggiedefender/torrent-client/internal/sorter"
"github.com/veggiedefender/torrent-client/internal/storage"
"github.com/veggiedefender/torrent-client/internal/torrentfile"
"github.com/veggiedefender/torrent-client/internal/tracker"
@ -42,6 +44,7 @@ type Engine struct {
peerID [20]byte
dhtServer *dht.Server
portManager *portforward.PortManager
cancel context.CancelFunc
@ -66,6 +69,8 @@ type Engine struct {
// rate limiting
downloadLimitBps atomic.Int64
uploadLimitBps atomic.Int64
scheduler *pieceScheduler // For toggling sequential mode
}
type PieceState uint8
@ -295,6 +300,7 @@ func NewEngine() *Engine {
peerID: generatePeerID(),
phase: "idle",
incomingConns: make(chan net.Conn, 128),
portManager: portforward.NewPortManager(),
}
}
@ -334,7 +340,7 @@ func (e *Engine) LoadTorrent(path, outputRoot string) error {
e.resetStateForNewLoad()
e.setPhase("loading_metadata")
outputRoot = normalizeOutputRoot(outputRoot)
log.Printf("loading torrent metadata from %s", path)
logger.Info("ENGINE", "loading torrent metadata from %s", path)
debugf("download root selected: %s", outputRoot)
downloadCtx, cancel := context.WithCancel(context.Background())
@ -347,7 +353,7 @@ func (e *Engine) LoadTorrent(path, outputRoot string) error {
tf, err := torrentfile.Open(path)
if err != nil {
log.Printf("failed to parse torrent %s: %v", path, err)
logger.Info("ENGINE", "failed to parse torrent %s: %v", path, err)
e.setError(err)
e.setPhase("failed")
cancel()
@ -433,18 +439,18 @@ func (e *Engine) downloadMetadataFromPeers(ctx context.Context, infoHash [20]byt
pc, err := newPeerClient(dialCtx, addr, infoHash, e.peerID, 0)
cancel()
if err != nil {
log.Printf("peer dial failed %s: %v", addr, err)
logger.Info("ENGINE", "peer dial failed %s: %v", addr, err)
return
}
defer pc.Close()
if pc.PeerUtMetadataID() == 0 {
log.Printf("peer %s: no ut_metadata support", addr)
logger.Info("ENGINE", "peer %s: no ut_metadata support", addr)
return
}
size := pc.MetadataSize()
log.Printf("peer %s: metadata size=%d, ut_metadata=%d", addr, size, pc.PeerUtMetadataID())
logger.Info("ENGINE", "peer %s: metadata size=%d, ut_metadata=%d", addr, size, pc.PeerUtMetadataID())
if size <= 0 || size > 10*1024*1024 {
return
}
@ -540,7 +546,7 @@ func (e *Engine) startTorrent(downloadCtx context.Context, tf *torrentfile.Torre
}
e.phase = "ready"
e.mu.Unlock()
log.Printf("torrent loaded: name=%s size=%d peers=%d output=%s", tf.Name, tf.Length, len(peers), e.outputPath)
logger.Info("ENGINE", "torrent loaded: name=%s size=%d peers=%d output=%s", tf.Name, tf.Length, len(peers), e.outputPath)
go e.runDownload(downloadCtx, tf)
}
@ -626,7 +632,7 @@ func (e *Engine) queryTrackers(ctx context.Context, trackersToTry []string, info
go func() {
defer wg.Done()
started := time.Now()
log.Printf("tracker request started: %s", announce)
logger.Info("ENGINE", "tracker request started: %s", announce)
perTrackerCtx, cancel := context.WithTimeout(ctx, opts.Timeout)
peers, err := tracker.GetPeersFromURL(perTrackerCtx, announce, infoHash, length, opts)
@ -650,7 +656,7 @@ func (e *Engine) queryTrackers(ctx context.Context, trackersToTry []string, info
if res.err != nil {
st.State = "error"
st.Error = res.err.Error()
log.Printf("tracker response received: %s state=error elapsed=%s err=%v", res.announce, res.elapsed.Round(time.Millisecond), res.err)
logger.Info("ENGINE", "tracker response received: %s state=error elapsed=%s err=%v", res.announce, res.elapsed.Round(time.Millisecond), res.err)
statuses = append(statuses, st)
continue
}
@ -662,7 +668,7 @@ func (e *Engine) queryTrackers(ctx context.Context, trackersToTry []string, info
st.PeerCount = len(res.peers)
}
statuses = append(statuses, st)
log.Printf("tracker response received: %s state=%s peers=%d elapsed=%s", res.announce, st.State, len(res.peers), res.elapsed.Round(time.Millisecond))
logger.Info("ENGINE", "tracker response received: %s state=%s peers=%d elapsed=%s", res.announce, st.State, len(res.peers), res.elapsed.Round(time.Millisecond))
for _, peer := range res.peers {
addPeer(peerMap, peer, res.announce)
@ -696,7 +702,7 @@ func (e *Engine) ensureDHTServer(ctx context.Context) {
if err := srv.Start(ctx, dht.Port); err == nil {
e.dhtServer = srv
} else {
log.Printf("Failed to start DHT server: %v", err)
logger.Info("ENGINE", "Failed to start DHT server: %v", err)
}
}
@ -728,17 +734,63 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
go e.listenIncoming(ctx, tf)
e.setPhase("preparing_download")
log.Printf("preparing download to %s", e.outputRoot)
logger.Info("ENGINE", "preparing download to %s", e.outputRoot)
partPath, partFile, resumedPieces, err := createOrOpenPartFile(tf, e.outputRoot)
if err != nil {
log.Printf("failed to create part file: %v", err)
logger.Info("ENGINE", "failed to create part file: %v", err)
e.setTerminalError("failed", fmt.Errorf("create temp file: %w", err))
return
}
e.mu.Lock()
e.outputPath = partPath
e.mu.Unlock()
defer partFile.Close()
// Восстанавливаем состояние уже скачанных кусков
// Определяем куски, которые нужно пропустить (состоят только из файлов с Priority == 0)
skippedPieces := make(map[int]bool)
{
fileOffset := int64(0)
fileSkippedRanges := make([][2]int64, 0)
for _, f := range tf.Files {
if f.Priority == 0 {
fileSkippedRanges = append(fileSkippedRanges, [2]int64{fileOffset, fileOffset + int64(f.Length)})
}
fileOffset += int64(f.Length)
}
for i := 0; i < len(tf.PieceHashes); i++ {
pStart := int64(i * tf.PieceLength)
pEnd := pStart + int64(tf.PieceLength)
if pEnd > int64(tf.Length) {
pEnd = int64(tf.Length)
}
// Проверяем, покрыт ли кусок полностью skipped файлами
covered := int64(0)
for _, r := range fileSkippedRanges {
// пересечение [pStart, pEnd) и [r[0], r[1])
start := pStart
if r[0] > start {
start = r[0]
}
end := pEnd
if r[1] < end {
end = r[1]
}
if start < end {
covered += (end - start)
}
}
if covered >= (pEnd - pStart) {
skippedPieces[i] = true
resumedPieces = append(resumedPieces, i)
}
}
}
// Восстанавливаем состояние уже скачанных кусков (включая пропущенные)
if len(resumedPieces) > 0 {
e.mu.Lock()
for _, idx := range resumedPieces {
@ -752,13 +804,22 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
e.downloadedBytes = int64(tf.Length)
}
e.mu.Unlock()
log.Printf("resumed download: %d/%d pieces already done", len(resumedPieces), len(tf.PieceHashes))
logger.Info("ENGINE", "resumed download: %d/%d pieces already done", len(resumedPieces), len(tf.PieceHashes))
}
pieceWriter := newBufferedPieceWriter(partFile, tf.PieceLength, 8, 140*time.Millisecond)
defer pieceWriter.Close()
scheduler := newPieceSchedulerWithResume(len(tf.PieceHashes), resumedPieces)
e.mu.Lock()
e.scheduler = scheduler
e.mu.Unlock()
defer func() {
e.mu.Lock()
e.scheduler = nil
e.mu.Unlock()
scheduler.Stop()
}()
workerCtx, workerCancel := context.WithCancel(ctx)
var workersWG sync.WaitGroup
@ -816,7 +877,7 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
defer cleanupWorkers()
e.setPhase("downloading")
log.Printf("download started: pieces=%d peers=%d", len(tf.PieceHashes), len(e.snapshotPeers()))
logger.Info("ENGINE", "download started: pieces=%d peers=%d", len(tf.PieceHashes), len(e.snapshotPeers()))
lastProgressAt := time.Now()
lastReannounceAt := time.Now()
startWorkers(e.snapshotPeers())
@ -867,31 +928,39 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
startWorkers(e.snapshotPeers())
if time.Since(lastProgressAt) >= downloadStallTimeout {
log.Printf("download stalled: no progress for %s", downloadStallTimeout.Round(time.Second))
logger.Info("ENGINE", "download stalled: no progress for %s", downloadStallTimeout.Round(time.Second))
e.setTerminalError("stalled", fmt.Errorf("download stalled: no piece progress for %s", downloadStallTimeout.Round(time.Second)))
return
}
}
}
cleanupWorkers()
if err := pieceWriter.Close(); err != nil {
log.Printf("failed to flush piece writer: %v", err)
logger.Info("ENGINE", "failed to flush piece writer: %v", err)
e.setTerminalError("failed", fmt.Errorf("flush buffered pieces: %w", err))
return
}
e.setPhase("writing_files")
if err := materializeDownloadedFiles(tf, partPath, e.outputRoot); err != nil {
log.Printf("failed to materialize files: %v", err)
logger.Info("ENGINE", "failed to materialize files: %v", err)
e.setTerminalError("failed", fmt.Errorf("write output files: %w", err))
return
}
// Smart Sorter
e.setPhase("sorting_files")
if newRoot, err := sorter.SortAndMove(e.outputRoot, tf.Name); err != nil {
logger.Warn("ENGINE", "Smart Sorter failed to move files: %v", err)
} else if newRoot != "" {
e.mu.Lock()
e.outputRoot = newRoot
e.mu.Unlock()
}
if err := os.Remove(partPath); err != nil && !errors.Is(err, os.ErrNotExist) {
log.Printf("failed to remove part file %s: %v", partPath, err)
logger.Info("ENGINE", "failed to remove part file %s: %v", partPath, err)
}
// Удаляем bitmap вместе с .part
_ = os.Remove(partPath + ".bitmap")
@ -910,15 +979,16 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
e.downloadSpeed = 0
e.uploadSpeed = 0
e.mu.Unlock()
log.Printf("download completed, transitioning to seeding: %s", e.outputPath)
logger.Info("ENGINE", "download completed, transitioning to seeding: %s", e.outputPath)
e.runSeeding(ctx, tf)
}
func (e *Engine) runSeeding(ctx context.Context, tf *torrentfile.TorrentFile) {
log.Printf("entering seeding phase")
logger.Info("ENGINE", "entering seeding phase")
pieceReader := storage.NewPieceReader(tf.Files, tf.PieceLength, tf.Length, e.outputRoot)
defer pieceReader.Close()
workerCtx, workerCancel := context.WithCancel(ctx)
defer workerCancel()
@ -1060,7 +1130,7 @@ func (e *Engine) runSeedingWorker(ctx context.Context, tf *torrentfile.TorrentFi
return
}
msg, err := pc.ReadMessage(ctx)
msg, ptr, err := pc.ReadMessage(ctx)
if err != nil {
if isTimeout(err) {
continue
@ -1071,12 +1141,19 @@ func (e *Engine) runSeedingWorker(ctx context.Context, tf *torrentfile.TorrentFi
switch msg.ID {
case 6: // msgRequest
if len(msg.Payload) < 12 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue
}
idx := int(binary.BigEndian.Uint32(msg.Payload[0:4]))
begin := int(binary.BigEndian.Uint32(msg.Payload[4:8]))
length := int(binary.BigEndian.Uint32(msg.Payload[8:12]))
if ptr != nil {
wireMsgPool.Put(ptr)
}
if length > 16384*2 {
return
}
@ -1092,6 +1169,9 @@ func (e *Engine) runSeedingWorker(ctx context.Context, tf *torrentfile.TorrentFi
e.addUploadedBytes(int64(length))
default:
if ptr != nil {
wireMsgPool.Put(ptr)
}
}
}
}
@ -1109,14 +1189,14 @@ func (e *Engine) runPeerWorker(
e.setPeerState(peer.Address, peer.Port, "connecting", "")
peerAddr := net.JoinHostPort(peer.Address, strconv.Itoa(int(peer.Port)))
log.Printf("peer connection attempt: %s", peerAddr)
logger.Info("ENGINE", "peer connection attempt: %s", peerAddr)
debugf("starting peer worker for %s", peerAddr)
client, err := newPeerClient(ctx, peerAddr, tf.InfoHash, e.peerID, len(tf.PieceHashes))
if err != nil {
if ctx.Err() != nil {
e.setPeerState(peer.Address, peer.Port, "stopped", "")
} else {
log.Printf("peer failed: %s err=%v", peerAddr, err)
logger.Info("ENGINE", "peer failed: %s err=%v", peerAddr, err)
e.setPeerError(peer.Address, peer.Port, err.Error())
}
return
@ -1148,11 +1228,14 @@ func (e *Engine) runPeerWorker(
}
}()
log.Printf("peer connected: %s", peerAddr)
logger.Info("ENGINE", "peer connected: %s", peerAddr)
e.setPeerState(peer.Address, peer.Port, "ready", "")
consecutiveFailures := 0
endgameCancelCh := scheduler.SubscribeCancel(peerKey)
defer scheduler.UnsubscribeCancel(peerKey)
for {
if ctx.Err() != nil {
e.setPeerState(peer.Address, peer.Port, "stopped", "")
@ -1174,7 +1257,7 @@ func (e *Engine) runPeerWorker(
if ctx.Err() != nil {
e.setPeerState(peer.Address, peer.Port, "stopped", "")
} else {
log.Printf("peer %s scheduler acquire failed: %v", peerAddr, err)
logger.Info("ENGINE", "peer %s scheduler acquire failed: %v", peerAddr, err)
e.setPeerError(peer.Address, peer.Port, err.Error())
}
return
@ -1196,10 +1279,31 @@ func (e *Engine) runPeerWorker(
e.setPeerState(peer.Address, peer.Port, "requesting", "")
pieceSize := pieceSizeForIndex(tf, task.Index)
pieceData, transferStats, err := client.DownloadPiece(ctx, task.Index, pieceSize)
pieceCancelCh := make(chan struct{})
pieceDoneCh := make(chan struct{})
go func() {
for {
select {
case canceledPiece := <-endgameCancelCh:
if canceledPiece == task.Index {
close(pieceCancelCh)
return
}
case <-pieceDoneCh:
return
case <-ctx.Done():
return
}
}
}()
pieceData, transferStats, err := client.DownloadPiece(ctx, task.Index, pieceSize, pieceCancelCh)
close(pieceDoneCh)
if err != nil {
if _, reportErr := scheduler.Report(ctx, task.Index, false); reportErr != nil && ctx.Err() == nil {
log.Printf("scheduler report failure for piece %d after peer error: %v", task.Index, reportErr)
logger.Info("ENGINE", "scheduler report failure for piece %d after peer error: %v", task.Index, reportErr)
}
e.setPieceState(task.Index, PieceMissing)
if ctx.Err() != nil {
@ -1207,8 +1311,13 @@ func (e *Engine) runPeerWorker(
return
}
if errors.Is(err, ErrPieceCanceled) {
// Soft cancel: piece was downloaded by another peer, continue.
continue
}
consecutiveFailures++
log.Printf("peer %s disconnected: %v", peerAddr, err)
logger.Info("ENGINE", "peer %s disconnected: %v", peerAddr, err)
e.setPeerError(peer.Address, peer.Port, err.Error())
if shouldDropPeer(err) || consecutiveFailures >= 3 {
return
@ -1219,11 +1328,11 @@ func (e *Engine) runPeerWorker(
hash := sha1.Sum(pieceData)
if hash != tf.PieceHashes[task.Index] {
if _, reportErr := scheduler.Report(ctx, task.Index, false); reportErr != nil && ctx.Err() == nil {
log.Printf("scheduler report hash mismatch for piece %d failed: %v", task.Index, reportErr)
logger.Info("ENGINE", "scheduler report hash mismatch for piece %d failed: %v", task.Index, reportErr)
}
e.setPieceState(task.Index, PieceMissing)
consecutiveFailures++
log.Printf("peer %s piece %d hash mismatch", peerAddr, task.Index)
logger.Info("ENGINE", "peer %s piece %d hash mismatch", peerAddr, task.Index)
e.setPeerError(peer.Address, peer.Port, fmt.Sprintf("piece %d hash mismatch", task.Index))
if consecutiveFailures >= 3 {
return
@ -1233,10 +1342,10 @@ func (e *Engine) runPeerWorker(
if err := pieceWriter.WritePiece(ctx, task.Index, pieceData); err != nil {
if _, reportErr := scheduler.Report(ctx, task.Index, false); reportErr != nil && ctx.Err() == nil {
log.Printf("scheduler report write failure for piece %d failed: %v", task.Index, reportErr)
logger.Info("ENGINE", "scheduler report write failure for piece %d failed: %v", task.Index, reportErr)
}
e.setPieceState(task.Index, PieceMissing)
log.Printf("io error buffering piece %d from peer %s: %v", task.Index, peerAddr, err)
logger.Info("ENGINE", "io error buffering piece %d from peer %s: %v", task.Index, peerAddr, err)
e.setPeerError(peer.Address, peer.Port, fmt.Sprintf("write piece %d: %v", task.Index, err))
return
}
@ -1392,13 +1501,12 @@ func (e *Engine) refreshPeersFromTrackers(ctx context.Context, tf *torrentfile.T
})
// Пересобираем индекс после сортировки
e.peerIdx = buildPeerIdx(e.peers)
log.Printf("discovered %d new peers (total=%d)", added, len(e.peers))
logger.Info("ENGINE", "discovered %d new peers (total=%d)", added, len(e.peers))
}
return added
}
func (e *Engine) currentDownloadedBytes() int64 {
e.mu.RLock()
defer e.mu.RUnlock()
@ -1453,7 +1561,6 @@ func (e *Engine) recordPieceComplete(address string, port uint16, pieceIndex int
}
}
// asyncSaveBitmap сохраняет bitmap в фоне (вызывается из runDownload heartbeat).
func (e *Engine) asyncSaveBitmap(bitmapPath string) {
e.mu.RLock()
@ -1468,7 +1575,6 @@ func (e *Engine) asyncSaveBitmap(bitmapPath string) {
go func() { _ = saveBitmap(bitmapPath, completed, total) }()
}
// peerByKey возвращает указатель на PeerStatus по ключу (O(1)).
// Вызывать только под e.mu.
func (e *Engine) peerByKey(key string) *PeerStatus {
@ -1659,7 +1765,6 @@ func buildPeerIdx(peers []PeerStatus) map[string]int {
return idx
}
func (e *Engine) updateSpeed(now time.Time) {
e.mu.Lock()
defer e.mu.Unlock()
@ -1852,11 +1957,11 @@ func createOrOpenPartFile(tf *torrentfile.TorrentFile, outputRoot string) (strin
if bitmapPieces, bErr := loadBitmap(bitmapPath, len(tf.PieceHashes)); bErr == nil && len(bitmapPieces) > 0 {
// Быстрый путь: bitmap есть, только верифицируем упомянутые куски
completed = verifyPieces(existing, tf, bitmapPieces)
log.Printf("resume via bitmap: %d pieces verified", len(completed))
logger.Info("ENGINE", "resume via bitmap: %d pieces verified", len(completed))
} else {
// Медленный путь: сканируем все куски через SHA-1
completed = verifyAllPieces(existing, tf)
log.Printf("resume via full scan: %d/%d pieces verified", len(completed), len(tf.PieceHashes))
logger.Info("ENGINE", "resume via full scan: %d/%d pieces verified", len(completed), len(tf.PieceHashes))
}
// Перезаписываем bitmap актуальными данными
_ = saveBitmap(bitmapPath, completed, len(tf.PieceHashes))
@ -1950,8 +2055,6 @@ func loadBitmap(path string, expectedTotal int) ([]int, error) {
return data.Completed, nil
}
func materializeDownloadedFiles(tf *torrentfile.TorrentFile, partPath, outputRoot string) error {
partFile, err := os.Open(partPath)
if err != nil {
@ -2057,10 +2160,14 @@ func (e *Engine) listenIncoming(ctx context.Context, tf *torrentfile.TorrentFile
return
}
defer ln.Close()
log.Printf("listening for incoming peers on %s", ln.Addr())
logger.Info("ENGINE", "listening for incoming peers on %s", ln.Addr())
// Open UPnP mapping
go e.portManager.OpenPort(ctx, 6881, "Ztorrent")
go func() {
<-ctx.Done()
e.portManager.ClosePort()
ln.Close()
}()
@ -2149,7 +2256,7 @@ func (e *Engine) handleIncoming(ctx context.Context, rawConn net.Conn, tf *torre
}
e.mu.Unlock()
log.Printf("incoming peer: %s", remoteAddr)
logger.Info("ENGINE", "incoming peer: %s", remoteAddr)
// Delegate to the active loop via channel without closing
select {
@ -2207,7 +2314,7 @@ func (e *Engine) handleIncomingSeeding(ctx context.Context, tf *torrentfile.Torr
return
}
msg, err := pc.ReadMessage(ctx)
msg, ptr, err := pc.ReadMessage(ctx)
if err != nil {
if isTimeout(err) {
continue
@ -2218,12 +2325,19 @@ func (e *Engine) handleIncomingSeeding(ctx context.Context, tf *torrentfile.Torr
switch msg.ID {
case 6: // msgRequest
if len(msg.Payload) < 12 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue
}
idx := int(binary.BigEndian.Uint32(msg.Payload[0:4]))
begin := int(binary.BigEndian.Uint32(msg.Payload[4:8]))
length := int(binary.BigEndian.Uint32(msg.Payload[8:12]))
if ptr != nil {
wireMsgPool.Put(ptr)
}
if length > 16384*2 {
return
}
@ -2238,8 +2352,54 @@ func (e *Engine) handleIncomingSeeding(ctx context.Context, tf *torrentfile.Torr
}
e.addUploadedBytes(int64(length))
default:
if ptr != nil {
wireMsgPool.Put(ptr)
}
}
}
}()
}
// SetSequentialMode toggles sequential piece downloading mode on the fly
func (e *Engine) SetSequentialMode(mode bool) {
e.mu.RLock()
scheduler := e.scheduler
e.mu.RUnlock()
if scheduler != nil {
scheduler.SetSequential(mode)
}
}
// HasPiece is a thread-safe check to see if a piece is fully downloaded
func (e *Engine) HasCompletedPiece(index int) bool {
e.mu.RLock()
defer e.mu.RUnlock()
if index < 0 || index >= len(e.pieceStates) {
return false
}
return e.pieceStates[index] == PieceCompleted
}
// ToggleFilePriority toggles a file's priority between 0 (skip) and 1 (normal).
func (e *Engine) ToggleFilePriority(fileIdx int) {
e.mu.Lock()
defer e.mu.Unlock()
if e.torrent == nil || fileIdx < 0 || fileIdx >= len(e.torrent.Files) {
return
}
if e.torrent.Files[fileIdx].Priority == 0 {
e.torrent.Files[fileIdx].Priority = 1
} else {
e.torrent.Files[fileIdx].Priority = 0
}
}
// PartFilePath returns the path to the current .part file
func (e *Engine) PartFilePath() string {
e.mu.RLock()
defer e.mu.RUnlock()
return e.outputPath
}

View file

@ -7,7 +7,6 @@ import (
"errors"
"fmt"
"io"
"log"
"net"
"sync"
"sync/atomic"
@ -16,6 +15,7 @@ import (
"github.com/jackpal/bencode-go"
"golang.org/x/time/rate"
"github.com/veggiedefender/torrent-client/internal/logger"
"github.com/veggiedefender/torrent-client/internal/torrentfile"
"github.com/veggiedefender/torrent-client/internal/tracker"
)
@ -52,6 +52,14 @@ const (
peerSocketWriteBuffer = 512 * 1024
)
var wireMsgPool = sync.Pool{
New: func() any {
// typical piece msg: 4(len) + 1(id) + 8(header) + 16384(block) = 16397
b := make([]byte, requestBlockSize+128)
return &b
},
}
type extendedHandshake struct {
M map[string]int `bencode:"m"`
MetadataSize int `bencode:"metadata_size"`
@ -192,15 +200,14 @@ type peerClient struct {
kaOnce sync.Once
}
func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [20]byte, pieceCount int) (*peerClient, error) {
dialer := net.Dialer{Timeout: peerConnectTimeout}
rawConn, err := dialer.DialContext(ctx, "tcp", addr)
if err != nil {
log.Printf("peer dial failed %s: %v", addr, err)
logger.Info("PEER", "peer dial failed %s: %v", addr, err)
return nil, err
}
log.Printf("connected to peer %s", addr)
logger.Info("PEER", "connected to peer %s", addr)
if tcpConn, ok := rawConn.(*net.TCPConn); ok {
_ = tcpConn.SetNoDelay(true)
@ -229,7 +236,7 @@ func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [
if pc.supportsExtensions {
if err := pc.sendExtendedHandshake(ctx); err != nil {
log.Printf("failed to send extended handshake to %s: %v", addr, err)
logger.Info("PEER", "failed to send extended handshake to %s: %v", addr, err)
}
}
@ -240,7 +247,7 @@ func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [
debugf("sent interested to %s", addr)
if err := pc.readInitialMessages(ctx); err != nil {
log.Printf("peer %s initial message read failed: %v", addr, err)
logger.Info("PEER", "peer %s initial message read failed: %v", addr, err)
measured.Close()
return nil, err
}
@ -249,7 +256,6 @@ func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [
return pc, nil
}
func (pc *peerClient) PeerUtPexID() int {
return pc.peerUtPexID
}
@ -268,7 +274,7 @@ func newIncomingPeerClient(ctx context.Context, rawConn net.Conn, extensions boo
if pc.supportsExtensions {
if err := pc.sendExtendedHandshake(ctx); err != nil {
log.Printf("failed to send extended handshake to incoming peer: %v", err)
logger.Info("PEER", "failed to send extended handshake to incoming peer: %v", err)
}
}
@ -308,14 +314,15 @@ func (pc *peerClient) keepaliveLoop() {
}
}
func (pc *peerClient) PieceAvailability() ([]bool, bool) {
have := make([]bool, len(pc.have))
copy(have, pc.have)
return have, pc.hasPieceInfo
}
func (pc *peerClient) DownloadPiece(ctx context.Context, pieceIndex int, pieceLength int) ([]byte, pieceTransferStats, error) {
var ErrPieceCanceled = errors.New("piece download canceled")
func (pc *peerClient) DownloadPiece(ctx context.Context, pieceIndex int, pieceLength int, cancelCh <-chan struct{}) ([]byte, pieceTransferStats, error) {
var transfer pieceTransferStats
if pieceLength <= 0 {
return nil, transfer, fmt.Errorf("invalid piece length %d", pieceLength)
@ -342,6 +349,17 @@ func (pc *peerClient) DownloadPiece(ctx context.Context, pieceIndex int, pieceLe
depth := initPipelineDepth // adaptive, пересчитывается каждые 4 блока
for received < pieceLength {
if cancelCh != nil {
select {
case <-cancelCh:
for begin, req := range pending {
pc.SendCancel(pieceIndex, begin, req.length)
}
return nil, transfer, ErrPieceCanceled
default:
}
}
for len(pending) < depth && offset < pieceLength {
blockLength := requestBlockSize
if remaining := pieceLength - offset; remaining < blockLength {
@ -358,22 +376,34 @@ func (pc *peerClient) DownloadPiece(ctx context.Context, pieceIndex int, pieceLe
offset += blockLength
}
gotIndex, gotBegin, block, err := pc.readPieceMessage(ctx)
gotIndex, gotBegin, block, bufPtr, err := pc.readPieceMessage(ctx)
if err != nil {
return nil, transfer, err
}
if gotIndex != pieceIndex {
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
continue
}
req, ok := pending[gotBegin]
if !ok {
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
continue
}
if len(block) != req.length {
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
return nil, transfer, fmt.Errorf("unexpected block size for piece=%d begin=%d: got=%d want=%d", pieceIndex, gotBegin, len(block), req.length)
}
if gotBegin < 0 || gotBegin+len(block) > len(piece) {
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
return nil, transfer, fmt.Errorf("piece block bounds invalid for piece=%d begin=%d block=%d", pieceIndex, gotBegin, len(block))
}
@ -381,6 +411,10 @@ func (pc *peerClient) DownloadPiece(ctx context.Context, pieceIndex int, pieceLe
delete(pending, gotBegin)
received += len(block)
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
blocksCompleted++
blockLatency := time.Since(req.requestedAt)
latencySum += blockLatency
@ -438,7 +472,6 @@ func (pc *peerClient) sendMessageDirect(msgID int, payload []byte) error {
return err
}
func (pc *peerClient) readInitialMessages(ctx context.Context) error {
if err := pc.setReadDeadlineFromContext(ctx, 2*time.Second); err != nil {
return err
@ -446,7 +479,7 @@ func (pc *peerClient) readInitialMessages(ctx context.Context) error {
defer pc.conn.SetReadDeadline(time.Time{})
for {
msg, err := readWireMessage(pc.conn)
msg, bufPtr, err := readWireMessage(pc.conn)
if err != nil {
if isTimeout(err) {
debugf("peer initial message window finished")
@ -455,6 +488,9 @@ func (pc *peerClient) readInitialMessages(ctx context.Context) error {
return err
}
pc.consumeMessage(msg)
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
}
}
@ -475,7 +511,7 @@ func (pc *peerClient) waitForUnchoke(ctx context.Context) error {
if err := pc.setReadDeadlineFromContext(ctx, peerReadTimeout); err != nil {
return err
}
msg, err := readWireMessage(pc.conn)
msg, bufPtr, err := readWireMessage(pc.conn)
if err != nil {
if isTimeout(err) {
continue
@ -483,7 +519,12 @@ func (pc *peerClient) waitForUnchoke(ctx context.Context) error {
return err
}
pc.consumeMessage(msg)
if msg.ID == msgUnchoke {
msgID := msg.ID
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
if msgID == msgUnchoke {
debugf("peer unchoked")
return nil
}
@ -499,41 +540,53 @@ func (pc *peerClient) sendRequest(ctx context.Context, pieceIndex, begin, length
return pc.sendMessage(ctx, msgRequest, payload)
}
func (pc *peerClient) readPieceMessage(ctx context.Context) (pieceIndex int, begin int, block []byte, err error) {
func (pc *peerClient) readPieceMessage(ctx context.Context) (pieceIndex int, begin int, block []byte, bufPtr *[]byte, err error) {
for {
if err := ctx.Err(); err != nil {
return 0, 0, nil, err
return 0, 0, nil, nil, err
}
if err := pc.setReadDeadlineFromContext(ctx, peerReadTimeout); err != nil {
return 0, 0, nil, err
return 0, 0, nil, nil, err
}
msg, err := readWireMessage(pc.conn)
msg, ptr, err := readWireMessage(pc.conn)
if err != nil {
if isTimeout(err) {
continue
}
return 0, 0, nil, err
return 0, 0, nil, nil, err
}
switch msg.ID {
case msgPiece:
if len(msg.Payload) < 8 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue
}
gotIndex := int(binary.BigEndian.Uint32(msg.Payload[0:4]))
gotBegin := int(binary.BigEndian.Uint32(msg.Payload[4:8]))
data := msg.Payload[8:]
if len(data) == 0 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue
}
debugf("received piece %d offset %d block=%d", gotIndex, gotBegin, len(data))
return gotIndex, gotBegin, data, nil
return gotIndex, gotBegin, data, ptr, nil
case msgChoke:
pc.consumeMessage(msg)
return 0, 0, nil, errors.New("peer choked")
if ptr != nil {
wireMsgPool.Put(ptr)
}
return 0, 0, nil, nil, errors.New("peer choked")
default:
pc.consumeMessage(msg)
if ptr != nil {
wireMsgPool.Put(ptr)
}
}
}
}
@ -724,9 +777,9 @@ func (pc *peerClient) SendPex(ctx context.Context, added []tracker.Peer) error {
return pc.sendMessage(ctx, msgExtended, buf.Bytes())
}
func (pc *peerClient) ReadMessage(ctx context.Context) (wireMessage, error) {
func (pc *peerClient) ReadMessage(ctx context.Context) (wireMessage, *[]byte, error) {
if err := pc.setReadDeadlineFromContext(ctx, peerReadTimeout); err != nil {
return wireMessage{}, err
return wireMessage{}, nil, err
}
return readWireMessage(pc.conn)
}
@ -737,34 +790,58 @@ func (pc *peerClient) sendMessage(ctx context.Context, msgID int, payload []byte
}
length := uint32(1 + len(payload))
buf := make([]byte, 4+length)
// Get buffer from pool
bufPtr := wireMsgPool.Get().(*[]byte)
buf := *bufPtr
if uint32(len(buf)) < 4+length {
// Fallback to allocation if payload is larger than typical
buf = make([]byte, 4+length)
}
binary.BigEndian.PutUint32(buf[0:4], length)
buf[4] = byte(msgID)
copy(buf[5:], payload)
_, err := pc.conn.Write(buf)
_, err := pc.conn.Write(buf[:4+length])
// Return buffer to pool
wireMsgPool.Put(bufPtr)
return err
}
func readWireMessage(r io.Reader) (wireMessage, error) {
func readWireMessage(r io.Reader) (wireMessage, *[]byte, error) {
var lengthBuf [4]byte
if _, err := io.ReadFull(r, lengthBuf[:]); err != nil {
return wireMessage{}, err
return wireMessage{}, nil, err
}
length := binary.BigEndian.Uint32(lengthBuf[:])
if length == 0 {
return wireMessage{ID: -1}, nil
return wireMessage{ID: -1}, nil, nil
}
if length > maxWireMessageSize {
return wireMessage{}, fmt.Errorf("wire message too large: %d", length)
return wireMessage{}, nil, fmt.Errorf("wire message too large: %d", length)
}
msg := make([]byte, length)
if _, err := io.ReadFull(r, msg); err != nil {
return wireMessage{}, err
var msg []byte
var bufPtr *[]byte
if length <= requestBlockSize+128 {
bufPtr = wireMsgPool.Get().(*[]byte)
msg = (*bufPtr)[:length]
} else {
msg = make([]byte, length)
}
return wireMessage{ID: int(msg[0]), Payload: msg[1:]}, nil
if _, err := io.ReadFull(r, msg); err != nil {
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
return wireMessage{}, nil, err
}
return wireMessage{ID: int(msg[0]), Payload: msg[1:]}, bufPtr, nil
}
func bitfieldHasPiece(bitfield []byte, index int) bool {
@ -822,7 +899,7 @@ func (pc *peerClient) ReadMetadataMessage(ctx context.Context) (piece int, data
if err := pc.setReadDeadlineFromContext(ctx, peerReadTimeout); err != nil {
return 0, nil, false, err
}
msg, err := readWireMessage(pc.conn)
msg, ptr, err := readWireMessage(pc.conn)
if err != nil {
if isTimeout(err) {
continue
@ -831,6 +908,9 @@ func (pc *peerClient) ReadMetadataMessage(ctx context.Context) (piece int, data
}
if msg.ID == msgExtended {
if len(msg.Payload) == 0 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue
}
extID := int(msg.Payload[0])
@ -838,20 +918,34 @@ func (pc *peerClient) ReadMetadataMessage(ctx context.Context) (piece int, data
reader := bytes.NewReader(msg.Payload[1:])
var dict map[string]int
if err := bencode.Unmarshal(reader, &dict); err != nil {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue
}
msgType, ok := dict["msg_type"]
if !ok {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue
}
pieceIdx := dict["piece"]
if msgType == 2 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
return pieceIdx, nil, true, nil
}
if msgType == 1 {
bytesRead := len(msg.Payload[1:]) - reader.Len()
data := msg.Payload[1+bytesRead:]
srcData := msg.Payload[1+bytesRead:]
data := make([]byte, len(srcData))
copy(data, srcData)
if ptr != nil {
wireMsgPool.Put(ptr)
}
return pieceIdx, data, false, nil
}
} else if extID == 0 {
@ -860,6 +954,10 @@ func (pc *peerClient) ReadMetadataMessage(ctx context.Context) (piece int, data
} else {
pc.consumeMessage(msg)
}
if ptr != nil {
wireMsgPool.Put(ptr)
}
}
}

View file

@ -42,6 +42,8 @@ type pieceScheduler struct {
stopCh chan struct{}
stopOnce sync.Once
setSequentialCh chan bool
// cancelCh рассылает сигнал отмены конкретному воркеру в endgame-режиме.
// Ключ — peerID, значение — канал с индексом куска для отмены.
cancelMu sync.RWMutex
@ -68,9 +70,10 @@ func newPieceSchedulerWithResume(pieceCount int, completedIndices []int) *pieceS
assignCh: make(chan assignPieceRequest, 128),
reportCh: make(chan reportPieceRequest, 128),
releasePeerCh: make(chan string, 128),
progressCh: make(chan int, 128),
progressCh: make(chan int, 1),
doneCh: make(chan struct{}),
stopCh: make(chan struct{}),
setSequentialCh: make(chan bool),
cancelSubs: make(map[string]chan int),
}
@ -189,15 +192,21 @@ func (ps *pieceScheduler) Stop() {
})
}
func (ps *pieceScheduler) SetSequential(mode bool) {
select {
case ps.setSequentialCh <- mode:
case <-ps.stopCh:
}
}
func (ps *pieceScheduler) run(pieceCount int, completedIndices []int) {
states := make([]pieceState, pieceCount)
availability := make([]int, pieceCount)
peerAvailability := make(map[string][]bool, 256)
// endgamePeers — множество пиров, которым назначен кусок в endgame режиме
endgamePeers := make(map[int][]string) // pieceIndex -> []peerID
endgamePeers := make(map[int][]string)
completed := 0
sequentialMode := false
// Восстанавливаем уже завершённые куски из resume
for _, idx := range completedIndices {
if idx >= 0 && idx < pieceCount {
states[idx] = pieceDone
@ -221,7 +230,6 @@ func (ps *pieceScheduler) run(pieceCount int, completedIndices []int) {
return
}
// Подсчёт оставшихся кусков для определения endgame
remaining := pieceCount - completed
endgame := remaining <= endgameThreshold
@ -238,19 +246,20 @@ func (ps *pieceScheduler) run(pieceCount int, completedIndices []int) {
pieceIndex := -1
if endgame {
// Endgame: разрешаем брать куски в состоянии InProgress тоже
pieceIndex = selectPieceEndgame(states, availability, req.have, req.hasInfo, endgamePeers, req.peerID)
} else {
if sequentialMode {
pieceIndex = selectPendingPieceSequential(states, req.have, req.hasInfo)
} else {
pieceIndex = selectPendingPieceRarest(states, availability, req.have, req.hasInfo)
}
}
if pieceIndex >= 0 {
if !endgame {
states[pieceIndex] = pieceInProgress
}
// В endgame: запоминаем всех пиров, которые качают этот кусок
endgamePeers[pieceIndex] = append(endgamePeers[pieceIndex], req.peerID)
debugf("scheduler assigned piece %d to %s (endgame=%v)", pieceIndex, req.peerID, endgame)
req.responseCh <- assignPieceResponse{
task: pieceTask{Index: pieceIndex},
ok: true,
@ -260,25 +269,21 @@ func (ps *pieceScheduler) run(pieceCount int, completedIndices []int) {
req.responseCh <- assignPieceResponse{ok: false}
case req := <-ps.reportCh:
if req.pieceIndex < 0 || req.pieceIndex >= len(states) {
if req.pieceIndex < 0 || req.pieceIndex >= pieceCount {
req.responseCh <- false
continue
}
if req.success {
if states[req.pieceIndex] == pieceDone {
debugf("scheduler report piece %d ignored (already done)", req.pieceIndex)
req.responseCh <- false
continue
}
states[req.pieceIndex] = pieceDone
completed++
debugf("scheduler report piece %d success (%d/%d)", req.pieceIndex, completed, pieceCount)
// Endgame: уведомить других пиров отменить этот кусок
if peers, ok := endgamePeers[req.pieceIndex]; ok && len(peers) > 1 {
// Находим winner — последний репортнувший (req не содержит peerID,
// поэтому рассылаем всем — воркер проверяет индекс)
go ps.broadcastCancel(req.pieceIndex, "")
}
delete(endgamePeers, req.pieceIndex)
@ -287,30 +292,15 @@ func (ps *pieceScheduler) run(pieceCount int, completedIndices []int) {
case ps.progressCh <- req.pieceIndex:
default:
}
req.responseCh <- true
continue
}
// Неуспех: возвращаем кусок в pending
if states[req.pieceIndex] == pieceInProgress {
states[req.pieceIndex] = piecePending
debugf("scheduler report piece %d failed, re-queued", req.pieceIndex)
}
// В endgame: убираем только этого пира из списка
if peers, ok := endgamePeers[req.pieceIndex]; ok {
filtered := peers[:0]
for _, p := range peers {
if p != "" { // убираем все (peerID недоступен в req)
filtered = append(filtered, p)
}
}
if len(filtered) == 0 {
delete(endgamePeers, req.pieceIndex)
} else {
endgamePeers[req.pieceIndex] = filtered
if states[req.pieceIndex] != pieceDone {
states[req.pieceIndex] = piecePending
}
}
req.responseCh <- false
req.responseCh <- true
case mode := <-ps.setSequentialCh:
sequentialMode = mode
}
}
}
@ -448,6 +438,35 @@ func selectPendingPieceRarest(states []pieceState, availability []int, have []bo
return -1
}
func selectPendingPieceSequential(states []pieceState, have []bool, hasInfo bool) int {
firstPending := firstPendingPiece(states)
if firstPending < 0 {
return -1
}
// If peer has not sent bitfield/have yet, optimistically probe.
if !hasInfo {
return firstPending
}
for pieceIndex, state := range states {
if state != piecePending {
continue
}
if pieceIndex >= len(have) || !have[pieceIndex] {
continue
}
return pieceIndex // В последовательном режиме сразу берём первый доступный
}
// Some peers send truncated availability info; allow fallback probing.
if len(have) == 0 || len(have) < len(states) {
return firstPending
}
return -1
}
func firstPendingPiece(states []pieceState) int {
for pieceIndex, state := range states {
if state == piecePending {

View file

@ -33,6 +33,7 @@ type file struct {
type File struct {
Path string
Length int
Priority int // 0 = skip, 1 = normal
}
type TorrentFile struct {
@ -198,7 +199,7 @@ func splitPieceHashes(rawPieces string) ([][20]byte, error) {
func deriveFilesAndLength(info bencodeInfo) ([]File, int, error) {
if info.Length > 0 {
return []File{{Path: info.Name, Length: info.Length}}, info.Length, nil
return []File{{Path: info.Name, Length: info.Length, Priority: 1}}, info.Length, nil
}
files := make([]File, 0, len(info.Files))
@ -208,20 +209,18 @@ func deriveFilesAndLength(info bencodeInfo) ([]File, int, error) {
return nil, 0, errors.New("torrent file length must be greater than zero")
}
if len(f.Path) == 0 {
return nil, 0, errors.New("torrent file path is empty")
return nil, 0, errors.New("torrent file path cannot be empty")
}
parts := make([]string, 0, len(f.Path)+1)
parts = append(parts, info.Name)
parts = append(parts, f.Path...)
filePath := path.Join(parts...)
if filePath == "." {
return nil, 0, errors.New("torrent file path is invalid")
}
files = append(files, File{
Path: filePath,
Length: f.Length,
Priority: 1,
})
totalLength += f.Length
}

View file

@ -153,47 +153,77 @@ func getPeersUDP(ctx context.Context, announceURL *url.URL, infoHash [20]byte, l
if err != nil {
return nil, err
}
connectReq := buildUDPConnectRequest(connectTx)
var connectionID uint64
var connOK bool
for n := 0; n <= 8; n++ {
if _, err := conn.Write(connectReq[:]); err != nil {
return nil, err
}
conn.SetReadDeadline(time.Now().Add(15 * time.Second * time.Duration(1<<n)))
resp := make([]byte, 65535)
n, err := conn.Read(resp)
if err != nil {
return nil, err
nRead, err := conn.Read(resp)
if err == nil {
cid, errParse := parseUDPConnectResponse(resp[:nRead], connectTx)
if errParse == nil {
connectionID = cid
connOK = true
break
}
}
connectionID, err := parseUDPConnectResponse(resp[:n], connectTx)
if err != nil {
return nil, err
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
}
if !connOK {
return nil, fmt.Errorf("udp tracker connect timeout")
}
announceTx, err := randomUint32()
if err != nil {
return nil, err
}
key, err := randomUint32()
if err != nil {
return nil, err
}
announceReq, err := buildUDPAnnounceRequest(connectionID, announceTx, infoHash, length, opts, key)
if err != nil {
return nil, err
}
for n := 0; n <= 8; n++ {
if _, err := conn.Write(announceReq); err != nil {
return nil, err
}
n, err = conn.Read(resp)
conn.SetReadDeadline(time.Now().Add(15 * time.Second * time.Duration(1<<n)))
resp := make([]byte, 65535)
for {
nRead, err := conn.Read(resp)
if err != nil {
return nil, err
break // read timeout or error, retransmit
}
peers, errParse := parseUDPAnnounceResponse(resp[:nRead], announceTx)
if errParse == nil {
return peers, nil
}
// if errParse != nil, it might be a delayed packet from another transaction, so we just loop and Read again
}
return parseUDPAnnounceResponse(resp[:n], announceTx)
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
}
return nil, fmt.Errorf("udp tracker announce timeout")
}
func normalizeOptions(opts AnnounceOptions) AnnounceOptions {
@ -312,8 +342,8 @@ func buildUDPAnnounceRequest(connectionID uint64, transactionID uint32, infoHash
binary.BigEndian.PutUint64(req[56:64], uint64(downloaded))
binary.BigEndian.PutUint64(req[64:72], uint64(left))
binary.BigEndian.PutUint64(req[72:80], uint64(uploaded))
binary.BigEndian.PutUint32(req[80:84], 0)
binary.BigEndian.PutUint32(req[84:88], 0)
binary.BigEndian.PutUint32(req[80:84], 1) // event: 1 (started) instead of 0 (none)
binary.BigEndian.PutUint32(req[84:88], 0) // IP address (default 0)
binary.BigEndian.PutUint32(req[88:92], key)
binary.BigEndian.PutUint32(req[92:96], uint32(int32(opts.NumWant)))
binary.BigEndian.PutUint16(req[96:98], opts.Port)

View file

@ -225,6 +225,26 @@ type LogoEngine struct {
PhaseTime float64
TargetNodeQueue []int
SpawnAccumulator float64
Grid [][]string
// Pre-baked frames
frames []string
frameIndex int
}
func (e *LogoEngine) ensureGrid(w, h int) {
if len(e.Grid) < h {
newGrid := make([][]string, h)
copy(newGrid, e.Grid)
e.Grid = newGrid
}
for y := 0; y < h; y++ {
if len(e.Grid[y]) < w {
newRow := make([]string, w)
copy(newRow, e.Grid[y])
e.Grid[y] = newRow
}
}
}
func NewLogoEngine(width, height int, offsetX, offsetY float64) *LogoEngine {
@ -236,9 +256,138 @@ func NewLogoEngine(width, height int, offsetX, offsetY float64) *LogoEngine {
LastUpdate: time.Now(),
}
e.startAssemblePhase(e.LastUpdate)
e.bakeFrames()
return e
}
const logoBakeFrames = 180 // ~18 секунд полного цикла при 10 FPS
func (e *LogoEngine) bakeFrames() {
now := time.Unix(0, 0)
e.frames = make([]string, logoBakeFrames)
for f := 0; f < logoBakeFrames; f++ {
e.simulateStep(now, 0.1)
now = now.Add(100 * time.Millisecond)
e.frames[f] = e.renderNow(now)
}
}
func (e *LogoEngine) simulateStep(now time.Time, dt float64) {
e.PhaseTime += dt
switch e.Phase {
case logoPhaseAssemble:
spawnRate := float64(len(e.Nodes)) / 2.0
e.SpawnAccumulator += spawnRate * dt
spawnCount := int(e.SpawnAccumulator)
e.SpawnAccumulator -= float64(spawnCount)
for i := 0; i < spawnCount && len(e.TargetNodeQueue) > 0; i++ {
targetID := e.TargetNodeQueue[0]
e.TargetNodeQueue = e.TargetNodeQueue[1:]
e.spawnDownloadParticleTo(e.Nodes[targetID])
}
activeDL := 0
for _, p := range e.Particles {
if p.Type == logoTypeDownload && p.Active {
activeDL++
}
}
if len(e.TargetNodeQueue) == 0 && activeDL == 0 {
e.Phase = logoPhaseVerify
e.PhaseTime = 0
for _, n := range e.Nodes {
if !n.IsOutline {
n.setState(logoStateVerifying, now)
}
}
}
case logoPhaseVerify:
if e.PhaseTime >= 3.0 {
e.Phase = logoPhaseDisperse
e.PhaseTime = 0
for _, n := range e.Nodes {
if !n.IsOutline {
n.setState(logoStateCompleted, now)
e.spawnUploadParticle(n)
}
}
}
case logoPhaseDisperse:
if e.PhaseTime >= 1.0 {
for _, n := range e.Nodes {
if !n.IsOutline {
n.setState(logoStateFading, now)
}
}
e.Phase = logoPhaseWait
e.PhaseTime = 0
}
case logoPhaseWait:
if e.PhaseTime >= 2.0 {
e.startAssemblePhase(now)
}
}
activeParticles := e.Particles[:0]
for _, p := range e.Particles {
p.Update(dt)
if !p.Active && p.Type == logoTypeDownload {
if p.TargetID >= 0 && p.TargetID < len(e.Nodes) {
node := e.Nodes[p.TargetID]
if node.State == logoStateIdle {
node.setState(logoStateDownloading, now)
}
}
}
if p.Active {
activeParticles = append(activeParticles, p)
}
}
e.Particles = activeParticles
}
func (e *LogoEngine) renderNow(now time.Time) string {
e.ensureGrid(e.Width, e.Height)
for y := 0; y < e.Height; y++ {
for x := 0; x < e.Width; x++ {
e.Grid[y][x] = " "
}
}
for _, n := range e.Nodes {
if !n.IsOutline {
x, y := int(n.X+0.5), int(n.Y+0.5)
if x >= 0 && x < e.Width && y >= 0 && y < e.Height {
char, style := n.Render(now)
e.Grid[y][x] = style.Render(char)
}
}
}
for _, p := range e.Particles {
x, y := int(p.X+0.5), int(p.Y+0.5)
if x >= 0 && x < e.Width && y >= 0 && y < e.Height {
char, style := p.Render()
e.Grid[y][x] = style.Render(char)
}
}
for _, n := range e.Nodes {
if n.IsOutline {
x, y := int(n.X+0.5), int(n.Y+0.5)
if x >= 0 && x < e.Width && y >= 0 && y < e.Height {
char, style := n.Render(now)
e.Grid[y][x] = style.Render(char)
}
}
}
var sb strings.Builder
sb.Grow(e.Height * e.Width * 5)
for y := 0; y < e.Height; y++ {
sb.WriteString(strings.Join(e.Grid[y][:e.Width], ""))
if y < e.Height-1 {
sb.WriteString("\n")
}
}
return sb.String()
}
func (e *LogoEngine) startAssemblePhase(now time.Time) {
e.Phase = logoPhaseAssemble
e.PhaseTime = 0
@ -257,96 +406,11 @@ func (e *LogoEngine) startAssemblePhase(now time.Time) {
})
}
// Update переключает кадр — нулевой CPU.
func (e *LogoEngine) Update(now time.Time) {
if e.LastUpdate.IsZero() {
e.LastUpdate = now
if len(e.frames) > 0 {
e.frameIndex = (e.frameIndex + 1) % len(e.frames)
}
dt := now.Sub(e.LastUpdate).Seconds()
if dt > 0.1 {
dt = 0.1
}
e.LastUpdate = now
e.PhaseTime += dt
switch e.Phase {
case logoPhaseAssemble:
spawnRate := float64(len(e.Nodes)) / 2.0 // Заполнение за ~2 секунды
e.SpawnAccumulator += spawnRate * dt
spawnCount := int(e.SpawnAccumulator)
e.SpawnAccumulator -= float64(spawnCount)
for i := 0; i < spawnCount && len(e.TargetNodeQueue) > 0; i++ {
targetID := e.TargetNodeQueue[0]
e.TargetNodeQueue = e.TargetNodeQueue[1:]
e.spawnDownloadParticleTo(e.Nodes[targetID])
}
activeDL := 0
for _, p := range e.Particles {
if p.Type == logoTypeDownload && p.Active {
activeDL++
}
}
if len(e.TargetNodeQueue) == 0 && activeDL == 0 {
e.Phase = logoPhaseVerify
e.PhaseTime = 0
for _, n := range e.Nodes {
if !n.IsOutline {
n.setState(logoStateVerifying, now)
}
}
}
case logoPhaseVerify:
if e.PhaseTime >= 3.0 {
e.Phase = logoPhaseDisperse
e.PhaseTime = 0
for _, n := range e.Nodes {
if !n.IsOutline {
n.setState(logoStateCompleted, now)
e.spawnUploadParticle(n)
}
}
}
case logoPhaseDisperse:
if e.PhaseTime >= 1.0 {
for _, n := range e.Nodes {
if !n.IsOutline {
n.setState(logoStateFading, now)
}
}
e.Phase = logoPhaseWait
e.PhaseTime = 0
}
case logoPhaseWait:
if e.PhaseTime >= 2.0 {
e.startAssemblePhase(now)
}
}
activeParticles := make([]*logoParticle, 0)
for _, p := range e.Particles {
p.Update(dt)
if !p.Active && p.Type == logoTypeDownload {
if p.TargetID >= 0 && p.TargetID < len(e.Nodes) {
node := e.Nodes[p.TargetID]
if node.State == logoStateIdle {
node.setState(logoStateDownloading, now)
}
}
}
if p.Active {
activeParticles = append(activeParticles, p)
}
}
e.Particles = activeParticles
}
func (e *LogoEngine) spawnDownloadParticleTo(target *logoNode) {
@ -371,51 +435,12 @@ func (e *LogoEngine) spawnUploadParticle(n *logoNode) {
e.Particles = append(e.Particles, NewLogoParticle(logoTypeUpload, n.X, n.Y, targetX, targetY, -1))
}
// Render возвращает pre-baked кадр — нулевой CPU.
func (e *LogoEngine) Render(now time.Time) string {
grid := make([][]string, e.Height)
for y := 0; y < e.Height; y++ {
grid[y] = make([]string, e.Width)
for x := 0; x < e.Width; x++ {
grid[y][x] = " "
if len(e.frames) == 0 {
return ""
}
}
for _, n := range e.Nodes {
if !n.IsOutline {
x, y := int(n.X+0.5), int(n.Y+0.5)
if x >= 0 && x < e.Width && y >= 0 && y < e.Height {
char, style := n.Render(now)
grid[y][x] = style.Render(char)
}
}
}
for _, p := range e.Particles {
x, y := int(p.X+0.5), int(p.Y+0.5)
if x >= 0 && x < e.Width && y >= 0 && y < e.Height {
char, style := p.Render()
grid[y][x] = style.Render(char)
}
}
for _, n := range e.Nodes {
if n.IsOutline {
x, y := int(n.X+0.5), int(n.Y+0.5)
if x >= 0 && x < e.Width && y >= 0 && y < e.Height {
char, style := n.Render(now)
grid[y][x] = style.Render(char)
}
}
}
var sb strings.Builder
for y := 0; y < e.Height; y++ {
sb.WriteString(strings.Join(grid[y], ""))
if y < e.Height-1 {
sb.WriteString("\n")
}
}
return sb.String()
return e.frames[e.frameIndex]
}
// generateZNodes генерирует координаты для Z-образного логотипа.

175
ui/matrix.go Normal file
View file

@ -0,0 +1,175 @@
package ui
import (
"fmt"
"math/rand"
"strings"
"github.com/charmbracelet/lipgloss"
)
type matrixDrop struct {
X int
Y float64
Speed float64
Length int
CharType int // 0 = Z, 1 = V, 2 = O
}
// MatrixEngine рендерит кадры заранее и хранит их как 2D-сетки ячеек.
// RenderRegion вырезает любой прямоугольник напрямую — без ANSI-парсинга.
type MatrixEngine struct {
Width int
Height int
// frames[f] — плоский массив len=Width*Height, доступ: cell = frames[f][y*Width+x]
frames [][]string
frameIndex int
}
const matrixNumFrames = 120
func NewMatrixEngine(width, height int) *MatrixEngine {
e := &MatrixEngine{
Width: width,
Height: height,
}
e.bakeFrames()
return e
}
func (e *MatrixEngine) bakeFrames() {
drops := []*matrixDrop{}
step := func(dt float64) {
for _, d := range drops {
d.Y += d.Speed * dt
}
var active []*matrixDrop
for _, d := range drops {
if int(d.Y)-d.Length < e.Height {
active = append(active, d)
}
}
drops = active
targetDrops := e.Width / 6
if targetDrops < 1 {
targetDrops = 1
}
if len(drops) < targetDrops && rand.Float64() < 0.15 {
drops = append(drops, &matrixDrop{
X: rand.Intn(e.Width),
Y: float64(rand.Intn(5)) - 5.0,
Speed: 3.0 + rand.Float64()*4.0,
Length: 4 + rand.Intn(5),
CharType: rand.Intn(3),
})
}
}
renderGrid := func() []string {
grid := make([]string, e.Width*e.Height)
for i := range grid {
grid[i] = " "
}
for _, d := range drops {
headY := int(d.Y)
for i := 0; i < d.Length; i++ {
y := headY - i
x := d.X
if y >= 0 && y < e.Height && x >= 0 && x < e.Width {
opacity := 1.0 - float64(i)/float64(d.Length)
var r, g, b int
if i == 0 {
r, g, b = 255, 255, 255
} else if d.CharType == 0 {
g = int(255.0 * opacity)
} else if d.CharType == 1 {
r = int(255.0 * opacity)
} else {
r = int(255.0 * opacity)
g = int(255.0 * opacity)
}
char := "Z"
if d.CharType == 1 {
char = "V"
} else if d.CharType == 2 {
char = "O"
}
grid[y*e.Width+x] = getMatrixStyledChar(char, r, g, b, i == 0)
}
}
}
return grid
}
// Прогрев: заполняем экран каплями
for i := 0; i < 100; i++ {
step(0.1)
}
e.frames = make([][]string, matrixNumFrames)
for f := 0; f < matrixNumFrames; f++ {
// 5 шагов симуляции на кадр (100ms / 5 = 20ms шаг)
for s := 0; s < 5; s++ {
step(0.02)
}
e.frames[f] = renderGrid()
}
}
// Update переключает кадр — нулевой CPU.
func (e *MatrixEngine) Update() {
e.frameIndex = (e.frameIndex + 1) % len(e.frames)
}
// RenderRegion вырезает прямоугольник [rx,ry,rw,rh] из текущего кадра.
// Работает напрямую с ячейками — нет ANSI-парсинга, нет аллокаций.
func (e *MatrixEngine) RenderRegion(rx, ry, rw, rh int) string {
if len(e.frames) == 0 || rw <= 0 || rh <= 0 {
return strings.Repeat(" \n", rh)
}
frame := e.frames[e.frameIndex]
var sb strings.Builder
sb.Grow(rh * rw * 8)
for row := 0; row < rh; row++ {
y := ry + row
for col := 0; col < rw; col++ {
x := rx + col
if y >= 0 && y < e.Height && x >= 0 && x < e.Width {
sb.WriteString(frame[y*e.Width+x])
} else {
sb.WriteString(" ")
}
}
if row < rh-1 {
sb.WriteString("\n")
}
}
return sb.String()
}
// CurrentFrame — весь кадр целиком (для совместимости).
func (e *MatrixEngine) CurrentFrame() string {
return e.RenderRegion(0, 0, e.Width, e.Height)
}
var matrixStyleCache = make(map[string]string)
func getMatrixStyledChar(char string, r, g, b int, bold bool) string {
hex := fmt.Sprintf("#%02x%02x%02x", r, g, b)
key := char + hex
if bold {
key += "B"
}
if val, ok := matrixStyleCache[key]; ok {
return val
}
style := lipgloss.NewStyle().Foreground(lipgloss.Color(hex))
if bold {
style = style.Bold(true)
}
res := style.Render(char)
matrixStyleCache[key] = res
return res
}

View file

@ -1,6 +1,10 @@
package ui
import "github.com/charmbracelet/lipgloss"
import (
"strings"
"github.com/charmbracelet/lipgloss"
)
// Color definitions (Hex palette)
const (
@ -14,6 +18,7 @@ const (
ColorGreen = "#3cd58c" // Vibrant green (completed)
ColorYellow = "#f4b942" // Amber/yellow (downloading)
ColorRed = "#ff5555" // Bright red (error/missing)
ColorBlue = "#00aeff" // Vibrant blue (dht)
)
var (
@ -129,4 +134,132 @@ var (
StylePieceCompleted = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorGreen))
StylePieceDownloading = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorYellow))
StylePieceMissing = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorBorder))
// ── Кэшированные стили для горячих путей рендеринга ──────────────────────
// Dashboard header
StyleHeaderBrand = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorPink)).Bold(true)
StyleHeaderSep = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorBorder))
StyleHeaderName = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorText))
StyleHeaderIcon = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued))
StyleHeaderLine = lipgloss.NewStyle().
Border(lipgloss.NormalBorder(), false, false, true, false).
BorderForeground(lipgloss.Color(ColorBorder)).
PaddingBottom(0)
// Phase status styles
StylePhaseStopped = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued)).Bold(true)
StylePhaseFailed = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorRed)).Bold(true)
StylePhaseSeeding = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorGreen)).Bold(true)
StylePhaseDownload = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorCyan)).Bold(true)
StylePhaseDefault = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorYellow)).Bold(true)
// KPI cards
StyleKPIProgress = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color(ColorCyan))
StyleKPIDown = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color(ColorGreen))
StyleKPIUp = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color(ColorYellow))
StyleKPIEta = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color(ColorPink))
StyleKPIPeers = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color(ColorText))
StyleKPIBarFill = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorCyan))
StyleKPIBarEmpty = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorBorder))
// Menu styles
StyleMenuText = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorText))
StyleMenuDesc = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued))
StyleMenuActive = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorCyan)).Bold(true)
StyleMenuCursor = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorPink))
StyleMenuHelp = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued)).MarginTop(2)
StyleMenuBorder = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorBorder))
StyleMenuPadding = lipgloss.NewStyle().Padding(2, 4)
StyleMenuBox = lipgloss.NewStyle().Padding(2, 0)
// Log overlay styles
StyleLogTime = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued))
StyleLogEngine = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorPink)).Bold(true)
StyleLogDHT = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorBlue)).Bold(true)
StyleLogPeer = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorGreen)).Bold(true)
StyleLogSys = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorCyan)).Bold(true)
StyleLogDefault = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorText)).Bold(true)
StyleLogError = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorRed)).Bold(true)
StyleLogWarn = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorYellow)).Bold(true)
StyleLogInfo = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorCyan))
StyleLogDebug = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued))
StyleLogMsg = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorText))
StyleLogMsgError = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorRed))
// Overview styles
StyleSectionTitle = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color(ColorCyan))
StyleDim = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued))
StyleVal = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorText))
StyleSpeedDown = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorGreen)).Bold(true)
StyleSpeedUp = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorYellow)).Bold(true)
StyleETA = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorPink)).Bold(true)
StyleErrText = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorRed)).Bold(true)
// HelpBar
StyleHelpKey = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorPink)).Bold(true)
StyleHelpDesc = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued))
StyleHelpSep = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorBorder))
StyleHelpBar = lipgloss.NewStyle().
Foreground(lipgloss.Color(ColorSubdued)).
Border(lipgloss.NormalBorder(), true, false, false, false).
BorderForeground(lipgloss.Color(ColorBorder)).
MarginTop(1).
PaddingTop(0)
// Logo
StyleLogoTorrent = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorText)).Bold(true)
StyleLogoZ = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorGreen)).Bold(true)
// Settings indicator
StyleSettingsCursor = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorPink))
StyleSettingsActive = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorCyan)).Bold(true)
StyleSettingsLabel = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorSubdued)).Width(26)
// Additional menu
StyleAdditCyan = lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color(ColorCyan))
StyleAdditPink = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorPink)).Bold(true)
StyleAdditVal = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorText))
StyleAdditShort = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorYellow))
)
// Статичные строки, вычисляемые один раз
var (
// Статичный "torrent" ASCII-арт (не меняется никогда)
cachedTorrentArt = StyleLogoTorrent.Render(strings.Join([]string{
"████████╗ ██████╗ ██████╗ ██████╗ ███████╗███╗ ██╗████████╗",
"╚══██╔══╝██╔═══██╗██╔══██╗██╔══██╗██╔════╝████╗ ██║╚══██╔══╝",
" ██║ ██║ ██║██████╔╝██████╔╝█████╗ ██╔██╗ ██║ ██║ ",
" ██║ ██║ ██║██╔══██╗██╔══██╗██╔══╝ ██║╚████║ ██║ ",
" ██║ ╚██████╔╝██║ ██║██║ ██║███████╗██║ ╚███║ ██║ ",
" ╚═╝ ╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝╚═╝ ╚══╝ ╚═╝ ",
}, "\n"))
// Статичный Z-логотип (при отключённых анимациях)
cachedStaticZArt = StyleLogoZ.Render(strings.Join([]string{
" ",
" ",
"███████╗",
"╚════██║",
" ██╔╝",
" ██╔╝ ",
" ██╔╝ ",
" ███████╗",
" ╚══════╝",
" ",
" ",
" ",
}, "\n"))
// Статичный help bar главного меню
cachedMenuHelp = StyleMenuHelp.Render(" ↓↑ — выбор Enter — подтвердить Q — выход")
// Статичные строки рамки меню (зависят от boxWidth=85, не меняются)
cachedMenuTopBorder = StyleMenuBorder.Render("╭" + strings.Repeat("─", 85) + "╮")
cachedMenuBottomBorder = StyleMenuBorder.Render("╰" + strings.Repeat("─", 85) + "╯")
cachedMenuLeftEdge = StyleMenuBorder.Render("│")
cachedMenuRightEdge = StyleMenuBorder.Render("│")
// Help bar dashboard
cachedHelpSep = StyleHelpSep.Render(" │ ")
)

1311
ui/ui.go

File diff suppressed because it is too large Load diff