[wip] daemon separation

This commit is contained in:
royalcat 2024-11-24 20:33:44 +03:00
parent 98ee1dc6f1
commit fa084118c3
48 changed files with 48 additions and 35 deletions

View file

@ -0,0 +1,27 @@
package qbittorrent
import (
"context"
"git.kmsign.ru/royalcat/tstor/pkg/qbittorrent"
)
func (d *Daemon) ListTorrents(ctx context.Context) ([]*qbittorrent.TorrentInfo, error) {
return d.client.qb.Torrent().GetTorrents(ctx, &qbittorrent.TorrentOption{})
}
func (d *Daemon) SourceFiles(ctx context.Context, hash string) ([]string, error) {
d.sourceFilesMu.Lock()
defer d.sourceFilesMu.Unlock()
out := make([]string, 0, 1)
for k, h := range d.sourceFiles {
if h != hash {
continue
}
out = append(out, k)
}
return out, nil
}

View file

@ -0,0 +1,66 @@
package qbittorrent
import (
"context"
"errors"
"fmt"
"log/slog"
"os"
"path"
"git.kmsign.ru/royalcat/tstor/pkg/qbittorrent"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
)
func (d *Daemon) Cleanup(ctx context.Context, run bool) ([]string, error) {
d.log.Info(ctx, "cleanup started")
torrentInfos, err := d.client.qb.Torrent().GetTorrents(ctx, &qbittorrent.TorrentOption{})
if err != nil {
d.log.Error(ctx, "failed to get torrents", rlog.Error(err))
return nil, fmt.Errorf("failed to get torrents: %w", err)
}
torrentToDelete := make([]string, 0, 5)
for _, info := range torrentInfos {
if d.registeredTorrents.Contains(info.Hash) {
continue
}
d.log.Info(ctx, "torrent not found in registry", slog.String("infohash", info.Hash))
torrentToDelete = append(torrentToDelete, info.Hash)
}
d.log.Info(ctx, "marked torrents to delete",
slog.Int("count", len(torrentToDelete)),
slog.Any("infohashes", torrentToDelete),
)
if !run {
d.log.Info(ctx, "dry run, skipping deletion")
return torrentToDelete, nil
}
err = d.client.qb.Torrent().DeleteTorrents(ctx, torrentToDelete, true)
if err != nil {
d.log.Error(ctx, "failed to delete torrents", slog.Any("infohashes", torrentToDelete), rlog.Error(err))
return nil, fmt.Errorf("failed to delete torrents: %w", err)
}
d.log.Info(ctx, "torrents deleted from qbittorrent", slog.Int("count", len(torrentToDelete)))
for _, hash := range torrentToDelete {
torrentPath := path.Join(d.dataDir, hash)
_, err := os.Stat(torrentPath)
if errors.Is(err, os.ErrNotExist) {
continue
}
if err != nil {
d.log.Error(ctx, "failed to get torrent path", slog.String("path", torrentPath), rlog.Error(err))
continue
}
d.log.Warn(ctx, "leftover data for torrent detected, cleaning up", slog.String("infohash", hash), slog.String("path", torrentPath))
}
return torrentToDelete, nil
}

View file

@ -0,0 +1,164 @@
package qbittorrent
import (
"context"
"fmt"
"slices"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/qbittorrent"
"github.com/hashicorp/golang-lru/v2/expirable"
"github.com/royalcat/btrgo/btrsync"
)
type cacheClient struct {
qb qbittorrent.Client
propertiesCache *expirable.LRU[string, qbittorrent.TorrentProperties]
torrentsCache *expirable.LRU[string, qbittorrent.TorrentInfo]
pieceCache btrsync.MapOf[pieceKey, int]
}
type pieceKey struct {
hash string
index int
}
func wrapClient(qb qbittorrent.Client) *cacheClient {
const (
cacheSize = 5000
cacheTTL = time.Minute
)
return &cacheClient{
qb: qb,
propertiesCache: expirable.NewLRU[string, qbittorrent.TorrentProperties](cacheSize, nil, cacheTTL),
torrentsCache: expirable.NewLRU[string, qbittorrent.TorrentInfo](cacheSize, nil, cacheTTL),
pieceCache: btrsync.MapOf[pieceKey, int]{},
}
}
func (f *cacheClient) getInfo(ctx context.Context, hash string) (*qbittorrent.TorrentInfo, error) {
if v, ok := f.torrentsCache.Get(hash); ok {
return &v, nil
}
infos, err := f.qb.Torrent().GetTorrents(ctx, &qbittorrent.TorrentOption{
Hashes: []string{hash},
})
if err != nil {
return nil, fmt.Errorf("error to check torrent existence: %w", err)
}
if len(infos) == 0 {
return nil, nil
}
if len(infos) > 1 {
return nil, fmt.Errorf("multiple torrents with the same hash")
}
f.torrentsCache.Add(hash, *infos[0])
return infos[0], nil
}
func (f *cacheClient) getProperties(ctx context.Context, hash string) (*qbittorrent.TorrentProperties, error) {
if v, ok := f.propertiesCache.Get(hash); ok {
return &v, nil
}
info, err := f.qb.Torrent().GetProperties(ctx, hash)
if err != nil {
return nil, err
}
f.propertiesCache.Add(hash, *info)
return info, nil
}
func (f *cacheClient) listContent(ctx context.Context, hash string) ([]*qbittorrent.TorrentContent, error) {
contents, err := f.qb.Torrent().GetContents(ctx, hash)
if err != nil {
return nil, err
}
return contents, nil
}
func (f *cacheClient) getContent(ctx context.Context, hash string, contentIndex int) (*qbittorrent.TorrentContent, error) {
contents, err := f.qb.Torrent().GetContents(ctx, hash, contentIndex)
if err != nil {
return nil, err
}
contentI := slices.IndexFunc(contents, func(c *qbittorrent.TorrentContent) bool {
return c.Index == contentIndex
})
if contentI == -1 {
return nil, fmt.Errorf("content not found")
}
return contents[contentI], nil
}
func (f *cacheClient) isPieceComplete(ctx context.Context, hash string, pieceIndex int) (bool, error) {
cachedPieceState, ok := f.pieceCache.Load(pieceKey{hash: hash, index: pieceIndex})
if ok && cachedPieceState == 2 {
return true, nil
}
completion, err := f.qb.Torrent().GetPiecesStates(ctx, hash)
if err != nil {
return false, err
}
for i, v := range completion {
f.pieceCache.Store(pieceKey{hash: hash, index: i}, v)
}
if completion[pieceIndex] == 2 {
return true, nil
}
return false, nil
}
func (f *cacheClient) waitPieceToComplete(ctx context.Context, hash string, pieceIndex int) error {
const checkingInterval = 1 * time.Second
ok, err := f.isPieceComplete(ctx, hash, pieceIndex)
if err != nil {
return err
}
if ok {
return nil
}
if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) < checkingInterval {
return context.DeadlineExceeded
}
ticker := time.NewTicker(checkingInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
ok, err := f.isPieceComplete(ctx, hash, pieceIndex)
if err != nil {
return err
}
if ok {
return nil
}
if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) < checkingInterval {
return context.DeadlineExceeded
}
}
}
}

View file

@ -0,0 +1,284 @@
package qbittorrent
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"os"
"path"
"path/filepath"
"sync"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/qbittorrent"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/src/config"
"git.kmsign.ru/royalcat/tstor/src/logwrap"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/types/infohash"
infohash_v2 "github.com/anacrolix/torrent/types/infohash-v2"
mapset "github.com/deckarep/golang-set/v2"
"github.com/iceber/iouring-go"
"github.com/royalcat/ctxio"
"go.opentelemetry.io/otel"
)
var trace = otel.Tracer("git.kmsign.ru/royalcat/tstor/daemons/qbittorrent")
type Daemon struct {
proc *os.Process
qb qbittorrent.Client
client *cacheClient
sourceFilesMu sync.Mutex
sourceFiles map[string]string // [sourcePath]infohash
registeredTorrents mapset.Set[string] // infohash list
dataDir string
ur *iouring.IOURing
log *rlog.Logger
}
const defaultConf = `
[LegalNotice]
Accepted=true
[Preferences]
WebUI\LocalHostAuth=false
WebUI\Password_PBKDF2="@ByteArray(qef5I4wZBkDG+PP6/5mQwA==:LoTmorQM/QM5RHI4+dOiu6xfAz9xak6fhR4ZGpRtJF3JNCGG081Yrtva4G71kXz//ODUuWQKTLlrZPuIDvzqUQ==)"
`
func NewDaemon(conf config.QBittorrent) (*Daemon, error) {
ctx := context.Background()
log := rlog.Component("qbittorrent")
binPath := conf.MetadataFolder + "/qbittorrent-nox"
err := downloadLatestQbitRelease(ctx, binPath)
if err != nil {
return nil, err
}
daemonLog := log.WithComponent("process")
outLog := logwrap.NewSlogWriter(ctx, slog.LevelInfo, daemonLog.Slog())
errLog := logwrap.NewSlogWriter(ctx, slog.LevelError, daemonLog.Slog())
_, err = os.Stat(conf.MetadataFolder + "/profile/qBittorrent/config/qBittorrent.conf")
if errors.Is(err, os.ErrNotExist) {
err = os.MkdirAll(conf.MetadataFolder+"/profile/qBittorrent/config", 0744)
if err != nil {
return nil, err
}
err = os.WriteFile(conf.MetadataFolder+"/profile/qBittorrent/config/qBittorrent.conf", []byte(defaultConf), 0644)
if err != nil {
return nil, err
}
}
err = os.MkdirAll(conf.DataFolder, 0744)
if err != nil {
return nil, err
}
const port = 25436
proc, err := runQBittorrent(binPath, conf.MetadataFolder+"/profile", port, outLog, errLog)
if err != nil {
return nil, err
}
time.Sleep(time.Second)
qb, err := qbittorrent.NewClient(ctx, &qbittorrent.Config{
Address: fmt.Sprintf("http://localhost:%d", port),
})
if err != nil {
return nil, err
}
for { // wait for qbittorrent to start
ver, err := qb.Application().Version(ctx)
log.Info(ctx, "qbittorrent started", slog.String("version", ver))
if err == nil {
break
}
log.Warn(ctx, "waiting for qbittorrent to start", rlog.Error(err))
time.Sleep(time.Second)
}
dataDir, err := filepath.Abs(conf.DataFolder)
if err != nil {
return nil, err
}
err = qb.Application().SetPreferences(ctx, &qbittorrent.Preferences{
SavePath: dataDir,
})
if err != nil {
return nil, err
}
ur, err := iouring.New(8, iouring.WithAsync())
if err != nil {
return nil, err
}
return &Daemon{
qb: qb,
proc: proc,
dataDir: conf.DataFolder,
ur: ur,
sourceFiles: make(map[string]string),
registeredTorrents: mapset.NewSet[string](),
client: wrapClient(qb),
log: rlog.Component("qbittorrent"),
}, nil
}
func (d *Daemon) Close(ctx context.Context) error {
err := d.proc.Signal(os.Interrupt)
if err != nil {
return err
}
_, err = d.proc.Wait()
if err != nil {
return err
}
return nil
}
func torrentDataPath(dataDir string, ih string) (string, error) {
return filepath.Abs(path.Join(dataDir, ih))
}
func (fs *Daemon) GetTorrentFS(ctx context.Context, sourcePath string, file vfs.File) (vfs.Filesystem, error) {
ctx, span := trace.Start(ctx, "GetTorrentFS")
defer span.End()
log := fs.log.With(slog.String("file", file.Name()))
ih, err := readInfoHash(ctx, file)
if err != nil {
return nil, err
}
log = log.With(slog.String("infohash", ih.HexString()))
torrentPath, err := torrentDataPath(fs.dataDir, ih.HexString())
if err != nil {
return nil, fmt.Errorf("error getting torrent path: %w", err)
}
log = log.With(slog.String("torrentPath", torrentPath))
log.Debug(ctx, "creating fs for torrent")
err = fs.syncTorrentState(ctx, file, ih, torrentPath)
if err != nil {
return nil, fmt.Errorf("error syncing torrent state: %w", err)
}
fs.sourceFilesMu.Lock()
fs.sourceFiles[sourcePath] = ih.HexString()
fs.sourceFilesMu.Unlock()
return newTorrentFS(ctx, fs.ur, fs.client, file.Name(), ih.HexString(), torrentPath)
}
func (d *Daemon) syncTorrentState(ctx context.Context, file vfs.File, ih metainfo.Hash, torrentPath string) error {
ctx, span := trace.Start(ctx, "syncTorrentState")
defer span.End()
log := d.log.With(slog.String("file", file.Name()), slog.String("infohash", ih.HexString()))
info, err := d.client.getInfo(ctx, ih.HexString())
if err != nil {
return err
}
log = log.With(slog.String("torrentPath", torrentPath))
if info == nil {
_, err := file.Seek(0, io.SeekStart)
if err != nil {
return err
}
data, err := ctxio.ReadAll(ctx, file)
if err != nil {
return err
}
err = d.qb.Torrent().AddNewTorrent(ctx, &qbittorrent.TorrentAddOption{
Torrents: []*qbittorrent.TorrentAddFileMetadata{
{
Data: data,
},
},
SavePath: torrentPath,
// SequentialDownload: "true",
FirstLastPiecePrio: "true",
})
if err != nil {
d.log.Error(ctx, "error adding torrent", rlog.Error(err))
return err
}
var props *qbittorrent.TorrentProperties
for {
props, err = d.client.getProperties(ctx, ih.HexString())
if err == nil {
break
}
if errors.Is(err, context.DeadlineExceeded) {
return err
}
log.Error(ctx, "waiting for torrent to be added", rlog.Error(err))
time.Sleep(time.Millisecond * 15)
}
log.Info(ctx, "added torrent", slog.String("infohash", ih.HexString()))
d.registeredTorrents.Add(props.Hash)
return nil
} else {
// info := existing[0]
props, err := d.client.getProperties(ctx, ih.HexString())
if err != nil {
return err
}
d.registeredTorrents.Add(props.Hash)
if props.SavePath != torrentPath {
log.Info(ctx, "moving torrent to correct location", slog.String("oldPath", props.SavePath))
err = d.qb.Torrent().SetLocation(ctx, []string{ih.HexString()}, torrentPath)
if err != nil {
return err
}
}
return nil
}
}
// TODO caching
func readInfoHash(ctx context.Context, file vfs.File) (infohash.T, error) {
mi, err := metainfo.Load(ctxio.IoReader(ctx, file))
if err != nil {
return infohash.T{}, err
}
info, err := mi.UnmarshalInfo()
if err != nil {
return infohash.T{}, err
}
if info.HasV2() {
ih := infohash_v2.HashBytes(mi.InfoBytes)
return *(&ih).ToShort(), nil
}
return infohash.HashBytes(mi.InfoBytes), nil
}

409
daemons/qbittorrent/fs.go Normal file
View file

