tstor/src/sources/torrent/storage_open.go

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(info *metainfo.Info, infoHash infohash.T) (storage.TorrentImpl, error) {
	ctx := context.Background()
	log := me.log.With(slog.String("infohash", infoHash.HexString()))

	// dir := torrentDir(me.baseDir, infoHash)
	// legacyDir := filepath.Join(me.baseDir, info.Name)

	// log = log.With(slog.String("legacy_dir", legacyDir), slog.String("dir", dir))
	// if _, err := os.Stat(legacyDir); err == nil {
	// 	log.Warn(ctx, "legacy torrent dir found, renaming", slog.String("dir", dir))
	// 	err = os.Rename(legacyDir, dir)
	// 	if err != nil {
	// 		return storage.TorrentImpl{}, fmt.Errorf("error renaming legacy torrent dir: %w", err)
	// 	}
	// }

	// if _, err := os.Stat(dir); errors.Is(err, fs.ErrNotExist) {
	// 	log.Info(ctx, "new torrent, trying copy files from existing")
	// 	dups := me.dupIndex.Includes(infoHash, info.Files)

	// 	for _, dup := range dups {
	// 		err := me.copyDup(ctx, infoHash, dup)
	// 		if err != nil {
	// 			log.Error(ctx, "error copying file", slog.String("file", dup.fileinfo.DisplayPath(info)), rlog.Error(err))
	// 		}
	// 	}

	// }

	impl, err := me.client.OpenTorrent(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
}