royalcat/tstor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages1 Wiki Activity Actions

qbittorrent source files

Browse source
This commit is contained in:
royalcat 2024-10-19 04:24:14 +03:00
parent 63e63c1c37
commit aa0affb019
30 changed files with 1338 additions and 242 deletions

1258
src/delivery/graphql/generated.go
View file

File diff suppressed because it is too large Load diff

10
src/delivery/graphql/model/models_gen.go
View file

@ -96,6 +96,16 @@ type Pagination struct {
Limit int64 `json:"limit"`
}
type QBitTorrentDaemonQuery struct {
Torrents []*QTorrent `json:"torrents"`
}
type QTorrent struct {
Name string `json:"name"`
Hash string `json:"hash"`
SourceFiles []string `json:"sourceFiles"`
}
type Query struct {
}

2
src/delivery/graphql/resolver/fs.resolvers.go
View file

@ -2,7 +2,7 @@ package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.49
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"

4
src/delivery/graphql/resolver/mutation.resolvers.go
View file

@ -2,7 +2,7 @@ package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.49
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"
@ -49,7 +49,7 @@ func (r *mutationResolver) UploadFile(ctx context.Context, dir string, file grap
// DedupeStorage is the resolver for the dedupeStorage field.
func (r *mutationResolver) DedupeStorage(ctx context.Context) (int64, error) {
deduped, err := r.Service.Storage.Dedupe(ctx)
deduped, err := r.ATorrentDaemon.Storage.Dedupe(ctx)
if err != nil {
return 0, err
}

38
src/delivery/graphql/resolver/qbittorrent_query.resolvers.go Normal file
View file

@ -0,0 +1,38 @@
package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"
"fmt"
graph "git.kmsign.ru/royalcat/tstor/src/delivery/graphql"
"git.kmsign.ru/royalcat/tstor/src/delivery/graphql/model"
)
// Torrents is the resolver for the torrents field.
func (r *qBitTorrentDaemonQueryResolver) Torrents(ctx context.Context, obj *model.QBitTorrentDaemonQuery) ([]*model.QTorrent, error) {
info, err := r.QBitTorrentDaemon.ListTorrents(ctx)
if err != nil {
return nil, fmt.Errorf("error listing torrents: %w", err)
}
out := make([]*model.QTorrent, len(info))
for i, v := range info {
out[i] = &model.QTorrent{
Name: v.Name,
Hash: v.Hash,
}
}
return out, nil
}
// QBitTorrentDaemonQuery returns graph.QBitTorrentDaemonQueryResolver implementation.
func (r *Resolver) QBitTorrentDaemonQuery() graph.QBitTorrentDaemonQueryResolver {
return &qBitTorrentDaemonQueryResolver{r}
}
type qBitTorrentDaemonQueryResolver struct{ *Resolver }

22
src/delivery/graphql/resolver/qbittorrent_types.resolvers.go Normal file
View file

@ -0,0 +1,22 @@
package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"
graph "git.kmsign.ru/royalcat/tstor/src/delivery/graphql"
"git.kmsign.ru/royalcat/tstor/src/delivery/graphql/model"
)
// SourceFiles is the resolver for the sourceFiles field.
func (r *qTorrentResolver) SourceFiles(ctx context.Context, obj *model.QTorrent) ([]string, error) {
return r.QBitTorrentDaemon.SourceFiles(ctx, obj.Hash)
}
// QTorrent returns graph.QTorrentResolver implementation.
func (r *Resolver) QTorrent() graph.QTorrentResolver { return &qTorrentResolver{r} }
type qTorrentResolver struct{ *Resolver }

7
src/delivery/graphql/resolver/query.resolvers.go
View file