@ -0,0 +1,409 @@
package qbittorrent
import (
"context"
"errors"
"fmt"
"io"
"io/fs"
"log/slog"
"os"
"path"
"strings"
"sync"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/qbittorrent"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/pkg/uring"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/iceber/iouring-go"
)
type FS struct {
mu sync.Mutex
client *cacheClient
name string
hash string
dataDir string // directory where torrent files are stored
ur *iouring.IOURing
entries map[string]fileEntry
log *rlog.Logger
vfs.FilesystemPrototype
}
type fileEntry struct {
fs.FileInfo
Content *qbittorrent.TorrentContent
}
var _ vfs.Filesystem = (*FS)(nil)
func newTorrentFS(ctx context.Context, ur *iouring.IOURing, client *cacheClient, name string, hash string, dataDir string) (*FS, error) {
ctx, span := trace.Start(ctx, "newTorrentFS")
defer span.End()
cnts, err := client.listContent(ctx, hash)
if err != nil {
return nil, fmt.Errorf("failed to list content for hash %s: %w", hash, err)
}
entries := make(map[string]fileEntry, len(cnts))
for _, cnt := range cnts {
if cnt.Priority == qbittorrent.PriorityDoNotDownload {
continue
}
entries[vfs.AbsPath(cnt.Name)] = fileEntry{
Content: cnt,
FileInfo: vfs.NewFileInfo(cnt.Name, cnt.Size),
}
}
return &FS{
client: client,
name: name,
hash: hash,
dataDir: dataDir,
entries: entries,
ur: ur,
log: rlog.Component("qbittorrent", "fs"),
FilesystemPrototype: vfs.FilesystemPrototype(name),
}, nil
}
// Open implements vfs.Filesystem.
func (f *FS) Open(ctx context.Context, name string) (vfs.File, error) {
if name == vfs.Separator {
return vfs.NewDirFile(name), nil
}
if entry, ok := f.entries[name]; ok {
return openFile(ctx, f.ur, f.client, f.dataDir, f.hash, entry.Content)
}
for p := range f.entries {
if strings.HasPrefix(p, name) {
return vfs.NewDirFile(name), nil
}
}
return nil, vfs.ErrNotExist
}
// ReadDir implements vfs.Filesystem.
func (f *FS) ReadDir(ctx context.Context, name string) ([]fs.DirEntry, error) {
infos := make(map[string]fs.FileInfo, len(f.entries))
for k, v := range f.entries {
infos[k] = v.FileInfo
}
return vfs.ListDirFromInfo(infos, name)
}
// Stat implements vfs.Filesystem.
func (f *FS) Stat(ctx context.Context, name string) (fs.FileInfo, error) {
name = vfs.AbsPath(path.Clean(name))
if vfs.IsRoot(name) {
return vfs.NewDirInfo(f.name), nil
}
if entry, ok := f.entries[name]; ok {
return entry.FileInfo, nil
}
for p := range f.entries {
if strings.HasPrefix(p, name) {
return vfs.NewDirInfo(name), nil
}
}
return nil, vfs.ErrNotExist
}
// Unlink implements vfs.Filesystem.
func (f *FS) Unlink(ctx context.Context, filename string) error {
filename = vfs.AbsPath(path.Clean(filename))
// we cannot delete a torrent itself, cause it will be added on next source scan and all delited files will be restored
if entry, ok := f.entries[filename]; ok {
return f.removeFile(ctx, f.hash, entry.Content)
}
for p, entry := range f.entries {
if strings.HasPrefix(p, filename) {
return f.removeFile(ctx, f.hash, entry.Content)
}
}
return vfs.ErrNotExist
}
func (f *FS) Rename(ctx context.Context, oldpath string, newpath string) error {
oldpath = vfs.AbsPath(path.Clean(oldpath))
newpath = vfs.AbsPath(path.Clean(newpath))
if _, ok := f.entries[oldpath]; ok {
err := f.client.qb.Torrent().RenameFile(ctx, f.hash, vfs.RelPath(oldpath), vfs.RelPath(newpath))
if err != nil {
return fmt.Errorf("failed to rename file %s to %s: %w", oldpath, newpath, err)
}
f.mu.Lock()
defer f.mu.Unlock()
f.entries[newpath] = f.entries[oldpath]
return nil
}
return vfs.ErrNotExist
}
func (f *FS) removeFile(ctx context.Context, hash string, content *qbittorrent.TorrentContent) error {
log := f.log.With(slog.String("hash", hash), slog.String("file", content.Name))
f.mu.Lock()
defer f.mu.Unlock()
fpath := vfs.AbsPath(content.Name)
if _, ok := f.entries[fpath]; !ok {
return fmt.Errorf("file %s is does not found", fpath)
}
delete(f.entries, fpath)
err := f.client.qb.Torrent().SetFilePriority(ctx, f.hash, content.Index, qbittorrent.PriorityDoNotDownload)
if err != nil {
return fmt.Errorf("failed to set priority for torrent %s for file %s: %w", hash, content.Name, err)
}
err = os.Remove(path.Join(f.dataDir, vfs.RelPath(content.Name)))
if err != nil && !errors.Is(err, fs.ErrNotExist) {
log.Warn(ctx, "failed to remove file", rlog.Error(err))
return fmt.Errorf("failed to remove file %s: %w", content.Name, err)
}
return nil
}
func openFile(ctx context.Context, ur *iouring.IOURing, client *cacheClient, torrentDir string, hash string, content *qbittorrent.TorrentContent) (*File, error) {
props, err := client.getProperties(ctx, hash)
if err != nil {
return nil, err
}
// FIXME error when file not started downloading
file, err := os.OpenFile(path.Join(torrentDir, content.Name), os.O_RDONLY, 0)
if err != nil {
return nil, err
}
return &File{
client: client,
hash: hash,
torrentDir: torrentDir,
filePath: content.Name,
contentIndex: content.Index,
pieceSize: props.PieceSize,
fileSize: content.Size,
file: uring.NewFile(ur, file),
offset: 0,
}, nil
}
type File struct {
client *cacheClient
hash string
torrentDir string
filePath string // path inside a torrent directory
contentIndex int
pieceSize int
fileSize int64
mu sync.Mutex
file *uring.File
offset int64
}
var _ vfs.File = (*File)(nil)
// Info implements vfs.File.
func (f *File) Info() (fs.FileInfo, error) {
return &fileInfo{name: path.Base(f.filePath), size: f.fileSize}, nil
}
// IsDir implements vfs.File.
func (f *File) IsDir() bool {
return false
}
// Seek implements vfs.File.
func (f *File) Seek(offset int64, whence int) (int64, error) {
switch whence {
case io.SeekStart:
f.offset = offset
case io.SeekCurrent:
f.offset += offset
case io.SeekEnd:
f.offset = f.fileSize + offset
}
return f.offset, nil
}
// Name implements vfs.File.
func (f *File) Name() string {
return path.Base(f.filePath)
}
func (f *File) canExpectSoon(ctx context.Context) (bool, error) {
info, err := f.client.getInfo(ctx, f.hash)
if err != nil {
return false, err
}
return info.Completed == info.Size || info.State == qbittorrent.TorrentStateCheckingUP || info.State == qbittorrent.TorrentStateDownloading || info.State == qbittorrent.TorrentStateForcedDL, nil
}
func (f *File) isRangeComplete(ctx context.Context, offset int64, size int) (bool, error) {
startPieceIndex := int(offset / int64(f.pieceSize))
pieceCount := (size + f.pieceSize - 1) / f.pieceSize // rouding up
for i := range pieceCount {
ok, err := f.client.isPieceComplete(ctx, f.hash, startPieceIndex+i)
if err != nil {
return false, err
}
if !ok {
return false, nil
}
}
return true, nil
}
func (f *File) waitPieceAvailable(ctx context.Context, offset int64, size int) error {
complete, err := f.isRangeComplete(ctx, offset, size)
if err != nil {
return err
}
if complete {
return nil
}
canExpectSoon, err := f.canExpectSoon(ctx)
if err != nil {
return err
}
if !canExpectSoon {
return fmt.Errorf("torrent is not downloading")
}
const checkingInterval = 1 * time.Second
ticker := time.NewTicker(checkingInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
complete, err := f.isRangeComplete(ctx, offset, size)
if err != nil {
return err
}
if complete {
return nil
}
}
}
}
// Read implements vfs.File.
func (f *File) Read(ctx context.Context, p []byte) (int, error) {
f.mu.Lock()
defer f.mu.Unlock()
if err := f.waitPieceAvailable(ctx, f.offset, len(p)); err != nil {
return 0, err
}
n, err := f.file.ReadAt(ctx, p, f.offset)
f.offset += int64(n)
return n, err
}
// ReadAt implements vfs.File.
func (f *File) ReadAt(ctx context.Context, p []byte, off int64) (int, error) {
if err := f.waitPieceAvailable(ctx, f.offset, len(p)); err != nil {
return 0, err
}
return f.file.ReadAt(ctx, p, off)
}
// Size implements vfs.File.
func (f *File) Size() int64 {
return f.fileSize
}
// Type implements vfs.File.
func (f *File) Type() fs.FileMode {
return fs.ModeDir
}
// Close implements vfs.File.
func (f *File) Close(ctx context.Context) error {
return f.file.Close(ctx)
}
type fileInfo struct {
name string
size int64
}
var _ fs.FileInfo = (*fileInfo)(nil)
// IsDir implements fs.FileInfo.
func (f *fileInfo) IsDir() bool {
return false
}
// ModTime implements fs.FileInfo.
func (f *fileInfo) ModTime() time.Time {
return time.Time{}
}
// Mode implements fs.FileInfo.
func (f *fileInfo) Mode() fs.FileMode {
return vfs.ModeFileRO
}
// Name implements fs.FileInfo.
func (f *fileInfo) Name() string {
return f.name
}
// Size implements fs.FileInfo.
func (f *fileInfo) Size() int64 {
return f.size
}
// Sys implements fs.FileInfo.
func (f *fileInfo) Sys() any {
return nil
}

View file

@ -0,0 +1,150 @@
package qbittorrent
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"os"
"os/exec"
"path"
"runtime"
"time"
"github.com/google/go-github/v63/github"
"golang.org/x/sys/cpu"
)
const (
repoOwner = "userdocs"
repoName = "qbittorrent-nox-static"
)
func runQBittorrent(binPath string, profileDir string, port int, stdout, stderr io.Writer) (*os.Process, error) {
err := os.Chmod(binPath, 0755)
if err != nil {
return nil, err
}
cmd := exec.Command(binPath, fmt.Sprintf("--profile=%s", profileDir), fmt.Sprintf("--webui-port=%d", port))
cmd.Stdin = bytes.NewReader([]byte("y\n"))
cmd.Stdout = stdout
cmd.Stderr = stderr
err = cmd.Start()
if err != nil {
return nil, err
}
return cmd.Process, nil
}
func downloadLatestQbitRelease(ctx context.Context, binPath string) error {
client := github.NewClient(nil)
rel, _, err := client.Repositories.GetLatestRelease(ctx, repoOwner, repoName)
if err != nil {
return err
}
arch := ""
switch runtime.GOARCH {
case "amd64":
arch = "x86_64"
case "arm":
arch = "armhf" // this is a safe version, go does not distinguish between armv6 and armv7
if cpu.ARM.HasNEON {
arch = "armv7"
}
case "arm64":
arch = "aarch64"
}
if arch == "" {
return errors.New("unsupported architecture")
}
binName := arch + "-qbittorrent-nox"
var targetRelease *github.ReleaseAsset
for _, v := range rel.Assets {
if v.GetName() == binName {
targetRelease = v
break
}
}
if targetRelease == nil {
return fmt.Errorf("target asset %s not found", binName)
}
downloadUrl := targetRelease.GetBrowserDownloadURL()
if downloadUrl == "" {
return errors.New("download url is empty")
}
err = os.MkdirAll(path.Dir(binPath), 0755)
if err != nil {
return err
}
slog.InfoContext(ctx, "downloading latest qbittorrent-nox release", slog.String("url", downloadUrl))
return downloadFile(ctx, binPath, downloadUrl)
}
func downloadFile(ctx context.Context, filepath string, webUrl string) error {
if stat, err := os.Stat(filepath); err == nil {
resp, err := http.Head(webUrl)
if err != nil {
return err
}
defer resp.Body.Close()
var lastModified time.Time
lastModifiedHeader := resp.Header.Get("Last-Modified")
if lastModifiedHeader != "" {
lastModified, err = time.Parse(http.TimeFormat, lastModifiedHeader)
if err != nil {
return err
}
}
if resp.ContentLength == stat.Size() && lastModified.Before(stat.ModTime()) {
slog.InfoContext(ctx, "there is already newest version of the file", slog.String("filepath", filepath))
return nil
}
}
// Create the file
out, err := os.Create(filepath)
if err != nil {
return err
}
defer out.Close()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, webUrl, nil)
if err != nil {
return err
}
// Get the data
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
// Check server response
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("bad status: %s", resp.Status)
}
// Writer the body to file
_, err = io.Copy(out, resp.Body)
if err != nil {
return err
}
return nil
}

View file

@ -0,0 +1,18 @@
package qbittorrent
import (
"context"
"testing"
"github.com/stretchr/testify/require"
)
func TestDownloadQBittorent(t *testing.T) {
ctx := context.Background()
tempDir := t.TempDir()
require := require.New(t)
err := downloadLatestQbitRelease(ctx, tempDir)
require.NoError(err)
err = downloadLatestQbitRelease(ctx, tempDir)
require.NoError(err)
}

110
daemons/torrent/client.go Normal file
View file

@ -0,0 +1,110 @@
package torrent
import (
"crypto/rand"
"log/slog"
"os"
"git.kmsign.ru/royalcat/tstor/src/config"
"git.kmsign.ru/royalcat/tstor/src/logwrap"
"github.com/anacrolix/dht/v2/bep44"
tlog "github.com/anacrolix/log"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/storage"
"github.com/anacrolix/torrent/types/infohash"
)
func newClientConfig(st storage.ClientImpl, fis bep44.Store, cfg *config.TorrentClient, id [20]byte) *torrent.ClientConfig {
l := slog.With("component", "torrent-client")
// TODO download and upload limits
torrentCfg := torrent.NewDefaultClientConfig()
torrentCfg.PeerID = string(id[:])
torrentCfg.DefaultStorage = st
// torrentCfg.AlwaysWantConns = true
torrentCfg.DropMutuallyCompletePeers = true
// torrentCfg.TorrentPeersLowWater = 100
// torrentCfg.TorrentPeersHighWater = 1000
// torrentCfg.AcceptPeerConnections = true
torrentCfg.Seed = true
torrentCfg.DisableAggressiveUpload = false
torrentCfg.PeriodicallyAnnounceTorrentsToDht = true
// torrentCfg.ConfigureAnacrolixDhtServer = func(cfg *dht.ServerConfig) {
// cfg.Store = fis
// cfg.Exp = dhtTTL
// cfg.PeerStore = fis
// }
tl := tlog.NewLogger("torrent-client")
tl.SetHandlers(&logwrap.Torrent{L: l})
torrentCfg.Logger = tl
torrentCfg.Callbacks.NewPeer = append(torrentCfg.Callbacks.NewPeer, func(p *torrent.Peer) {
l.With(peerAttrs(p)...).Debug("new peer")
})
torrentCfg.Callbacks.PeerClosed = append(torrentCfg.Callbacks.PeerClosed, func(p *torrent.Peer) {
l.With(peerAttrs(p)...).Debug("peer closed")
})
torrentCfg.Callbacks.CompletedHandshake = func(pc *torrent.PeerConn, ih infohash.T) {
attrs := append(peerAttrs(&pc.Peer), slog.String("infohash", ih.HexString()))
l.With(attrs...).Debug("completed handshake")
}
torrentCfg.Callbacks.PeerConnAdded = append(torrentCfg.Callbacks.PeerConnAdded, func(pc *torrent.PeerConn) {
l.With(peerAttrs(&pc.Peer)...).Debug("peer conn added")
})
torrentCfg.Callbacks.PeerConnClosed = func(pc *torrent.PeerConn) {
l.With(peerAttrs(&pc.Peer)...).Debug("peer conn closed")
}
torrentCfg.Callbacks.CompletedHandshake = func(pc *torrent.PeerConn, ih infohash.T) {
attrs := append(peerAttrs(&pc.Peer), slog.String("infohash", ih.HexString()))
l.With(attrs...).Debug("completed handshake")
}
torrentCfg.Callbacks.ReceivedRequested = append(torrentCfg.Callbacks.ReceivedRequested, func(pme torrent.PeerMessageEvent) {
l.With(peerAttrs(pme.Peer)...).Debug("received requested")
})
torrentCfg.Callbacks.ReceivedUsefulData = append(torrentCfg.Callbacks.ReceivedUsefulData, func(pme torrent.PeerMessageEvent) {
l.With(peerAttrs(pme.Peer)...).Debug("received useful data")
})
return torrentCfg
}
var emptyBytes [20]byte
func getOrCreatePeerID(p string) ([20]byte, error) {
idb, err := os.ReadFile(p)
if err == nil {
var out [20]byte
copy(out[:], idb)
return out, nil
}
if !os.IsNotExist(err) {
return emptyBytes, err
}
var out [20]byte
_, err = rand.Read(out[:])
if err != nil {
return emptyBytes, err
}
return out, os.WriteFile(p, out[:], 0755)
}
func peerAttrs(peer *torrent.Peer) []any {
out := []any{
slog.String("ip", peer.RemoteAddr.String()),
slog.String("discovery", string(peer.Discovery)),
slog.Int("max-requests", peer.PeerMaxRequests),
slog.Bool("prefers-encryption", peer.PeerPrefersEncryption),
}
if peer.Torrent() != nil {
out = append(out, slog.String("torrent", peer.Torrent().Name()))
}
return out
}

