diff --git a/docs/openapi/downloader.json b/docs/openapi/downloader.json index a9b54d04..366a0165 100644 --- a/docs/openapi/downloader.json +++ b/docs/openapi/downloader.json @@ -348,6 +348,17 @@ ] }, "phase": { + "type": "string", + "enum": [ + "metadata", + "downloading", + "uploading", + "seeding", + "completed", + "error" + ] + }, + "engineState": { "type": "string", "maxLength": 80 }, @@ -684,6 +695,17 @@ ] }, "phase": { + "type": "string", + "enum": [ + "metadata", + "downloading", + "uploading", + "seeding", + "completed", + "error" + ] + }, + "engineState": { "type": "string", "maxLength": 80 }, @@ -1004,6 +1026,17 @@ ] }, "phase": { + "type": "string", + "enum": [ + "metadata", + "downloading", + "uploading", + "seeding", + "completed", + "error" + ] + }, + "engineState": { "type": "string", "maxLength": 80 }, @@ -1267,6 +1300,17 @@ ] }, "phase": { + "type": "string", + "enum": [ + "metadata", + "downloading", + "uploading", + "seeding", + "completed", + "error" + ] + }, + "engineState": { "type": "string", "maxLength": 80 }, @@ -1498,6 +1542,17 @@ ] }, "phase": { + "type": "string", + "enum": [ + "metadata", + "downloading", + "uploading", + "seeding", + "completed", + "error" + ] + }, + "engineState": { "type": "string", "maxLength": 80 }, diff --git a/downloader/internal/client/client.go b/downloader/internal/client/client.go index 9032053b..24eb7f7c 100644 --- a/downloader/internal/client/client.go +++ b/downloader/internal/client/client.go @@ -46,6 +46,7 @@ type DownloadTask struct { type DownloadTaskDetail struct { Engine string `json:"engine,omitempty"` Phase string `json:"phase,omitempty"` + EngineState string `json:"engineState,omitempty"` Message string `json:"message,omitempty"` ETASeconds *int64 `json:"etaSeconds,omitempty"` Connections *int64 `json:"connections,omitempty"` diff --git a/downloader/internal/engine/engine.go b/downloader/internal/engine/engine.go index bae6e946..09478218 100644 --- a/downloader/internal/engine/engine.go +++ b/downloader/internal/engine/engine.go @@ -27,10 +27,18 @@ type Result struct { } type Seed struct { - Engine string - ID string - Path string - Cleanup func(context.Context) error + Engine string + ID string + Path string + Snapshot func(context.Context) (SeedSnapshot, error) + Cleanup func(context.Context) error +} + +type SeedSnapshot struct { + Downloaded int64 + Total *int64 + Bps int64 + Detail *client.DownloadTaskDetail } type Progress func(downloaded int64, total *int64, bps int64, detail *client.DownloadTaskDetail) error @@ -184,10 +192,11 @@ func (a Aria2) Download(ctx context.Context, task client.DownloadTask, progress } if a.RetainSeed && task.SourceType != "http" { result.Seed = &Seed{ - Engine: "aria2", - ID: status.GID, - Path: taskDir, - Cleanup: a.cleanupSeed(status.GID, taskDir), + Engine: "aria2", + ID: status.GID, + Path: taskDir, + Snapshot: a.seedSnapshot(status.GID), + Cleanup: a.cleanupSeed(status.GID, taskDir), } return result, nil } @@ -200,6 +209,34 @@ func (a Aria2) client(ctx context.Context) (*arigo.Client, error) { return arigo.DialContext(ctx, a.URL, a.Secret) } +func (a Aria2) seedSnapshot(gid string) func(context.Context) (SeedSnapshot, error) { + return func(ctx context.Context) (SeedSnapshot, error) { + aria, err := a.client(ctx) + if err != nil { + return SeedSnapshot{}, err + } + defer aria.Close() + status, err := aria.TellStatus(gid) + if err != nil { + return SeedSnapshot{}, err + } + peers := a.getAria2Peers(ctx, &aria, gid) + total := int64(status.TotalLength) + var totalPtr *int64 + if total > 0 { + totalPtr = &total + } + detail := aria2Detail(status, peers) + detail.Phase = "seeding" + return SeedSnapshot{ + Downloaded: int64(status.CompletedLength), + Total: totalPtr, + Bps: int64(status.DownloadSpeed), + Detail: detail, + }, nil + } +} + func (a Aria2) cleanupSeed(gid string, localPath string) func(context.Context) error { return func(ctx context.Context) error { var errs []error @@ -302,10 +339,11 @@ func (q QBittorrent) Download(ctx context.Context, task client.DownloadTask, pro } if q.RetainSeed { result.Seed = &Seed{ - Engine: "qbittorrent", - ID: torrent.Hash, - Path: taskDir, - Cleanup: q.cleanupSeed(torrent.Hash, taskDir), + Engine: "qbittorrent", + ID: torrent.Hash, + Path: taskDir, + Snapshot: q.seedSnapshot(torrent.Hash), + Cleanup: q.cleanupSeed(torrent.Hash, taskDir), } return result, nil } @@ -329,6 +367,39 @@ func (q QBittorrent) cleanupSeed(hash string, localPath string) func(context.Con } } +func (q QBittorrent) seedSnapshot(hash string) func(context.Context) (SeedSnapshot, error) { + return func(ctx context.Context) (SeedSnapshot, error) { + qbt, err := q.login(ctx) + if err != nil { + return SeedSnapshot{}, err + } + torrents, err := qbt.GetTorrentsCtx(ctx, qbittorrent.TorrentFilterOptions{Hashes: []string{hash}}) + if err != nil { + return SeedSnapshot{}, err + } + if len(torrents) == 0 { + return SeedSnapshot{}, fmt.Errorf("qbittorrent torrent %s not found", hash) + } + torrent := torrents[0] + total := torrent.TotalSize + if total <= 0 { + total = torrent.Size + } + var totalPtr *int64 + if total > 0 { + totalPtr = &total + } + detail := qbittorrentDetail(ctx, qbt, torrent) + detail.Phase = "seeding" + return SeedSnapshot{ + Downloaded: torrent.Completed, + Total: totalPtr, + Bps: torrent.DlSpeed, + Detail: detail, + }, nil + } +} + type progressWriter struct { progress Progress total *int64 @@ -457,7 +528,8 @@ func aria2Detail(status arigo.Status, peers []arigo.Peer) *client.DownloadTaskDe uploadBps := int64(status.UploadSpeed) detail := &client.DownloadTaskDetail{ Engine: "aria2", - Phase: string(status.Status), + Phase: aria2Phase(string(status.Status), status.FollowedBy), + EngineState: string(status.Status), Connections: &connections, InfoHash: status.InfoHash, TorrentName: status.BitTorrent.Info.Name, @@ -476,6 +548,24 @@ func aria2Detail(status arigo.Status, peers []arigo.Peer) *client.DownloadTaskDe return detail } +func aria2Phase(state string, followedBy []string) string { + switch state { + case string(arigo.StatusWaiting): + if len(followedBy) > 0 { + return "metadata" + } + return "downloading" + case string(arigo.StatusActive): + return "downloading" + case "complete", string(arigo.StatusCompleted): + return "completed" + case string(arigo.StatusError), string(arigo.StatusRemoved): + return "error" + default: + return "downloading" + } +} + func aria2Trackers(announceList [][]string) []client.DownloadTaskTracker { trackers := make([]client.DownloadTaskTracker, 0, 20) seen := map[string]struct{}{} @@ -621,7 +711,8 @@ func qbittorrentDetail(ctx context.Context, qbt *qbittorrent.Client, torrent qbi } return &client.DownloadTaskDetail{ Engine: "qbittorrent", - Phase: string(torrent.State), + Phase: qbittorrentPhase(string(torrent.State)), + EngineState: string(torrent.State), ETASeconds: eta, Connections: &connections, InfoHash: torrent.Hash, @@ -636,6 +727,24 @@ func qbittorrentDetail(ctx context.Context, qbt *qbittorrent.Client, torrent qbi } } +func qbittorrentPhase(state string) string { + normalized := strings.ToLower(state) + switch { + case strings.Contains(normalized, "meta"): + return "metadata" + case strings.Contains(normalized, "up"), strings.Contains(normalized, "seed"): + return "seeding" + case strings.Contains(normalized, "error"), strings.Contains(normalized, "missing"): + return "error" + case strings.Contains(normalized, "paused"): + return "downloading" + case normalized == "uploading": + return "seeding" + default: + return "downloading" + } +} + func qbittorrentTrackers(ctx context.Context, qbt *qbittorrent.Client, torrent qbittorrent.Torrent) []client.DownloadTaskTracker { trackers := torrent.Trackers if len(trackers) == 0 && torrent.Hash != "" { diff --git a/downloader/internal/openapi/client.gen.go b/downloader/internal/openapi/client.gen.go index e8c2489d..0d4fa0ca 100644 --- a/downloader/internal/openapi/client.gen.go +++ b/downloader/internal/openapi/client.gen.go @@ -250,6 +250,36 @@ func (e GetApiDownloadTasks200JSONResponseBodyItemsDetailEngine) Valid() bool { } } +// Defines values for GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase. +const ( + GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseCompleted GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase = "completed" + GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseDownloading GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase = "downloading" + GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseError GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase = "error" + GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseMetadata GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase = "metadata" + GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseSeeding GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase = "seeding" + GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseUploading GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase = "uploading" +) + +// Valid indicates whether the value is a known member of the GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase enum. +func (e GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase) Valid() bool { + switch e { + case GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseCompleted: + return true + case GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseDownloading: + return true + case GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseError: + return true + case GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseMetadata: + return true + case GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseSeeding: + return true + case GetApiDownloadTasks200JSONResponseBodyItemsDetailPhaseUploading: + return true + default: + return false + } +} + // Defines values for GetApiDownloadTasks200JSONResponseBodyItemsSourceType. const ( GetApiDownloadTasks200JSONResponseBodyItemsSourceTypeHttp GetApiDownloadTasks200JSONResponseBodyItemsSourceType = "http" @@ -349,6 +379,36 @@ func (e PostApiDownloadTasks201JSONResponseBodyDetailEngine) Valid() bool { } } +// Defines values for PostApiDownloadTasks201JSONResponseBodyDetailPhase. +const ( + PostApiDownloadTasks201JSONResponseBodyDetailPhaseCompleted PostApiDownloadTasks201JSONResponseBodyDetailPhase = "completed" + PostApiDownloadTasks201JSONResponseBodyDetailPhaseDownloading PostApiDownloadTasks201JSONResponseBodyDetailPhase = "downloading" + PostApiDownloadTasks201JSONResponseBodyDetailPhaseError PostApiDownloadTasks201JSONResponseBodyDetailPhase = "error" + PostApiDownloadTasks201JSONResponseBodyDetailPhaseMetadata PostApiDownloadTasks201JSONResponseBodyDetailPhase = "metadata" + PostApiDownloadTasks201JSONResponseBodyDetailPhaseSeeding PostApiDownloadTasks201JSONResponseBodyDetailPhase = "seeding" + PostApiDownloadTasks201JSONResponseBodyDetailPhaseUploading PostApiDownloadTasks201JSONResponseBodyDetailPhase = "uploading" +) + +// Valid indicates whether the value is a known member of the PostApiDownloadTasks201JSONResponseBodyDetailPhase enum. +func (e PostApiDownloadTasks201JSONResponseBodyDetailPhase) Valid() bool { + switch e { + case PostApiDownloadTasks201JSONResponseBodyDetailPhaseCompleted: + return true + case PostApiDownloadTasks201JSONResponseBodyDetailPhaseDownloading: + return true + case PostApiDownloadTasks201JSONResponseBodyDetailPhaseError: + return true + case PostApiDownloadTasks201JSONResponseBodyDetailPhaseMetadata: + return true + case PostApiDownloadTasks201JSONResponseBodyDetailPhaseSeeding: + return true + case PostApiDownloadTasks201JSONResponseBodyDetailPhaseUploading: + return true + default: + return false + } +} + // Defines values for PostApiDownloadTasks201JSONResponseBodySourceType. const ( PostApiDownloadTasks201JSONResponseBodySourceTypeHttp PostApiDownloadTasks201JSONResponseBodySourceType = "http" @@ -427,6 +487,36 @@ func (e GetApiDownloadTasksId200JSONResponseBodyDetailEngine) Valid() bool { } } +// Defines values for GetApiDownloadTasksId200JSONResponseBodyDetailPhase. +const ( + GetApiDownloadTasksId200JSONResponseBodyDetailPhaseCompleted GetApiDownloadTasksId200JSONResponseBodyDetailPhase = "completed" + GetApiDownloadTasksId200JSONResponseBodyDetailPhaseDownloading GetApiDownloadTasksId200JSONResponseBodyDetailPhase = "downloading" + GetApiDownloadTasksId200JSONResponseBodyDetailPhaseError GetApiDownloadTasksId200JSONResponseBodyDetailPhase = "error" + GetApiDownloadTasksId200JSONResponseBodyDetailPhaseMetadata GetApiDownloadTasksId200JSONResponseBodyDetailPhase = "metadata" + GetApiDownloadTasksId200JSONResponseBodyDetailPhaseSeeding GetApiDownloadTasksId200JSONResponseBodyDetailPhase = "seeding" + GetApiDownloadTasksId200JSONResponseBodyDetailPhaseUploading GetApiDownloadTasksId200JSONResponseBodyDetailPhase = "uploading" +) + +// Valid indicates whether the value is a known member of the GetApiDownloadTasksId200JSONResponseBodyDetailPhase enum. +func (e GetApiDownloadTasksId200JSONResponseBodyDetailPhase) Valid() bool { + switch e { + case GetApiDownloadTasksId200JSONResponseBodyDetailPhaseCompleted: + return true + case GetApiDownloadTasksId200JSONResponseBodyDetailPhaseDownloading: + return true + case GetApiDownloadTasksId200JSONResponseBodyDetailPhaseError: + return true + case GetApiDownloadTasksId200JSONResponseBodyDetailPhaseMetadata: + return true + case GetApiDownloadTasksId200JSONResponseBodyDetailPhaseSeeding: + return true + case GetApiDownloadTasksId200JSONResponseBodyDetailPhaseUploading: + return true + default: + return false + } +} + // Defines values for GetApiDownloadTasksId200JSONResponseBodySourceType. const ( GetApiDownloadTasksId200JSONResponseBodySourceTypeHttp GetApiDownloadTasksId200JSONResponseBodySourceType = "http" @@ -505,6 +595,36 @@ func (e PatchApiDownloadTasksIdJSONBodyDetailEngine) Valid() bool { } } +// Defines values for PatchApiDownloadTasksIdJSONBodyDetailPhase. +const ( + PatchApiDownloadTasksIdJSONBodyDetailPhaseCompleted PatchApiDownloadTasksIdJSONBodyDetailPhase = "completed" + PatchApiDownloadTasksIdJSONBodyDetailPhaseDownloading PatchApiDownloadTasksIdJSONBodyDetailPhase = "downloading" + PatchApiDownloadTasksIdJSONBodyDetailPhaseError PatchApiDownloadTasksIdJSONBodyDetailPhase = "error" + PatchApiDownloadTasksIdJSONBodyDetailPhaseMetadata PatchApiDownloadTasksIdJSONBodyDetailPhase = "metadata" + PatchApiDownloadTasksIdJSONBodyDetailPhaseSeeding PatchApiDownloadTasksIdJSONBodyDetailPhase = "seeding" + PatchApiDownloadTasksIdJSONBodyDetailPhaseUploading PatchApiDownloadTasksIdJSONBodyDetailPhase = "uploading" +) + +// Valid indicates whether the value is a known member of the PatchApiDownloadTasksIdJSONBodyDetailPhase enum. +func (e PatchApiDownloadTasksIdJSONBodyDetailPhase) Valid() bool { + switch e { + case PatchApiDownloadTasksIdJSONBodyDetailPhaseCompleted: + return true + case PatchApiDownloadTasksIdJSONBodyDetailPhaseDownloading: + return true + case PatchApiDownloadTasksIdJSONBodyDetailPhaseError: + return true + case PatchApiDownloadTasksIdJSONBodyDetailPhaseMetadata: + return true + case PatchApiDownloadTasksIdJSONBodyDetailPhaseSeeding: + return true + case PatchApiDownloadTasksIdJSONBodyDetailPhaseUploading: + return true + default: + return false + } +} + // Defines values for PatchApiDownloadTasksIdJSONBodyStatus. const ( PatchApiDownloadTasksIdJSONBodyStatusAssigned PatchApiDownloadTasksIdJSONBodyStatus = "assigned" @@ -562,6 +682,36 @@ func (e PatchApiDownloadTasksId200JSONResponseBodyDetailEngine) Valid() bool { } } +// Defines values for PatchApiDownloadTasksId200JSONResponseBodyDetailPhase. +const ( + PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseCompleted PatchApiDownloadTasksId200JSONResponseBodyDetailPhase = "completed" + PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseDownloading PatchApiDownloadTasksId200JSONResponseBodyDetailPhase = "downloading" + PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseError PatchApiDownloadTasksId200JSONResponseBodyDetailPhase = "error" + PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseMetadata PatchApiDownloadTasksId200JSONResponseBodyDetailPhase = "metadata" + PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseSeeding PatchApiDownloadTasksId200JSONResponseBodyDetailPhase = "seeding" + PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseUploading PatchApiDownloadTasksId200JSONResponseBodyDetailPhase = "uploading" +) + +// Valid indicates whether the value is a known member of the PatchApiDownloadTasksId200JSONResponseBodyDetailPhase enum. +func (e PatchApiDownloadTasksId200JSONResponseBodyDetailPhase) Valid() bool { + switch e { + case PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseCompleted: + return true + case PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseDownloading: + return true + case PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseError: + return true + case PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseMetadata: + return true + case PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseSeeding: + return true + case PatchApiDownloadTasksId200JSONResponseBodyDetailPhaseUploading: + return true + default: + return false + } +} + // Defines values for PatchApiDownloadTasksId200JSONResponseBodySourceType. const ( Http PatchApiDownloadTasksId200JSONResponseBodySourceType = "http" @@ -585,34 +735,34 @@ func (e PatchApiDownloadTasksId200JSONResponseBodySourceType) Valid() bool { // Defines values for PatchApiDownloadTasksId200JSONResponseBodyStatus. const ( - PatchApiDownloadTasksId200JSONResponseBodyStatusAssigned PatchApiDownloadTasksId200JSONResponseBodyStatus = "assigned" - PatchApiDownloadTasksId200JSONResponseBodyStatusBillingPaused PatchApiDownloadTasksId200JSONResponseBodyStatus = "billing_paused" - PatchApiDownloadTasksId200JSONResponseBodyStatusCanceled PatchApiDownloadTasksId200JSONResponseBodyStatus = "canceled" - PatchApiDownloadTasksId200JSONResponseBodyStatusCompleted PatchApiDownloadTasksId200JSONResponseBodyStatus = "completed" - PatchApiDownloadTasksId200JSONResponseBodyStatusFailed PatchApiDownloadTasksId200JSONResponseBodyStatus = "failed" - PatchApiDownloadTasksId200JSONResponseBodyStatusQueued PatchApiDownloadTasksId200JSONResponseBodyStatus = "queued" - PatchApiDownloadTasksId200JSONResponseBodyStatusRunning PatchApiDownloadTasksId200JSONResponseBodyStatus = "running" - PatchApiDownloadTasksId200JSONResponseBodyStatusUploading PatchApiDownloadTasksId200JSONResponseBodyStatus = "uploading" + Assigned PatchApiDownloadTasksId200JSONResponseBodyStatus = "assigned" + BillingPaused PatchApiDownloadTasksId200JSONResponseBodyStatus = "billing_paused" + Canceled PatchApiDownloadTasksId200JSONResponseBodyStatus = "canceled" + Completed PatchApiDownloadTasksId200JSONResponseBodyStatus = "completed" + Failed PatchApiDownloadTasksId200JSONResponseBodyStatus = "failed" + Queued PatchApiDownloadTasksId200JSONResponseBodyStatus = "queued" + Running PatchApiDownloadTasksId200JSONResponseBodyStatus = "running" + Uploading PatchApiDownloadTasksId200JSONResponseBodyStatus = "uploading" ) // Valid indicates whether the value is a known member of the PatchApiDownloadTasksId200JSONResponseBodyStatus enum. func (e PatchApiDownloadTasksId200JSONResponseBodyStatus) Valid() bool { switch e { - case PatchApiDownloadTasksId200JSONResponseBodyStatusAssigned: + case Assigned: return true - case PatchApiDownloadTasksId200JSONResponseBodyStatusBillingPaused: + case BillingPaused: return true - case PatchApiDownloadTasksId200JSONResponseBodyStatusCanceled: + case Canceled: return true - case PatchApiDownloadTasksId200JSONResponseBodyStatusCompleted: + case Completed: return true - case PatchApiDownloadTasksId200JSONResponseBodyStatusFailed: + case Failed: return true - case PatchApiDownloadTasksId200JSONResponseBodyStatusQueued: + case Queued: return true - case PatchApiDownloadTasksId200JSONResponseBodyStatusRunning: + case Running: return true - case PatchApiDownloadTasksId200JSONResponseBodyStatusUploading: + case Uploading: return true default: return false @@ -820,6 +970,9 @@ type GetApiDownloadTasksParamsAssignedTo string // GetApiDownloadTasks200JSONResponseBodyItemsDetailEngine defines parameters for GetApiDownloadTasks. type GetApiDownloadTasks200JSONResponseBodyItemsDetailEngine string +// GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase defines parameters for GetApiDownloadTasks. +type GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase string + // GetApiDownloadTasks200JSONResponseBodyItemsSourceType defines parameters for GetApiDownloadTasks. type GetApiDownloadTasks200JSONResponseBodyItemsSourceType string @@ -842,6 +995,9 @@ type PostApiDownloadTasksJSONBodySourceType string // PostApiDownloadTasks201JSONResponseBodyDetailEngine defines parameters for PostApiDownloadTasks. type PostApiDownloadTasks201JSONResponseBodyDetailEngine string +// PostApiDownloadTasks201JSONResponseBodyDetailPhase defines parameters for PostApiDownloadTasks. +type PostApiDownloadTasks201JSONResponseBodyDetailPhase string + // PostApiDownloadTasks201JSONResponseBodySourceType defines parameters for PostApiDownloadTasks. type PostApiDownloadTasks201JSONResponseBodySourceType string @@ -851,6 +1007,9 @@ type PostApiDownloadTasks201JSONResponseBodyStatus string // GetApiDownloadTasksId200JSONResponseBodyDetailEngine defines parameters for GetApiDownloadTasksId. type GetApiDownloadTasksId200JSONResponseBodyDetailEngine string +// GetApiDownloadTasksId200JSONResponseBodyDetailPhase defines parameters for GetApiDownloadTasksId. +type GetApiDownloadTasksId200JSONResponseBodyDetailPhase string + // GetApiDownloadTasksId200JSONResponseBodySourceType defines parameters for GetApiDownloadTasksId. type GetApiDownloadTasksId200JSONResponseBodySourceType string @@ -862,6 +1021,7 @@ type PatchApiDownloadTasksIdJSONBody struct { Detail *struct { Connections *int `json:"connections,omitempty"` Engine *PatchApiDownloadTasksIdJSONBodyDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -879,12 +1039,12 @@ type PatchApiDownloadTasksIdJSONBody struct { Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *PatchApiDownloadTasksIdJSONBodyDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -907,12 +1067,18 @@ type PatchApiDownloadTasksIdJSONBody struct { // PatchApiDownloadTasksIdJSONBodyDetailEngine defines parameters for PatchApiDownloadTasksId. type PatchApiDownloadTasksIdJSONBodyDetailEngine string +// PatchApiDownloadTasksIdJSONBodyDetailPhase defines parameters for PatchApiDownloadTasksId. +type PatchApiDownloadTasksIdJSONBodyDetailPhase string + // PatchApiDownloadTasksIdJSONBodyStatus defines parameters for PatchApiDownloadTasksId. type PatchApiDownloadTasksIdJSONBodyStatus string // PatchApiDownloadTasksId200JSONResponseBodyDetailEngine defines parameters for PatchApiDownloadTasksId. type PatchApiDownloadTasksId200JSONResponseBodyDetailEngine string +// PatchApiDownloadTasksId200JSONResponseBodyDetailPhase defines parameters for PatchApiDownloadTasksId. +type PatchApiDownloadTasksId200JSONResponseBodyDetailPhase string + // PatchApiDownloadTasksId200JSONResponseBodySourceType defines parameters for PatchApiDownloadTasksId. type PatchApiDownloadTasksId200JSONResponseBodySourceType string @@ -2310,6 +2476,7 @@ type GetApiDownloadTasksResponse struct { Detail *struct { Connections *int `json:"connections,omitempty"` Engine *GetApiDownloadTasks200JSONResponseBodyItemsDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -2327,12 +2494,12 @@ type GetApiDownloadTasksResponse struct { Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -2398,6 +2565,7 @@ type PostApiDownloadTasksResponse struct { Detail *struct { Connections *int `json:"connections,omitempty"` Engine *PostApiDownloadTasks201JSONResponseBodyDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -2415,12 +2583,12 @@ type PostApiDownloadTasksResponse struct { Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *PostApiDownloadTasks201JSONResponseBodyDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -2488,6 +2656,7 @@ type GetApiDownloadTasksIdResponse struct { Detail *struct { Connections *int `json:"connections,omitempty"` Engine *GetApiDownloadTasksId200JSONResponseBodyDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -2505,12 +2674,12 @@ type GetApiDownloadTasksIdResponse struct { Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *GetApiDownloadTasksId200JSONResponseBodyDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -2572,6 +2741,7 @@ type PatchApiDownloadTasksIdResponse struct { Detail *struct { Connections *int `json:"connections,omitempty"` Engine *PatchApiDownloadTasksId200JSONResponseBodyDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -2589,12 +2759,12 @@ type PatchApiDownloadTasksIdResponse struct { Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *PatchApiDownloadTasksId200JSONResponseBodyDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -3245,6 +3415,7 @@ func ParseGetApiDownloadTasksResponse(rsp *http.Response) (*GetApiDownloadTasksR Detail *struct { Connections *int `json:"connections,omitempty"` Engine *GetApiDownloadTasks200JSONResponseBodyItemsDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -3262,12 +3433,12 @@ func ParseGetApiDownloadTasksResponse(rsp *http.Response) (*GetApiDownloadTasksR Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *GetApiDownloadTasks200JSONResponseBodyItemsDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -3335,6 +3506,7 @@ func ParsePostApiDownloadTasksResponse(rsp *http.Response) (*PostApiDownloadTask Detail *struct { Connections *int `json:"connections,omitempty"` Engine *PostApiDownloadTasks201JSONResponseBodyDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -3352,12 +3524,12 @@ func ParsePostApiDownloadTasksResponse(rsp *http.Response) (*PostApiDownloadTask Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *PostApiDownloadTasks201JSONResponseBodyDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -3439,6 +3611,7 @@ func ParseGetApiDownloadTasksIdResponse(rsp *http.Response) (*GetApiDownloadTask Detail *struct { Connections *int `json:"connections,omitempty"` Engine *GetApiDownloadTasksId200JSONResponseBodyDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -3456,12 +3629,12 @@ func ParseGetApiDownloadTasksIdResponse(rsp *http.Response) (*GetApiDownloadTask Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *GetApiDownloadTasksId200JSONResponseBodyDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` @@ -3525,6 +3698,7 @@ func ParsePatchApiDownloadTasksIdResponse(rsp *http.Response) (*PatchApiDownload Detail *struct { Connections *int `json:"connections,omitempty"` Engine *PatchApiDownloadTasksId200JSONResponseBodyDetailEngine `json:"engine,omitempty"` + EngineState *string `json:"engineState,omitempty"` EtaSeconds *int `json:"etaSeconds,omitempty"` Files *[]struct { CompletedBytes *int `json:"completedBytes,omitempty"` @@ -3542,12 +3716,12 @@ func ParsePatchApiDownloadTasksIdResponse(rsp *http.Response) (*PatchApiDownload Progress *float32 `json:"progress,omitempty"` UploadBps *int `json:"uploadBps,omitempty"` } `json:"peerSamples,omitempty"` - PeerUploadBps *int `json:"peerUploadBps,omitempty"` - PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` - Peers *int `json:"peers,omitempty"` - Phase *string `json:"phase,omitempty"` - Seeders *int `json:"seeders,omitempty"` - TorrentName *string `json:"torrentName,omitempty"` + PeerUploadBps *int `json:"peerUploadBps,omitempty"` + PeerUploadedBytes *int `json:"peerUploadedBytes,omitempty"` + Peers *int `json:"peers,omitempty"` + Phase *PatchApiDownloadTasksId200JSONResponseBodyDetailPhase `json:"phase,omitempty"` + Seeders *int `json:"seeders,omitempty"` + TorrentName *string `json:"torrentName,omitempty"` Trackers *[]struct { Leechers *int `json:"leechers,omitempty"` Message *string `json:"message,omitempty"` diff --git a/downloader/internal/worker/worker.go b/downloader/internal/worker/worker.go index 34dccaef..a2c2f9b2 100644 --- a/downloader/internal/worker/worker.go +++ b/downloader/internal/worker/worker.go @@ -26,6 +26,7 @@ import ( const Version = "0.1.0" const maxTaskErrorMessageLength = 1000 +const retainedSeedReportInterval = 5 * time.Second var errBillingPaused = errors.New("billing paused") @@ -47,6 +48,7 @@ type retainedSeed struct { path string retainedAt time.Time expiresAt time.Time + snapshot func(context.Context) (engine.SeedSnapshot, error) cleanup func(context.Context) error } @@ -91,6 +93,8 @@ func (w *Worker) Run(ctx context.Context) error { defer ticker.Stop() seedCleanupTicker := time.NewTicker(time.Minute) defer seedCleanupTicker.Stop() + seedReportTicker := time.NewTicker(retainedSeedReportInterval) + defer seedReportTicker.Stop() if err := w.tick(ctx); err != nil { w.logger.Error("downloader tick failed", "error", err) @@ -106,6 +110,8 @@ func (w *Worker) Run(ctx context.Context) error { } case <-seedCleanupTicker.C: w.cleanupRetainedSeeds(ctx) + case <-seedReportTicker.C: + w.reportRetainedSeeds(ctx) } } } @@ -215,6 +221,7 @@ func (w *Worker) process(ctx context.Context, task client.DownloadTask) { } log.Debug("task completed", "object_id", resultObjectID) if w.retainSeed(task, result, log) { + w.reportRetainedSeeds(ctx) w.cleanupRetainedSeeds(ctx) return } @@ -231,7 +238,7 @@ func cleanupDownloadedResult(ctx context.Context, result engine.Result) error { } func (w *Worker) retainSeed(task client.DownloadTask, result engine.Result, log *slog.Logger) bool { - if !w.cfg.SeedEnabled || result.Seed == nil || result.Seed.Cleanup == nil { + if !w.cfg.SeedEnabled || result.Seed == nil || result.Seed.Cleanup == nil || result.Seed.Snapshot == nil { return false } now := time.Now() @@ -241,6 +248,7 @@ func (w *Worker) retainSeed(task client.DownloadTask, result engine.Result, log seedID: result.Seed.ID, path: result.Seed.Path, retainedAt: now, + snapshot: result.Seed.Snapshot, cleanup: result.Seed.Cleanup, } if w.cfg.SeedDuration > 0 { @@ -261,6 +269,34 @@ func (w *Worker) retainSeed(task client.DownloadTask, result engine.Result, log return true } +func (w *Worker) reportRetainedSeeds(ctx context.Context) { + for _, seed := range w.retainedSeedSnapshot() { + log := w.logger.With("task_id", seed.taskID, "engine", seed.engine, "seed_id", seed.seedID) + snapshot, err := seed.snapshot(ctx) + if err != nil { + log.Warn("failed to inspect retained bt seed", "error", err) + continue + } + if snapshot.Detail == nil { + continue + } + snapshot.Detail.Phase = "seeding" + zero := int64(0) + _, err = w.api.UpdateTask(ctx, seed.taskID, client.TaskPatch{ + DownloadedBytes: &snapshot.Downloaded, + TotalBytes: snapshot.Total, + DownloadBps: &snapshot.Bps, + StorageUploadBps: &zero, + Detail: snapshot.Detail, + }) + if err != nil { + log.Warn("failed to report retained bt seed", "error", err) + continue + } + log.Debug("reported retained bt seed", "downloaded_bytes", snapshot.Downloaded, "bps", snapshot.Bps) + } +} + func (w *Worker) cleanupRetainedSeeds(ctx context.Context) { seeds := w.retainedSeedSnapshot() if len(seeds) == 0 { diff --git a/downloader/internal/worker/worker_test.go b/downloader/internal/worker/worker_test.go index 237b87a1..b0b685f6 100644 --- a/downloader/internal/worker/worker_test.go +++ b/downloader/internal/worker/worker_test.go @@ -102,6 +102,9 @@ func TestRetainSeedKeepsDownloadedResult(t *testing.T) { Engine: "aria2", ID: "gid", Path: dir, + Snapshot: func(context.Context) (engine.SeedSnapshot, error) { + return engine.SeedSnapshot{}, nil + }, Cleanup: func(context.Context) error { cleaned = true return nil @@ -132,6 +135,9 @@ func TestCleanupRetainedSeedsRemovesExpiredSeed(t *testing.T) { seedID: "gid", path: dir, expiresAt: time.Now().Add(-time.Second), + snapshot: func(context.Context) (engine.SeedSnapshot, error) { + return engine.SeedSnapshot{}, nil + }, cleanup: func(context.Context) error { cleaned = true return nil @@ -174,6 +180,9 @@ func TestCleanupRetainedSeedsRemovesOldestWhenCacheLimitExceeded(t *testing.T) { seedID: "old-hash", path: oldDir, retainedAt: time.Now().Add(-time.Hour), + snapshot: func(context.Context) (engine.SeedSnapshot, error) { + return engine.SeedSnapshot{}, nil + }, cleanup: func(context.Context) error { cleaned = append(cleaned, "old") return nil @@ -185,6 +194,9 @@ func TestCleanupRetainedSeedsRemovesOldestWhenCacheLimitExceeded(t *testing.T) { seedID: "new-hash", path: newDir, retainedAt: time.Now(), + snapshot: func(context.Context) (engine.SeedSnapshot, error) { + return engine.SeedSnapshot{}, nil + }, cleanup: func(context.Context) error { cleaned = append(cleaned, "new") return nil diff --git a/server/routes/download-tasks.integration.test.ts b/server/routes/download-tasks.integration.test.ts index 04e9f07b..a9721b07 100644 --- a/server/routes/download-tasks.integration.test.ts +++ b/server/routes/download-tasks.integration.test.ts @@ -205,7 +205,8 @@ describe('Download tasks API integration', () => { downloadBps: 512_000, detail: { engine: 'aria2', - phase: 'active', + phase: 'downloading', + engineState: 'active', infoHash: 'abc123', torrentName: 'fixture', connections: 8, diff --git a/server/routes/download-tasks.ts b/server/routes/download-tasks.ts index a443bd57..3f601433 100644 --- a/server/routes/download-tasks.ts +++ b/server/routes/download-tasks.ts @@ -58,6 +58,9 @@ const downloadTaskPageSchema = z.object({ pageSize: z.number().int(), }) +const sseEncoder = new TextEncoder() +const downloadTaskEventIntervalMs = 2000 + function jsonResponse(schema: z.ZodType, description: string) { return { content: { 'application/json': { schema } }, description } } @@ -85,6 +88,19 @@ const createRouteDoc = createRoute({ }, }) +const eventsRoute = createRoute({ + method: 'get', + path: '/events', + middleware: [requireAuth] as const, + responses: { + 200: { + content: { 'text/event-stream': { schema: z.string() } }, + description: 'Download task events', + }, + 401: jsonResponse(errorSchema, 'Unauthorized'), + }, +}) + const getRoute = createRoute({ method: 'get', path: '/{id}', @@ -154,6 +170,52 @@ const downloadTasksRoute = new OpenAPIHono() 201, ) }) as never) + .openapi(eventsRoute, (async (c: OpenAPIContext) => { + const orgId = c.get('orgId') + if (!orgId) return c.json({ error: 'Unauthorized' }, 401) + const signal = c.req.raw.signal + let closed = false + let lastFingerprint = '' + + const stream = new ReadableStream({ + start(controller) { + const send = (event: string, data: unknown) => { + controller.enqueue(sseEncoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)) + } + const tick = async () => { + if (closed) return + try { + const result = await listDownloadTasks(c.get('platform'), { orgId, page: 1, pageSize: 50 }) + const fingerprint = result.items.map((task) => `${task.id}:${task.updatedAt}`).join('|') + if (fingerprint !== lastFingerprint) { + lastFingerprint = fingerprint + send('snapshot', { items: result.items, total: result.total, page: 1, pageSize: 50 }) + } else { + send('heartbeat', { at: new Date().toISOString() }) + } + } catch (error) { + send('error', { message: error instanceof Error ? error.message : 'unknown error' }) + } + if (!closed) setTimeout(tick, downloadTaskEventIntervalMs) + } + signal.addEventListener('abort', () => { + closed = true + controller.close() + }) + void tick() + }, + cancel() { + closed = true + }, + }) + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream; charset=utf-8', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }, + }) + }) as never) .openapi(getRoute, (async (c: OpenAPIContext) => { const orgId = c.get('orgId') if (!orgId) return c.json({ error: 'Unauthorized' }, 401) diff --git a/server/services/downloads.ts b/server/services/downloads.ts index 754a728a..4382e866 100644 --- a/server/services/downloads.ts +++ b/server/services/downloads.ts @@ -124,6 +124,7 @@ export async function deleteDownloader(platform: Platform, id: string): Promise< const rows = await platform.db.select({ id: downloaders.id }).from(downloaders).where(eq(downloaders.id, id)).limit(1) if (!rows[0]) throw new DownloadError('not_found') const now = new Date() + await platform.db .update(downloadTasks) .set({ @@ -342,6 +343,9 @@ export async function updateDownloadTask( } } + const nextFinishedAt = + task.finishedAt ?? (input.status !== undefined && ['completed', 'failed', 'canceled'].includes(status) ? now : null) + await platform.db .update(downloadTasks) .set({ @@ -359,7 +363,7 @@ export async function updateDownloadTask( resultObjectId: input.resultObjectId === undefined ? task.resultObjectId : input.resultObjectId, detail: input.detail === undefined ? task.detail : JSON.stringify(input.detail), startedAt: task.startedAt ?? (status === 'running' ? now : null), - finishedAt: ['completed', 'failed', 'canceled'].includes(status) ? now : task.finishedAt, + finishedAt: nextFinishedAt, updatedAt: now, }) .where(eq(downloadTasks.id, id)) diff --git a/shared/schemas/downloads.ts b/shared/schemas/downloads.ts index 0bdcd788..80e5451c 100644 --- a/shared/schemas/downloads.ts +++ b/shared/schemas/downloads.ts @@ -13,6 +13,7 @@ export const downloadTaskStatusSchema = z.enum([ 'canceled', ]) export const downloadSourceTypeSchema = z.enum(['http', 'magnet', 'torrent_url']) +export const downloadTaskPhaseSchema = z.enum(['metadata', 'downloading', 'uploading', 'seeding', 'completed', 'error']) const downloadTaskTrackerSchema = z.object({ url: z.string().max(1024), @@ -40,7 +41,8 @@ const downloadTaskFileSchema = z.object({ export const downloadTaskDetailSchema = z.object({ engine: downloaderEngineSchema.optional(), - phase: z.string().max(80).optional(), + phase: downloadTaskPhaseSchema.optional(), + engineState: z.string().max(80).optional(), message: z.string().max(500).optional(), etaSeconds: z.number().int().min(0).nullable().optional(), connections: z.number().int().min(0).optional(), diff --git a/shared/types/index.ts b/shared/types/index.ts index 5e1752c3..b94bf742 100644 --- a/shared/types/index.ts +++ b/shared/types/index.ts @@ -297,7 +297,8 @@ export interface DownloadTaskFile { export interface DownloadTaskDetail { engine?: Downloader['engine'] - phase?: string + phase?: 'metadata' | 'downloading' | 'uploading' | 'seeding' | 'completed' | 'error' + engineState?: string message?: string etaSeconds?: number | null connections?: number diff --git a/src/i18n/locales/en.json b/src/i18n/locales/en.json index c1994766..b2ece355 100644 --- a/src/i18n/locales/en.json +++ b/src/i18n/locales/en.json @@ -147,8 +147,15 @@ "downloads.status.billing_paused": "Billing paused", "downloads.status.uploading": "Uploading", "downloads.status.completed": "Completed", + "downloads.status.seeding": "Seeding", "downloads.status.failed": "Failed", "downloads.status.canceled": "Canceled", + "downloads.phase.metadata": "Metadata", + "downloads.phase.downloading": "Downloading", + "downloads.phase.uploading": "Uploading", + "downloads.phase.seeding": "Seeding", + "downloads.phase.completed": "Completed", + "downloads.phase.error": "Error", "downloads.detail.downloadSpeed": "Download speed", "downloads.detail.uploadSpeed": "Upload speed", "downloads.detail.connections": "Connections", @@ -163,6 +170,7 @@ "downloads.detail.target": "Target folder", "downloads.detail.engine": "Engine", "downloads.detail.phase": "Phase", + "downloads.detail.engineState": "Engine state", "downloads.detail.sourceType": "Source type", "downloads.detail.source": "Source", "downloads.detail.size": "Size", diff --git a/src/i18n/locales/zh.json b/src/i18n/locales/zh.json index 6dffaf33..61cd61a1 100644 --- a/src/i18n/locales/zh.json +++ b/src/i18n/locales/zh.json @@ -147,8 +147,15 @@ "downloads.status.billing_paused": "计费暂停", "downloads.status.uploading": "上传中", "downloads.status.completed": "已完成", + "downloads.status.seeding": "做种中", "downloads.status.failed": "失败", "downloads.status.canceled": "已取消", + "downloads.phase.metadata": "获取元数据", + "downloads.phase.downloading": "下载中", + "downloads.phase.uploading": "上传中", + "downloads.phase.seeding": "做种中", + "downloads.phase.completed": "已完成", + "downloads.phase.error": "错误", "downloads.detail.downloadSpeed": "下载速度", "downloads.detail.uploadSpeed": "上传速度", "downloads.detail.connections": "连接数", @@ -163,6 +170,7 @@ "downloads.detail.target": "目标目录", "downloads.detail.engine": "下载引擎", "downloads.detail.phase": "阶段", + "downloads.detail.engineState": "引擎状态", "downloads.detail.sourceType": "来源类型", "downloads.detail.source": "来源", "downloads.detail.size": "大小", diff --git a/src/lib/api.test.ts b/src/lib/api.test.ts index 73d2c788..00a224c9 100644 --- a/src/lib/api.test.ts +++ b/src/lib/api.test.ts @@ -44,6 +44,7 @@ import { deleteUser, disableCloudGiftCard, disconnectCloud, + downloadTaskEventsUrl, emptyTrash, enableIhostFeature, getAnnouncement, @@ -1036,6 +1037,10 @@ describe('api', () => { expect(init.body).toBe(JSON.stringify(body)) }) + it('builds the download task events URL from RPC client', () => { + expect(downloadTaskEventsUrl().pathname).toBe('/api/download-tasks/events') + }) + it('lists admin downloaders', async () => { const payload = { items: [{ id: 'downloader-1', name: 'vps-1' }], total: 1 } vi.mocked(fetch).mockResolvedValueOnce(makeResponse(payload)) diff --git a/src/lib/api.ts b/src/lib/api.ts index 64960619..9b260b9d 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -71,6 +71,7 @@ import { cloudStoreApi, downloaderSelfApi, downloadTasksApi, + downloadTasksUrlApi, emailConfig, ihostApi, ihostConfigApi, @@ -292,6 +293,10 @@ export function updateDownloadTask(id: string, data: UpdateDownloadTaskInput) { return unwrap(downloadTasksApi[':id'].$patch({ param: { id }, json: data })) } +export function downloadTaskEventsUrl() { + return downloadTasksUrlApi.events.$url() +} + export function listDownloaders() { return unwrap>(adminDownloadersApi.index.$get()) } diff --git a/src/lib/rpc.ts b/src/lib/rpc.ts index d19f77c9..eb797cca 100644 --- a/src/lib/rpc.ts +++ b/src/lib/rpc.ts @@ -38,9 +38,14 @@ import type { import { hc } from 'hono/client' const opts = { init: { credentials: 'include' as RequestCredentials } } +const absoluteUrlBase = (path: string) => { + const origin = typeof window === 'undefined' ? 'http://localhost' : window.location.origin + return new URL(path, origin).toString() +} export const objects = hc('/api/objects', opts) export const downloadTasksApi = hc('/api/download-tasks', opts) +export const downloadTasksUrlApi = hc(absoluteUrlBase('/api/download-tasks'), opts) export const downloaderSelfApi = hc('/api/downloader', opts) export const trash = hc('/api/trash', opts) export const storages = hc('/api/admin/storages', opts) diff --git a/src/routes/_authenticated/downloads/index.tsx b/src/routes/_authenticated/downloads/index.tsx index fcebe194..6a1ea29f 100644 --- a/src/routes/_authenticated/downloads/index.tsx +++ b/src/routes/_authenticated/downloads/index.tsx @@ -22,7 +22,7 @@ import { Upload, Users, } from 'lucide-react' -import { type FormEvent, type ReactNode, useState } from 'react' +import { type FormEvent, type ReactNode, useEffect, useState } from 'react' import { useTranslation } from 'react-i18next' import { toast } from 'sonner' import { useFilesQuery } from '@/components/files/hooks/use-files-query' @@ -37,7 +37,7 @@ import { Progress } from '@/components/ui/progress' import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from '@/components/ui/table' import { Textarea } from '@/components/ui/textarea' import { ToggleGroup, ToggleGroupItem } from '@/components/ui/toggle-group' -import { createDownloadTask, listDownloadTasks, updateDownloadTask } from '@/lib/api' +import { createDownloadTask, downloadTaskEventsUrl, listDownloadTasks, updateDownloadTask } from '@/lib/api' import { cn } from '@/lib/utils' export const Route = createFileRoute('/_authenticated/downloads/')({ @@ -46,6 +46,8 @@ export const Route = createFileRoute('/_authenticated/downloads/')({ const QUERY_KEY = ['download-tasks'] const ACTIVE_STATUSES = new Set(['queued', 'assigned', 'running', 'billing_paused', 'uploading']) +type DownloadTaskDisplayStatus = DownloadTaskStatus | 'seeding' +type DownloadTaskPhase = NonNullable['phase']> type DetailTab = 'overview' | 'trackers' | 'peers' | 'files' | 'log' const DETAIL_TABS: Array<{ id: DetailTab; labelKey: string; icon: ReactNode }> = [ @@ -70,9 +72,17 @@ function DownloadsPage() { const tasksQuery = useQuery({ queryKey: QUERY_KEY, queryFn: () => listDownloadTasks({ page: 1, pageSize: 50 }), - refetchInterval: 2000, }) + useEffect(() => { + const events = new EventSource(downloadTaskEventsUrl(), { withCredentials: true }) + events.addEventListener('snapshot', (event) => { + const data = JSON.parse((event as MessageEvent).data) + queryClient.setQueryData(QUERY_KEY, data) + }) + return () => events.close() + }, [queryClient]) + const createMutation = useMutation({ mutationFn: createDownloadTask, onSuccess: () => { @@ -366,7 +376,6 @@ function TaskRow({ const { t } = useTranslation() const progress = transferProgress(task) const active = ACTIVE_STATUSES.has(task.status) - const detail = task.detail return ( - -
- {task.status === 'billing_paused' ? t('downloads.billingPaused') : detail?.phase || '-'} -
+
@@ -402,7 +408,7 @@ function TaskRow({ {formatBytes(task.downloadedBytes)} / {task.totalBytes ? formatBytes(task.totalBytes) : t('downloads.unknown')} - {formatDuration(detail?.etaSeconds)} + {formatDuration(task.detail?.etaSeconds)} {active ? ( @@ -553,7 +559,8 @@ function OverviewPanel({ task }: { task: DownloadTask }) {
- + + {t(`downloads.status.${status}`)} } +function formatPhase(phase: DownloadTaskPhase | undefined, t: ReturnType['t']) { + if (!phase) return '-' + return t(`downloads.phase.${phase}`) +} + function formatBytes(bytes: number) { if (!Number.isFinite(bytes) || bytes <= 0) return '0 B' const units = ['B', 'KB', 'MB', 'GB', 'TB']