opensource/zpan
3
0
Fork 0
mirror of https://github.com/saltbo/zpan.git synced 2026-09-19 01:51:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(downloads): stream normalized task details

Browse Source
This commit is contained in:
saltbo
2026-06-03 13:08:14 -04:00
parent c9e0a9b0f2
commit 1b4f149d05
17 changed files with 610 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/openapi/downloader.json
+55
View File
@@ -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
},
downloader/internal/client/client.go
+1
View File
@@ -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"`
downloader/internal/engine/engine.go
+123 -14
View File
@@ -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 != "" {
downloader/internal/openapi/client.gen.go
+244 -70
View File
@@ -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"`
downloader/internal/worker/worker.go
+37 -1
View File
@@ -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 {
downloader/internal/worker/worker_test.go
+12
View File
@@ -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
server/routes/download-tasks.integration.test.ts
+2 -1
View File
@@ -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,
server/routes/download-tasks.ts
+62
View File
@@ -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<Env>()
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<Uint8Array>({
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)
server/services/downloads.ts
+5 -1
View File
@@ -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))
shared/schemas/downloads.ts
+3 -1
View File
@@ -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(),
shared/types/index.ts
+2 -1
View File
@@ -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
src/i18n/locales/en.json
+8
View File
@@ -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",
src/i18n/locales/zh.json
+8
View File
@@ -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": "大小",
src/lib/api.test.ts
+5
View File
@@ -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))
src/lib/api.ts
+5
View File
@@ -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<DownloadTask>(downloadTasksApi[':id'].$patch({ param: { id }, json: data }))
}
export function downloadTaskEventsUrl() {
return downloadTasksUrlApi.events.$url()
}
export function listDownloaders() {
return unwrap<PaginatedResponse<Downloader>>(adminDownloadersApi.index.$get())
}
src/lib/rpc.ts
+5
View File
@@ -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<ObjectsRoute>('/api/objects', opts)
export const downloadTasksApi = hc<DownloadTasksRoute>('/api/download-tasks', opts)
export const downloadTasksUrlApi = hc<DownloadTasksRoute>(absoluteUrlBase('/api/download-tasks'), opts)
export const downloaderSelfApi = hc<DownloaderSelfRoute>('/api/downloader', opts)
export const trash = hc<TrashRoute>('/api/trash', opts)
export const storages = hc<StoragesRoute>('/api/admin/storages', opts)
src/routes/_authenticated/downloads/index.tsx
+33 -12
View File
@@ -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<DownloadTaskStatus>(['queued', 'assigned', 'running', 'billing_paused', 'uploading'])
type DownloadTaskDisplayStatus = DownloadTaskStatus | 'seeding'
type DownloadTaskPhase = NonNullable<NonNullable<DownloadTask['detail']>['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<string>).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 (
<TableRow
@@ -383,10 +392,7 @@ function TaskRow({
</div>
</TableCell>
<TableCell className="py-1">
<StatusBadge status={task.status} />
<div className="mt-0.5 max-w-32 truncate text-[11px] text-muted-foreground">
{task.status === 'billing_paused' ? t('downloads.billingPaused') : detail?.phase || '-'}
</div>
<StatusBadge status={displayStatus(task)} />
</TableCell>
<TableCell className="min-w-48 py-1">
<div className="flex items-center gap-2">
@@ -402,7 +408,7 @@ function TaskRow({
{formatBytes(task.downloadedBytes)} / {task.totalBytes ? formatBytes(task.totalBytes) : t('downloads.unknown')}
</TableCell>
<TableCell className="whitespace-nowrap py-1 text-[11px] tabular-nums text-muted-foreground">
{formatDuration(detail?.etaSeconds)}
{formatDuration(task.detail?.etaSeconds)}
</TableCell>
<TableCell className="py-1 text-right">
{active ? (
@@ -553,7 +559,8 @@ function OverviewPanel({ task }: { task: DownloadTask }) {
<div className="grid gap-x-5 gap-y-2 text-xs sm:grid-cols-2 xl:grid-cols-4">
<InspectorField label={t('downloads.detail.progress')} value={`${progress.overall}%`} />
<InspectorField label={t('downloads.detail.engine')} value={detail?.engine || t('downloads.unknown')} />
<InspectorField label={t('downloads.detail.phase')} value={detail?.phase || '-'} />
<InspectorField label={t('downloads.detail.phase')} value={formatPhase(detail?.phase, t)} />
<InspectorField label={t('downloads.detail.engineState')} value={detail?.engineState || '-'} />
<InspectorField
label={t('downloads.detail.target')}
value={task.targetFolder || t('downloads.targetFolderRoot')}
@@ -808,13 +815,27 @@ function formatDate(value: string | null | undefined) {
}).format(date)
}
function StatusBadge({ status }: { status: DownloadTaskStatus }) {
function displayStatus(task: DownloadTask): DownloadTaskDisplayStatus {
if (task.status === 'completed' && task.detail?.phase === 'seeding') return 'seeding'
return task.status
}
function StatusBadge({ status }: { status: DownloadTaskDisplayStatus }) {
const { t } = useTranslation()
const variant =
status === 'completed' ? 'default' : status === 'failed' || status === 'canceled' ? 'destructive' : 'secondary'
status === 'completed' || status === 'seeding'
? 'default'
: status === 'failed' || status === 'canceled'
? 'destructive'
: 'secondary'
return <Badge variant={variant}>{t(`downloads.status.${status}`)}</Badge>
}
function formatPhase(phase: DownloadTaskPhase | undefined, t: ReturnType<typeof useTranslation>['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']