View file

@ -0,0 +1,265 @@
package torrent
import (
"context"
"log/slog"
"strings"
"git.kmsign.ru/royalcat/tstor/pkg/kvsingle"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/types"
"github.com/royalcat/kv"
)
type TorrentFileDeleter interface {
DeleteFile(file *torrent.File) error
}
type FileProperties struct {
Excluded bool `json:"excluded"`
Priority types.PiecePriority `json:"priority"`
}
type Controller struct {
torrentFilePath string
t *torrent.Torrent
storage TorrentFileDeleter
fileProperties kv.Store[string, FileProperties]
log *rlog.Logger
}
func newController(t *torrent.Torrent, torrentFileProperties kv.Store[string, FileProperties], storage TorrentFileDeleter, log *rlog.Logger) *Controller {
return &Controller{
t: t,
storage: storage,
fileProperties: torrentFileProperties,
log: log.WithComponent("controller").With(slog.String("infohash", t.InfoHash().HexString())),
}
}
func (s *Controller) TorrentFilePath() string {
return s.torrentFilePath
}
func (s *Controller) Torrent() *torrent.Torrent {
return s.t
}
func (c *Controller) Name() string {
<-c.t.GotInfo()
if name := c.t.Name(); name != "" {
return name
}
return c.InfoHash()
}
func (s *Controller) InfoHash() string {
<-s.t.GotInfo()
return s.t.InfoHash().HexString()
}
func (s *Controller) BytesCompleted() int64 {
<-s.t.GotInfo()
return s.t.BytesCompleted()
}
func (s *Controller) BytesMissing() int64 {
<-s.t.GotInfo()
return s.t.BytesMissing()
}
func (s *Controller) Length() int64 {
<-s.t.GotInfo()
return s.t.Length()
}
func (s *Controller) Files(ctx context.Context) ([]*FileController, error) {
ctx, span := tracer.Start(ctx, "Files")
defer span.End()
fps := map[string]FileProperties{}
err := s.fileProperties.Range(ctx, func(k string, v FileProperties) error {
fps[k] = v
return nil
})
if err != nil {
return nil, err
}
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-s.t.GotInfo():
}
files := make([]*FileController, 0)
for _, v := range s.t.Files() {
if strings.Contains(v.Path(), "/.pad/") {
continue
}
props := kvsingle.New(s.fileProperties, v.Path())
ctl := NewFileController(v, props, s.log)
files = append(files, ctl)
}
return files, nil
}
func (s *Controller) GetFile(ctx context.Context, file string) (*FileController, error) {
files, err := s.Files(ctx)
if err != nil {
return nil, err
}
for _, v := range files {
if v.Path() == file {
return v, nil
}
}
return nil, nil
}
func Map[T, U any](ts []T, f func(T) U) []U {
us := make([]U, len(ts))
for i := range ts {
us[i] = f(ts[i])
}
return us
}
func (s *Controller) ExcludeFile(ctx context.Context, f *torrent.File) error {
log := s.log.With(slog.String("file", f.Path()))
log.Info(ctx, "excluding file")
err := s.fileProperties.Edit(ctx, f.Path(), func(ctx context.Context, v FileProperties) (FileProperties, error) {
v.Excluded = true
return v, nil
})
if err == kv.ErrKeyNotFound {
err := s.fileProperties.Set(ctx, f.Path(), FileProperties{Excluded: true})
if err != nil {
return err
}
} else if err != nil {
return err
}
return s.storage.DeleteFile(f)
}
func (s *Controller) isFileComplete(startIndex int, endIndex int) bool {
for i := startIndex; i < endIndex; i++ {
if !s.t.Piece(i).State().Complete {
return false
}
}
return true
}
func (s *Controller) ValidateTorrent(ctx context.Context) error {
select {
case <-ctx.Done():
return ctx.Err()
case <-s.t.GotInfo():
}
for i := 0; i < s.t.NumPieces(); i++ {
if ctx.Err() != nil {
return ctx.Err()
}
s.t.Piece(i).VerifyData()
}
return nil
}
func (c *Controller) SetPriority(ctx context.Context, priority types.PiecePriority) error {
log := c.log.With(slog.Int("priority", int(priority)))
files, err := c.Files(ctx)
if err != nil {
return err
}
for _, f := range files {
excluded, err := f.Excluded(ctx)
if err != nil {
log.Error(ctx, "failed to get file exclusion status", rlog.Error(err))
}
if excluded {
continue
}
err = f.SetPriority(ctx, priority)
if err != nil {
log.Error(ctx, "failed to set file priority", rlog.Error(err))
}
}
return nil
}
const defaultPriority = types.PiecePriorityNone
func (c *Controller) Priority(ctx context.Context) (types.PiecePriority, error) {
prio := defaultPriority
files, err := c.Files(ctx)
if err != nil {
return 0, err
}
for _, v := range files {
filePriority := v.Priority()
if filePriority > prio {
prio = filePriority
}
}
return prio, nil
}
// func (c *Controller) setFilePriority(ctx context.Context, file *torrent.File, priority types.PiecePriority) error {
// err := c.fileProperties.Edit(ctx, file.Path(), func(ctx context.Context, v FileProperties) (FileProperties, error) {
// v.Priority = priority
// return v, nil
// })
// if err == kv.ErrKeyNotFound {
// seterr := c.fileProperties.Set(ctx, file.Path(), FileProperties{Priority: priority})
// if seterr != nil {
// return seterr
// }
// err = nil
// }
// if err != nil {
// return err
// }
// file.SetPriority(priority)
// return nil
// }
func (c *Controller) initializeTorrentPriories(ctx context.Context) error {
ctx, span := tracer.Start(ctx, "initializeTorrentPriories")
defer span.End()
log := c.log
files, err := c.Files(ctx)
if err != nil {
return err
}
for _, file := range files {
props, err := file.Properties(ctx)
if err != nil {
log.Error(ctx, "failed to get file properties", rlog.Error(err))
continue
}
file.file.SetPriority(props.Priority)
}
log.Debug(ctx, "torrent initialization complete", slog.String("infohash", c.InfoHash()), slog.String("torrent_name", c.Name()))
return nil
}

226
daemons/torrent/daemon.go Normal file
View file

@ -0,0 +1,226 @@
package torrent
import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
"sync"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/src/config"
"git.kmsign.ru/royalcat/tstor/src/tkv"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/trace"
"golang.org/x/exp/maps"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/bencode"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/types/infohash"
"github.com/go-git/go-billy/v5"
"github.com/royalcat/kv"
)
const instrument = "git.kmsign.ru/royalcat/tstor/daemons/torrent"
var (
tracer = otel.Tracer(instrument, trace.WithInstrumentationAttributes(attribute.String("component", "torrent-daemon")))
meter = otel.Meter(instrument, metric.WithInstrumentationAttributes(attribute.String("component", "torrent-daemon")))
)
type DirAquire struct {
Name string
Hashes []infohash.T
}
type Daemon struct {
client *torrent.Client
infoBytes *infoBytesStore
Storage *fileStorage
fis *dhtFileItemStore
dirsAquire kv.Store[string, DirAquire]
fileProperties kv.Store[string, FileProperties]
statsStore *statsStore
loadMutex sync.Mutex
sourceFs billy.Filesystem
log *rlog.Logger
}
const dhtTTL = 180 * 24 * time.Hour
func NewDaemon(sourceFs billy.Filesystem, conf config.TorrentClient) (*Daemon, error) {
s := &Daemon{
log: rlog.Component("torrent-service"),
sourceFs: sourceFs,
loadMutex: sync.Mutex{},
}
err := os.MkdirAll(conf.MetadataFolder, 0744)
if err != nil {
return nil, fmt.Errorf("error creating metadata folder: %w", err)
}
s.fis, err = newDHTStore(filepath.Join(conf.MetadataFolder, "dht-item-store"), dhtTTL)
if err != nil {
return nil, fmt.Errorf("error starting item store: %w", err)
}
s.Storage, _, err = setupStorage(conf)
if err != nil {
return nil, err
}
s.fileProperties, err = tkv.NewKV[string, FileProperties](conf.MetadataFolder, "file-properties")
if err != nil {
return nil, err
}
s.infoBytes, err = newInfoBytesStore(conf.MetadataFolder)
if err != nil {
return nil, err
}
id, err := getOrCreatePeerID(filepath.Join(conf.MetadataFolder, "ID"))
if err != nil {
return nil, fmt.Errorf("error creating node ID: %w", err)
}
s.statsStore, err = newStatsStore(conf.MetadataFolder, time.Hour*24*30)
if err != nil {
return nil, err
}
clientConfig := newClientConfig(s.Storage, s.fis, &conf, id)
s.client, err = torrent.NewClient(clientConfig)
if err != nil {
return nil, err
}
// TODO move to config
s.client.AddDhtNodes([]string{
"router.bittorrent.com:6881",
"router.utorrent.com:6881",
"dht.transmissionbt.com:6881",
"router.bitcomet.com:6881",
"dht.aelitis.com6881",
})
s.client.AddDhtNodes(conf.DHTNodes)
s.dirsAquire, err = tkv.NewKV[string, DirAquire](conf.MetadataFolder, "dir-acquire")
if err != nil {
return nil, err
}
// go func() {
// ctx := context.Background()
// err := s.backgroudFileLoad(ctx)
// if err != nil {
// s.log.Error(ctx, "initial torrent load failed", rlog.Error(err))
// }
// }()
go func() {
ctx := context.Background()
const period = time.Second * 10
err := registerTorrentMetrics(s.client)
if err != nil {
s.log.Error(ctx, "error registering torrent metrics", rlog.Error(err))
}
err = registerDhtMetrics(s.client)
if err != nil {
s.log.Error(ctx, "error registering dht metrics", rlog.Error(err))
}
timer := time.NewTicker(period)
for {
select {
case <-s.client.Closed():
return
case <-timer.C:
s.updateStats(ctx)
}
}
}()
return s, nil
}
var _ vfs.FsFactory = (*Daemon)(nil).NewTorrentFs
func (s *Daemon) Close(ctx context.Context) error {
return errors.Join(append(
s.client.Close(),
s.Storage.Close(),
s.dirsAquire.Close(ctx),
// s.excludedFiles.Close(ctx),
s.infoBytes.Close(),
s.fis.Close(),
)...)
}
func isValidInfoHashBytes(d []byte) bool {
var info metainfo.Info
err := bencode.Unmarshal(d, &info)
return err == nil
}
func (s *Daemon) Stats() torrent.ConnStats {
return s.client.Stats().ConnStats
}
func storeByTorrent[K kv.Bytes, V any](s kv.Store[K, V], infohash infohash.T) kv.Store[K, V] {
return kv.PrefixBytes[K, V](s, K(infohash.HexString()+"/"))
}
func (s *Daemon) newController(t *torrent.Torrent) *Controller {
return newController(t,
storeByTorrent(s.fileProperties, t.InfoHash()),
s.Storage,
s.log,
)
}
func (s *Daemon) ListTorrents(ctx context.Context) ([]*Controller, error) {
out := []*Controller{}
for _, v := range s.client.Torrents() {
out = append(out, s.newController(v))
}
return out, nil
}
func (s *Daemon) GetTorrent(infohashHex string) (*Controller, error) {
t, ok := s.client.Torrent(infohash.FromHexString(infohashHex))
if !ok {
return nil, nil
}
return s.newController(t), nil
}
func slicesUnique[S ~[]E, E comparable](in S) S {
m := map[E]struct{}{}
for _, v := range in {
m[v] = struct{}{}
}
return maps.Keys(m)
}
func apply[I, O any](in []I, f func(e I) O) []O {
out := []O{}
for _, v := range in {
out = append(out, f(v))
}
return out
}

View file

