tstor/src/sources/torrent/daemon.go

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/sources/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
}