Refactor engine, improve UI animations, secure Telegram bot, fix seeding bugs

This commit is contained in:
itexpert228 2026-07-19 19:25:02 +03:00
parent 425ced2c45
commit 368bc27eaa
24 changed files with 2506 additions and 1283 deletions

23
.gitignore vendored
View file

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

View file

@ -1,14 +1,39 @@
package main package main
import ( import (
"log"
"net/http"
_ "net/http/pprof"
"os"
"github.com/veggiedefender/torrent-client/internal/app" "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" "github.com/veggiedefender/torrent-client/ui"
) )
func main() { func main() {
go func() {
log.Println(http.ListenAndServe("localhost:6060", nil))
}()
controller := app.NewController() 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/bubbles v1.0.0
github.com/charmbracelet/bubbletea v1.3.10 github.com/charmbracelet/bubbletea v1.3.10
github.com/charmbracelet/lipgloss v1.1.0 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 github.com/jackpal/bencode-go v1.0.2
golang.org/x/time v0.15.0 golang.org/x/time v0.15.0
) )
@ -33,6 +35,7 @@ require (
) )
require ( require (
golang.org/x/sync v0.11.0 // indirect
golang.org/x/sys v0.38.0 // indirect golang.org/x/sys v0.38.0 // indirect
golang.org/x/text v0.22.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/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 h1:Y/CXytFA4m6baUTXGLOoWe4PQhGxaX0KpnayAqC48p4=
github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f/go.mod h1:vw97MGsxSvLiUE2X8qFplwetxpGLQrlU1Q9AUEIzCaM= 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 h1:LcCNfZ344u0LpBPOZNjpCLps/wUOuN4r87Fy9+5yU8g=
github.com/jackpal/bencode-go v1.0.2/go.mod h1:6jI9mUjO3GQbZti3JizEfxTzRfWOM8oBBcwbwlTfceI= github.com/jackpal/bencode-go v1.0.2/go.mod h1:6jI9mUjO3GQbZti3JizEfxTzRfWOM8oBBcwbwlTfceI=
github.com/lucasb-eyer/go-colorful v1.3.0 h1:2/yBRLdWBZKrf7gB40FoiKfAWYQ0lqNcbuQwVHXptag= 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= 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 h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI=
golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo= 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.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.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=

View file

@ -3,21 +3,22 @@ package app
import ( import (
"context" "context"
"encoding/hex" "encoding/hex"
"sync"
"time" "time"
"github.com/veggiedefender/torrent-client/internal/config"
"github.com/veggiedefender/torrent-client/internal/history" "github.com/veggiedefender/torrent-client/internal/history"
"github.com/veggiedefender/torrent-client/internal/stream" "github.com/veggiedefender/torrent-client/internal/tgbot"
"github.com/veggiedefender/torrent-client/internal/torrent" "github.com/veggiedefender/torrent-client/internal/torrent"
) )
type Controller struct { type Controller struct {
mu sync.Mutex
engine *torrent.Engine engine *torrent.Engine
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
currentPath string currentPath string
currentOut string currentOut string
streamServer *stream.Server
} }
func NewController() *Controller { func NewController() *Controller {
@ -28,18 +29,18 @@ func NewController() *Controller {
engine: engine, engine: engine,
ctx: ctx, ctx: ctx,
cancel: cancel, cancel: cancel,
streamServer: stream.NewServer(engine),
} }
} }
func (c *Controller) StartTorrent(path, outputRoot string) error { func (c *Controller) StartTorrent(path, outputRoot string) error {
// Cancel any previous history updater c.mu.Lock()
if c.cancel != nil { if c.cancel != nil {
c.cancel() c.cancel()
} }
c.ctx, c.cancel = context.WithCancel(context.Background()) c.ctx, c.cancel = context.WithCancel(context.Background())
c.currentPath = path c.currentPath = path
c.currentOut = outputRoot c.currentOut = outputRoot
c.mu.Unlock()
err := c.engine.LoadTorrent(path, outputRoot) err := c.engine.LoadTorrent(path, outputRoot)
if err != nil { if err != nil {
@ -48,26 +49,51 @@ func (c *Controller) StartTorrent(path, outputRoot string) error {
c.updateHistory() c.updateHistory()
go func() { go func(ctx context.Context) {
ticker := time.NewTicker(5 * time.Second) ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop() defer ticker.Stop()
for { for {
select { select {
case <-c.ctx.Done(): case <-ctx.Done():
return return
case <-ticker.C: case <-ticker.C:
c.updateHistory() c.updateHistory()
} }
} }
}() }(c.ctx)
return nil 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() { func (c *Controller) StopTorrent() {
c.mu.Lock()
if c.cancel != nil { if c.cancel != nil {
c.cancel() c.cancel()
} }
c.mu.Unlock()
c.updateHistory() c.updateHistory()
c.engine.Stop() c.engine.Stop()
} }
@ -83,11 +109,16 @@ func (c *Controller) updateHistory() {
hashHex = hex.EncodeToString(status.InfoHash[:]) hashHex = hex.EncodeToString(status.InfoHash[:])
} }
c.mu.Lock()
path := c.currentPath
out := c.currentOut
c.mu.Unlock()
hItem := history.Item{ hItem := history.Item{
InfoHash: hashHex, InfoHash: hashHex,
Name: status.Name, Name: status.Name,
TorrentPath: c.currentPath, TorrentPath: path,
OutputDir: c.currentOut, OutputDir: out,
Status: status.Phase, Status: status.Phase,
Progress: c.engine.Progress() * 100, // store as percentage 0-100 Progress: c.engine.Progress() * 100, // store as percentage 0-100
Size: status.TotalBytes, Size: status.TotalBytes,
@ -100,17 +131,14 @@ func (c *Controller) Progress() float64 {
} }
func (c *Controller) Status() torrent.Status { func (c *Controller) Status() torrent.Status {
if c.engine == nil {
return torrent.Status{}
}
return c.engine.Status() return c.engine.Status()
} }
func (c *Controller) SetSequentialMode(mode bool) { func (c *Controller) ToggleFilePriority(fileIdx int) {
c.engine.SetSequentialMode(mode) if c.engine != nil {
c.engine.ToggleFilePriority(fileIdx)
} }
func (c *Controller) StartStreamServer() {
c.streamServer.Start()
}
func (c *Controller) StopStreamServer() {
c.streamServer.Stop()
} }

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" "crypto/rand"
"encoding/hex" "encoding/hex"
"fmt" "fmt"
"log"
"net" "net"
"strings" "strings"
"sync" "sync"
"time" "time"
"github.com/veggiedefender/torrent-client/internal/logger"
"github.com/veggiedefender/torrent-client/internal/tracker" "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.readLoop(ctx)
go s.bootstrap(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 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) { 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) tidBytes := make([]byte, 2)
rand.Read(tidBytes) 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[:]) a["id"] = string(s.ID[:])
msg := NewQuery(tid, q, a) 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 return Msg{}, err
} }
ch := make(chan Msg, 1)
s.transactionsMu.Lock()
s.transactions[tid] = ch
s.transactionsMu.Unlock()
defer func() { defer func() {
s.transactionsMu.Lock() s.transactionsMu.Lock()
delete(s.transactions, tid) delete(s.transactions, tid)
@ -233,20 +238,20 @@ func (s *Server) bootstrap(ctx context.Context) {
"target": string(s.ID[:]), "target": string(s.ID[:]),
}) })
if err != nil { if err != nil {
log.Printf("DHT bootstrap %s failed: %v", name, err) logger.Info("DHT", "DHT bootstrap %s failed: %v", name, err)
return return
} }
if resp.R != nil { if resp.R != nil {
if nodesStr, ok := resp.R["nodes"].(string); ok { if nodesStr, ok := resp.R["nodes"].(string); ok {
count := len(nodesStr) / 26 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) s.parseAndAddNodes(nodesStr)
} }
} }
}(addr, addrStr) }(addr, addrStr)
} }
wg.Wait() 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. // SearchForPeers непрерывно ищет пиров для данного info_hash.
@ -259,7 +264,7 @@ func (s *Server) SearchForPeers(ctx context.Context, infoHash [20]byte) {
} }
time.Sleep(300 * time.Millisecond) 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) targetID := NodeID(infoHash)
queried := make(map[string]bool) // ключ = IP:port строка 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 { if len(toQuery) == 0 {
// Нет новых нод — сбрасываем карту уже запрошенных и пробуем снова logger.Info("DHT", "DHT: no new nodes to query (%d total queried), waiting before retry", len(queried))
// (новые ноды могли добавиться в routing table)
log.Printf("DHT: no new nodes to query (%d total queried), resetting and retrying", len(queried))
queried = make(map[string]bool) queried = make(map[string]bool)
// Если routing table совсем пуста — делаем повторный bootstrap
if s.routingTable.Len() == 0 { if s.routingTable.Len() == 0 {
go s.bootstrap(ctx) go s.bootstrap(ctx)
} }
select { select {
case <-ctx.Done(): case <-ctx.Done():
return return
case <-time.After(5 * time.Second): case <-time.After(30 * time.Second):
} }
continue continue
} }
var wg sync.WaitGroup
for _, n := range toQuery { for _, n := range toQuery {
wg.Add(1)
go func(node Node) { go func(node Node) {
defer wg.Done()
qctx, cancel := context.WithTimeout(ctx, 5*time.Second) qctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel() defer cancel()
resp, err := s.sendQuery(qctx, node.Addr, "get_peers", map[string]interface{}{ 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 { 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 { select {
case s.PeersFound <- peers: case s.PeersFound <- peers:
default: case <-ctx.Done():
// канал полный — пытаемся без блокировки return
case <-time.After(2 * time.Second):
} }
} }
} }
}(n) }(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[:]) 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 { type RoutingTable struct {
mu sync.RWMutex mu sync.RWMutex
ownID NodeID ownID NodeID
nodes []Node buckets [160]*Bucket
} }
// NewRoutingTable creates a new routing table. // NewRoutingTable creates a new routing table.
func NewRoutingTable(ownID NodeID) *RoutingTable { func NewRoutingTable(ownID NodeID) *RoutingTable {
return &RoutingTable{ rt := &RoutingTable{
ownID: ownID, 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. // AddNode adds a node to the routing table or updates it.
func (rt *RoutingTable) AddNode(n Node) { func (rt *RoutingTable) AddNode(n Node) {
idx := rt.bucketIndex(n.ID)
if idx == -1 {
return // Do not add ourselves
}
rt.mu.Lock() rt.mu.Lock()
defer rt.mu.Unlock() defer rt.mu.Unlock()
if n.ID == rt.ownID { bucket := rt.buckets[idx]
return
}
for i, existing := range rt.nodes { // Check if already exists and update
for i, existing := range bucket.nodes {
if existing.ID == n.ID { 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 return
} }
} }
rt.nodes = append(rt.nodes, n) // Add new node if bucket is not full
if len(bucket.nodes) < MaxNodes {
// Sort by distance to our own ID and keep top 1000 nodes for simplicity. bucket.nodes = append(bucket.nodes, n)
sort.Slice(rt.nodes, func(i, j int) bool { } else {
distI := Distance(rt.nodes[i].ID, rt.ownID) // In a full Kademlia implementation, we would ping the oldest node
distJ := Distance(rt.nodes[j].ID, rt.ownID) // and replace it if it doesn't respond. For now, just drop the new one.
return distI.Cmp(distJ) < 0
})
if len(rt.nodes) > 1000 {
rt.nodes = rt.nodes[:1000]
} }
} }
@ -83,26 +111,43 @@ func (rt *RoutingTable) ClosestNodes(target NodeID, count int) []Node {
rt.mu.RLock() rt.mu.RLock()
defer rt.mu.RUnlock() defer rt.mu.RUnlock()
// Since we don't have true k-buckets, we sort the entire list. var allNodes []Node
// 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)
sort.Slice(sortedNodes, func(i, j int) bool { // Fast path: find the target's bucket
distI := Distance(sortedNodes[i].ID, target) idx := rt.bucketIndex(target)
distJ := Distance(sortedNodes[j].ID, 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 return distI.Cmp(distJ) < 0
}) })
if len(sortedNodes) > count { if len(allNodes) > count {
return sortedNodes[: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 { func (rt *RoutingTable) Len() int {
rt.mu.RLock() rt.mu.RLock()
defer rt.mu.RUnlock() 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" "io"
"os" "os"
"path/filepath" "path/filepath"
"sync"
"github.com/veggiedefender/torrent-client/internal/torrentfile" "github.com/veggiedefender/torrent-client/internal/torrentfile"
) )
@ -15,6 +16,9 @@ type PieceReader struct {
outputRoot string outputRoot string
pieceLength int pieceLength int
totalLength int totalLength int
mu sync.Mutex
fdMap map[string]*os.File
} }
// NewPieceReader creates a new PieceReader. // NewPieceReader creates a new PieceReader.
@ -24,9 +28,36 @@ func NewPieceReader(files []torrentfile.File, pieceLength, totalLength int, outp
outputRoot: outputRoot, outputRoot: outputRoot,
pieceLength: pieceLength, pieceLength: pieceLength,
totalLength: totalLength, 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. // ReadBlock reads a specific block of data from the files.
func (pr *PieceReader) ReadBlock(pieceIndex, begin, length int) ([]byte, error) { func (pr *PieceReader) ReadBlock(pieceIndex, begin, length int) ([]byte, error) {
absoluteOffset := pieceIndex*pr.pieceLength + begin 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) filePath := filepath.Join(pr.outputRoot, file.Path)
f, err := os.Open(filePath) f, err := pr.getFile(filePath)
if err != nil { if err != nil {
return nil, err return nil, err
} }
_, err = f.ReadAt(buf[bytesRead:bytesRead+readLength], int64(fileOffset)) n, err := f.ReadAt(buf[bytesRead:bytesRead+readLength], int64(fileOffset))
f.Close()
if err != nil && err != io.EOF { if err != nil && err != io.EOF {
return nil, err return nil, err
} }
bytesRead += readLength bytesRead += n
if n < readLength {
break // EOF reached prematurely
}
currentOffset += file.Length currentOffset += file.Length
} }

View file

@ -1,142 +0,0 @@
package stream
import (
"context"
"fmt"
"io"
"log"
"net/http"
"os"
"time"
"github.com/veggiedefender/torrent-client/internal/torrent"
)
// Server handles HTTP streaming for a torrent
type Server struct {
server *http.Server
engine *torrent.Engine
}
func NewServer(engine *torrent.Engine) *Server {
s := &Server{
engine: engine,
}
mux := http.NewServeMux()
mux.HandleFunc("/stream", s.handleStream)
s.server = &http.Server{
Addr: ":8080",
Handler: mux,
}
return s
}
func (s *Server) Start() {
go func() {
log.Println("Starting stream server on http://localhost:8080/stream")
if err := s.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Printf("Stream server error: %v", err)
}
}()
}
func (s *Server) Stop() {
if s.server != nil {
s.server.Shutdown(context.Background())
}
}
func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) {
status := s.engine.Status()
if !status.Loaded {
http.Error(w, "Torrent not loaded", http.StatusNotFound)
return
}
partPath := s.engine.PartFilePath()
if partPath == "" {
http.Error(w, "Part file not ready", http.StatusServiceUnavailable)
return
}
file, err := os.Open(partPath)
if err != nil {
http.Error(w, "Failed to open file", http.StatusInternalServerError)
return
}
// Note: We don't defer file.Close() here because ServeContent uses the Seeker asynchronously,
// wait, http.ServeContent does NOT close it, but it finishes before returning!
// So we MUST defer file.Close().
defer file.Close()
reader := &streamReader{
file: file,
engine: s.engine,
pieceLength: status.PieceLength,
length: int64(status.Length),
offset: 0,
}
w.Header().Set("Content-Type", "video/mp4") // TODO: detect properly, default to mp4
w.Header().Set("Accept-Ranges", "bytes")
http.ServeContent(w, r, status.Name, time.Now(), reader)
}
// streamReader wraps a file and blocks until requested pieces are downloaded
type streamReader struct {
file *os.File
engine *torrent.Engine
pieceLength int
length int64
offset int64
}
func (r *streamReader) Read(p []byte) (n int, err error) {
if r.offset >= r.length {
return 0, io.EOF
}
// Calculate which piece we are trying to read
pieceIndex := int(r.offset / int64(r.pieceLength))
// Polling loop to wait for the piece
for !r.engine.HasPiece(pieceIndex) {
status := r.engine.Status()
if status.Phase == "failed" || status.Phase == "stopped" {
return 0, fmt.Errorf("torrent stopped or failed")
}
time.Sleep(200 * time.Millisecond)
}
// Read from the actual file
n, err = r.file.ReadAt(p, r.offset)
if n > 0 {
r.offset += int64(n)
}
return n, err
}
func (r *streamReader) Seek(offset int64, whence int) (int64, error) {
var newOffset int64
switch whence {
case io.SeekStart:
newOffset = offset
case io.SeekCurrent:
newOffset = r.offset + offset
case io.SeekEnd:
newOffset = r.length + offset
default:
return 0, fmt.Errorf("invalid whence: %d", whence)
}
if newOffset < 0 {
return 0, fmt.Errorf("negative offset")
}
r.offset = newOffset
return r.offset, nil
}

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

View file

@ -10,7 +10,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"io" "io"
"log"
mathrand "math/rand" mathrand "math/rand"
"net" "net"
"os" "os"
@ -26,7 +25,10 @@ import (
"golang.org/x/time/rate" "golang.org/x/time/rate"
"github.com/veggiedefender/torrent-client/internal/dht" "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/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/storage"
"github.com/veggiedefender/torrent-client/internal/torrentfile" "github.com/veggiedefender/torrent-client/internal/torrentfile"
"github.com/veggiedefender/torrent-client/internal/tracker" "github.com/veggiedefender/torrent-client/internal/tracker"
@ -42,6 +44,7 @@ type Engine struct {
peerID [20]byte peerID [20]byte
dhtServer *dht.Server dhtServer *dht.Server
portManager *portforward.PortManager
cancel context.CancelFunc cancel context.CancelFunc
@ -297,6 +300,7 @@ func NewEngine() *Engine {
peerID: generatePeerID(), peerID: generatePeerID(),
phase: "idle", phase: "idle",
incomingConns: make(chan net.Conn, 128), incomingConns: make(chan net.Conn, 128),
portManager: portforward.NewPortManager(),
} }
} }
@ -336,7 +340,7 @@ func (e *Engine) LoadTorrent(path, outputRoot string) error {
e.resetStateForNewLoad() e.resetStateForNewLoad()
e.setPhase("loading_metadata") e.setPhase("loading_metadata")
outputRoot = normalizeOutputRoot(outputRoot) 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) debugf("download root selected: %s", outputRoot)
downloadCtx, cancel := context.WithCancel(context.Background()) downloadCtx, cancel := context.WithCancel(context.Background())
@ -349,7 +353,7 @@ func (e *Engine) LoadTorrent(path, outputRoot string) error {
tf, err := torrentfile.Open(path) tf, err := torrentfile.Open(path)
if err != nil { 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.setError(err)
e.setPhase("failed") e.setPhase("failed")
cancel() cancel()
@ -435,18 +439,18 @@ func (e *Engine) downloadMetadataFromPeers(ctx context.Context, infoHash [20]byt
pc, err := newPeerClient(dialCtx, addr, infoHash, e.peerID, 0) pc, err := newPeerClient(dialCtx, addr, infoHash, e.peerID, 0)
cancel() cancel()
if err != nil { if err != nil {
log.Printf("peer dial failed %s: %v", addr, err) logger.Info("ENGINE", "peer dial failed %s: %v", addr, err)
return return
} }
defer pc.Close() defer pc.Close()
if pc.PeerUtMetadataID() == 0 { if pc.PeerUtMetadataID() == 0 {
log.Printf("peer %s: no ut_metadata support", addr) logger.Info("ENGINE", "peer %s: no ut_metadata support", addr)
return return
} }
size := pc.MetadataSize() 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 { if size <= 0 || size > 10*1024*1024 {
return return
} }
@ -542,7 +546,7 @@ func (e *Engine) startTorrent(downloadCtx context.Context, tf *torrentfile.Torre
} }
e.phase = "ready" e.phase = "ready"
e.mu.Unlock() 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) go e.runDownload(downloadCtx, tf)
} }
@ -628,7 +632,7 @@ func (e *Engine) queryTrackers(ctx context.Context, trackersToTry []string, info
go func() { go func() {
defer wg.Done() defer wg.Done()
started := time.Now() 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) perTrackerCtx, cancel := context.WithTimeout(ctx, opts.Timeout)
peers, err := tracker.GetPeersFromURL(perTrackerCtx, announce, infoHash, length, opts) peers, err := tracker.GetPeersFromURL(perTrackerCtx, announce, infoHash, length, opts)
@ -652,7 +656,7 @@ func (e *Engine) queryTrackers(ctx context.Context, trackersToTry []string, info
if res.err != nil { if res.err != nil {
st.State = "error" st.State = "error"
st.Error = res.err.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) statuses = append(statuses, st)
continue continue
} }
@ -664,7 +668,7 @@ func (e *Engine) queryTrackers(ctx context.Context, trackersToTry []string, info
st.PeerCount = len(res.peers) st.PeerCount = len(res.peers)
} }
statuses = append(statuses, st) 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 { for _, peer := range res.peers {
addPeer(peerMap, peer, res.announce) addPeer(peerMap, peer, res.announce)
@ -698,7 +702,7 @@ func (e *Engine) ensureDHTServer(ctx context.Context) {
if err := srv.Start(ctx, dht.Port); err == nil { if err := srv.Start(ctx, dht.Port); err == nil {
e.dhtServer = srv e.dhtServer = srv
} else { } else {
log.Printf("Failed to start DHT server: %v", err) logger.Info("ENGINE", "Failed to start DHT server: %v", err)
} }
} }
@ -730,11 +734,11 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
go e.listenIncoming(ctx, tf) go e.listenIncoming(ctx, tf)
e.setPhase("preparing_download") 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) partPath, partFile, resumedPieces, err := createOrOpenPartFile(tf, e.outputRoot)
if err != nil { 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)) e.setTerminalError("failed", fmt.Errorf("create temp file: %w", err))
return return
} }
@ -743,7 +747,50 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
e.mu.Unlock() e.mu.Unlock()
defer partFile.Close() 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 { if len(resumedPieces) > 0 {
e.mu.Lock() e.mu.Lock()
for _, idx := range resumedPieces { for _, idx := range resumedPieces {
@ -757,7 +804,7 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
e.downloadedBytes = int64(tf.Length) e.downloadedBytes = int64(tf.Length)
} }
e.mu.Unlock() 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) pieceWriter := newBufferedPieceWriter(partFile, tf.PieceLength, 8, 140*time.Millisecond)
@ -830,7 +877,7 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
defer cleanupWorkers() defer cleanupWorkers()
e.setPhase("downloading") 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() lastProgressAt := time.Now()
lastReannounceAt := time.Now() lastReannounceAt := time.Now()
startWorkers(e.snapshotPeers()) startWorkers(e.snapshotPeers())
@ -881,31 +928,39 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
startWorkers(e.snapshotPeers()) startWorkers(e.snapshotPeers())
if time.Since(lastProgressAt) >= downloadStallTimeout { 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))) e.setTerminalError("stalled", fmt.Errorf("download stalled: no piece progress for %s", downloadStallTimeout.Round(time.Second)))
return return
} }
} }
} }
cleanupWorkers() cleanupWorkers()
if err := pieceWriter.Close(); err != nil { 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)) e.setTerminalError("failed", fmt.Errorf("flush buffered pieces: %w", err))
return return
} }
e.setPhase("writing_files") e.setPhase("writing_files")
if err := materializeDownloadedFiles(tf, partPath, e.outputRoot); err != nil { 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)) e.setTerminalError("failed", fmt.Errorf("write output files: %w", err))
return 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) { 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 // Удаляем bitmap вместе с .part
_ = os.Remove(partPath + ".bitmap") _ = os.Remove(partPath + ".bitmap")
@ -924,15 +979,16 @@ func (e *Engine) runDownload(ctx context.Context, tf *torrentfile.TorrentFile) {
e.downloadSpeed = 0 e.downloadSpeed = 0
e.uploadSpeed = 0 e.uploadSpeed = 0
e.mu.Unlock() 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) e.runSeeding(ctx, tf)
} }
func (e *Engine) runSeeding(ctx context.Context, tf *torrentfile.TorrentFile) { 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) pieceReader := storage.NewPieceReader(tf.Files, tf.PieceLength, tf.Length, e.outputRoot)
defer pieceReader.Close()
workerCtx, workerCancel := context.WithCancel(ctx) workerCtx, workerCancel := context.WithCancel(ctx)
defer workerCancel() defer workerCancel()
@ -1074,7 +1130,7 @@ func (e *Engine) runSeedingWorker(ctx context.Context, tf *torrentfile.TorrentFi
return return
} }
msg, err := pc.ReadMessage(ctx) msg, ptr, err := pc.ReadMessage(ctx)
if err != nil { if err != nil {
if isTimeout(err) { if isTimeout(err) {
continue continue
@ -1085,12 +1141,19 @@ func (e *Engine) runSeedingWorker(ctx context.Context, tf *torrentfile.TorrentFi
switch msg.ID { switch msg.ID {
case 6: // msgRequest case 6: // msgRequest
if len(msg.Payload) < 12 { if len(msg.Payload) < 12 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue continue
} }
idx := int(binary.BigEndian.Uint32(msg.Payload[0:4])) idx := int(binary.BigEndian.Uint32(msg.Payload[0:4]))
begin := int(binary.BigEndian.Uint32(msg.Payload[4:8])) begin := int(binary.BigEndian.Uint32(msg.Payload[4:8]))
length := int(binary.BigEndian.Uint32(msg.Payload[8:12])) length := int(binary.BigEndian.Uint32(msg.Payload[8:12]))
if ptr != nil {
wireMsgPool.Put(ptr)
}
if length > 16384*2 { if length > 16384*2 {
return return
} }
@ -1106,6 +1169,9 @@ func (e *Engine) runSeedingWorker(ctx context.Context, tf *torrentfile.TorrentFi
e.addUploadedBytes(int64(length)) e.addUploadedBytes(int64(length))
default: default:
if ptr != nil {
wireMsgPool.Put(ptr)
}
} }
} }
} }
@ -1123,14 +1189,14 @@ func (e *Engine) runPeerWorker(
e.setPeerState(peer.Address, peer.Port, "connecting", "") e.setPeerState(peer.Address, peer.Port, "connecting", "")
peerAddr := net.JoinHostPort(peer.Address, strconv.Itoa(int(peer.Port))) 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) debugf("starting peer worker for %s", peerAddr)
client, err := newPeerClient(ctx, peerAddr, tf.InfoHash, e.peerID, len(tf.PieceHashes)) client, err := newPeerClient(ctx, peerAddr, tf.InfoHash, e.peerID, len(tf.PieceHashes))
if err != nil { if err != nil {
if ctx.Err() != nil { if ctx.Err() != nil {
e.setPeerState(peer.Address, peer.Port, "stopped", "") e.setPeerState(peer.Address, peer.Port, "stopped", "")
} else { } 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()) e.setPeerError(peer.Address, peer.Port, err.Error())
} }
return return
@ -1162,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", "") e.setPeerState(peer.Address, peer.Port, "ready", "")
consecutiveFailures := 0 consecutiveFailures := 0
endgameCancelCh := scheduler.SubscribeCancel(peerKey)
defer scheduler.UnsubscribeCancel(peerKey)
for { for {
if ctx.Err() != nil { if ctx.Err() != nil {
e.setPeerState(peer.Address, peer.Port, "stopped", "") e.setPeerState(peer.Address, peer.Port, "stopped", "")
@ -1188,7 +1257,7 @@ func (e *Engine) runPeerWorker(
if ctx.Err() != nil { if ctx.Err() != nil {
e.setPeerState(peer.Address, peer.Port, "stopped", "") e.setPeerState(peer.Address, peer.Port, "stopped", "")
} else { } 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()) e.setPeerError(peer.Address, peer.Port, err.Error())
} }
return return
@ -1210,10 +1279,31 @@ func (e *Engine) runPeerWorker(
e.setPeerState(peer.Address, peer.Port, "requesting", "") e.setPeerState(peer.Address, peer.Port, "requesting", "")
pieceSize := pieceSizeForIndex(tf, task.Index) 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 err != nil {
if _, reportErr := scheduler.Report(ctx, task.Index, false); reportErr != nil && ctx.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) e.setPieceState(task.Index, PieceMissing)
if ctx.Err() != nil { if ctx.Err() != nil {
@ -1221,8 +1311,13 @@ func (e *Engine) runPeerWorker(
return return
} }
if errors.Is(err, ErrPieceCanceled) {
// Soft cancel: piece was downloaded by another peer, continue.
continue
}
consecutiveFailures++ 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()) e.setPeerError(peer.Address, peer.Port, err.Error())
if shouldDropPeer(err) || consecutiveFailures >= 3 { if shouldDropPeer(err) || consecutiveFailures >= 3 {
return return
@ -1233,11 +1328,11 @@ func (e *Engine) runPeerWorker(
hash := sha1.Sum(pieceData) hash := sha1.Sum(pieceData)
if hash != tf.PieceHashes[task.Index] { if hash != tf.PieceHashes[task.Index] {
if _, reportErr := scheduler.Report(ctx, task.Index, false); reportErr != nil && ctx.Err() == nil { 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) e.setPieceState(task.Index, PieceMissing)
consecutiveFailures++ 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)) e.setPeerError(peer.Address, peer.Port, fmt.Sprintf("piece %d hash mismatch", task.Index))
if consecutiveFailures >= 3 { if consecutiveFailures >= 3 {
return return
@ -1247,10 +1342,10 @@ func (e *Engine) runPeerWorker(
if err := pieceWriter.WritePiece(ctx, task.Index, pieceData); err != nil { if err := pieceWriter.WritePiece(ctx, task.Index, pieceData); err != nil {
if _, reportErr := scheduler.Report(ctx, task.Index, false); reportErr != nil && ctx.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) 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)) e.setPeerError(peer.Address, peer.Port, fmt.Sprintf("write piece %d: %v", task.Index, err))
return return
} }
@ -1406,13 +1501,12 @@ func (e *Engine) refreshPeersFromTrackers(ctx context.Context, tf *torrentfile.T
}) })
// Пересобираем индекс после сортировки // Пересобираем индекс после сортировки
e.peerIdx = buildPeerIdx(e.peers) 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 return added
} }
func (e *Engine) currentDownloadedBytes() int64 { func (e *Engine) currentDownloadedBytes() int64 {
e.mu.RLock() e.mu.RLock()
defer e.mu.RUnlock() defer e.mu.RUnlock()
@ -1467,7 +1561,6 @@ func (e *Engine) recordPieceComplete(address string, port uint16, pieceIndex int
} }
} }
// asyncSaveBitmap сохраняет bitmap в фоне (вызывается из runDownload heartbeat). // asyncSaveBitmap сохраняет bitmap в фоне (вызывается из runDownload heartbeat).
func (e *Engine) asyncSaveBitmap(bitmapPath string) { func (e *Engine) asyncSaveBitmap(bitmapPath string) {
e.mu.RLock() e.mu.RLock()
@ -1482,7 +1575,6 @@ func (e *Engine) asyncSaveBitmap(bitmapPath string) {
go func() { _ = saveBitmap(bitmapPath, completed, total) }() go func() { _ = saveBitmap(bitmapPath, completed, total) }()
} }
// peerByKey возвращает указатель на PeerStatus по ключу (O(1)). // peerByKey возвращает указатель на PeerStatus по ключу (O(1)).
// Вызывать только под e.mu. // Вызывать только под e.mu.
func (e *Engine) peerByKey(key string) *PeerStatus { func (e *Engine) peerByKey(key string) *PeerStatus {
@ -1673,7 +1765,6 @@ func buildPeerIdx(peers []PeerStatus) map[string]int {
return idx return idx
} }
func (e *Engine) updateSpeed(now time.Time) { func (e *Engine) updateSpeed(now time.Time) {
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
@ -1866,11 +1957,11 @@ func createOrOpenPartFile(tf *torrentfile.TorrentFile, outputRoot string) (strin
if bitmapPieces, bErr := loadBitmap(bitmapPath, len(tf.PieceHashes)); bErr == nil && len(bitmapPieces) > 0 { if bitmapPieces, bErr := loadBitmap(bitmapPath, len(tf.PieceHashes)); bErr == nil && len(bitmapPieces) > 0 {
// Быстрый путь: bitmap есть, только верифицируем упомянутые куски // Быстрый путь: bitmap есть, только верифицируем упомянутые куски
completed = verifyPieces(existing, tf, bitmapPieces) 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 { } else {
// Медленный путь: сканируем все куски через SHA-1 // Медленный путь: сканируем все куски через SHA-1
completed = verifyAllPieces(existing, tf) 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 актуальными данными // Перезаписываем bitmap актуальными данными
_ = saveBitmap(bitmapPath, completed, len(tf.PieceHashes)) _ = saveBitmap(bitmapPath, completed, len(tf.PieceHashes))
@ -1964,8 +2055,6 @@ func loadBitmap(path string, expectedTotal int) ([]int, error) {
return data.Completed, nil return data.Completed, nil
} }
func materializeDownloadedFiles(tf *torrentfile.TorrentFile, partPath, outputRoot string) error { func materializeDownloadedFiles(tf *torrentfile.TorrentFile, partPath, outputRoot string) error {
partFile, err := os.Open(partPath) partFile, err := os.Open(partPath)
if err != nil { if err != nil {
@ -2071,10 +2160,14 @@ func (e *Engine) listenIncoming(ctx context.Context, tf *torrentfile.TorrentFile
return return
} }
defer ln.Close() 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() { go func() {
<-ctx.Done() <-ctx.Done()
e.portManager.ClosePort()
ln.Close() ln.Close()
}() }()
@ -2163,7 +2256,7 @@ func (e *Engine) handleIncoming(ctx context.Context, rawConn net.Conn, tf *torre
} }
e.mu.Unlock() 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 // Delegate to the active loop via channel without closing
select { select {
@ -2221,7 +2314,7 @@ func (e *Engine) handleIncomingSeeding(ctx context.Context, tf *torrentfile.Torr
return return
} }
msg, err := pc.ReadMessage(ctx) msg, ptr, err := pc.ReadMessage(ctx)
if err != nil { if err != nil {
if isTimeout(err) { if isTimeout(err) {
continue continue
@ -2232,12 +2325,19 @@ func (e *Engine) handleIncomingSeeding(ctx context.Context, tf *torrentfile.Torr
switch msg.ID { switch msg.ID {
case 6: // msgRequest case 6: // msgRequest
if len(msg.Payload) < 12 { if len(msg.Payload) < 12 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue continue
} }
idx := int(binary.BigEndian.Uint32(msg.Payload[0:4])) idx := int(binary.BigEndian.Uint32(msg.Payload[0:4]))
begin := int(binary.BigEndian.Uint32(msg.Payload[4:8])) begin := int(binary.BigEndian.Uint32(msg.Payload[4:8]))
length := int(binary.BigEndian.Uint32(msg.Payload[8:12])) length := int(binary.BigEndian.Uint32(msg.Payload[8:12]))
if ptr != nil {
wireMsgPool.Put(ptr)
}
if length > 16384*2 { if length > 16384*2 {
return return
} }
@ -2252,6 +2352,10 @@ func (e *Engine) handleIncomingSeeding(ctx context.Context, tf *torrentfile.Torr
} }
e.addUploadedBytes(int64(length)) e.addUploadedBytes(int64(length))
default:
if ptr != nil {
wireMsgPool.Put(ptr)
}
} }
} }
}() }()
@ -2268,7 +2372,7 @@ func (e *Engine) SetSequentialMode(mode bool) {
} }
// HasPiece is a thread-safe check to see if a piece is fully downloaded // HasPiece is a thread-safe check to see if a piece is fully downloaded
func (e *Engine) HasPiece(index int) bool { func (e *Engine) HasCompletedPiece(index int) bool {
e.mu.RLock() e.mu.RLock()
defer e.mu.RUnlock() defer e.mu.RUnlock()
if index < 0 || index >= len(e.pieceStates) { if index < 0 || index >= len(e.pieceStates) {
@ -2277,6 +2381,22 @@ func (e *Engine) HasPiece(index int) bool {
return e.pieceStates[index] == PieceCompleted 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 // PartFilePath returns the path to the current .part file
func (e *Engine) PartFilePath() string { func (e *Engine) PartFilePath() string {
e.mu.RLock() e.mu.RLock()

View file

@ -7,7 +7,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"io" "io"
"log"
"net" "net"
"sync" "sync"
"sync/atomic" "sync/atomic"
@ -16,6 +15,7 @@ import (
"github.com/jackpal/bencode-go" "github.com/jackpal/bencode-go"
"golang.org/x/time/rate" "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/torrentfile"
"github.com/veggiedefender/torrent-client/internal/tracker" "github.com/veggiedefender/torrent-client/internal/tracker"
) )
@ -52,6 +52,14 @@ const (
peerSocketWriteBuffer = 512 * 1024 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 { type extendedHandshake struct {
M map[string]int `bencode:"m"` M map[string]int `bencode:"m"`
MetadataSize int `bencode:"metadata_size"` MetadataSize int `bencode:"metadata_size"`
@ -192,15 +200,14 @@ type peerClient struct {
kaOnce sync.Once kaOnce sync.Once
} }
func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [20]byte, pieceCount int) (*peerClient, error) { func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [20]byte, pieceCount int) (*peerClient, error) {
dialer := net.Dialer{Timeout: peerConnectTimeout} dialer := net.Dialer{Timeout: peerConnectTimeout}
rawConn, err := dialer.DialContext(ctx, "tcp", addr) rawConn, err := dialer.DialContext(ctx, "tcp", addr)
if err != nil { 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 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 { if tcpConn, ok := rawConn.(*net.TCPConn); ok {
_ = tcpConn.SetNoDelay(true) _ = tcpConn.SetNoDelay(true)
@ -229,7 +236,7 @@ func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [
if pc.supportsExtensions { if pc.supportsExtensions {
if err := pc.sendExtendedHandshake(ctx); err != nil { 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) debugf("sent interested to %s", addr)
if err := pc.readInitialMessages(ctx); err != nil { 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() measured.Close()
return nil, err return nil, err
} }
@ -249,7 +256,6 @@ func newPeerClient(ctx context.Context, addr string, infoHash [20]byte, peerID [
return pc, nil return pc, nil
} }
func (pc *peerClient) PeerUtPexID() int { func (pc *peerClient) PeerUtPexID() int {
return pc.peerUtPexID return pc.peerUtPexID
} }
@ -268,7 +274,7 @@ func newIncomingPeerClient(ctx context.Context, rawConn net.Conn, extensions boo
if pc.supportsExtensions { if pc.supportsExtensions {
if err := pc.sendExtendedHandshake(ctx); err != nil { 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) { func (pc *peerClient) PieceAvailability() ([]bool, bool) {
have := make([]bool, len(pc.have)) have := make([]bool, len(pc.have))
copy(have, pc.have) copy(have, pc.have)
return have, pc.hasPieceInfo 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 var transfer pieceTransferStats
if pieceLength <= 0 { if pieceLength <= 0 {
return nil, transfer, fmt.Errorf("invalid piece length %d", pieceLength) 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 блока depth := initPipelineDepth // adaptive, пересчитывается каждые 4 блока
for received < pieceLength { 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 { for len(pending) < depth && offset < pieceLength {
blockLength := requestBlockSize blockLength := requestBlockSize
if remaining := pieceLength - offset; remaining < blockLength { if remaining := pieceLength - offset; remaining < blockLength {
@ -358,22 +376,34 @@ func (pc *peerClient) DownloadPiece(ctx context.Context, pieceIndex int, pieceLe
offset += blockLength offset += blockLength
} }
gotIndex, gotBegin, block, err := pc.readPieceMessage(ctx) gotIndex, gotBegin, block, bufPtr, err := pc.readPieceMessage(ctx)
if err != nil { if err != nil {
return nil, transfer, err return nil, transfer, err
} }
if gotIndex != pieceIndex { if gotIndex != pieceIndex {
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
continue continue
} }
req, ok := pending[gotBegin] req, ok := pending[gotBegin]
if !ok { if !ok {
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
continue continue
} }
if len(block) != req.length { 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) 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 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)) 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) delete(pending, gotBegin)
received += len(block) received += len(block)
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
blocksCompleted++ blocksCompleted++
blockLatency := time.Since(req.requestedAt) blockLatency := time.Since(req.requestedAt)
latencySum += blockLatency latencySum += blockLatency
@ -438,7 +472,6 @@ func (pc *peerClient) sendMessageDirect(msgID int, payload []byte) error {
return err return err
} }
func (pc *peerClient) readInitialMessages(ctx context.Context) error { func (pc *peerClient) readInitialMessages(ctx context.Context) error {
if err := pc.setReadDeadlineFromContext(ctx, 2*time.Second); err != nil { if err := pc.setReadDeadlineFromContext(ctx, 2*time.Second); err != nil {
return err return err
@ -446,7 +479,7 @@ func (pc *peerClient) readInitialMessages(ctx context.Context) error {
defer pc.conn.SetReadDeadline(time.Time{}) defer pc.conn.SetReadDeadline(time.Time{})
for { for {
msg, err := readWireMessage(pc.conn) msg, bufPtr, err := readWireMessage(pc.conn)
if err != nil { if err != nil {
if isTimeout(err) { if isTimeout(err) {
debugf("peer initial message window finished") debugf("peer initial message window finished")
@ -455,6 +488,9 @@ func (pc *peerClient) readInitialMessages(ctx context.Context) error {
return err return err
} }
pc.consumeMessage(msg) 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 { if err := pc.setReadDeadlineFromContext(ctx, peerReadTimeout); err != nil {
return err return err
} }
msg, err := readWireMessage(pc.conn) msg, bufPtr, err := readWireMessage(pc.conn)
if err != nil { if err != nil {
if isTimeout(err) { if isTimeout(err) {
continue continue
@ -483,7 +519,12 @@ func (pc *peerClient) waitForUnchoke(ctx context.Context) error {
return err return err
} }
pc.consumeMessage(msg) pc.consumeMessage(msg)
if msg.ID == msgUnchoke { msgID := msg.ID
if bufPtr != nil {
wireMsgPool.Put(bufPtr)
}
if msgID == msgUnchoke {
debugf("peer unchoked") debugf("peer unchoked")
return nil return nil
} }
@ -499,41 +540,53 @@ func (pc *peerClient) sendRequest(ctx context.Context, pieceIndex, begin, length
return pc.sendMessage(ctx, msgRequest, payload) 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 { for {
if err := ctx.Err(); err != nil { 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 { 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 err != nil {
if isTimeout(err) { if isTimeout(err) {
continue continue
} }
return 0, 0, nil, err return 0, 0, nil, nil, err
} }
switch msg.ID { switch msg.ID {
case msgPiece: case msgPiece:
if len(msg.Payload) < 8 { if len(msg.Payload) < 8 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue continue
} }
gotIndex := int(binary.BigEndian.Uint32(msg.Payload[0:4])) gotIndex := int(binary.BigEndian.Uint32(msg.Payload[0:4]))
gotBegin := int(binary.BigEndian.Uint32(msg.Payload[4:8])) gotBegin := int(binary.BigEndian.Uint32(msg.Payload[4:8]))
data := msg.Payload[8:] data := msg.Payload[8:]
if len(data) == 0 { if len(data) == 0 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue continue
} }
debugf("received piece %d offset %d block=%d", gotIndex, gotBegin, len(data)) 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: case msgChoke:
pc.consumeMessage(msg) 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: default:
pc.consumeMessage(msg) 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()) 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 { if err := pc.setReadDeadlineFromContext(ctx, peerReadTimeout); err != nil {
return wireMessage{}, err return wireMessage{}, nil, err
} }
return readWireMessage(pc.conn) return readWireMessage(pc.conn)
} }
@ -737,34 +790,58 @@ func (pc *peerClient) sendMessage(ctx context.Context, msgID int, payload []byte
} }
length := uint32(1 + len(payload)) 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) binary.BigEndian.PutUint32(buf[0:4], length)
buf[4] = byte(msgID) buf[4] = byte(msgID)
copy(buf[5:], payload) 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 return err
} }
func readWireMessage(r io.Reader) (wireMessage, error) { func readWireMessage(r io.Reader) (wireMessage, *[]byte, error) {
var lengthBuf [4]byte var lengthBuf [4]byte
if _, err := io.ReadFull(r, lengthBuf[:]); err != nil { if _, err := io.ReadFull(r, lengthBuf[:]); err != nil {
return wireMessage{}, err return wireMessage{}, nil, err
} }
length := binary.BigEndian.Uint32(lengthBuf[:]) length := binary.BigEndian.Uint32(lengthBuf[:])
if length == 0 { if length == 0 {
return wireMessage{ID: -1}, nil return wireMessage{ID: -1}, nil, nil
} }
if length > maxWireMessageSize { 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) var msg []byte
if _, err := io.ReadFull(r, msg); err != nil { var bufPtr *[]byte
return wireMessage{}, err
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 { 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 { if err := pc.setReadDeadlineFromContext(ctx, peerReadTimeout); err != nil {
return 0, nil, false, err return 0, nil, false, err
} }
msg, err := readWireMessage(pc.conn) msg, ptr, err := readWireMessage(pc.conn)
if err != nil { if err != nil {
if isTimeout(err) { if isTimeout(err) {
continue continue
@ -831,6 +908,9 @@ func (pc *peerClient) ReadMetadataMessage(ctx context.Context) (piece int, data
} }
if msg.ID == msgExtended { if msg.ID == msgExtended {
if len(msg.Payload) == 0 { if len(msg.Payload) == 0 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue continue
} }
extID := int(msg.Payload[0]) 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:]) reader := bytes.NewReader(msg.Payload[1:])
var dict map[string]int var dict map[string]int
if err := bencode.Unmarshal(reader, &dict); err != nil { if err := bencode.Unmarshal(reader, &dict); err != nil {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue continue
} }
msgType, ok := dict["msg_type"] msgType, ok := dict["msg_type"]
if !ok { if !ok {
if ptr != nil {
wireMsgPool.Put(ptr)
}
continue continue
} }
pieceIdx := dict["piece"] pieceIdx := dict["piece"]
if msgType == 2 { if msgType == 2 {
if ptr != nil {
wireMsgPool.Put(ptr)
}
return pieceIdx, nil, true, nil return pieceIdx, nil, true, nil
} }
if msgType == 1 { if msgType == 1 {
bytesRead := len(msg.Payload[1:]) - reader.Len() 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 return pieceIdx, data, false, nil
} }
} else if extID == 0 { } else if extID == 0 {
@ -860,6 +954,10 @@ func (pc *peerClient) ReadMetadataMessage(ctx context.Context) (piece int, data
} else { } else {
pc.consumeMessage(msg) pc.consumeMessage(msg)
} }
if ptr != nil {
wireMsgPool.Put(ptr)
}
} }
} }

View file

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

View file

@ -153,47 +153,77 @@ func getPeersUDP(ctx context.Context, announceURL *url.URL, infoHash [20]byte, l
if err != nil { if err != nil {
return nil, err return nil, err
} }
connectReq := buildUDPConnectRequest(connectTx) connectReq := buildUDPConnectRequest(connectTx)
var connectionID uint64
var connOK bool
for n := 0; n <= 8; n++ {
if _, err := conn.Write(connectReq[:]); err != nil { if _, err := conn.Write(connectReq[:]); err != nil {
return nil, err return nil, err
} }
conn.SetReadDeadline(time.Now().Add(15 * time.Second * time.Duration(1<<n)))
resp := make([]byte, 65535) resp := make([]byte, 65535)
n, err := conn.Read(resp) nRead, err := conn.Read(resp)
if err != nil { if err == nil {
return nil, err cid, errParse := parseUDPConnectResponse(resp[:nRead], connectTx)
if errParse == nil {
connectionID = cid
connOK = true
break
}
} }
connectionID, err := parseUDPConnectResponse(resp[:n], connectTx) select {
if err != nil { case <-ctx.Done():
return nil, err return nil, ctx.Err()
default:
}
}
if !connOK {
return nil, fmt.Errorf("udp tracker connect timeout")
} }
announceTx, err := randomUint32() announceTx, err := randomUint32()
if err != nil { if err != nil {
return nil, err return nil, err
} }
key, err := randomUint32() key, err := randomUint32()
if err != nil { if err != nil {
return nil, err return nil, err
} }
announceReq, err := buildUDPAnnounceRequest(connectionID, announceTx, infoHash, length, opts, key) announceReq, err := buildUDPAnnounceRequest(connectionID, announceTx, infoHash, length, opts, key)
if err != nil { if err != nil {
return nil, err return nil, err
} }
for n := 0; n <= 8; n++ {
if _, err := conn.Write(announceReq); err != nil { if _, err := conn.Write(announceReq); err != nil {
return nil, err 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 { 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 { 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[56:64], uint64(downloaded))
binary.BigEndian.PutUint64(req[64:72], uint64(left)) binary.BigEndian.PutUint64(req[64:72], uint64(left))
binary.BigEndian.PutUint64(req[72:80], uint64(uploaded)) binary.BigEndian.PutUint64(req[72:80], uint64(uploaded))
binary.BigEndian.PutUint32(req[80:84], 0) binary.BigEndian.PutUint32(req[80:84], 1) // event: 1 (started) instead of 0 (none)
binary.BigEndian.PutUint32(req[84:88], 0) binary.BigEndian.PutUint32(req[84:88], 0) // IP address (default 0)
binary.BigEndian.PutUint32(req[88:92], key) binary.BigEndian.PutUint32(req[88:92], key)
binary.BigEndian.PutUint32(req[92:96], uint32(int32(opts.NumWant))) binary.BigEndian.PutUint32(req[92:96], uint32(int32(opts.NumWant)))
binary.BigEndian.PutUint16(req[96:98], opts.Port) binary.BigEndian.PutUint16(req[96:98], opts.Port)

View file

@ -225,6 +225,26 @@ type LogoEngine struct {
PhaseTime float64 PhaseTime float64
TargetNodeQueue []int TargetNodeQueue []int
SpawnAccumulator float64 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 { 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(), LastUpdate: time.Now(),
} }
e.startAssemblePhase(e.LastUpdate) e.startAssemblePhase(e.LastUpdate)
e.bakeFrames()
return e 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) { func (e *LogoEngine) startAssemblePhase(now time.Time) {
e.Phase = logoPhaseAssemble e.Phase = logoPhaseAssemble
e.PhaseTime = 0 e.PhaseTime = 0
@ -257,96 +406,11 @@ func (e *LogoEngine) startAssemblePhase(now time.Time) {
}) })
} }
// Update переключает кадр — нулевой CPU.
func (e *LogoEngine) Update(now time.Time) { func (e *LogoEngine) Update(now time.Time) {
if e.LastUpdate.IsZero() { if len(e.frames) > 0 {
e.LastUpdate = now 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) { 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)) 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 { func (e *LogoEngine) Render(now time.Time) string {
grid := make([][]string, e.Height) if len(e.frames) == 0 {
for y := 0; y < e.Height; y++ { return ""
grid[y] = make([]string, e.Width)
for x := 0; x < e.Width; x++ {
grid[y][x] = " "
} }
} return e.frames[e.frameIndex]
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()
} }
// generateZNodes генерирует координаты для Z-образного логотипа. // 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 package ui
import "github.com/charmbracelet/lipgloss" import (
"strings"
"github.com/charmbracelet/lipgloss"
)
// Color definitions (Hex palette) // Color definitions (Hex palette)
const ( const (
@ -14,6 +18,7 @@ const (
ColorGreen = "#3cd58c" // Vibrant green (completed) ColorGreen = "#3cd58c" // Vibrant green (completed)
ColorYellow = "#f4b942" // Amber/yellow (downloading) ColorYellow = "#f4b942" // Amber/yellow (downloading)
ColorRed = "#ff5555" // Bright red (error/missing) ColorRed = "#ff5555" // Bright red (error/missing)
ColorBlue = "#00aeff" // Vibrant blue (dht)
) )
var ( var (
@ -129,4 +134,132 @@ var (
StylePieceCompleted = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorGreen)) StylePieceCompleted = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorGreen))
StylePieceDownloading = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorYellow)) StylePieceDownloading = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorYellow))
StylePieceMissing = lipgloss.NewStyle().Foreground(lipgloss.Color(ColorBorder)) 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(" │ ")
) )

1327
ui/ui.go

File diff suppressed because it is too large Load diff