@ -0,0 +1,246 @@
package torrent
import (
"bufio"
"context"
"fmt"
"io"
"log/slog"
"os"
"strings"
"sync"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/ctxbilly"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/metainfo"
"github.com/go-git/go-billy/v5/util"
"github.com/royalcat/ctxio"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
)
const activityTimeout = time.Minute * 15
func readInfoHash(ctx context.Context, f vfs.File) (metainfo.Hash, error) {
ctx, span := tracer.Start(ctx, "readInfoHash")
defer span.End()
mi, err := metainfo.Load(ctxio.IoReader(ctx, f))
if err != nil {
return metainfo.Hash{}, fmt.Errorf("loading metainfo: %w", err)
}
return mi.HashInfoBytes(), nil
}
func (s *Daemon) loadTorrent(ctx context.Context, f vfs.File) (*Controller, error) {
ctx, span := tracer.Start(ctx, "loadTorrent")
defer span.End()
log := s.log
stat, err := f.Info()
if err != nil {
return nil, fmt.Errorf("call stat failed: %w", err)
}
span.SetAttributes(attribute.String("filename", stat.Name()))
mi, err := metainfo.Load(bufio.NewReader(ctxio.IoReader(ctx, f)))
if err != nil {
return nil, fmt.Errorf("loading torrent metadata from file %s, error: %w", stat.Name(), err)
}
log = log.With(slog.String("info-hash", mi.HashInfoBytes().HexString()))
var ctl *Controller
t, ok := s.client.Torrent(mi.HashInfoBytes())
if ok {
log = log.With(slog.String("torrent-name", t.Name()))
ctl = s.newController(t)
} else {
span.AddEvent("torrent not found, loading from file")
log.Info(ctx, "torrent not found, loading from file")
spec, err := torrent.TorrentSpecFromMetaInfoErr(mi)
if err != nil {
return nil, fmt.Errorf("parse spec from metadata: %w", err)
}
infoBytes := spec.InfoBytes
if !isValidInfoHashBytes(infoBytes) {
log.Warn(ctx, "info loaded from spec not valid")
infoBytes = nil
}
if len(infoBytes) == 0 {
log.Info(ctx, "no info loaded from file, try to load from cache")
infoBytes, err = s.infoBytes.GetBytes(spec.InfoHash)
if err != nil && err != errNotFound {
return nil, fmt.Errorf("get info bytes from database: %w", err)
}
}
t, _ = s.client.AddTorrentOpt(torrent.AddTorrentOpts{
InfoHash: spec.InfoHash,
InfoHashV2: spec.InfoHashV2,
Storage: s.Storage,
InfoBytes: infoBytes,
ChunkSize: spec.ChunkSize,
})
log = log.With(slog.String("torrent-name", t.Name()))
t.AllowDataDownload()
t.AllowDataUpload()
span.AddEvent("torrent added to client")
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-t.GotInfo():
err := s.infoBytes.Set(t.InfoHash(), t.Metainfo())
if err != nil {
log.Error(ctx, "error setting info bytes for torrent",
slog.String("torrent-name", t.Name()),
rlog.Error(err),
)
}
}
span.AddEvent("got info")
ctl = s.newController(t)
err = ctl.initializeTorrentPriories(ctx)
if err != nil {
return nil, fmt.Errorf("initialize torrent priorities: %w", err)
}
// go func() {
// subscr := ctl.t.SubscribePieceStateChanges()
// defer subscr.Close()
// dropTimer := time.NewTimer(activityTimeout)
// defer dropTimer.Stop()
// for {
// select {
// case <-subscr.Values:
// dropTimer.Reset(activityTimeout)
// case <-dropTimer.C:
// log.Info(ctx, "torrent dropped by activity timeout")
// select {
// case <-ctl.t.Closed():
// return
// case <-time.After(time.Second):
// ctl.t.Drop()
// }
// case <-ctl.t.Closed():
// return
// }
// }
// }()
}
return ctl, nil
}
const loadWorkers = 5
func (s *Daemon) backgroudFileLoad(ctx context.Context) error {
ctx, span := tracer.Start(ctx, "loadTorrentFiles", trace.WithAttributes(
attribute.Int("workers", loadWorkers),
))
defer span.End()
log := s.log
loaderPaths := make(chan string, loadWorkers*5)
wg := sync.WaitGroup{}
defer func() {
close(loaderPaths)
wg.Wait()
}()
loaderWorker := func() {
for path := range loaderPaths {
info, err := s.sourceFs.Stat(path)
if err != nil {
log.Error(ctx, "error stat torrent file", slog.String("filename", path), rlog.Error(err))
continue
}
file, err := s.sourceFs.Open(path)
if err != nil {
log.Error(ctx, "error opening torrent file", slog.String("filename", path), rlog.Error(err))
continue
}
defer file.Close()
vfile := vfs.NewCtxBillyFile(info, ctxbilly.WrapFile(file))
ih, err := readInfoHash(ctx, vfile)
if err != nil {
log.Error(ctx, "error reading info hash", slog.String("filename", path), rlog.Error(err))
continue
}
props := storeByTorrent(s.fileProperties, ih)
_, err = vfile.Seek(0, io.SeekStart)
if err != nil {
log.Error(ctx, "error seeking file", slog.String("filename", path), rlog.Error(err))
continue
}
isPrioritized := false
err = props.Range(ctx, func(k string, v FileProperties) error {
if v.Priority > 0 {
isPrioritized = true
return io.EOF
}
return nil
})
if err != nil && err != io.EOF {
log.Error(ctx, "error checking file priority", slog.String("filename", path), rlog.Error(err))
continue
}
if !isPrioritized {
log.Debug(ctx, "file not prioritized, skipping", slog.String("filename", path))
continue
}
_, err = s.loadTorrent(ctx, vfile)
if err != nil {
log.Error(ctx, "failed adding torrent", rlog.Error(err))
}
}
wg.Done()
}
wg.Add(loadWorkers)
for range loadWorkers {
go loaderWorker()
}
return util.Walk(s.sourceFs, "", func(path string, info os.FileInfo, err error) error {
if err != nil {
return fmt.Errorf("fs walk error: %w", err)
}
if ctx.Err() != nil {
return ctx.Err()
}
if info.IsDir() {
return nil
}
if strings.HasSuffix(path, ".torrent") {
loaderPaths <- path
}
return nil
})
}

View file

@ -0,0 +1,73 @@
package torrent
import (
"context"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/anacrolix/torrent/types/infohash"
)
func (s *Daemon) allStats(ctx context.Context) (map[infohash.T]TorrentStats, TorrentStats) {
totalPeers := 0
activePeers := 0
connectedSeeders := 0
perTorrentStats := map[infohash.T]TorrentStats{}
for _, v := range s.client.Torrents() {
stats := v.Stats()
perTorrentStats[v.InfoHash()] = TorrentStats{
Timestamp: time.Now(),
DownloadedBytes: uint64(stats.BytesRead.Int64()),
UploadedBytes: uint64(stats.BytesWritten.Int64()),
TotalPeers: uint16(stats.TotalPeers),
ActivePeers: uint16(stats.ActivePeers),
ConnectedSeeders: uint16(stats.ConnectedSeeders),
}
totalPeers += stats.TotalPeers
activePeers += stats.ActivePeers
connectedSeeders += stats.ConnectedSeeders
}
totalStats := s.client.Stats()
return perTorrentStats, TorrentStats{
Timestamp: time.Now(),
DownloadedBytes: uint64(totalStats.BytesRead.Int64()),
UploadedBytes: uint64(totalStats.BytesWritten.Int64()),
TotalPeers: uint16(totalPeers),
ActivePeers: uint16(activePeers),
ConnectedSeeders: uint16(connectedSeeders),
}
}
func (s *Daemon) updateStats(ctx context.Context) {
log := s.log
perTorrentStats, totalStats := s.allStats(ctx)
for ih, v := range perTorrentStats {
err := s.statsStore.AddTorrentStats(ih, v)
if err != nil {
log.Error(ctx, "error saving torrent stats", rlog.Error(err))
}
}
err := s.statsStore.AddTotalStats(totalStats)
if err != nil {
log.Error(ctx, "error saving total stats", rlog.Error(err))
}
}
func (s *Daemon) TotalStatsHistory(ctx context.Context, since time.Time) ([]TorrentStats, error) {
return s.statsStore.ReadTotalStatsHistory(ctx, since)
}
func (s *Daemon) TorrentStatsHistory(ctx context.Context, since time.Time, ih infohash.T) ([]TorrentStats, error) {
return s.statsStore.ReadTorrentStatsHistory(ctx, since, ih)
}
func (s *Daemon) StatsHistory(ctx context.Context, since time.Time) ([]TorrentStats, error) {
return s.statsStore.ReadStatsHistory(ctx, since)
}

View file

@ -0,0 +1,112 @@
package torrent
import (
"bytes"
"encoding/gob"
"time"
"git.kmsign.ru/royalcat/tstor/src/logwrap"
"github.com/anacrolix/dht/v2/bep44"
"github.com/dgraph-io/badger/v4"
)
var _ bep44.Store = &dhtFileItemStore{}
type dhtFileItemStore struct {
ttl time.Duration
db *badger.DB
}
func newDHTStore(path string, itemsTTL time.Duration) (*dhtFileItemStore, error) {
opts := badger.DefaultOptions(path).
WithLogger(logwrap.BadgerLogger("torrent-client", "dht-item-store")).
WithValueLogFileSize(1<<26 - 1)
db, err := badger.Open(opts)
if err != nil {
return nil, err
}
err = db.RunValueLogGC(0.5)
if err != nil && err != badger.ErrNoRewrite {
return nil, err
}
return &dhtFileItemStore{
db: db,
ttl: itemsTTL,
}, nil
}
func (fis *dhtFileItemStore) Put(i *bep44.Item) error {
tx := fis.db.NewTransaction(true)
defer tx.Discard()
key := i.Target()
var value bytes.Buffer
enc := gob.NewEncoder(&value)
if err := enc.Encode(i); err != nil {
return err
}
e := badger.NewEntry(key[:], value.Bytes()).WithTTL(fis.ttl)
if err := tx.SetEntry(e); err != nil {
return err
}
return tx.Commit()
}
func (fis *dhtFileItemStore) Get(t bep44.Target) (*bep44.Item, error) {
tx := fis.db.NewTransaction(false)
defer tx.Discard()
dbi, err := tx.Get(t[:])
if err == badger.ErrKeyNotFound {
return nil, bep44.ErrItemNotFound
}
if err != nil {
return nil, err
}
valb, err := dbi.ValueCopy(nil)
if err != nil {
return nil, err
}
buf := bytes.NewBuffer(valb)
dec := gob.NewDecoder(buf)
var i *bep44.Item
if err := dec.Decode(&i); err != nil {
return nil, err
}
return i, nil
}
func (fis *dhtFileItemStore) Del(t bep44.Target) error {
tx := fis.db.NewTransaction(true)
defer tx.Discard()
err := tx.Delete(t[:])
if err == badger.ErrKeyNotFound {
return nil
}
if err != nil {
return err
}
err = tx.Commit()
if err == badger.ErrKeyNotFound {
return nil
}
if err != nil {
return err
}
return nil
}
func (fis *dhtFileItemStore) Close() error {
return fis.db.Close()
}

View file

@ -0,0 +1,92 @@
package torrent
import (
"path"
"slices"
"sync"
"git.kmsign.ru/royalcat/tstor/pkg/slicesutils"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/types/infohash"
)
type dupInfo struct {
infohash infohash.T
fileinfo metainfo.FileInfo
}
type dupIndex struct {
mu sync.RWMutex
torrents map[infohash.T][]metainfo.FileInfo
sha1 map[string][]dupInfo // bittorrent v1
piecesRoot map[[32]byte][]dupInfo // bittorrent v2
}
func newDupIndex() *dupIndex {
return &dupIndex{
torrents: map[infohash.T][]metainfo.FileInfo{},
sha1: map[string][]dupInfo{},
piecesRoot: map[[32]byte][]dupInfo{},
}
}
func (c *dupIndex) AddFile(fileinfo metainfo.FileInfo, ih infohash.T) {
c.mu.Lock()
defer c.mu.Unlock()
c.torrents[ih] = append(c.torrents[ih], fileinfo)
if fileinfo.Sha1 != "" {
c.sha1[fileinfo.Sha1] = append(c.sha1[fileinfo.Sha1], dupInfo{fileinfo: fileinfo, infohash: ih})
}
if fileinfo.PiecesRoot.Ok {
c.piecesRoot[fileinfo.PiecesRoot.Value] = append(c.piecesRoot[fileinfo.PiecesRoot.Value], dupInfo{fileinfo: fileinfo, infohash: ih})
}
}
func (c *dupIndex) DuplicateFiles(fileinfo metainfo.FileInfo, ih infohash.T) []dupInfo {
c.mu.RLock()
defer c.mu.RUnlock()
if fileinfo.Sha1 != "" {
if dups, ok := c.sha1[fileinfo.Sha1]; ok {
return slices.Clone(dups)
}
}
if fileinfo.PiecesRoot.Ok {
if dups, ok := c.piecesRoot[fileinfo.PiecesRoot.Value]; ok {
return slices.Clone(dups)
}
}
return []dupInfo{}
}
func (c *dupIndex) Includes(ih infohash.T, files []metainfo.FileInfo) []dupInfo {
c.mu.RLock()
defer c.mu.RUnlock()
out := []dupInfo{}
for ih, v := range c.torrents {
intersection := slicesutils.IntersectionFunc(files, v, func(a, b metainfo.FileInfo) bool {
mostly := path.Join(a.BestPath()...) == path.Join(b.BestPath()...) && a.Length == b.Length
if a.Sha1 != "" && b.Sha1 != "" {
return mostly && a.Sha1 == b.Sha1
}
if a.PiecesRoot.Ok && b.PiecesRoot.Ok {
return mostly && a.PiecesRoot.Value == b.PiecesRoot.Value
}
return mostly
})
for _, v := range intersection {
out = append(out, dupInfo{infohash: ih, fileinfo: v})
}
}
return []dupInfo{}
}

View file

@ -0,0 +1,99 @@
package torrent
import (
"context"
"log/slog"
"git.kmsign.ru/royalcat/tstor/pkg/kvsingle"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/types"
"github.com/royalcat/kv"
)
type FileController struct {
file *torrent.File
properties *kvsingle.Value[string, FileProperties]
log *rlog.Logger
}
func NewFileController(f *torrent.File, properties *kvsingle.Value[string, FileProperties], log *rlog.Logger) *FileController {
return &FileController{
file: f,
properties: properties,
log: log.WithComponent("file-controller").With(slog.String("file", f.Path())),
}
}
func (s *FileController) Properties(ctx context.Context) (FileProperties, error) {
p, err := s.properties.Get(ctx)
if err == kv.ErrKeyNotFound {
return FileProperties{
Excluded: false,
Priority: defaultPriority,
}, nil
}
if err != nil {
return FileProperties{}, err
}
return p, nil
}
func (s *FileController) SetPriority(ctx context.Context, priority types.PiecePriority) error {
log := s.log.With(slog.Int("priority", int(priority)))
err := s.properties.Edit(ctx, func(ctx context.Context, v FileProperties) (FileProperties, error) {
v.Priority = priority
return v, nil
})
if err == kv.ErrKeyNotFound {
seterr := s.properties.Set(ctx, FileProperties{
Priority: priority,
})
if seterr != nil {
return err
}
err = nil
}
if err != nil {
return err
}
log.Debug(ctx, "file priority set")
s.file.SetPriority(priority)
return nil
}
func (s *FileController) FileInfo() metainfo.FileInfo {
return s.file.FileInfo()
}
func (s *FileController) Excluded(ctx context.Context) (bool, error) {
p, err := s.properties.Get(ctx)
if err == kv.ErrKeyNotFound {
return false, nil
}
if err != nil {
return false, err
}
return p.Excluded, nil
}
func (s *FileController) Path() string {
return s.file.Path()
}
func (s *FileController) Size() int64 {
return s.file.Length()
}
func (s *FileController) Priority() types.PiecePriority {
return s.file.Priority()
}
func (s *FileController) BytesCompleted() int64 {
return s.file.BytesCompleted()
}

567
daemons/torrent/fs.go Normal file
View file