@ -2,7 +2,7 @@ package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.49
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"
@ -16,6 +16,11 @@ func (r *queryResolver) TorrentDaemon(ctx context.Context) (*model.TorrentDaemon
return &model.TorrentDaemonQuery{}, nil
}
// QbittorrentDaemon is the resolver for the qbittorrentDaemon field.
func (r *queryResolver) QbittorrentDaemon(ctx context.Context) (*model.QBitTorrentDaemonQuery, error) {
return &model.QBitTorrentDaemonQuery{}, nil
}
// FsEntry is the resolver for the fsEntry field.
func (r *queryResolver) FsEntry(ctx context.Context, path string) (model.FsEntry, error) {
entry, err := r.VFS.Stat(ctx, path)

8
src/delivery/graphql/resolver/resolver.go
View file

@ -1,6 +1,7 @@
package resolver
import (
"git.kmsign.ru/royalcat/tstor/src/sources/qbittorrent"
"git.kmsign.ru/royalcat/tstor/src/sources/torrent"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/go-git/go-billy/v5"
@ -11,7 +12,8 @@ import (
// It serves as dependency injection for your app, add any dependencies you require here.
type Resolver struct {
Service *torrent.Daemon
VFS vfs.Filesystem
SourceFS billy.Filesystem
ATorrentDaemon *torrent.Daemon
QBitTorrentDaemon *qbittorrent.Daemon
VFS vfs.Filesystem
SourceFS billy.Filesystem
}

4
src/delivery/graphql/resolver/subscription.resolvers.go
View file

@ -2,7 +2,7 @@ package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.49
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"
@ -20,7 +20,7 @@ func (r *subscriptionResolver) TaskProgress(ctx context.Context, taskID string)
// TorrentDownloadUpdates is the resolver for the torrentDownloadUpdates field.
func (r *subscriptionResolver) TorrentDownloadUpdates(ctx context.Context) (<-chan *model.TorrentProgress, error) {
out := make(chan *model.TorrentProgress)
progress, err := r.Service.DownloadProgress(ctx)
progress, err := r.ATorrentDaemon.DownloadProgress(ctx)
if err != nil {
return nil, err
}

14
src/delivery/graphql/resolver/torrent_mutation.resolvers.go
View file

@ -2,7 +2,7 @@ package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.49
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"
@ -15,7 +15,7 @@ import (
// ValidateTorrent is the resolver for the validateTorrent field.
func (r *torrentDaemonMutationResolver) ValidateTorrent(ctx context.Context, obj *model.TorrentDaemonMutation, filter model.TorrentFilter) (bool, error) {
if filter.Infohash != nil {
t, err := r.Resolver.Service.GetTorrent(*filter.Infohash)
t, err := r.Resolver.ATorrentDaemon.GetTorrent(*filter.Infohash)
if err != nil {
return false, err
}
@ -28,7 +28,7 @@ func (r *torrentDaemonMutationResolver) ValidateTorrent(ctx context.Context, obj
}
if filter.Everything != nil && *filter.Everything {
torrents, err := r.Resolver.Service.ListTorrents(ctx)
torrents, err := r.Resolver.ATorrentDaemon.ListTorrents(ctx)
if err != nil {
return false, err
}
@ -45,7 +45,7 @@ func (r *torrentDaemonMutationResolver) ValidateTorrent(ctx context.Context, obj
// SetTorrentPriority is the resolver for the setTorrentPriority field.
func (r *torrentDaemonMutationResolver) SetTorrentPriority(ctx context.Context, obj *model.TorrentDaemonMutation, infohash string, file *string, priority types.PiecePriority) (bool, error) {
t, err := r.Resolver.Service.GetTorrent(infohash)
t, err := r.Resolver.ATorrentDaemon.GetTorrent(infohash)
if err != nil {
return false, err
}
@ -74,19 +74,19 @@ func (r *torrentDaemonMutationResolver) SetTorrentPriority(ctx context.Context,
// Cleanup is the resolver for the cleanup field.
func (r *torrentDaemonMutationResolver) Cleanup(ctx context.Context, obj *model.TorrentDaemonMutation, files *bool, dryRun bool) (*model.CleanupResponse, error) {
torrents, err := r.Service.ListTorrents(ctx)
torrents, err := r.ATorrentDaemon.ListTorrents(ctx)
if err != nil {
return nil, err
}
if files != nil && *files {
r, err := r.Service.Storage.CleanupFiles(ctx, torrents, dryRun)
r, err := r.ATorrentDaemon.Storage.CleanupFiles(ctx, torrents, dryRun)
return &model.CleanupResponse{
Count: int64(len(r)),
List: r,
}, err
} else {
r, err := r.Service.Storage.CleanupDirs(ctx, torrents, dryRun)
r, err := r.ATorrentDaemon.Storage.CleanupDirs(ctx, torrents, dryRun)
return &model.CleanupResponse{
Count: int64(len(r)),
List: r,

13
src/delivery/graphql/resolver/torrent_query.resolvers.go
View file

@ -2,7 +2,7 @@ package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.49
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"
@ -13,13 +13,12 @@ import (
graph "git.kmsign.ru/royalcat/tstor/src/delivery/graphql"
"git.kmsign.ru/royalcat/tstor/src/delivery/graphql/model"
"git.kmsign.ru/royalcat/tstor/src/sources/torrent"
tinfohash "github.com/anacrolix/torrent/types/infohash"
)
// Torrents is the resolver for the torrents field.
func (r *torrentDaemonQueryResolver) Torrents(ctx context.Context, obj *model.TorrentDaemonQuery, filter *model.TorrentsFilter) ([]*model.Torrent, error) {
torrents, err := r.Service.ListTorrents(ctx)
torrents, err := r.ATorrentDaemon.ListTorrents(ctx)
if err != nil {
return nil, err
}
@ -91,7 +90,7 @@ func (r *torrentDaemonQueryResolver) Torrents(ctx context.Context, obj *model.To
// ClientStats is the resolver for the clientStats field.
func (r *torrentDaemonQueryResolver) ClientStats(ctx context.Context, obj *model.TorrentDaemonQuery) (*model.TorrentClientStats, error) {
stats := r.Service.Stats()
stats := r.ATorrentDaemon.Stats()
return &model.TorrentClientStats{
BytesWritten: stats.BytesWritten.Int64(),
BytesRead: stats.BytesRead.Int64(),
@ -113,21 +112,21 @@ func (r *torrentDaemonQueryResolver) ClientStats(ctx context.Context, obj *model
func (r *torrentDaemonQueryResolver) StatsHistory(ctx context.Context, obj *model.TorrentDaemonQuery, since time.Time, infohash *string) ([]*model.TorrentStats, error) {
var stats []torrent.TorrentStats
if infohash == nil {
stats, err := r.Service.StatsHistory(ctx, since)
stats, err := r.ATorrentDaemon.StatsHistory(ctx, since)
if err != nil {
return nil, err
}
return model.Apply(stats, model.MapTorrentStats), nil
} else if *infohash == "total" {
var err error
stats, err = r.Service.TotalStatsHistory(ctx, since)
stats, err = r.ATorrentDaemon.TotalStatsHistory(ctx, since)
if err != nil {
return nil, err
}
} else {
ih := tinfohash.FromHexString(*infohash)
var err error
stats, err = r.Service.TorrentStatsHistory(ctx, since, ih)
stats, err = r.ATorrentDaemon.TorrentStatsHistory(ctx, since, ih)
if err != nil {
return nil, err
}

2
src/delivery/graphql/resolver/torrent_types.resolvers.go
View file

@ -2,7 +2,7 @@ package resolver
// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.
// Code generated by github.com/99designs/gqlgen version v0.17.49
// Code generated by github.com/99designs/gqlgen version v0.17.55
import (
"context"

5
src/delivery/http.go
View file

@ -7,6 +7,7 @@ import (
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/src/config"
"git.kmsign.ru/royalcat/tstor/src/sources/qbittorrent"
"git.kmsign.ru/royalcat/tstor/src/sources/torrent"
"git.kmsign.ru/royalcat/tstor/src/vfs"
echopprof "github.com/labstack/echo-contrib/pprof"
@ -15,7 +16,7 @@ import (
"github.com/prometheus/client_golang/prometheus/promhttp"
)
func Run(s *torrent.Daemon, vfs vfs.Filesystem, logPath string, cfg *config.Settings) error {
func Run(torrentdaemon *torrent.Daemon, qbitdaemon *qbittorrent.Daemon, vfs vfs.Filesystem, cfg *config.Settings) error {
log := slog.With()
r := echo.New()
@ -28,7 +29,7 @@ func Run(s *torrent.Daemon, vfs vfs.Filesystem, logPath string, cfg *config.Sett
echopprof.Register(r)
r.Any("/graphql", echo.WrapHandler((GraphQLHandler(s, vfs))))
r.Any("/graphql", echo.WrapHandler((GraphQLHandler(torrentdaemon, qbitdaemon, vfs))))
r.GET("/metrics", echo.WrapHandler(promhttp.Handler()))
log.Info("starting webserver", "host", fmt.Sprintf("%s:%d", cfg.WebUi.IP, cfg.WebUi.Port))

13
src/delivery/router.go
View file

@ -6,6 +6,7 @@ import (
graph "git.kmsign.ru/royalcat/tstor/src/delivery/graphql"
"git.kmsign.ru/royalcat/tstor/src/delivery/graphql/resolver"
"git.kmsign.ru/royalcat/tstor/src/sources/qbittorrent"
"git.kmsign.ru/royalcat/tstor/src/sources/torrent"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/99designs/gqlgen/graphql"
@ -20,11 +21,15 @@ func noopDirective(ctx context.Context, obj interface{}, next graphql.Resolver)
return next(ctx)
}
func GraphQLHandler(service *torrent.Daemon, vfs vfs.Filesystem) http.Handler {
func GraphQLHandler(service *torrent.Daemon, qbitdaemon *qbittorrent.Daemon, vfs vfs.Filesystem) http.Handler {
graphqlHandler := handler.NewDefaultServer(
graph.NewExecutableSchema(
graph.Config{
Resolvers: &resolver.Resolver{Service: service, VFS: vfs},
Resolvers: &resolver.Resolver{
ATorrentDaemon: service,
QBitTorrentDaemon: qbitdaemon,
VFS: vfs,
},
Directives: graph.DirectiveRoot{
OneOf: graph.OneOf,
Resolver: noopDirective,
@ -46,9 +51,9 @@ func GraphQLHandler(service *torrent.Daemon, vfs vfs.Filesystem) http.Handler {
graphqlHandler.AddTransport(&transport.Websocket{})
graphqlHandler.AddTransport(&transport.SSE{})
graphqlHandler.AddTransport(&transport.UrlEncodedForm{})
graphqlHandler.SetQueryCache(lru.New(1000))
// graphqlHandler.SetQueryCache(lru.New[*ast.QueryDocument](1000))
graphqlHandler.Use(extension.Introspection{})
graphqlHandler.Use(extension.AutomaticPersistedQuery{Cache: lru.New(100)})
graphqlHandler.Use(extension.AutomaticPersistedQuery{Cache: lru.New[string](100)})
graphqlHandler.Use(otelgqlgen.Middleware(
otelgqlgen.WithCreateSpanFromFields(func(ctx *graphql.FieldContext) bool {
return ctx.Field.Directives.ForName("link") != nil

27
src/sources/qbittorrent/api.go Normal file
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
}

30
src/sources/qbittorrent/daemon.go
View file

@ -9,6 +9,7 @@ import (
"os"
"path"
"path/filepath"
"sync"
"time"
"git.kmsign.ru/royalcat/tstor/pkg/qbittorrent"
@ -27,9 +28,13 @@ import (
var trace = otel.Tracer("git.kmsign.ru/royalcat/tstor/src/sources/qbittorrent")
type Daemon struct {
proc *os.Process
qb qbittorrent.Client
client *cacheClient
proc *os.Process
qb qbittorrent.Client
client *cacheClient
sourceFilesMu sync.Mutex
sourceFiles map[string]string // [sourcePath]infohash
dataDir string
ur *iouring.IOURing
log *rlog.Logger
@ -119,12 +124,13 @@ func NewDaemon(conf config.QBittorrent) (*Daemon, error) {
}
return &Daemon{
qb: qb,
proc: proc,
dataDir: conf.DataFolder,
ur: ur,
client: wrapClient(qb),
log: rlog.Component("qbittorrent"),
qb: qb,
proc: proc,
dataDir: conf.DataFolder,
ur: ur,
sourceFiles: make(map[string]string),
client: wrapClient(qb),
log: rlog.Component("qbittorrent"),
}, nil
}
@ -146,7 +152,7 @@ func torrentDataPath(dataDir string, ih string) (string, error) {
return filepath.Abs(path.Join(dataDir, ih))
}
func (fs *Daemon) GetTorrentFS(ctx context.Context, file vfs.File) (vfs.Filesystem, error) {
func (fs *Daemon) GetTorrentFS(ctx context.Context, sourcePath string, file vfs.File) (vfs.Filesystem, error) {
ctx, span := trace.Start(ctx, "GetTorrentFS")
defer span.End()
@ -171,6 +177,10 @@ func (fs *Daemon) GetTorrentFS(ctx context.Context, file vfs.File) (vfs.Filesyst
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)
}

1
src/sources/qbittorrent/fs.go
View file

@ -203,6 +203,7 @@ func openFile(ctx context.Context, ur *iouring.IOURing, client *cacheClient, tor
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

4
src/sources/storage.go
View file

@ -10,8 +10,8 @@ import (
func NewHostedFS(sourceFS vfs.Filesystem, tsrv *qbittorrent.Daemon, ytdlpsrv *ytdlp.Daemon) vfs.Filesystem {
factories := map[string]vfs.FsFactory{
".torrent": func(ctx context.Context, f vfs.File) (vfs.Filesystem, error) {
tfs, err := tsrv.GetTorrentFS(ctx, f)
".torrent": func(ctx context.Context, sourcePath string, f vfs.File) (vfs.Filesystem, error) {
tfs, err := tsrv.GetTorrentFS(ctx, sourcePath, f)
if err != nil {
return nil, err
}

2
src/sources/torrent/fs.go
View file

@ -39,7 +39,7 @@ var _ vfs.Filesystem = (*TorrentFS)(nil)
const shortTimeout = time.Millisecond
const lowTimeout = time.Second * 5
func (s *Daemon) NewTorrentFs(ctx context.Context, f vfs.File) (vfs.Filesystem, error) {
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

2
src/sources/ytdlp/daemon.go
View file

@ -51,7 +51,7 @@ func (c *Daemon) sourceDir(s Source) string {
return path.Join(c.dataDir, s.Name())
}
func (c *Daemon) BuildFS(ctx context.Context, f vfs.File) (vfs.Filesystem, error) {
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)

6
src/vfs/archive.go
View file

@ -19,21 +19,21 @@ import (
)
var ArchiveFactories = map[string]FsFactory{
".zip": func(ctx context.Context, f File) (Filesystem, error) {
".zip": func(ctx context.Context, _ string, f File) (Filesystem, error) {
stat, err := f.Info()
if err != nil {
return nil, err
}
return NewArchive(ctx, stat.Name(), f, stat.Size(), ZipLoader)
},
".rar": func(ctx context.Context, f File) (Filesystem, error) {
".rar": func(ctx context.Context, _ string, f File) (Filesystem, error) {
stat, err := f.Info()
if err != nil {
return nil, err
}
return NewArchive(ctx, stat.Name(), f, stat.Size(), RarLoader)
},
".7z": func(ctx context.Context, f File) (Filesystem, error) {
".7z": func(ctx context.Context, _ string, f File) (Filesystem, error) {
stat, err := f.Info()
if err != nil {
return nil, err

2
src/vfs/log.go
View file

@ -51,7 +51,7 @@ func WrapLogFS(vfs Filesystem) (*LogFS, error) {
fs: vfs,
log: rlog.Component("logfs"),
tel: &fsTelemetry{openedFiles: openedFiles},
timeout: time.Minute * 3,
timeout: time.Minute * 5,
readTimeout: time.Minute,
}, nil
}

6
src/vfs/resolver.go
View file

@ -223,7 +223,7 @@ func (r *ResolverFS) Type() fs.FileMode {
var _ Filesystem = &ResolverFS{}
// It factory responsobility to close file
type FsFactory func(ctx context.Context, f File) (Filesystem, error)
type FsFactory func(ctx context.Context, sourcePath string, f File) (Filesystem, error)
func NewResolver(factories map[string]FsFactory) *Resolver {
return &Resolver{
@ -272,7 +272,7 @@ func (r *Resolver) nestedFs(ctx context.Context, fsPath string, file File) (File
continue
}
nestedFs, err := nestFactory(ctx, file)
nestedFs, err := nestFactory(ctx, fsPath, file)
if err != nil {
return nil, fmt.Errorf("error calling nest factory: %s with error: %w", fsPath, err)
}
@ -348,7 +348,7 @@ PARTS_LOOP:
}
// it is factory responsibility to close file handler then needed
nestedFs, err := nestFactory(ctx, fsFile)
nestedFs, err := nestFactory(ctx, name, fsFile)
if err != nil {
return "", nil, "", fmt.Errorf("error creating filesystem from file: %s with error: %w", fsPath, err)
}