@ -0,0 +1,567 @@
package torrent
import (
"context"
"fmt"
"io"
"io/fs"
"log/slog"
"path"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/anacrolix/torrent"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"golang.org/x/exp/maps"
)
type TorrentFS struct {
name string
Torrent *Controller
filesCacheMu sync.Mutex
filesCache map[string]vfs.File
lastTorrentReadTimeout atomic.Pointer[time.Time]
resolver *vfs.Resolver
}
var _ vfs.Filesystem = (*TorrentFS)(nil)
const shortTimeout = time.Millisecond
const lowTimeout = time.Second * 5
func (s *Daemon) NewTorrentFs(ctx context.Context, _ string, f vfs.File) (vfs.Filesystem, error) {
c, err := s.loadTorrent(ctx, f)
if err != nil {
return nil, err
}
if err := f.Close(ctx); err != nil {
s.log.Error(ctx, "failed to close file", slog.String("name", f.Name()), rlog.Error(err))
}
return &TorrentFS{
name: f.Name(),
Torrent: c,
resolver: vfs.NewResolver(vfs.ArchiveFactories),
}, nil
}
var _ fs.DirEntry = (*TorrentFS)(nil)
// Name implements fs.DirEntry.
func (tfs *TorrentFS) Name() string {
return tfs.name
}
// Info implements fs.DirEntry.
func (tfs *TorrentFS) Info() (fs.FileInfo, error) {
return tfs, nil
}
// IsDir implements fs.DirEntry.
func (tfs *TorrentFS) IsDir() bool {
return true
}
// Type implements fs.DirEntry.
func (tfs *TorrentFS) Type() fs.FileMode {
return fs.ModeDir
}
// ModTime implements fs.FileInfo.
func (tfs *TorrentFS) ModTime() time.Time {
return time.Time{}
}
// Mode implements fs.FileInfo.
func (tfs *TorrentFS) Mode() fs.FileMode {
return fs.ModeDir
}
// Size implements fs.FileInfo.
func (tfs *TorrentFS) Size() int64 {
return 0
}
// Sys implements fs.FileInfo.
func (tfs *TorrentFS) Sys() any {
return nil
}
// FsName implements Filesystem.
func (tfs *TorrentFS) FsName() string {
return "torrentfs"
}
func (fs *TorrentFS) files(ctx context.Context) (map[string]vfs.File, error) {
fs.filesCacheMu.Lock()
defer fs.filesCacheMu.Unlock()
if fs.filesCache != nil {
return fs.filesCache, nil
}
ctx, span := tracer.Start(ctx, "files", fs.traceAttrs())
defer span.End()
files, err := fs.Torrent.Files(ctx)
if err != nil {
return nil, err
}
fs.filesCache = make(map[string]vfs.File)
for _, file := range files {
props, err := file.Properties(ctx)
if err != nil {
return nil, err
}
if props.Excluded {
continue
}
p := vfs.AbsPath(file.Path())
tf, err := openTorrentFile(ctx, path.Base(p), file.file, &fs.lastTorrentReadTimeout)
if err != nil {
return nil, err
}
fs.filesCache[p] = tf
}
// TODO optional
// if len(fs.filesCache) == 1 && fs.resolver.IsNestedFs(fs.Torrent.Name()) {
// filepath := "/" + fs.Torrent.Name()
// if file, ok := fs.filesCache[filepath]; ok {
// nestedFs, err := fs.resolver.NestedFs(ctx, filepath, file)
// if err != nil {
// return nil, err
// }
// if nestedFs == nil {
// goto DEFAULT_DIR // FIXME
// }
// fs.filesCache, err = listFilesRecursive(ctx, nestedFs, "/")
// if err != nil {
// return nil, err
// }
// return fs.filesCache, nil
// }
// }
// DEFAULT_DIR:
rootDir := "/" + fs.Torrent.Name() + "/"
singleDir := true
for k, _ := range fs.filesCache {
if !strings.HasPrefix(k, rootDir) {
singleDir = false
}
}
if singleDir {
for k, f := range fs.filesCache {
delete(fs.filesCache, k)
k, _ = strings.CutPrefix(k, rootDir)
k = vfs.AbsPath(k)
fs.filesCache[k] = f
}
}
return fs.filesCache, nil
}
func listFilesRecursive(ctx context.Context, fs vfs.Filesystem, start string) (map[string]vfs.File, error) {
out := make(map[string]vfs.File, 0)
entries, err := fs.ReadDir(ctx, start)
if err != nil {
return nil, err
}
for _, entry := range entries {
filename := path.Join(start, entry.Name())
if entry.IsDir() {
rec, err := listFilesRecursive(ctx, fs, filename)
if err != nil {
return nil, err
}
maps.Copy(out, rec)
} else {
file, err := fs.Open(ctx, filename)
if err != nil {
return nil, err
}
out[filename] = file
}
}
return out, nil
}
func (fs *TorrentFS) rawOpen(ctx context.Context, filename string) (file vfs.File, err error) {
ctx, span := tracer.Start(ctx, "rawOpen",
fs.traceAttrs(attribute.String("filename", filename)),
)
defer func() {
if err != nil {
span.RecordError(err)
}
span.End()
}()
files, err := fs.files(ctx)
if err != nil {
return nil, err
}
file, err = vfs.GetFile(files, filename)
return file, err
}
func (fs *TorrentFS) rawStat(ctx context.Context, filename string) (fs.FileInfo, error) {
ctx, span := tracer.Start(ctx, "rawStat",
fs.traceAttrs(attribute.String("filename", filename)),
)
defer span.End()
files, err := fs.files(ctx)
if err != nil {
return nil, err
}
file, err := vfs.GetFile(files, filename)
if err != nil {
return nil, err
}
return file.Info()
}
func (fs *TorrentFS) traceAttrs(add ...attribute.KeyValue) trace.SpanStartOption {
return trace.WithAttributes(append([]attribute.KeyValue{
attribute.String("fs", fs.FsName()),
attribute.String("torrent", fs.Torrent.Name()),
attribute.String("infohash", fs.Torrent.InfoHash()),
}, add...)...)
}
func (tfs *TorrentFS) readContext(ctx context.Context) (context.Context, context.CancelFunc) {
lastReadTimeout := tfs.lastTorrentReadTimeout.Load()
if lastReadTimeout != nil && time.Since(*lastReadTimeout) < secondaryTimeout { // make short timeout for already faliled files
trace.SpanFromContext(ctx).SetAttributes(attribute.Bool("short_timeout", true))
return context.WithTimeout(ctx, shortTimeout)
}
return ctx, func() {}
}
// Stat implements Filesystem.
func (tfs *TorrentFS) Stat(ctx context.Context, filename string) (fs.FileInfo, error) {
ctx, span := tracer.Start(ctx, "Stat",
tfs.traceAttrs(attribute.String("filename", filename)),
)
defer span.End()
if vfs.IsRoot(filename) {
return tfs, nil
}
var err error
ctx, cancel := tfs.readContext(ctx)
defer func() {
cancel()
if err == context.DeadlineExceeded {
now := time.Now()
tfs.lastTorrentReadTimeout.Store(&now)
}
}()
fsPath, nestedFs, nestedFsPath, err := tfs.resolver.ResolvePath(ctx, filename, tfs.rawOpen)
if err != nil {
return nil, err
}
if nestedFs != nil {
return nestedFs.Stat(ctx, nestedFsPath)
}
return tfs.rawStat(ctx, fsPath)
}
func (tfs *TorrentFS) Open(ctx context.Context, filename string) (file vfs.File, err error) {
ctx, span := tracer.Start(ctx, "Open",
tfs.traceAttrs(attribute.String("filename", filename)),
)
defer span.End()
if vfs.IsRoot(filename) {
return vfs.NewDirFile(tfs.name), nil
}
ctx, cancel := tfs.readContext(ctx)
defer func() {
cancel()
if err == context.DeadlineExceeded {
now := time.Now()
tfs.lastTorrentReadTimeout.Store(&now)
}
}()
fsPath, nestedFs, nestedFsPath, err := tfs.resolver.ResolvePath(ctx, filename, tfs.rawOpen)
if err != nil {
return nil, err
}
if nestedFs != nil {
return nestedFs.Open(ctx, nestedFsPath)
}
return tfs.rawOpen(ctx, fsPath)
}
func (tfs *TorrentFS) ReadDir(ctx context.Context, name string) ([]fs.DirEntry, error) {
ctx, span := tracer.Start(ctx, "ReadDir",
tfs.traceAttrs(attribute.String("name", name)),
)
defer span.End()
var err error
ctx, cancel := tfs.readContext(ctx)
defer func() {
cancel()
if err == context.DeadlineExceeded {
now := time.Now()
tfs.lastTorrentReadTimeout.Store(&now)
}
}()
fsPath, nestedFs, nestedFsPath, err := tfs.resolver.ResolvePath(ctx, name, tfs.rawOpen)
if err != nil {
return nil, err
}
if nestedFs != nil {
return nestedFs.ReadDir(ctx, nestedFsPath)
}
files, err := tfs.files(ctx)
if err != nil {
return nil, err
}
return vfs.ListDirFromFiles(files, fsPath)
}
func (fs *TorrentFS) Unlink(ctx context.Context, name string) error {
ctx, span := tracer.Start(ctx, "Unlink",
fs.traceAttrs(attribute.String("name", name)),
)
defer span.End()
name = vfs.AbsPath(name)
files, err := fs.files(ctx)
if err != nil {
return err
}
if !slices.Contains(maps.Keys(files), name) {
return vfs.ErrNotExist
}
file := files[name]
fs.filesCacheMu.Lock()
delete(fs.filesCache, name)
fs.filesCacheMu.Unlock()
tfile, ok := file.(*torrentFile)
if !ok {
return vfs.ErrNotImplemented
}
return fs.Torrent.ExcludeFile(ctx, tfile.file)
}
// Rename implements vfs.Filesystem.
func (s *TorrentFS) Rename(ctx context.Context, oldpath string, newpath string) error {
return vfs.ErrNotImplemented
}
var _ vfs.File = (*torrentFile)(nil)
type torrentFile struct {
name string
mu sync.RWMutex
tr torrent.Reader
lastReadTimeout atomic.Pointer[time.Time]
lastTorrentReadTimeout *atomic.Pointer[time.Time]
file *torrent.File
}
const secondaryTimeout = time.Hour * 24
func openTorrentFile(ctx context.Context, name string, file *torrent.File, lastTorrentReadTimeout *atomic.Pointer[time.Time]) (*torrentFile, error) {
select {
case <-file.Torrent().GotInfo():
break
case <-ctx.Done():
return nil, ctx.Err()
}
r := file.NewReader()
_, err := r.ReadContext(ctx, make([]byte, 128))
if err != nil && err != io.EOF {
return nil, fmt.Errorf("failed initial file read: %w", err)
}
_, err = r.Seek(0, io.SeekStart)
if err != nil {
return nil, fmt.Errorf("failed seeking to start, after initial read: %w", err)
}
return &torrentFile{
name: name,
tr: r,
file: file,
lastTorrentReadTimeout: lastTorrentReadTimeout,
}, nil
}
// Name implements File.
func (tf *torrentFile) Name() string {
return tf.name
}
// Seek implements vfs.File.
func (tf *torrentFile) Seek(offset int64, whence int) (int64, error) {
tf.mu.Lock()
defer tf.mu.Unlock()
return tf.tr.Seek(offset, whence)
}
// Type implements File.
func (tf *torrentFile) Type() fs.FileMode {
return vfs.ModeFileRO | fs.ModeDir
}
func (tf *torrentFile) Info() (fs.FileInfo, error) {
return vfs.NewFileInfo(tf.name, tf.file.Length()), nil
}
func (tf *torrentFile) Size() int64 {
return tf.file.Length()
}
func (tf *torrentFile) IsDir() bool {
return false
}
func (rw *torrentFile) Close(ctx context.Context) error {
rw.mu.Lock()
defer rw.mu.Unlock()
return rw.tr.Close()
}
func (tf *torrentFile) readTimeout(ctx context.Context) (context.Context, context.CancelFunc) {
lastReadTimeout := tf.lastReadTimeout.Load()
if lastReadTimeout != nil && time.Since(*lastReadTimeout) < secondaryTimeout { // make short timeout for already faliled files
trace.SpanFromContext(ctx).SetAttributes(attribute.Bool("short_timeout", true))
return context.WithTimeout(ctx, shortTimeout)
}
lastTorrentReadTimeout := tf.lastTorrentReadTimeout.Load()
if lastTorrentReadTimeout != nil && time.Since(*lastTorrentReadTimeout) < secondaryTimeout { // make short timeout for already faliled files
trace.SpanFromContext(ctx).SetAttributes(attribute.Bool("low_timeout", true))
return context.WithTimeout(ctx, lowTimeout)
}
return ctx, func() {}
}
// Read implements ctxio.Reader.
func (tf *torrentFile) Read(ctx context.Context, p []byte) (n int, err error) {
ctx, span := tracer.Start(ctx, "Read",
trace.WithAttributes(attribute.Int("length", len(p))),
)
defer func() {
span.SetAttributes(attribute.Int("read", n))
span.End()
}()
tf.mu.Lock()
defer tf.mu.Unlock()
ctx, cancel := tf.readTimeout(ctx)
defer cancel()
defer func() {
if err == context.DeadlineExceeded {
now := time.Now()
tf.lastReadTimeout.Store(&now)
tf.lastTorrentReadTimeout.Store(&now)
}
}()
return tf.tr.ReadContext(ctx, p)
}
func (tf *torrentFile) ReadAt(ctx context.Context, p []byte, off int64) (n int, err error) {
ctx, span := tracer.Start(ctx, "ReadAt",
trace.WithAttributes(attribute.Int("length", len(p)), attribute.Int64("offset", off)),
)
defer func() {
span.SetAttributes(attribute.Int("read", n))
span.End()
}()
tf.mu.RLock()
defer tf.mu.RUnlock()
ctx, cancel := tf.readTimeout(ctx)
defer cancel()
defer func() {
if err == context.DeadlineExceeded {
now := time.Now()
tf.lastReadTimeout.Store(&now)
tf.lastTorrentReadTimeout.Store(&now)
}
}()
_, err = tf.tr.Seek(off, io.SeekStart)
if err != nil {
return 0, err
}
// return tf.tr.ReadContext(ctx, p)
n, err = readAtLeast(ctx, tf.tr, p, len(p))
_, err = tf.tr.Seek(0, io.SeekStart)
if err != nil {
return 0, err
}
return n, err
}
func readAtLeast(ctx context.Context, r torrent.Reader, buf []byte, min int) (n int, err error) {
if len(buf) < min {
return 0, io.ErrShortBuffer
}
for n < min && err == nil {
var nn int
nn, err = r.ReadContext(ctx, buf[n:])
n += nn
}
if n >= min {
err = nil
} else if n > 0 && err == io.EOF {
err = io.ErrUnexpectedEOF
}
return
}

142
daemons/torrent/fs_test.go Normal file
View file

@ -0,0 +1,142 @@
package torrent
import (
"os"
"testing"
"github.com/anacrolix/torrent"
)
const testMagnet = "magnet:?xt=urn:btih:a88fda5954e89178c372716a6a78b8180ed4dad3&dn=The+WIRED+CD+-+Rip.+Sample.+Mash.+Share&tr=udp%3A%2F%2Fexplodie.org%3A6969&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969&tr=udp%3A%2F%2Ftracker.empire-js.us%3A1337&tr=udp%3A%2F%2Ftracker.leechers-paradise.org%3A6969&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=wss%3A%2F%2Ftracker.btorrent.xyz&tr=wss%3A%2F%2Ftracker.fastcast.nz&tr=wss%3A%2F%2Ftracker.openwebtorrent.com&ws=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2F&xs=https%3A%2F%2Fwebtorrent.io%2Ftorrents%2Fwired-cd.torrent"
var Cli *torrent.Client
func TestMain(m *testing.M) {
cfg := torrent.NewDefaultClientConfig()
cfg.DataDir = os.TempDir()
// disable webseeds to avoid a panic when closing client on tests
cfg.DisableWebseeds = true
client, err := torrent.NewClient(cfg)
if err != nil {
panic(err)
}
Cli = client
exitVal := m.Run()
client.Close()
os.Exit(exitVal)
}
// func TestTorrentFilesystem(t *testing.T) {
// require := require.New(t)
// to, err := Cli.AddMagnet(testMagnet)
// require.NoError(err)
// tfs := NewTorrentFs(600)
// tfs.AddTorrent(to)
// files, err := tfs.ReadDir("/")
// require.NoError(err)
// require.Len(files, 1)
// require.Contains(files, "The WIRED CD - Rip. Sample. Mash. Share")
// files, err = tfs.ReadDir("/The WIRED CD - Rip. Sample. Mash. Share")
// require.NoError(err)
// require.Len(files, 18)
// f, err := tfs.Open("/The WIRED CD - Rip. Sample. Mash. Share/not_existing_file.txt")
// require.Equal(os.ErrNotExist, err)
// require.Nil(f)
// f, err = tfs.Open("/The WIRED CD - Rip. Sample. Mash. Share/01 - Beastie Boys - Now Get Busy.mp3")
// require.NoError(err)
// require.NotNil(f)
// require.Equal(f.Size(), int64(1964275))
// b := make([]byte, 10)
// n, err := f.Read(b)
// require.NoError(err)
// require.Equal(10, n)
// require.Equal([]byte{0x49, 0x44, 0x33, 0x3, 0x0, 0x0, 0x0, 0x0, 0x1f, 0x76}, b)
// n, err = f.ReadAt(b, 10)
// require.NoError(err)
// require.Equal(10, n)
// n, err = f.ReadAt(b, 10000)
// require.NoError(err)
// require.Equal(10, n)
// tfs.RemoveTorrent(to.InfoHash().String())
// files, err = tfs.ReadDir("/")
// require.NoError(err)
// require.Len(files, 0)
// require.NoError(f.Close())
// }
// func TestReadAtTorrent(t *testing.T) {
// t.Parallel()
// ctx := context.Background()
// require := require.New(t)
// to, err := Cli.AddMagnet(testMagnet)
// require.NoError(err)
// <-to.GotInfo()
// torrFile := to.Files()[0]
// tf, err := openTorrentFile(ctx, "torr", torrFile)
// require.NoError(err)
// defer tf.Close(ctx)
// toRead := make([]byte, 5)
// n, err := tf.ReadAt(ctx, toRead, 6)
// require.NoError(err)
// require.Equal(5, n)
// require.Equal([]byte{0x0, 0x0, 0x1f, 0x76, 0x54}, toRead)
// n, err = tf.ReadAt(ctx, toRead, 0)
// require.NoError(err)
// require.Equal(5, n)
// require.Equal([]byte{0x49, 0x44, 0x33, 0x3, 0x0}, toRead)
// }
// func TestReadAtWrapper(t *testing.T) {
// t.Parallel()
// ctx := context.Background()
// require := require.New(t)
// to, err := Cli.AddMagnet(testMagnet)
// require.NoError(err)
// <-to.GotInfo()
// torrFile := to.Files()[0]
// r, err := openTorrentFile(ctx, "file", torrFile)
// require.NoError(err)
// defer r.Close(ctx)
// toRead := make([]byte, 5)
// n, err := r.ReadAt(ctx, toRead, 6)
// require.NoError(err)
// require.Equal(5, n)
// require.Equal([]byte{0x0, 0x0, 0x1f, 0x76, 0x54}, toRead)
// n, err = r.ReadAt(ctx, toRead, 0)
// require.NoError(err)
// require.Equal(5, n)
// require.Equal([]byte{0x49, 0x44, 0x33, 0x3, 0x0}, toRead)
// }

View file

@ -0,0 +1,90 @@
package torrent
import (
"bytes"
"errors"
"fmt"
"path/filepath"
"git.kmsign.ru/royalcat/tstor/src/logwrap"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/types/infohash"
"github.com/dgraph-io/badger/v4"
)
var errNotFound = errors.New("not found")
type infoBytesStore struct {
db *badger.DB
}
func newInfoBytesStore(metaDir string) (*infoBytesStore, error) {
opts := badger.
DefaultOptions(filepath.Join(metaDir, "infobytes")).
WithLogger(logwrap.BadgerLogger("torrent-client", "infobytes"))
db, err := badger.Open(opts)
if err != nil {
return nil, err
}
return &infoBytesStore{db}, nil
}
func (k *infoBytesStore) GetBytes(ih infohash.T) ([]byte, error) {
var data []byte
err := k.db.View(func(tx *badger.Txn) error {
item, err := tx.Get(ih.Bytes())
if err != nil {
if err == badger.ErrKeyNotFound {
return errNotFound
}
return fmt.Errorf("error getting value: %w", err)
}
data, err = item.ValueCopy(data)
return err
})
return data, err
}
func (k *infoBytesStore) Get(ih infohash.T) (*metainfo.MetaInfo, error) {
data, err := k.GetBytes(ih)
if err != nil {
return nil, err
}
return metainfo.Load(bytes.NewReader(data))
}
func (me *infoBytesStore) SetBytes(ih infohash.T, data []byte) error {
return me.db.Update(func(txn *badger.Txn) error {
item, err := txn.Get(ih.Bytes())
if err != nil {
if err == badger.ErrKeyNotFound {
return txn.Set(ih.Bytes(), data)
}
return err
}
return item.Value(func(val []byte) error {
if !bytes.Equal(val, data) {
return txn.Set(ih.Bytes(), data)
}
return nil
})
})
}
func (me *infoBytesStore) Set(ih infohash.T, info metainfo.MetaInfo) error {
return me.SetBytes(ih, info.InfoBytes)
}
func (k *infoBytesStore) Delete(ih infohash.T) error {
return k.db.Update(func(txn *badger.Txn) error {
return txn.Delete(ih.Bytes())
})
}
func (me *infoBytesStore) Close() error {
return me.db.Close()
}

View file

@ -0,0 +1,69 @@
package torrent
import (
"context"
"encoding/base64"
"github.com/anacrolix/dht/v2"
"github.com/anacrolix/torrent"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
)
func registerTorrentMetrics(client *torrent.Client) error {
meterTotalPeers, _ := meter.Int64ObservableGauge("torrent.peers.total")
meterActivePeers, _ := meter.Int64ObservableGauge("torrent.peers.active")
meterSeeders, _ := meter.Int64ObservableGauge("torrent.seeders")
meterDownloaded, _ := meter.Int64ObservableGauge("torrent.downloaded", metric.WithUnit("By"))
meterIO, _ := meter.Int64ObservableGauge("torrent.io", metric.WithUnit("By"))
meterLoaded, _ := meter.Int64ObservableGauge("torrent.loaded")
_, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
o.ObserveInt64(meterLoaded, int64(len(client.Torrents())))
for _, v := range client.Torrents() {
as := attribute.NewSet(
attribute.String("infohash", v.InfoHash().HexString()),
attribute.String("name", v.Name()),
attribute.Int64("size", v.Length()),
)
stats := v.Stats()
o.ObserveInt64(meterTotalPeers, int64(stats.TotalPeers), metric.WithAttributeSet(as))
o.ObserveInt64(meterActivePeers, int64(stats.ActivePeers), metric.WithAttributeSet(as))
o.ObserveInt64(meterSeeders, int64(stats.ConnectedSeeders), metric.WithAttributeSet(as))
o.ObserveInt64(meterIO, stats.BytesRead.Int64(), metric.WithAttributeSet(as), metric.WithAttributes(attribute.String("direction", "download")))
o.ObserveInt64(meterIO, stats.BytesWritten.Int64(), metric.WithAttributeSet(as), metric.WithAttributes(attribute.String("direction", "upload")))
o.ObserveInt64(meterDownloaded, v.BytesCompleted(), metric.WithAttributeSet(as))
}
return nil
}, meterTotalPeers, meterActivePeers, meterSeeders, meterIO, meterDownloaded, meterLoaded)
if err != nil {
return err
}
return nil
}
func registerDhtMetrics(client *torrent.Client) error {
meterDhtNodes, _ := meter.Int64ObservableGauge("torrent.dht.nodes")
_, err := meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
servers := client.DhtServers()
for _, dhtSrv := range servers {
stats, ok := dhtSrv.Stats().(dht.ServerStats)
if !ok {
continue
}
id := dhtSrv.ID()
as := attribute.NewSet(
attribute.String("id", base64.StdEncoding.EncodeToString(id[:])),
attribute.String("address", dhtSrv.Addr().String()),
)
o.ObserveInt64(meterDhtNodes, int64(stats.Nodes), metric.WithAttributeSet(as))
}
return nil
}, meterDhtNodes)
return err
}

View file

@ -0,0 +1,24 @@
package torrent
import (
"github.com/anacrolix/dht/v2/krpc"
peer_store "github.com/anacrolix/dht/v2/peer-store"
"github.com/anacrolix/torrent/types/infohash"
"github.com/royalcat/kv"
)
type peerStore struct {
store kv.Store[infohash.T, []krpc.NodeAddr]
}
var _ peer_store.Interface = (*peerStore)(nil)
// AddPeer implements peer_store.Interface.
func (p *peerStore) AddPeer(ih infohash.T, node krpc.NodeAddr) {
panic("unimplemented")
}
// GetPeers implements peer_store.Interface.
func (p *peerStore) GetPeers(ih infohash.T) []krpc.NodeAddr {
panic("unimplemented")
}

View file

@ -0,0 +1,137 @@
package torrent
import (
"context"
"encoding/binary"
"fmt"
"git.kmsign.ru/royalcat/tstor/src/logwrap"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/storage"
"github.com/royalcat/kv"
"github.com/royalcat/kv/kvbadger"
)
type PieceCompletionState byte
const (
PieceNotComplete PieceCompletionState = 0
PieceComplete PieceCompletionState = 1<<8 - 1
)
var _ kv.Binary = (*PieceCompletionState)(nil)
// MarshalBinary implements kv.Binary.
func (p PieceCompletionState) MarshalBinary() (data []byte, err error) {
return []byte{byte(p)}, nil
}
// UnmarshalBinary implements kv.Binary.
func (p *PieceCompletionState) UnmarshalBinary(data []byte) error {
if len(data) != 1 {
return fmt.Errorf("bad length")
}
switch PieceCompletionState(data[0]) {
case PieceComplete:
*p = PieceComplete
case PieceNotComplete:
*p = PieceNotComplete
default:
*p = PieceNotComplete
}
return nil
}
func pieceCompletionState(i bool) PieceCompletionState {
if i {
return PieceComplete
}
return PieceNotComplete
}
type pieceKey metainfo.PieceKey
const pieceKeySize = metainfo.HashSize + 4
var _ kv.Binary = (*pieceKey)(nil)
// const delimeter rune = 0x1F
// MarshalBinary implements kv.Binary.
func (pk pieceKey) MarshalBinary() (data []byte, err error) {
key := make([]byte, 0, pieceKeySize)
key = append(key, pk.InfoHash.Bytes()...)
key = binary.BigEndian.AppendUint32(key, uint32(pk.Index))
return key, nil
}
// UnmarshalBinary implements kv.Binary.
func (p *pieceKey) UnmarshalBinary(data []byte) error {
if len(data) < pieceKeySize {
return fmt.Errorf("data too short")
}
p.InfoHash = metainfo.Hash(data[:metainfo.HashSize])
p.Index = int(binary.BigEndian.Uint32(data[metainfo.HashSize:]))
return nil
}
type badgerPieceCompletion struct {
db kv.Store[pieceKey, PieceCompletionState]
}
var _ storage.PieceCompletion = (*badgerPieceCompletion)(nil)
func newPieceCompletion(dir string) (storage.PieceCompletion, error) {
opts := kvbadger.DefaultOptions[PieceCompletionState](dir)
opts.Codec = kv.CodecBinary[PieceCompletionState, *PieceCompletionState]{}
opts.BadgerOptions = opts.BadgerOptions.WithLogger(logwrap.BadgerLogger("torrent-client", "piece-completion"))
db, err := kvbadger.NewBagerKVBinaryKey[pieceKey, PieceCompletionState](opts)
if err != nil {
return nil, err
}
return &badgerPieceCompletion{
db: db,
}, nil
}
func (c *badgerPieceCompletion) Get(pk metainfo.PieceKey) (completion storage.Completion, err error) {
ctx := context.Background()
state, err := c.db.Get(ctx, pieceKey(pk))
if err != nil {
if err == kv.ErrKeyNotFound {
return completion, nil
}
return completion, err
}
if state == PieceComplete {
return storage.Completion{
Complete: true,
Ok: true,
}, nil
}
return storage.Completion{
Complete: false,
Ok: true,
}, nil
}
func (me badgerPieceCompletion) Set(pk metainfo.PieceKey, b bool) error {
ctx := context.Background()
if c, err := me.Get(pk); err == nil && c.Ok && c.Complete == b {
return nil
}
return me.db.Set(ctx, pieceKey(pk), pieceCompletionState(b))
}
func (me *badgerPieceCompletion) Close() error {
return me.db.Close(context.Background())
}

View file

@ -0,0 +1,36 @@
package torrent
import (
"testing"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/storage"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestBoltPieceCompletion(t *testing.T) {
td := t.TempDir()
pc, err := newPieceCompletion(td)
require.NoError(t, err)
defer pc.Close()
pk := metainfo.PieceKey{}
b, err := pc.Get(pk)
require.NoError(t, err)
assert.False(t, b.Ok)
require.NoError(t, pc.Set(pk, false))
b, err = pc.Get(pk)
require.NoError(t, err)
assert.Equal(t, storage.Completion{Complete: false, Ok: true}, b)
require.NoError(t, pc.Set(pk, true))
b, err = pc.Get(pk)
require.NoError(t, err)
assert.Equal(t, storage.Completion{Complete: true, Ok: true}, b)
}

135
daemons/torrent/queue.go Normal file
View file

@ -0,0 +1,135 @@
package torrent
import (
"context"
"fmt"
"git.kmsign.ru/royalcat/tstor/pkg/uuid"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/types/infohash"
)
type DownloadTask struct {
ID uuid.UUID
InfoHash infohash.T
File string
}
func (s *Daemon) Download(ctx context.Context, task *DownloadTask) error {
t, ok := s.client.Torrent(task.InfoHash)
if !ok {
return fmt.Errorf("torrent with IH %s not found", task.InfoHash.HexString())
}
if task.File != "" {
var file *torrent.File
for _, tf := range t.Files() {
if tf.Path() == task.File {
file = tf
break
}
}
if file == nil {
return fmt.Errorf("file %s not found in torrent torrent with IH %s", task.File, task.InfoHash.HexString())
}
file.Download()
} else {
for _, file := range t.Files() {
file.Download()
}
}
return nil
}
// func (s *Service) DownloadAndWait(ctx context.Context, task *TorrentDownloadTask) error {
// t, ok := s.c.Torrent(task.InfoHash)
// if !ok {
// return fmt.Errorf("torrent with IH %s not found", task.InfoHash.HexString())
// }
// if task.File != "" {
// var file *torrent.File
// for _, tf := range t.Files() {
// if tf.Path() == task.File {
// file = tf
// break
// }
// }
// if file == nil {
// return fmt.Errorf("file %s not found in torrent torrent with IH %s", task.File, task.InfoHash.HexString())
// }
// file.Download()
// return waitPieceRange(ctx, t, file.BeginPieceIndex(), file.EndPieceIndex())
// }
// t.DownloadAll()
// select {
// case <-ctx.Done():
// return ctx.Err()
// case <-t.Complete.On():
// return nil
// }
// }
// func waitPieceRange(ctx context.Context, t *torrent.Torrent, start, end int) error {
// for i := start; i < end; i++ {
// timer := time.NewTimer(time.Millisecond)
// for {
// select {
// case <-ctx.Done():
// return ctx.Err()
// case <-timer.C:
// if t.PieceState(i).Complete {
// continue
// }
// }
// }
// }
// return nil
// }
type TorrentProgress struct {
Torrent *Controller
Current int64
Total int64
}
func (s *Daemon) DownloadProgress(ctx context.Context) (<-chan TorrentProgress, error) {
torrents, err := s.ListTorrents(ctx)
if err != nil {
return nil, err
}
out := make(chan TorrentProgress, 1)
go func() {
defer close(out)
for _, t := range torrents {
sub := t.Torrent().SubscribePieceStateChanges()
go func(t *Controller) {
for stateChange := range sub.Values {
if !stateChange.Complete && !stateChange.Partial {
continue
}
out <- TorrentProgress{
Torrent: t,
Current: t.BytesCompleted(),
Total: t.Length(),
}
}
}(t)
defer sub.Close()
}
<-ctx.Done()
}()
return out, nil
}

56
daemons/torrent/setup.go Normal file
View file

@ -0,0 +1,56 @@
package torrent
import (
"fmt"
"os"
"path/filepath"
"git.kmsign.ru/royalcat/tstor/src/config"
"github.com/anacrolix/torrent/storage"
)
func setupStorage(cfg config.TorrentClient) (*fileStorage, storage.PieceCompletion, error) {
pcp := filepath.Join(cfg.MetadataFolder, "piece-completion")
if err := os.MkdirAll(pcp, 0744); err != nil {
return nil, nil, fmt.Errorf("error creating piece completion folder: %w", err)
}
// pc, err := storage.NewBoltPieceCompletion(pcp)
// if err != nil {
// return nil, nil, err
// }
pc, err := newPieceCompletion(pcp)
if err != nil {
return nil, nil, fmt.Errorf("error creating servers piece completion: %w", err)
}
// TODO implement cache/storage switching
// cacheDir := filepath.Join(tcfg.DataFolder, "cache")
// if err := os.MkdirAll(cacheDir, 0744); err != nil {
// return nil, nil, fmt.Errorf("error creating piece completion folder: %w", err)
// }
// fc, err := filecache.NewCache(cacheDir)
// if err != nil {
// return nil, nil, fmt.Errorf("error creating cache: %w", err)
// }
// log.Info().Msg(fmt.Sprintf("setting cache size to %d MB", 1024))
// fc.SetCapacity(1024 * 1024 * 1024)
// rp := storage.NewResourcePieces(fc.AsResourceProvider())
// st := &stc{rp}
filesDir := cfg.DataFolder
if err := os.MkdirAll(filesDir, 0744); err != nil {
return nil, nil, fmt.Errorf("error creating piece completion folder: %w", err)
}
st := NewFileStorage(filesDir, pc)
// piecesDir := filepath.Join(cfg.DataFolder, ".pieces")
// if err := os.MkdirAll(piecesDir, 0744); err != nil {
// return nil, nil, fmt.Errorf("error creating piece completion folder: %w", err)
// }
// st := storage.NewMMapWithCompletion(piecesDir, pc)
return st, pc, nil
}

207
daemons/torrent/stats.go Normal file
View file

@ -0,0 +1,207 @@
package torrent
import (
"context"
"encoding/json"
"path"
"slices"
"time"
"git.kmsign.ru/royalcat/tstor/src/logwrap"
"github.com/anacrolix/torrent/types/infohash"
"github.com/dgraph-io/badger/v4"
)
func newStatsStore(metaDir string, lifetime time.Duration) (*statsStore, error) {
db, err := badger.OpenManaged(
badger.
DefaultOptions(path.Join(metaDir, "stats")).
WithNumVersionsToKeep(int(^uint(0) >> 1)).
WithLogger(logwrap.BadgerLogger("stats")), // Infinity
)
if err != nil {
return nil, err
}
go func() {
for n := range time.NewTimer(lifetime / 2).C {
db.SetDiscardTs(uint64(n.Add(-lifetime).Unix()))
}
}()
return &statsStore{
db: db,
}, nil
}
type statsStore struct {
db *badger.DB
}
type TorrentStats struct {
Timestamp time.Time
DownloadedBytes uint64
UploadedBytes uint64
TotalPeers uint16
ActivePeers uint16
ConnectedSeeders uint16
}
func (s TorrentStats) Same(o TorrentStats) bool {
return s.DownloadedBytes == o.DownloadedBytes &&
s.UploadedBytes == o.UploadedBytes &&
s.TotalPeers == o.TotalPeers &&
s.ActivePeers == o.ActivePeers &&
s.ConnectedSeeders == o.ConnectedSeeders
}
func (r *statsStore) addStats(key []byte, stat TorrentStats) error {
ts := uint64(stat.Timestamp.Unix())
txn := r.db.NewTransactionAt(ts, true)
defer txn.Discard()
item, err := txn.Get(key)
if err != nil && err != badger.ErrKeyNotFound {
return err
}
if err != badger.ErrKeyNotFound {
var prevStats TorrentStats
err = item.Value(func(val []byte) error {
return json.Unmarshal(val, &prevStats)
})
if err != nil {
return err
}
if prevStats.Same(stat) {
return nil
}
}
data, err := json.Marshal(stat)
if err != nil {
return err
}
err = txn.Set(key, data)
if err != nil {
return err
}
return txn.CommitAt(ts, nil)
}
func (r *statsStore) AddTorrentStats(ih infohash.T, stat TorrentStats) error {
return r.addStats(ih.Bytes(), stat)
}
const totalKey = "total"
func (r *statsStore) AddTotalStats(stat TorrentStats) error {
return r.addStats([]byte(totalKey), stat)
}
func (r *statsStore) ReadTotalStatsHistory(ctx context.Context, since time.Time) ([]TorrentStats, error) {
stats := []TorrentStats{}
err := r.db.View(func(txn *badger.Txn) error {
opts := badger.DefaultIteratorOptions
opts.AllVersions = true
opts.SinceTs = uint64(since.Unix())
it := txn.NewKeyIterator([]byte(totalKey), opts)
defer it.Close()
for it.Rewind(); it.Valid(); it.Next() {
item := it.Item()
var stat TorrentStats
err := item.Value(func(v []byte) error {
return json.Unmarshal(v, &stat)
})
if err != nil {
return err
}
stats = append(stats, stat)
}
return nil
})
if err != nil {
return nil, err
}
slices.SortFunc(stats, func(a, b TorrentStats) int {
return a.Timestamp.Compare(b.Timestamp)
})
stats = slices.Compact(stats)
return stats, nil
}
func (r *statsStore) ReadTorrentStatsHistory(ctx context.Context, since time.Time, ih infohash.T) ([]TorrentStats, error) {
stats := []TorrentStats{}
err := r.db.View(func(txn *badger.Txn) error {
opts := badger.DefaultIteratorOptions
opts.AllVersions = true
opts.SinceTs = uint64(since.Unix())
it := txn.NewKeyIterator(ih.Bytes(), opts)
defer it.Close()
for it.Rewind(); it.Valid(); it.Next() {
item := it.Item()
var stat TorrentStats
err := item.Value(func(v []byte) error {
return json.Unmarshal(v, &stat)
})
if err != nil {
return err
}
stats = append(stats, stat)
}
return nil
})
if err != nil {
return nil, err
}
slices.SortFunc(stats, func(a, b TorrentStats) int {
return a.Timestamp.Compare(b.Timestamp)
})
stats = slices.Compact(stats)
return stats, nil
}
func (r *statsStore) ReadStatsHistory(ctx context.Context, since time.Time) ([]TorrentStats, error) {
stats := []TorrentStats{}
err := r.db.View(func(txn *badger.Txn) error {
opts := badger.DefaultIteratorOptions
opts.AllVersions = true
opts.SinceTs = uint64(since.Unix())
it := txn.NewIterator(opts)
defer it.Close()
for it.Rewind(); it.Valid(); it.Next() {
item := it.Item()
var stat TorrentStats
err := item.Value(func(v []byte) error {
return json.Unmarshal(v, &stat)
})
if err != nil {
return err
}
stats = append(stats, stat)
}
return nil
})
if err != nil {
return nil, err
}
slices.SortFunc(stats, func(a, b TorrentStats) int {
return a.Timestamp.Compare(b.Timestamp)
})
stats = slices.Compact(stats)
return stats, nil
}

196
daemons/torrent/storage.go Normal file
View file

@ -0,0 +1,196 @@
package torrent
import (
"context"
"errors"
"io/fs"
"log/slog"
"os"
"path"
"path/filepath"
"slices"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/storage"
)
// NewFileStorage creates a new ClientImplCloser that stores files using the OS native filesystem.
func NewFileStorage(baseDir string, pc storage.PieceCompletion) *fileStorage {
return &fileStorage{
client: storage.NewFileOpts(storage.NewFileClientOpts{
ClientBaseDir: baseDir,
PieceCompletion: pc,
TorrentDirMaker: func(baseDir string, info *metainfo.Info, infoHash metainfo.Hash) string {
return torrentDir(baseDir, infoHash)
},
FilePathMaker: func(opts storage.FilePathMakerOpts) string {
return filePath(*opts.File)
},
}),
baseDir: baseDir,
pieceCompletion: pc,
dupIndex: newDupIndex(),
log: rlog.Component("daemon", "torrent"),
}
}
// File-based storage for torrents, that isn't yet bound to a particular torrent.
type fileStorage struct {
baseDir string
client storage.ClientImplCloser
pieceCompletion storage.PieceCompletion
dupIndex *dupIndex
log *rlog.Logger
}
var _ storage.ClientImplCloser = (*fileStorage)(nil)
func (me *fileStorage) Close() error {
return errors.Join(
me.client.Close(),
me.pieceCompletion.Close(),
)
}
func (fs *fileStorage) fullFilePath(infoHash metainfo.Hash, fileInfo metainfo.FileInfo) string {
return filepath.Join(
torrentDir(fs.baseDir, infoHash),
filePath(fileInfo),
)
}
func (fs *fileStorage) DeleteFile(file *torrent.File) error {
infoHash := file.Torrent().InfoHash()
torrentDir := torrentDir(fs.baseDir, infoHash)
fileInfo := file.FileInfo()
relFilePath := filePath(fileInfo)
filePath := path.Join(torrentDir, relFilePath)
for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ {
pk := metainfo.PieceKey{InfoHash: infoHash, Index: i}
err := fs.pieceCompletion.Set(pk, false)
if err != nil {
return err
}
}
return os.Remove(filePath)
}
func (fs *fileStorage) CleanupDirs(ctx context.Context, expected []*Controller, dryRun bool) ([]string, error) {
log := fs.log.With(slog.Int("expectedTorrents", len(expected)), slog.Bool("dryRun", dryRun))
expectedEntries := []string{}
for _, e := range expected {
expectedEntries = append(expectedEntries, e.Torrent().InfoHash().HexString())
}
entries, err := os.ReadDir(fs.baseDir)
if err != nil {
return nil, err
}
toDelete := []string{}
for _, v := range entries {
if !slices.Contains(expectedEntries, v.Name()) {
toDelete = append(toDelete, v.Name())
}
}
if ctx.Err() != nil {
return nil, ctx.Err()
}
log.Info(ctx, "deleting trash data", slog.Int("dirsCount", len(toDelete)))
if !dryRun {
for i, name := range toDelete {
p := path.Join(fs.baseDir, name)
log.Warn(ctx, "deleting trash data", slog.String("path", p))
err := os.RemoveAll(p)
if err != nil {
return toDelete[:i], err
}
}
}
return toDelete, nil
}
func (s *fileStorage) CleanupFiles(ctx context.Context, expected []*Controller, dryRun bool) ([]string, error) {
log := s.log.With(slog.Int("expectedTorrents", len(expected)), slog.Bool("dryRun", dryRun))
expectedEntries := []string{}
{
for _, e := range expected {
files, err := e.Files(ctx)
if err != nil {
return nil, err
}
for _, f := range files {
expectedEntries = append(expectedEntries, s.fullFilePath(e.Torrent().InfoHash(), f.FileInfo()))
}
}
}
entries := []string{}
err := filepath.WalkDir(s.baseDir,
func(path string, info fs.DirEntry, err error) error {
if err != nil {
return err
}
if ctx.Err() != nil {
return ctx.Err()
}
if info.IsDir() {
return nil
}
entries = append(entries, path)
return nil
})
if err != nil {
return nil, err
}
toDelete := []string{}
for _, v := range entries {
if !slices.Contains(expectedEntries, v) {
toDelete = append(toDelete, v)
}
}
if ctx.Err() != nil {
return toDelete, ctx.Err()
}
log.Info(ctx, "deleting trash data", slog.Int("filesCount", len(toDelete)))
if !dryRun {
for i, p := range toDelete {
s.log.Warn(ctx, "deleting trash data", slog.String("path", p))
err := os.Remove(p)
if err != nil {
return toDelete[i:], err
}
}
}
return toDelete, nil
}
func (s *fileStorage) iterFiles(ctx context.Context, iter func(ctx context.Context, path string, entry fs.FileInfo) error) error {
return filepath.Walk(s.baseDir,
func(path string, info fs.FileInfo, err error) error {
if err != nil {
return err
}
if ctx.Err() != nil {
return ctx.Err()
}
if info.IsDir() {
return nil
}
return iter(ctx, path, info)
})
}

View file

@ -0,0 +1,225 @@
package torrent
import (
"context"
"crypto/sha1"
"fmt"
"io"
"io/fs"
"log/slog"
"os"
"path/filepath"
"slices"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/dustin/go-humanize"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"golang.org/x/exp/maps"
"golang.org/x/sys/unix"
)
func (s *fileStorage) Dedupe(ctx context.Context) (uint64, error) {
ctx, span := tracer.Start(ctx, fmt.Sprintf("Dedupe"))
defer span.End()
log := s.log
sizeMap := map[int64][]string{}
err := s.iterFiles(ctx, func(ctx context.Context, path string, info fs.FileInfo) error {
size := info.Size()
sizeMap[size] = append(sizeMap[size], path)
return nil
})
if err != nil {
return 0, err
}
maps.DeleteFunc(sizeMap, func(k int64, v []string) bool {
return len(v) <= 1
})
span.AddEvent("collected files with same size", trace.WithAttributes(
attribute.Int("count", len(sizeMap)),
))
var deduped uint64 = 0
i := 0
for _, paths := range sizeMap {
if i%100 == 0 {
log.Info(ctx, "deduping in progress", slog.Int("current", i), slog.Int("total", len(sizeMap)))
}
i++
if ctx.Err() != nil {
return deduped, ctx.Err()
}
slices.Sort(paths)
paths = slices.Compact(paths)
if len(paths) <= 1 {
continue
}
paths, err = applyErr(paths, filepath.Abs)
if err != nil {
return deduped, err
}
dedupedGroup, err := s.dedupeFiles(ctx, paths)
if err != nil {
log.Error(ctx, "Error applying dedupe", slog.Any("files", paths), rlog.Error(err))
continue
}
if dedupedGroup > 0 {
deduped += dedupedGroup
log.Info(ctx, "deduped file group",
slog.String("files", fmt.Sprint(paths)),
slog.String("deduped", humanize.Bytes(dedupedGroup)),
slog.String("deduped_total", humanize.Bytes(deduped)),
)
}
}
return deduped, nil
}
func applyErr[E, O any](in []E, apply func(E) (O, error)) ([]O, error) {
out := make([]O, 0, len(in))
for _, p := range in {
o, err := apply(p)
if err != nil {
return out, err
}
out = append(out, o)
}
return out, nil
}
// const blockSize uint64 = 4096
func (s *fileStorage) dedupeFiles(ctx context.Context, paths []string) (deduped uint64, err error) {
ctx, span := tracer.Start(ctx, fmt.Sprintf("dedupeFiles"), trace.WithAttributes(
attribute.StringSlice("files", paths),
))
defer func() {
span.SetAttributes(attribute.Int64("deduped", int64(deduped)))
if err != nil {
span.RecordError(err)
}
span.End()
}()
log := s.log
srcF, err := os.Open(paths[0])
if err != nil {
return deduped, fmt.Errorf("error opening file %s: %w", paths[0], err)
}
defer srcF.Close()
srcStat, err := srcF.Stat()
if err != nil {
return deduped, fmt.Errorf("error stat file %s: %w", paths[0], err)
}
srcFd := int(srcF.Fd())
srcSize := srcStat.Size()
fsStat := unix.Statfs_t{}
err = unix.Fstatfs(srcFd, &fsStat)
if err != nil {
span.RecordError(err)
return deduped, fmt.Errorf("error statfs file %s: %w", paths[0], err)
}
srcHash, err := filehash(srcF)
if err != nil {
return deduped, fmt.Errorf("error hashing file %s: %w", paths[0], err)
}
if int64(fsStat.Bsize) > srcSize { // for btrfs it means file in residing in not deduplicatable metadata
return deduped, nil
}
blockSize := uint64((srcSize % int64(fsStat.Bsize)) * int64(fsStat.Bsize))
span.SetAttributes(attribute.Int64("blocksize", int64(blockSize)))
rng := unix.FileDedupeRange{
Src_offset: 0,
Src_length: blockSize,
Info: []unix.FileDedupeRangeInfo{},
}
for _, dst := range paths[1:] {
if ctx.Err() != nil {
return deduped, ctx.Err()
}
destF, err := os.OpenFile(dst, os.O_RDWR, os.ModePerm)
if err != nil {
return deduped, fmt.Errorf("error opening file %s: %w", dst, err)
}
defer destF.Close()
dstHash, err := filehash(destF)
if err != nil {
return deduped, fmt.Errorf("error hashing file %s: %w", dst, err)
}
if srcHash != dstHash {
destF.Close()
continue
}
rng.Info = append(rng.Info, unix.FileDedupeRangeInfo{
Dest_fd: int64(destF.Fd()),
Dest_offset: 0,
})
}
if len(rng.Info) == 0 {
return deduped, nil
}
log.Info(ctx, "found same files, deduping", slog.Any("files", paths), slog.String("size", humanize.Bytes(uint64(srcStat.Size()))))
if ctx.Err() != nil {
return deduped, ctx.Err()
}
rng.Src_offset = 0
for i := range rng.Info {
rng.Info[i].Dest_offset = 0
}
err = unix.IoctlFileDedupeRange(srcFd, &rng)
if err != nil {
return deduped, fmt.Errorf("error calling FIDEDUPERANGE: %w", err)
}
for i := range rng.Info {
deduped += rng.Info[i].Bytes_deduped
rng.Info[i].Status = 0
rng.Info[i].Bytes_deduped = 0
}
return deduped, nil
}
const compareBlockSize = 1024 * 128
func filehash(r io.Reader) ([20]byte, error) {
buf := make([]byte, compareBlockSize)
_, err := r.Read(buf)
if err != nil && err != io.EOF {
return [20]byte{}, err
}
return sha1.Sum(buf), nil
}

View file

@ -0,0 +1,179 @@
package torrent
import (
"context"
"errors"
"fmt"
"log/slog"
"os"
"path/filepath"
"slices"
"strings"
"git.kmsign.ru/royalcat/tstor/pkg/cowutils"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/storage"
"github.com/anacrolix/torrent/types/infohash"
"github.com/royalcat/kv"
)
// OpenTorrent implements storage.ClientImplCloser.
func (me *fileStorage) OpenTorrent(ctx context.Context, info *metainfo.Info, infoHash infohash.T) (storage.TorrentImpl, error) {
ctx, span := tracer.Start(ctx, "OpenTorrent")
defer span.End()
log := me.log.With(slog.String("infohash", infoHash.HexString()), slog.String("name", info.BestName()))
log.Debug(ctx, "opening torrent")
impl, err := me.client.OpenTorrent(ctx, info, infoHash)
if err != nil {
log.Error(ctx, "error opening torrent", rlog.Error(err))
}
return impl, err
}
func (me *fileStorage) copyDup(ctx context.Context, infoHash infohash.T, dup dupInfo) error {
log := me.log.With(slog.String("infohash", infoHash.HexString()), slog.String("dup_infohash", dup.infohash.HexString()))
srcPath := me.fullFilePath(dup.infohash, dup.fileinfo)
src, err := os.Open(me.fullFilePath(dup.infohash, dup.fileinfo))
if err != nil {
return err
}
dstPath := me.fullFilePath(infoHash, dup.fileinfo)
dst, err := os.OpenFile(dstPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)
if err != nil {
return err
}
log.Info(ctx, "copying duplicate file", slog.String("src", srcPath), slog.String("dst", dstPath))
err = cowutils.Reflink(ctx, dst, src, true)
if err != nil {
return fmt.Errorf("error copying file: %w", err)
}
return nil
}
func torrentDir(baseDir string, infoHash metainfo.Hash) string {
return filepath.Join(baseDir, infoHash.HexString())
}
func filePath(file metainfo.FileInfo) string {
return filepath.Join(file.BestPath()...)
}
func (s *Daemon) checkTorrentCompatable(ctx context.Context, ih infohash.T, info metainfo.Info) (compatable bool, tryLater bool, err error) {
log := s.log.With(
slog.String("new-name", info.BestName()),
slog.String("new-infohash", ih.String()),
)
name := info.BestName()
aq, err := s.dirsAquire.Get(ctx, info.BestName())
if errors.Is(err, kv.ErrKeyNotFound) {
err = s.dirsAquire.Set(ctx, name, DirAquire{
Name: name,
Hashes: slices.Compact([]infohash.T{ih}),
})
if err != nil {
return false, false, err
}
log.Debug(ctx, "acquiring was not found, so created")
return true, false, nil
} else if err != nil {
return false, false, err
}
if slices.Contains(aq.Hashes, ih) {
log.Debug(ctx, "hash already know to be compatable")
return true, false, nil
}
for _, existingTorrent := range s.client.Torrents() {
if existingTorrent.Name() != name || existingTorrent.InfoHash() == ih {
continue
}
existingInfo := existingTorrent.Info()
existingFiles := slices.Clone(existingInfo.Files)
newFiles := slices.Clone(info.Files)
if !s.checkTorrentFilesCompatable(ctx, aq, existingFiles, newFiles) {
return false, false, nil
}
aq.Hashes = slicesUnique(append(aq.Hashes, ih))
err = s.dirsAquire.Set(ctx, aq.Name, aq)
if err != nil {
log.Warn(ctx, "torrent not compatible")
return false, false, err
}
}
if slices.Contains(aq.Hashes, ih) {
log.Debug(ctx, "hash is compatable")
return true, false, nil
}
log.Debug(ctx, "torrent with same name not found, try later")
return false, true, nil
}
func (s *Daemon) checkTorrentFilesCompatable(ctx context.Context, aq DirAquire, existingFiles, newFiles []metainfo.FileInfo) bool {
log := s.log.With(slog.String("name", aq.Name))
pathCmp := func(a, b metainfo.FileInfo) int {
return slices.Compare(a.BestPath(), b.BestPath())
}
slices.SortStableFunc(existingFiles, pathCmp)
slices.SortStableFunc(newFiles, pathCmp)
// torrents basically equals
if slices.EqualFunc(existingFiles, newFiles, func(fi1, fi2 metainfo.FileInfo) bool {
return fi1.Length == fi2.Length && slices.Equal(fi1.BestPath(), fi1.BestPath())
}) {
return true
}
if len(newFiles) > len(existingFiles) {
type fileInfo struct {
Path string
Length int64
}
mapInfo := func(fi metainfo.FileInfo) fileInfo {
return fileInfo{
Path: strings.Join(fi.BestPath(), "/"),
Length: fi.Length,
}
}
existingFiles := apply(existingFiles, mapInfo)
newFiles := apply(newFiles, mapInfo)
for _, n := range newFiles {
if slices.Contains(existingFiles, n) {
continue
}
for _, e := range existingFiles {
if e.Path == n.Path && e.Length != n.Length {
log.Warn(ctx, "torrents not compatible, has files with different length",
slog.String("path", n.Path),
slog.Int64("existing-length", e.Length),
slog.Int64("new-length", e.Length),
)
return false
}
}
}
}
return true
}

111
daemons/ytdlp/controller.go Normal file
View file

@ -0,0 +1,111 @@
package ytdlp
import (
"context"
"os"
"git.kmsign.ru/royalcat/tstor/pkg/ctxbilly"
"git.kmsign.ru/royalcat/tstor/pkg/kvsingle"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/pkg/ytdlp"
"git.kmsign.ru/royalcat/tstor/src/tasks"
"github.com/royalcat/ctxio"
"github.com/royalcat/ctxprogress"
"github.com/royalcat/kv"
)
type Controller struct {
datafs ctxbilly.Filesystem
source Source
client *ytdlp.Client
cachedinfo *kvsingle.Value[string, ytdlp.Info]
}
func newYtdlpController(datafs ctxbilly.Filesystem, source Source, client *ytdlp.Client) *Controller {
return &Controller{
datafs: datafs,
source: source,
client: client,
}
}
func (c *Controller) Source() Source {
return c.source
}
const sizeApprox = 1024 * 1024 * 1024
func (c *Controller) Update(ctx context.Context, updater tasks.Updater) error {
log := updater.Logger()
ctxprogress.New(ctx)
ctxprogress.Set(ctx, ctxprogress.RangeProgress{Current: 0, Total: 10})
plst, err := c.client.Playlist(ctx, c.source.Url)
ctxprogress.Set(ctx, ctxprogress.RangeProgress{Current: 1, Total: 10})
ctxprogress.Range(ctx, plst, func(ctx context.Context, _ int, e ytdlp.Entry) bool {
if e.OriginalURL == "" {
log.Error("no URL in entry", rlog.Error(err))
return true
}
info, err := c.Info(ctx)
if err != nil {
log.Error("error getting info", rlog.Error(err))
return true
}
dwl := info.RequestedDownloads[0]
fileinfo, err := c.datafs.Stat(ctx, dwl.Filename)
if err != nil {
log.Error("error getting file info", rlog.Error(err))
return true
}
if fileinfo.Size()+sizeApprox > dwl.FilesizeApprox && fileinfo.Size()-sizeApprox < dwl.FilesizeApprox {
log.Debug("file already downloaded", "filename", dwl.Filename)
return true
}
file, err := c.datafs.OpenFile(ctx, dwl.Filename, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)
if err != nil {
log.Error("error opening destination file", rlog.Error(err))
return true
}
err = c.client.Download(ctx, info.OriginalURL, ctxio.IoWriter(ctx, file))
if err != nil {
return false
}
return true
})
ctxprogress.Set(ctx, ctxprogress.RangeProgress{Current: 2, Total: 2})
if err != nil {
return err
}
return nil
}
func (c *Controller) Info(ctx context.Context) (ytdlp.Info, error) {
info, err := c.cachedinfo.Get(ctx)
if err == nil {
return info, nil
}
if err != kv.ErrKeyNotFound {
return info, err
}
info, err = c.Info(ctx)
if err != nil {
return info, err
}
err = c.cachedinfo.Set(ctx, info)
if err != nil {
return info, err
}
return info, nil
}
func (c *Controller) Downloaded() error {
return nil
}

71
daemons/ytdlp/daemon.go Normal file
View file

@ -0,0 +1,71 @@
package ytdlp
import (
"context"
"encoding/json"
"fmt"
"path"
"sync"
"git.kmsign.ru/royalcat/tstor/pkg/ctxbilly"
"git.kmsign.ru/royalcat/tstor/pkg/ytdlp"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/go-git/go-billy/v5/osfs"
"github.com/royalcat/ctxio"
)
func NewService(dataDir string) (*Daemon, error) {
client, err := ytdlp.New()
if err != nil {
return nil, err
}
s := &Daemon{
mu: sync.Mutex{},
client: client,
dataDir: dataDir,
controllers: make(map[string]*Controller, 0),
}
return s, nil
}
type Daemon struct {
mu sync.Mutex
dataDir string
client *ytdlp.Client
controllers map[string]*Controller
}
func (c *Daemon) addSource(s Source) {
c.mu.Lock()
defer c.mu.Unlock()
ctl := newYtdlpController(ctxbilly.WrapFileSystem(osfs.New(c.sourceDir(s))), s, c.client)
c.controllers[s.Name()] = ctl
}
func (c *Daemon) sourceDir(s Source) string {
return path.Join(c.dataDir, s.Name())
}
func (c *Daemon) BuildFS(ctx context.Context, sourcePath string, f vfs.File) (vfs.Filesystem, error) {
data, err := ctxio.ReadAll(ctx, f)
if err != nil {
return nil, fmt.Errorf("failed to read source file: %w", err)
}
var s Source
err = json.Unmarshal(data, &s)
if err != nil {
return nil, err
}
c.addSource(s)
downloadFS := ctxbilly.WrapFileSystem(osfs.New(c.sourceDir(s)))
return newSourceFS(path.Base(f.Name()), downloadFS, c, s), nil
}

75
daemons/ytdlp/fs.go Normal file
View file

@ -0,0 +1,75 @@
package ytdlp
import (
"context"
"io/fs"
"os"
"git.kmsign.ru/royalcat/tstor/pkg/ctxbilly"
"git.kmsign.ru/royalcat/tstor/src/vfs"
)
type SourceFS struct {
service *Daemon
source Source
fs ctxbilly.Filesystem
vfs.DefaultFS
}
var _ vfs.Filesystem = (*SourceFS)(nil)
func newSourceFS(name string, fs ctxbilly.Filesystem, service *Daemon, source Source) *SourceFS {
return &SourceFS{
fs: fs,
service: service,
source: source,
DefaultFS: vfs.DefaultFS(name),
}
}
// Open implements vfs.Filesystem.
func (s *SourceFS) Open(ctx context.Context, filename string) (vfs.File, error) {
info, err := s.fs.Stat(ctx, filename)
if err != nil {
return nil, err
}
f, err := s.fs.OpenFile(ctx, filename, os.O_RDONLY, 0)
if err != nil {
return nil, err
}
return vfs.NewCtxBillyFile(info, f), nil
}
// ReadDir implements vfs.Filesystem.
func (s *SourceFS) ReadDir(ctx context.Context, path string) ([]fs.DirEntry, error) {
infos, err := s.fs.ReadDir(ctx, path)
if err != nil {
return nil, err
}
entries := make([]fs.DirEntry, 0, len(infos))
for _, info := range infos {
entries = append(entries, vfs.NewFileInfo(info.Name(), info.Size()))
}
return entries, nil
}
// Stat implements vfs.Filesystem.
func (s *SourceFS) Stat(ctx context.Context, filename string) (fs.FileInfo, error) {
return s.fs.Stat(ctx, filename)
}
// Unlink implements vfs.Filesystem.
func (s *SourceFS) Unlink(ctx context.Context, filename string) error {
return vfs.ErrNotImplemented
}
// Rename implements vfs.Filesystem.
func (s *SourceFS) Rename(ctx context.Context, oldpath string, newpath string) error {
return vfs.ErrNotImplemented
}

37
daemons/ytdlp/tasks.go Normal file
View file

@ -0,0 +1,37 @@
package ytdlp
import (
"context"
"fmt"
"git.kmsign.ru/royalcat/tstor/src/tasks"
)
const executorName = "ytdlp"
type DownloadTask struct {
Name string
}
var _ tasks.Task = (*DownloadTask)(nil)
// Executor implements tasks.Task.
func (d *DownloadTask) Executor() string {
return executorName
}
var _ tasks.TaskExecutor = (*Daemon)(nil)
// ExecutorName implements tasks.TaskExecutor.
func (c *Daemon) ExecutorName() string {
return executorName
}
func (c *Daemon) RunTask(ctx context.Context, upd tasks.Updater, task tasks.Task) error {
switch t := task.(type) {
case *DownloadTask:
return c.controllers[t.Name].Update(ctx, upd)
default:
return fmt.Errorf("unknown task type: %T", task)
}
}

29
daemons/ytdlp/ytdlp.go Normal file
View file

@ -0,0 +1,29 @@
package ytdlp
import (
"crypto/sha1"
"encoding/base64"
"strings"
)
type Source struct {
Url string `json:"url"`
}
var hasher = sha1.New()
var prefixCutset = [...]string{
"https://", "http://", "www.",
}
func urlHash(url string) string {
for _, v := range prefixCutset {
url = strings.TrimPrefix(url, v)
}
return base64.URLEncoding.EncodeToString(hasher.Sum([]byte(url)))
}
func (s *Source) Name() string {
return urlHash(s.Url)
}