diff --git a/.env.example b/.env.example index 3a3a384..a8c749a 100755 --- a/.env.example +++ b/.env.example @@ -14,4 +14,5 @@ DOWNLOAD_HOST_DIR=./data/downloads MIN_FREE_GB=10 MAX_CONCURRENT_DOWNLOADS=2 +CONNECTIONS_PER_DOWNLOAD=4 ALLOWED_PORTS=80,443 diff --git a/README.md b/README.md index 6dd6330..4aec011 100755 --- a/README.md +++ b/README.md @@ -18,11 +18,12 @@ Self-hosted direct download client that fetches files onto the server, extracts | `internal/libraries` | Discovers and validates first level library folders | | `internal/auth` | Sessions, credential check, login rate limiting | | `internal/guard` | URL validation and SSRF-safe HTTP client | -| `internal/fetch` | Streaming download with progress, size and stall limits | +| `internal/fetch` | Resumable multi-connection downloads with progress, size and stall limits | +| `internal/history` | Move history saved to `.history.json` | | `internal/extract` | Zip and tar extraction with Zip Slip and bomb protection | | `internal/files` | Safe path resolution, file tree, cross-device moves | | `internal/diskspace` | Free space checks | -| `internal/jobs` | Download job lifecycle and workflows | +| `internal/jobs` | Download job lifecycle, pause and resume, state saved to `.job.json` per download | | `internal/server` | HTTP routes, middleware, handlers | | `web/static` | Single page frontend | | `deploy/nginx` | Reverse proxy example | diff --git a/internal/config/config.go b/internal/config/config.go index 8465c84..bdac0bb 100755 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -38,6 +38,7 @@ func Load() (Settings, error) { DownloadDir: filepath.Clean(reader.text("DOWNLOAD_DIR", "/data/downloads")), Defaults: preferences.Preferences{ MaxConcurrentDownloads: reader.integer("MAX_CONCURRENT_DOWNLOADS", 2), + ConnectionsPerDownload: reader.integer("CONNECTIONS_PER_DOWNLOAD", 4), MinFreeGB: reader.integer("MIN_FREE_GB", 10), AllowedPorts: reader.ports("ALLOWED_PORTS", "80,443"), AutoExtract: true, diff --git a/internal/fetch/copy.go b/internal/fetch/copy.go new file mode 100644 index 0000000..7e17d17 --- /dev/null +++ b/internal/fetch/copy.go @@ -0,0 +1,55 @@ +package fetch + +import ( + "errors" + "fmt" + "io" + "sync/atomic" + "time" +) + +const ( + bufferSize = 1 << 20 + stallTimeout = 2 * time.Minute +) + +func copyBody(destination io.Writer, body io.ReadCloser, remaining int64, record func(int64) error) error { + var stalled atomic.Bool + stallTimer := time.AfterFunc(stallTimeout, func() { + stalled.Store(true) + body.Close() + }) + defer stallTimer.Stop() + + buffer := make([]byte, bufferSize) + for remaining != 0 { + readSize := len(buffer) + if remaining > 0 && remaining < int64(readSize) { + readSize = int(remaining) + } + + count, readErr := body.Read(buffer[:readSize]) + if count > 0 { + stallTimer.Reset(stallTimeout) + if _, err := destination.Write(buffer[:count]); err != nil { + return err + } + if err := record(int64(count)); err != nil { + return err + } + if remaining > 0 { + remaining -= int64(count) + } + } + if errors.Is(readErr, io.EOF) { + return nil + } + if readErr != nil { + if stalled.Load() { + return ErrStalled + } + return fmt.Errorf("%w: %w", errConnectionFailed, readErr) + } + } + return nil +} diff --git a/internal/fetch/downloader.go b/internal/fetch/downloader.go index b068e1b..7e8d7e9 100755 --- a/internal/fetch/downloader.go +++ b/internal/fetch/downloader.go @@ -3,25 +3,15 @@ package fetch import ( "context" "errors" - "fmt" - "io" "net/http" "os" "path/filepath" - "time" "ddc/internal/diskspace" "ddc/internal/files" "ddc/internal/guard" ) -const ( - userAgent = "ddc/1.0" - bufferSize = 1 << 20 - stallTimeout = 2 * time.Minute - spaceCheckInterval = int64(256) << 20 -) - var ( ErrIncomplete = errors.New("download ended before the full file was received") ErrStalled = errors.New("download stalled, no data received for too long") @@ -29,8 +19,21 @@ var ( type Progress interface { SetTotal(total int64) + SetDone(done int64) Add(delta int64) SetFileName(name string) + SetNotice(text string) +} + +type Request struct { + URL string + Directory string + FileName string + ReservedBytes int64 + Connections int + State *State + Progress Progress + Checkpoint func(State) } type Downloader struct { @@ -42,111 +45,99 @@ func NewDownloader(policy *guard.Policy) *Downloader { return &Downloader{policy: policy, client: policy.NewHTTPClient()} } -func (downloader *Downloader) Download(ctx context.Context, rawURL, directory string, reservedBytes int64, progress Progress) (string, error) { - target, err := downloader.policy.ValidateURL(rawURL) +func (downloader *Downloader) Download(ctx context.Context, request Request) (string, error) { + target, err := downloader.policy.ValidateURL(request.URL) if err != nil { return "", err } - ctx, cancel := context.WithCancelCause(ctx) - defer cancel(nil) - - response, err := downloader.open(ctx, target.String()) + source := &remoteFile{url: target.String()} + partPath := filepath.Join(request.Directory, files.PartFileName) + state, probe, err := downloader.prepare(ctx, source, request, partPath) if err != nil { return "", err } - defer response.Body.Close() - fileName := FileNameFromResponse(response) - progress.SetFileName(fileName) - if err := diskspace.EnsureFree(directory, max(response.ContentLength, 0), reservedBytes); err != nil { + progress := request.Progress + progress.SetFileName(state.FileName) + progress.SetTotal(state.TotalBytes) + progress.SetDone(state.DoneBytes()) + remainingBytes := max(state.TotalBytes-state.DoneBytes(), 0) + if err := diskspace.EnsureFree(request.Directory, remainingBytes, request.ReservedBytes); err != nil { + closeBody(probe) return "", err } - progress.SetTotal(max(response.ContentLength, 0)) - partPath := filepath.Join(directory, files.PartFileName) - written, err := writePart(cancel, partPath, response.Body, progress, directory, reservedBytes) + file, err := openPartFile(partPath, probe != nil) if err != nil { - os.Remove(partPath) - if errors.Is(context.Cause(ctx), ErrStalled) { - return "", ErrStalled - } + closeBody(probe) return "", err } - if response.ContentLength > 0 && written != response.ContentLength { - os.Remove(partPath) - return "", ErrIncomplete + runErr := newSession(downloader, source, file, request, state).run(ctx, probe) + closeErr := file.Close() + if runErr != nil { + return "", runErr + } + if closeErr != nil { + return "", closeErr } - finalPath := files.UniquePath(filepath.Join(directory, fileName)) + finalPath := files.UniquePath(filepath.Join(request.Directory, state.FileName)) if err := os.Rename(partPath, finalPath); err != nil { - os.Remove(partPath) return "", err } return finalPath, nil } -func (downloader *Downloader) open(ctx context.Context, target string) (*http.Response, error) { - request, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) - if err != nil { - return nil, err +func (downloader *Downloader) prepare(ctx context.Context, source *remoteFile, request Request, partPath string) (State, *http.Response, error) { + if request.State != nil && request.State.valid() && fileExists(partPath) { + state := *request.State + source.validator = state.Validator + source.totalBytes = state.TotalBytes + return state, nil, nil } - request.Header.Set("User-Agent", userAgent) - request.Header.Set("Accept-Encoding", "identity") - response, err := downloader.client.Do(request) + probe, err := downloader.open(ctx, source, 0, 0) if err != nil { - return nil, err + return State{}, nil, err } - if response.StatusCode != http.StatusOK { + source.describe(probe) + + connections := 1 + if probe.Header.Get("Accept-Ranges") == "bytes" { + connections = min(max(request.Connections, 1), MaxConnections) + } + state := State{ + FileName: chooseFileName(request.FileName, probe), + Validator: source.validator, + TotalBytes: source.totalBytes, + Segments: planSegments(source.totalBytes, connections), + } + return state, probe, nil +} + +func chooseFileName(requested string, response *http.Response) string { + if name := SanitizeFileName(requested); name != "" { + return name + } + return FileNameFromResponse(response) +} + +func openPartFile(partPath string, fresh bool) (*os.File, error) { + flags := os.O_WRONLY | os.O_CREATE + if fresh { + flags |= os.O_TRUNC + } + return os.OpenFile(partPath, flags, 0o644) +} + +func fileExists(path string) bool { + info, err := os.Stat(path) + return err == nil && info.Mode().IsRegular() +} + +func closeBody(response *http.Response) { + if response != nil { response.Body.Close() - return nil, fmt.Errorf("server responded with %s", response.Status) - } - return response, nil -} - -func writePart(cancel context.CancelCauseFunc, partPath string, body io.Reader, progress Progress, directory string, reservedBytes int64) (int64, error) { - file, err := os.OpenFile(partPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644) - if err != nil { - return 0, err - } - written, copyErr := copyBody(cancel, file, body, progress, directory, reservedBytes) - closeErr := file.Close() - if copyErr != nil { - return written, copyErr - } - return written, closeErr -} - -func copyBody(cancel context.CancelCauseFunc, destination io.Writer, body io.Reader, progress Progress, directory string, reservedBytes int64) (int64, error) { - stallTimer := time.AfterFunc(stallTimeout, func() { cancel(ErrStalled) }) - defer stallTimer.Stop() - - buffer := make([]byte, bufferSize) - var written, sinceSpaceCheck int64 - for { - count, readErr := body.Read(buffer) - if count > 0 { - stallTimer.Reset(stallTimeout) - if _, err := destination.Write(buffer[:count]); err != nil { - return written, err - } - written += int64(count) - sinceSpaceCheck += int64(count) - progress.Add(int64(count)) - - if sinceSpaceCheck >= spaceCheckInterval { - sinceSpaceCheck = 0 - if err := diskspace.EnsureFree(directory, 0, reservedBytes); err != nil { - return written, err - } - } - } - if errors.Is(readErr, io.EOF) { - return written, nil - } - if readErr != nil { - return written, readErr - } } } diff --git a/internal/fetch/request.go b/internal/fetch/request.go new file mode 100644 index 0000000..70be287 --- /dev/null +++ b/internal/fetch/request.go @@ -0,0 +1,141 @@ +package fetch + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + + "ddc/internal/guard" +) + +const userAgent = "ddc/1.0" + +var ( + ErrResumeUnsupported = errors.New("connection was lost and the server does not support resuming") + ErrSourceChanged = errors.New("file changed on the server while downloading") + errConnectionFailed = errors.New("connection to the server failed") + errServerUnavailable = errors.New("server is temporarily unavailable") +) + +type remoteFile struct { + url string + validator string + totalBytes int64 +} + +func (source *remoteFile) describe(response *http.Response) { + source.totalBytes = max(response.ContentLength, 0) + source.validator = validatorOf(response.Header) +} + +func validatorOf(header http.Header) string { + if entityTag := header.Get("ETag"); entityTag != "" && !strings.HasPrefix(entityTag, "W/") { + return entityTag + } + return header.Get("Last-Modified") +} + +type cancelingBody struct { + io.ReadCloser + cancel context.CancelFunc +} + +func (body *cancelingBody) Close() error { + body.cancel() + return body.ReadCloser.Close() +} + +func (downloader *Downloader) open(ctx context.Context, source *remoteFile, position, end int64) (*http.Response, error) { + requestContext, cancelRequest := context.WithCancel(ctx) + response, err := downloader.send(requestContext, source, position, end) + if err != nil { + cancelRequest() + return nil, err + } + response.Body = &cancelingBody{ReadCloser: response.Body, cancel: cancelRequest} + return response, nil +} + +func (downloader *Downloader) send(ctx context.Context, source *remoteFile, position, end int64) (*http.Response, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, source.url, nil) + if err != nil { + return nil, err + } + request.Header.Set("User-Agent", userAgent) + request.Header.Set("Accept-Encoding", "identity") + + ranged := position > 0 || (end > 0 && end < source.totalBytes) + if ranged { + request.Header.Set("Range", rangeHeader(position, end)) + if source.validator != "" { + request.Header.Set("If-Range", source.validator) + } + } + + response, err := downloader.client.Do(request) + if err != nil { + return nil, connectionError(err) + } + if err := checkResponse(response, source, position, ranged); err != nil { + response.Body.Close() + return nil, err + } + return response, nil +} + +func rangeHeader(position, end int64) string { + if end > 0 { + return fmt.Sprintf("bytes=%d-%d", position, end-1) + } + return fmt.Sprintf("bytes=%d-", position) +} + +func checkResponse(response *http.Response, source *remoteFile, position int64, ranged bool) error { + status := response.StatusCode + switch { + case !ranged && status == http.StatusOK: + if source.totalBytes > 0 && response.ContentLength != source.totalBytes { + return ErrSourceChanged + } + return nil + case ranged && status == http.StatusPartialContent: + return checkContentRange(response.Header.Get("Content-Range"), source.totalBytes, position) + case ranged && status == http.StatusOK: + if source.validator != "" && validatorOf(response.Header) != source.validator { + return ErrSourceChanged + } + return ErrResumeUnsupported + case status == http.StatusTooManyRequests || status >= http.StatusInternalServerError: + return fmt.Errorf("%w (%s)", errServerUnavailable, response.Status) + default: + return fmt.Errorf("server responded with %s", response.Status) + } +} + +func checkContentRange(header string, totalBytes, offset int64) error { + rangeText, found := strings.CutPrefix(header, "bytes ") + if !found { + return ErrResumeUnsupported + } + span, totalText, found := strings.Cut(rangeText, "/") + startText, _, _ := strings.Cut(span, "-") + start, err := strconv.ParseInt(startText, 10, 64) + if !found || err != nil || start != offset { + return ErrResumeUnsupported + } + if totalBytes > 0 && totalText != "*" && totalText != strconv.FormatInt(totalBytes, 10) { + return ErrSourceChanged + } + return nil +} + +func connectionError(err error) error { + if errors.Is(err, guard.ErrBlockedAddress) || errors.Is(err, guard.ErrBlockedPort) || errors.Is(err, guard.ErrTooManyRedirects) { + return err + } + return fmt.Errorf("%w: %w", errConnectionFailed, err) +} diff --git a/internal/fetch/session.go b/internal/fetch/session.go new file mode 100644 index 0000000..89a4df1 --- /dev/null +++ b/internal/fetch/session.go @@ -0,0 +1,128 @@ +package fetch + +import ( + "context" + "net/http" + "os" + "sync" + "sync/atomic" + "time" + + "ddc/internal/diskspace" +) + +const ( + checkpointInterval = 10 * time.Second + spaceCheckInterval = int64(256) << 20 +) + +type session struct { + downloader *Downloader + source *remoteFile + file *os.File + request Request + state State + segments []*liveSegment + sinceSpaceCheck atomic.Int64 +} + +func newSession(downloader *Downloader, source *remoteFile, file *os.File, request Request, state State) *session { + segments := make([]*liveSegment, len(state.Segments)) + for index, segment := range state.Segments { + segments[index] = newLiveSegment(segment) + } + return &session{ + downloader: downloader, + source: source, + file: file, + request: request, + state: state, + segments: segments, + } +} + +func (session *session) run(ctx context.Context, probe *http.Response) error { + session.saveCheckpoint() + stopSaving := make(chan struct{}) + savingStopped := make(chan struct{}) + go session.saveCheckpoints(stopSaving, savingStopped) + + err := session.runWorkers(ctx, probe) + close(stopSaving) + <-savingStopped + session.saveCheckpoint() + return err +} + +func (session *session) runWorkers(ctx context.Context, probe *http.Response) error { + ctx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + + var workers sync.WaitGroup + var firstFailure sync.Once + var failure error + for index, segment := range session.segments { + if segment.complete() { + continue + } + var initial *http.Response + if index == 0 { + initial, probe = probe, nil + } + worker := &segmentWorker{session: session, segment: segment} + workers.Add(1) + go func() { + defer workers.Done() + if err := worker.run(ctx, initial); err != nil { + firstFailure.Do(func() { + failure = err + cancel(err) + }) + } + }() + } + closeBody(probe) + workers.Wait() + return failure +} + +func (session *session) record(count int64) error { + session.request.Progress.Add(count) + if session.sinceSpaceCheck.Add(count) < spaceCheckInterval { + return nil + } + session.sinceSpaceCheck.Store(0) + return diskspace.EnsureFree(session.request.Directory, 0, session.request.ReservedBytes) +} + +func (session *session) snapshot() State { + state := session.state + state.Segments = make([]Segment, len(session.segments)) + for index, segment := range session.segments { + state.Segments[index] = segment.snapshot() + } + return state +} + +func (session *session) saveCheckpoints(stop <-chan struct{}, stopped chan<- struct{}) { + defer close(stopped) + ticker := time.NewTicker(checkpointInterval) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ticker.C: + session.saveCheckpoint() + } + } +} + +func (session *session) saveCheckpoint() { + if session.request.Checkpoint == nil { + return + } + state := session.snapshot() + session.file.Sync() + session.request.Checkpoint(state) +} diff --git a/internal/fetch/state.go b/internal/fetch/state.go new file mode 100644 index 0000000..3b3b0d8 --- /dev/null +++ b/internal/fetch/state.go @@ -0,0 +1,97 @@ +package fetch + +import "sync/atomic" + +const ( + MaxConnections = 16 + minSegmentBytes = int64(32) << 20 +) + +type Segment struct { + Start int64 `json:"start"` + End int64 `json:"end"` + Done int64 `json:"done"` +} + +type State struct { + FileName string `json:"fileName"` + Validator string `json:"validator"` + TotalBytes int64 `json:"totalBytes"` + Segments []Segment `json:"segments"` +} + +func (state State) DoneBytes() int64 { + var done int64 + for _, segment := range state.Segments { + done += segment.Done + } + return done +} + +func (state State) valid() bool { + if state.FileName == "" || len(state.Segments) == 0 { + return false + } + for _, segment := range state.Segments { + if segment.Start < 0 || segment.Done < 0 { + return false + } + if segment.End == 0 && state.TotalBytes > 0 { + return false + } + if segment.End > 0 && segment.Start+segment.Done > segment.End { + return false + } + } + return true +} + +func planSegments(totalBytes int64, connections int) []Segment { + if totalBytes <= 0 { + return []Segment{{}} + } + count := min(int64(max(connections, 1)), max(totalBytes/minSegmentBytes, 1)) + size := totalBytes / count + segments := make([]Segment, count) + for index := range segments { + start := int64(index) * size + segments[index] = Segment{Start: start, End: start + size} + } + segments[count-1].End = totalBytes + return segments +} + +type liveSegment struct { + start int64 + end int64 + done atomic.Int64 +} + +func newLiveSegment(segment Segment) *liveSegment { + live := &liveSegment{start: segment.Start, end: segment.End} + live.done.Store(segment.Done) + return live +} + +func (segment *liveSegment) position() int64 { + return segment.start + segment.done.Load() +} + +func (segment *liveSegment) bounded() bool { + return segment.end > 0 +} + +func (segment *liveSegment) remaining() int64 { + if !segment.bounded() { + return -1 + } + return segment.end - segment.position() +} + +func (segment *liveSegment) complete() bool { + return segment.bounded() && segment.position() >= segment.end +} + +func (segment *liveSegment) snapshot() Segment { + return Segment{Start: segment.start, End: segment.end, Done: segment.done.Load()} +} diff --git a/internal/fetch/worker.go b/internal/fetch/worker.go new file mode 100644 index 0000000..be6a919 --- /dev/null +++ b/internal/fetch/worker.go @@ -0,0 +1,107 @@ +package fetch + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "time" +) + +const ( + maxRetriesWithoutProgress = 8 + retryBaseDelay = 2 * time.Second + retryMaxDelay = 30 * time.Second +) + +type segmentWorker struct { + session *session + segment *liveSegment + failedAttempts int +} + +func (worker *segmentWorker) run(ctx context.Context, response *http.Response) error { + var err error + if response == nil { + response, err = worker.open(ctx) + } + for { + if err != nil { + response, err = worker.reconnect(ctx, err) + if err != nil { + return err + } + } + + doneBefore := worker.segment.done.Load() + err = worker.copyFrom(response) + if err == nil { + return nil + } + if worker.segment.done.Load() > doneBefore { + worker.failedAttempts = 0 + } + } +} + +func (worker *segmentWorker) open(ctx context.Context) (*http.Response, error) { + return worker.session.downloader.open(ctx, worker.session.source, worker.segment.position(), worker.segment.end) +} + +func (worker *segmentWorker) copyFrom(response *http.Response) error { + defer response.Body.Close() + writer := io.NewOffsetWriter(worker.session.file, worker.segment.position()) + if err := copyBody(writer, response.Body, worker.segment.remaining(), worker.record); err != nil { + return err + } + if worker.segment.bounded() && !worker.segment.complete() { + return ErrIncomplete + } + return nil +} + +func (worker *segmentWorker) record(count int64) error { + worker.segment.done.Add(count) + return worker.session.record(count) +} + +func (worker *segmentWorker) reconnect(ctx context.Context, cause error) (*http.Response, error) { + progress := worker.session.request.Progress + for { + worker.failedAttempts++ + if ctx.Err() != nil || !isRetryable(cause) || worker.failedAttempts > maxRetriesWithoutProgress { + return nil, cause + } + progress.SetNotice(fmt.Sprintf("Connection lost, reconnecting (attempt %d of %d)", worker.failedAttempts, maxRetriesWithoutProgress)) + if err := waitBeforeRetry(ctx, worker.failedAttempts); err != nil { + return nil, err + } + + response, err := worker.open(ctx) + if err == nil { + progress.SetNotice("") + return response, nil + } + cause = err + } +} + +func isRetryable(err error) bool { + return errors.Is(err, errConnectionFailed) || + errors.Is(err, errServerUnavailable) || + errors.Is(err, ErrStalled) || + errors.Is(err, ErrIncomplete) +} + +func waitBeforeRetry(ctx context.Context, attempt int) error { + delay := min(retryBaseDelay*time.Duration(attempt), retryMaxDelay) + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/internal/files/internal.go b/internal/files/internal.go new file mode 100644 index 0000000..a13b26e --- /dev/null +++ b/internal/files/internal.go @@ -0,0 +1,41 @@ +package files + +import ( + "errors" + "os" + "path/filepath" + "strings" +) + +const ( + PartFileName = ".download.part" + JobFileName = ".job.json" + temporaryPrefix = ".ddc-tmp-" + temporaryWildcard = temporaryPrefix + "*" +) + +func IsInternal(name string) bool { + return name == PartFileName || name == JobFileName || strings.HasPrefix(name, temporaryPrefix) +} + +func WriteAtomically(path string, data []byte) error { + temporary, err := os.CreateTemp(filepath.Dir(path), temporaryWildcard) + if err != nil { + return err + } + _, writeErr := temporary.Write(data) + closeErr := temporary.Close() + if writeErr != nil || closeErr != nil { + os.Remove(temporary.Name()) + return errors.Join(writeErr, closeErr) + } + if err := os.Chmod(temporary.Name(), 0o644); err != nil { + os.Remove(temporary.Name()) + return err + } + if err := os.Rename(temporary.Name(), path); err != nil { + os.Remove(temporary.Name()) + return err + } + return nil +} diff --git a/internal/files/paths.go b/internal/files/paths.go index 2be8fee..53f06c6 100755 --- a/internal/files/paths.go +++ b/internal/files/paths.go @@ -9,7 +9,6 @@ import ( ) const ( - PartFileName = ".download.part" maxSubfolderDepth = 4 maxNameBytes = 255 ) diff --git a/internal/files/tree.go b/internal/files/tree.go index 6fb753d..2118ed0 100755 --- a/internal/files/tree.go +++ b/internal/files/tree.go @@ -41,7 +41,7 @@ func readLevel(root, relative string, depth int, entryCount *int) ([]Node, error break } entryType := entry.Type() - if entry.Name() == PartFileName || (!entryType.IsDir() && !entryType.IsRegular()) { + if IsInternal(entry.Name()) || (!entryType.IsDir() && !entryType.IsRegular()) { continue } *entryCount++ diff --git a/internal/history/history.go b/internal/history/history.go new file mode 100644 index 0000000..8c0e649 --- /dev/null +++ b/internal/history/history.go @@ -0,0 +1,82 @@ +package history + +import ( + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" + "slices" + "sync" + "time" + + "ddc/internal/files" +) + +const maxEntries = 500 + +type Entry struct { + FileName string `json:"fileName"` + SourceURL string `json:"sourceUrl"` + Destination string `json:"destination"` + Items []string `json:"items"` + MovedAt time.Time `json:"movedAt"` +} + +type Store struct { + mutex sync.Mutex + path string + entries []Entry +} + +func Open(path string) (*Store, error) { + store := &Store{path: path, entries: []Entry{}} + data, err := os.ReadFile(path) + if errors.Is(err, fs.ErrNotExist) { + return store, nil + } + if err != nil { + return nil, err + } + if err := json.Unmarshal(data, &store.entries); err != nil { + return nil, fmt.Errorf("history file %s: %w", path, err) + } + if store.entries == nil { + store.entries = []Entry{} + } + return store, nil +} + +func (store *Store) List() []Entry { + store.mutex.Lock() + defer store.mutex.Unlock() + return slices.Clone(store.entries) +} + +func (store *Store) Add(entry Entry) error { + store.mutex.Lock() + defer store.mutex.Unlock() + next := append([]Entry{entry}, store.entries...) + if len(next) > maxEntries { + next = next[:maxEntries] + } + return store.save(next) +} + +func (store *Store) Clear() error { + store.mutex.Lock() + defer store.mutex.Unlock() + return store.save([]Entry{}) +} + +func (store *Store) save(entries []Entry) error { + data, err := json.MarshalIndent(entries, "", " ") + if err != nil { + return err + } + if err := files.WriteAtomically(store.path, data); err != nil { + return err + } + store.entries = entries + return nil +} diff --git a/internal/jobs/download.go b/internal/jobs/download.go new file mode 100644 index 0000000..cd18376 --- /dev/null +++ b/internal/jobs/download.go @@ -0,0 +1,122 @@ +package jobs + +import ( + "context" + "errors" + "path/filepath" + + "ddc/internal/extract" + "ddc/internal/fetch" +) + +func (manager *Manager) Pause(id string) error { + job, err := manager.lookup(id) + if err != nil { + return err + } + if !job.pause() { + return ErrJobBusy + } + return nil +} + +func (manager *Manager) Resume(id string) error { + job, err := manager.lookup(id) + if err != nil { + return err + } + if !job.begin(StatusQueued, StatusPaused, StatusFailed) { + return ErrJobBusy + } + job.saveRecord() + manager.launchDownload(job) + return nil +} + +func (manager *Manager) Reorder(id, direction string) error { + if direction != queueFirst && direction != queueUp && direction != queueDown { + return invalid(errors.New("direction must be first, up or down")) + } + job, err := manager.lookup(id) + if err != nil { + return err + } + if job.currentStatus() != StatusQueued || !manager.slots.reorder(id, direction) { + return ErrJobBusy + } + return nil +} + +func (manager *Manager) launchDownload(job *Job) { + runContext, cancelRun := job.startRun() + manager.slots.enqueue(job.id) + go func() { + defer job.workers.Done() + defer cancelRun(nil) + manager.download(job, runContext) + }() +} + +func (manager *Manager) download(job *Job, runContext context.Context) { + if !manager.slots.acquire(runContext, job.id, manager.concurrencyLimit) { + manager.handleDownloadError(job, runContext, context.Cause(runContext)) + return + } + defer manager.slots.release() + + job.setStatus(StatusDownloading, "") + filePath, err := manager.downloader.Download(runContext, manager.fetchRequest(job)) + if err != nil { + manager.handleDownloadError(job, runContext, err) + return + } + job.markDownloaded(filepath.Base(filePath)) + + if job.autoExtract && extract.IsArchive(filePath) { + job.setStatus(StatusExtracting, "") + destination, err := manager.extractArchive(job, filePath) + if err != nil { + manager.finishWithError(job, StatusReady, "Extraction failed: ", err) + job.saveRecord() + return + } + job.SetFileName(filepath.Base(destination)) + } + + job.setStatus(StatusReady, "") + job.saveRecord() + manager.moveToTarget(job) +} + +func (manager *Manager) fetchRequest(job *Job) fetch.Request { + current := manager.preferences.Get() + return fetch.Request{ + URL: job.sourceURL, + Directory: job.directory, + FileName: job.requestedName, + ReservedBytes: current.MinFreeBytes(), + Connections: current.ConnectionsPerDownload, + State: job.currentDownloadState(), + Progress: job, + Checkpoint: func(state fetch.State) { + job.setDownloadState(&state) + job.saveRecord() + }, + } +} + +func (manager *Manager) handleDownloadError(job *Job, runContext context.Context, err error) { + switch { + case errors.Is(context.Cause(runContext), errPaused): + job.setStatus(StatusPaused, "") + case job.jobContext.Err() != nil: + job.setStatus(StatusCancelled, "") + return + default: + if errors.Is(err, fetch.ErrSourceChanged) || errors.Is(err, fetch.ErrResumeUnsupported) { + job.setDownloadState(nil) + } + job.setStatus(StatusFailed, err.Error()) + } + job.saveRecord() +} diff --git a/internal/jobs/workflows.go b/internal/jobs/extraction.go old mode 100755 new mode 100644 similarity index 68% rename from internal/jobs/workflows.go rename to internal/jobs/extraction.go index 07b3a1c..051aa50 --- a/internal/jobs/workflows.go +++ b/internal/jobs/extraction.go @@ -14,35 +14,6 @@ const maxArchiveEntries = 100000 var errNotArchive = errors.New("selected file is not a supported archive") -func (manager *Manager) download(job *Job) { - defer job.workers.Done() - - if !manager.slots.acquire(job.jobContext, manager.concurrencyLimit) { - return - } - defer manager.slots.release() - - job.setStatus(StatusDownloading, "") - filePath, err := manager.downloader.Download(job.jobContext, job.sourceURL, job.directory, manager.reservedBytes(), job) - if err != nil { - manager.finishWithError(job, StatusFailed, "", err) - return - } - job.SetFileName(filepath.Base(filePath)) - - if job.autoExtract && extract.IsArchive(filePath) { - job.setStatus(StatusExtracting, "") - destination, err := manager.extractArchive(job, filePath) - if err != nil { - manager.finishWithError(job, StatusReady, "Extraction failed: ", err) - return - } - job.SetFileName(filepath.Base(destination)) - } - - job.setStatus(StatusReady, "") -} - func (manager *Manager) Extract(id, relativePath string) error { job, err := manager.lookup(id) if err != nil { @@ -58,7 +29,7 @@ func (manager *Manager) Extract(id, relativePath string) error { return invalid(errNotArchive) } - if !job.begin(StatusReady, StatusExtracting) { + if !job.begin(StatusExtracting, StatusReady) { return ErrJobBusy } go func() { diff --git a/internal/jobs/history.go b/internal/jobs/history.go new file mode 100644 index 0000000..b83e69f --- /dev/null +++ b/internal/jobs/history.go @@ -0,0 +1,33 @@ +package jobs + +import ( + "log" + "time" + + "ddc/internal/history" +) + +func (manager *Manager) History() []history.Entry { + return manager.history.List() +} + +func (manager *Manager) ClearHistory() error { + return manager.history.Clear() +} + +func (manager *Manager) recordHistory(job *Job, destination string, movedNames []string) { + if len(movedNames) == 0 { + return + } + snapshot := job.snapshot() + entry := history.Entry{ + FileName: snapshot.FileName, + SourceURL: snapshot.SourceURL, + Destination: destination, + Items: movedNames, + MovedAt: time.Now(), + } + if err := manager.history.Add(entry); err != nil { + log.Printf("could not save history: %v", err) + } +} diff --git a/internal/jobs/job.go b/internal/jobs/job.go index 565e76e..379293d 100755 --- a/internal/jobs/job.go +++ b/internal/jobs/job.go @@ -2,9 +2,12 @@ package jobs import ( "context" + "slices" "sync" "sync/atomic" "time" + + "ddc/internal/fetch" ) type Status string @@ -12,6 +15,7 @@ type Status string const ( StatusQueued Status = "queued" StatusDownloading Status = "downloading" + StatusPaused Status = "paused" StatusExtracting Status = "extracting" StatusReady Status = "ready" StatusMoving Status = "moving" @@ -21,56 +25,46 @@ const ( ) type Snapshot struct { - ID string `json:"id"` - SourceURL string `json:"sourceUrl"` - FileName string `json:"fileName"` - Status Status `json:"status"` - Message string `json:"message"` - BytesDone int64 `json:"bytesDone"` - BytesTotal int64 `json:"bytesTotal"` - CreatedAt time.Time `json:"createdAt"` + ID string `json:"id"` + SourceURL string `json:"sourceUrl"` + FileName string `json:"fileName"` + Status Status `json:"status"` + Message string `json:"message"` + BytesDone int64 `json:"bytesDone"` + BytesTotal int64 `json:"bytesTotal"` + CreatedAt time.Time `json:"createdAt"` + MoveTarget string `json:"moveTarget,omitempty"` + QueuePosition int `json:"queuePosition,omitempty"` } type Job struct { - id string - sourceURL string - displayURL string - directory string - autoExtract bool - createdAt time.Time - jobContext context.Context - cancel context.CancelFunc - workers sync.WaitGroup + id string + sourceURL string + displayURL string + directory string + requestedName string + autoExtract bool + target *moveTarget + createdAt time.Time + jobContext context.Context + cancel context.CancelFunc + workers sync.WaitGroup - mutex sync.Mutex - status Status - message string - fileName string - deleted bool + mutex sync.Mutex + status Status + message string + fileName string + deleted bool + downloaded bool + downloadState *fetch.State + cancelRun context.CancelCauseFunc + + recordMutex sync.Mutex bytesDone atomic.Int64 bytesTotal atomic.Int64 } -func (job *Job) SetTotal(total int64) { - job.bytesTotal.Store(total) -} - -func (job *Job) Add(delta int64) { - job.bytesDone.Add(delta) -} - -func (job *Job) SetFileName(name string) { - job.mutex.Lock() - defer job.mutex.Unlock() - job.fileName = name -} - -func (job *Job) resetProgress() { - job.bytesDone.Store(0) - job.bytesTotal.Store(0) -} - func (job *Job) setStatus(status Status, message string) { job.mutex.Lock() defer job.mutex.Unlock() @@ -87,10 +81,10 @@ func (job *Job) currentStatus() Status { return job.status } -func (job *Job) begin(from, to Status) bool { +func (job *Job) begin(to Status, allowedFrom ...Status) bool { job.mutex.Lock() defer job.mutex.Unlock() - if job.deleted || job.status != from { + if job.deleted || !slices.Contains(allowedFrom, job.status) { return false } job.status = to @@ -113,6 +107,10 @@ func (job *Job) markDeleted() bool { func (job *Job) snapshot() Snapshot { job.mutex.Lock() defer job.mutex.Unlock() + moveTargetPath := "" + if job.target != nil { + moveTargetPath = job.target.displayPath() + } return Snapshot{ ID: job.id, SourceURL: job.displayURL, @@ -122,5 +120,6 @@ func (job *Job) snapshot() Snapshot { BytesDone: job.bytesDone.Load(), BytesTotal: job.bytesTotal.Load(), CreatedAt: job.createdAt, + MoveTarget: moveTargetPath, } } diff --git a/internal/jobs/manager.go b/internal/jobs/manager.go index 4a42894..fe46dbd 100755 --- a/internal/jobs/manager.go +++ b/internal/jobs/manager.go @@ -10,6 +10,7 @@ import ( "path" "path/filepath" "sort" + "strings" "sync" "time" @@ -18,6 +19,7 @@ import ( "ddc/internal/fetch" "ddc/internal/files" "ddc/internal/guard" + "ddc/internal/history" "ddc/internal/libraries" "ddc/internal/preferences" ) @@ -31,6 +33,7 @@ var ( type Manager struct { settings config.Settings preferences *preferences.Store + history *history.Store policy *guard.Policy downloader *fetch.Downloader rootContext context.Context @@ -40,12 +43,13 @@ type Manager struct { jobs map[string]*Job } -func NewManager(settings config.Settings, store *preferences.Store) *Manager { +func NewManager(settings config.Settings, store *preferences.Store, historyStore *history.Store) *Manager { policy := guard.NewPolicy(store.Get().AllowedPorts) rootContext, cancelAll := context.WithCancel(context.Background()) return &Manager{ settings: settings, preferences: store, + history: historyStore, policy: policy, downloader: fetch.NewDownloader(policy), rootContext: rootContext, @@ -63,6 +67,11 @@ func (manager *Manager) Preferences() preferences.Preferences { } func (manager *Manager) UpdatePreferences(next preferences.Preferences) error { + downloadPaths, err := manager.normalizeDownloadPaths(next.DownloadPaths) + if err != nil { + return err + } + next.DownloadPaths = downloadPaths if err := manager.preferences.Update(next); err != nil { return err } @@ -82,11 +91,27 @@ func (manager *Manager) AvailableBytes() (int64, error) { return diskspace.Available(manager.settings.DownloadDir) } -func (manager *Manager) Create(rawURL string, autoExtract bool) (Snapshot, error) { - target, err := manager.policy.ValidateURL(rawURL) +type CreateRequest struct { + URL string `json:"url"` + FileName string `json:"fileName"` + AutoExtract bool `json:"autoExtract"` + Destination string `json:"destination"` + Subfolder string `json:"subfolder"` +} + +func (manager *Manager) Create(request CreateRequest) (Snapshot, error) { + target, err := manager.policy.ValidateURL(request.URL) if err != nil { return Snapshot{}, invalid(err) } + requestedName := fetch.SanitizeFileName(request.FileName) + if strings.TrimSpace(request.FileName) != "" && requestedName == "" { + return Snapshot{}, invalid(errors.New("file name is not allowed")) + } + moveTarget, err := manager.resolveTarget(request.Destination, request.Subfolder) + if err != nil { + return Snapshot{}, err + } if err := diskspace.EnsureFree(manager.settings.DownloadDir, 0, manager.reservedBytes()); err != nil { return Snapshot{}, err } @@ -103,13 +128,16 @@ func (manager *Manager) Create(rawURL string, autoExtract bool) (Snapshot, error job := manager.newJob(id, directory, time.Now()) job.sourceURL = target.String() job.displayURL = target.Redacted() - job.autoExtract = autoExtract + job.requestedName = requestedName + job.autoExtract = request.AutoExtract + job.target = moveTarget job.status = StatusQueued - job.fileName = initialFileName(target.Path, target.Host) + job.fileName = initialFileName(requestedName, target.Path, target.Host) job.workers.Add(1) manager.store(job) - go manager.download(job) + job.saveRecord() + manager.launchDownload(job) return job.snapshot(), nil } @@ -117,7 +145,11 @@ func (manager *Manager) List() []Snapshot { manager.mutex.RLock() snapshots := make([]Snapshot, 0, len(manager.jobs)) for _, job := range manager.jobs { - snapshots = append(snapshots, job.snapshot()) + snapshot := job.snapshot() + if snapshot.Status == StatusQueued { + snapshot.QueuePosition = manager.slots.position(job.id) + } + snapshots = append(snapshots, snapshot) } manager.mutex.RUnlock() @@ -157,8 +189,26 @@ func (manager *Manager) Delete(id string) error { return os.RemoveAll(job.directory) } -func (manager *Manager) CancelAll() { +func (manager *Manager) Shutdown(ctx context.Context) { manager.cancelAll() + manager.mutex.RLock() + allJobs := make([]*Job, 0, len(manager.jobs)) + for _, job := range manager.jobs { + allJobs = append(allJobs, job) + } + manager.mutex.RUnlock() + + finished := make(chan struct{}) + go func() { + for _, job := range allJobs { + job.workers.Wait() + } + close(finished) + }() + select { + case <-finished: + case <-ctx.Done(): + } } func (manager *Manager) newJob(id, directory string, createdAt time.Time) *Job { @@ -196,7 +246,10 @@ func newJobID() (string, error) { return hex.EncodeToString(buffer), nil } -func initialFileName(urlPath, host string) string { +func initialFileName(requestedName, urlPath, host string) string { + if requestedName != "" { + return requestedName + } if name := fetch.SanitizeFileName(path.Base(urlPath)); name != "" { return name } diff --git a/internal/jobs/move.go b/internal/jobs/move.go index 4b2557b..e455a33 100755 --- a/internal/jobs/move.go +++ b/internal/jobs/move.go @@ -7,7 +7,6 @@ import ( "path/filepath" "ddc/internal/files" - "ddc/internal/libraries" ) type MoveRequest struct { @@ -29,19 +28,18 @@ func (manager *Manager) Move(id string, request MoveRequest) error { return invalid(errors.New("select at least one file or folder")) } - library, err := libraries.Resolve(manager.settings.LibraryRoots, request.Destination) + target, err := manager.resolveTarget(request.Destination, request.Subfolder) if err != nil { - return invalid(err) + return err } - subfolder, err := files.SanitizeSubfolder(request.Subfolder) - if err != nil { - return invalid(err) + if target == nil { + return invalid(errors.New("choose a destination")) } sources, err := resolveSources(job.directory, request.Paths) if err != nil { return err } - destination, err := files.EnsureSubdirectory(library.Path, subfolder) + destination, err := files.EnsureSubdirectory(target.destination, target.subfolder) if err != nil { return invalid(err) } @@ -49,12 +47,12 @@ func (manager *Manager) Move(id string, request MoveRequest) error { return err } - if !job.begin(StatusReady, StatusMoving) { + if !job.begin(StatusMoving, StatusReady) { return ErrJobBusy } go func() { defer job.workers.Done() - manager.moveFiles(job, library, sources, destination, request.Cleanup) + manager.moveFiles(job, sources, destination, request.Cleanup) }() return nil } @@ -69,6 +67,9 @@ func resolveSources(root string, relativePaths []string) ([]string, error) { return nil, invalid(err) } name := filepath.Base(source) + if files.IsInternal(name) { + return nil, invalid(files.ErrUnsafePath) + } if seenNames[name] { return nil, invalid(fmt.Errorf("more than one selected item is named %q", name)) } @@ -96,7 +97,7 @@ func checkTargets(sources []string, destination string) error { return nil } -func (manager *Manager) moveFiles(job *Job, library libraries.Destination, sources []string, destination string, cleanup bool) { +func (manager *Manager) moveFiles(job *Job, sources []string, destination string, cleanup bool) { job.resetProgress() var totalBytes int64 for _, source := range sources { @@ -106,14 +107,17 @@ func (manager *Manager) moveFiles(job *Job, library libraries.Destination, sourc } job.SetTotal(totalBytes) + movedNames := make([]string, 0, len(sources)) + defer func() { manager.recordHistory(job, destination, movedNames) }() for _, source := range sources { if err := files.MoveInto(source, destination, manager.reservedBytes(), job.Add); err != nil { job.setStatus(StatusReady, "Move failed: "+err.Error()) return } + movedNames = append(movedNames, filepath.Base(source)) } - message := fmt.Sprintf("Moved %d item(s) to %s", len(sources), library.Path) + message := fmt.Sprintf("Moved %d item(s) to %s", len(sources), destination) if !cleanup { job.setStatus(StatusReady, message) return diff --git a/internal/jobs/progress.go b/internal/jobs/progress.go new file mode 100644 index 0000000..4d67a78 --- /dev/null +++ b/internal/jobs/progress.go @@ -0,0 +1,79 @@ +package jobs + +import ( + "context" + "errors" + + "ddc/internal/fetch" +) + +var errPaused = errors.New("download paused") + +func (job *Job) SetTotal(total int64) { + job.bytesTotal.Store(total) +} + +func (job *Job) SetDone(done int64) { + job.bytesDone.Store(done) +} + +func (job *Job) Add(delta int64) { + job.bytesDone.Add(delta) +} + +func (job *Job) SetFileName(name string) { + job.mutex.Lock() + defer job.mutex.Unlock() + job.fileName = name +} + +func (job *Job) SetNotice(text string) { + job.mutex.Lock() + defer job.mutex.Unlock() + if !job.deleted && job.status == StatusDownloading { + job.message = text + } +} + +func (job *Job) resetProgress() { + job.bytesDone.Store(0) + job.bytesTotal.Store(0) +} + +func (job *Job) startRun() (context.Context, context.CancelCauseFunc) { + runContext, cancelRun := context.WithCancelCause(job.jobContext) + job.mutex.Lock() + defer job.mutex.Unlock() + job.cancelRun = cancelRun + return runContext, cancelRun +} + +func (job *Job) pause() bool { + job.mutex.Lock() + defer job.mutex.Unlock() + if job.cancelRun == nil || (job.status != StatusQueued && job.status != StatusDownloading) { + return false + } + job.cancelRun(errPaused) + return true +} + +func (job *Job) currentDownloadState() *fetch.State { + job.mutex.Lock() + defer job.mutex.Unlock() + return job.downloadState +} + +func (job *Job) setDownloadState(state *fetch.State) { + job.mutex.Lock() + defer job.mutex.Unlock() + job.downloadState = state +} + +func (job *Job) markDownloaded(fileName string) { + job.mutex.Lock() + defer job.mutex.Unlock() + job.downloaded = true + job.downloadState = nil + job.fileName = fileName +} diff --git a/internal/jobs/record.go b/internal/jobs/record.go new file mode 100644 index 0000000..b8b49cc --- /dev/null +++ b/internal/jobs/record.go @@ -0,0 +1,106 @@ +package jobs + +import ( + "encoding/json" + "log" + "os" + "path/filepath" + "time" + + "ddc/internal/fetch" + "ddc/internal/files" +) + +const ( + phaseDownloading = "downloading" + phasePaused = "paused" + phaseFailed = "failed" + phaseDone = "done" +) + +type jobRecord struct { + SourceURL string `json:"sourceUrl"` + DisplayURL string `json:"displayUrl"` + RequestedName string `json:"requestedName,omitempty"` + FileName string `json:"fileName"` + AutoExtract bool `json:"autoExtract"` + Destination string `json:"destination,omitempty"` + Subfolder string `json:"subfolder,omitempty"` + Phase string `json:"phase"` + Message string `json:"message,omitempty"` + Download *fetch.State `json:"download,omitempty"` + CreatedAt time.Time `json:"createdAt"` +} + +func (job *Job) record() jobRecord { + job.mutex.Lock() + defer job.mutex.Unlock() + record := jobRecord{ + SourceURL: job.sourceURL, + DisplayURL: job.displayURL, + RequestedName: job.requestedName, + FileName: job.fileName, + AutoExtract: job.autoExtract, + Phase: phaseFor(job.status, job.downloaded), + Message: job.message, + Download: job.downloadState, + CreatedAt: job.createdAt, + } + if job.target != nil { + record.Destination = job.target.destination + record.Subfolder = job.target.subfolder + } + return record +} + +func phaseFor(status Status, downloaded bool) string { + switch { + case downloaded: + return phaseDone + case status == StatusPaused: + return phasePaused + case status == StatusFailed: + return phaseFailed + default: + return phaseDownloading + } +} + +func (job *Job) saveRecord() { + data, err := json.MarshalIndent(job.record(), "", " ") + if err != nil { + log.Printf("could not encode download %s: %v", job.id, err) + return + } + job.recordMutex.Lock() + defer job.recordMutex.Unlock() + if err := files.WriteAtomically(filepath.Join(job.directory, files.JobFileName), data); err != nil && !os.IsNotExist(err) { + log.Printf("could not save download %s: %v", job.id, err) + } +} + +func readRecord(directory string) (jobRecord, bool) { + data, err := os.ReadFile(filepath.Join(directory, files.JobFileName)) + if err != nil { + return jobRecord{}, false + } + var record jobRecord + if err := json.Unmarshal(data, &record); err != nil { + return jobRecord{}, false + } + return record, true +} + +func (job *Job) applyRecord(record jobRecord) { + job.sourceURL = record.SourceURL + job.displayURL = record.DisplayURL + job.requestedName = record.RequestedName + job.fileName = record.FileName + job.autoExtract = record.AutoExtract + if !record.CreatedAt.IsZero() { + job.createdAt = record.CreatedAt + } + if record.Destination != "" { + job.target = &moveTarget{destination: record.Destination, subfolder: record.Subfolder} + } +} diff --git a/internal/jobs/restore.go b/internal/jobs/restore.go index 0f4a3e6..fc0cf9c 100755 --- a/internal/jobs/restore.go +++ b/internal/jobs/restore.go @@ -1,9 +1,11 @@ package jobs import ( + "io/fs" "os" "path/filepath" "regexp" + "sort" "time" "ddc/internal/files" @@ -17,33 +19,72 @@ func (manager *Manager) RestoreExisting() error { return err } + var queuedJobs []*Job for _, entry := range entries { if !entry.IsDir() || !jobIDPattern.MatchString(entry.Name()) { continue } - directory := filepath.Join(manager.settings.DownloadDir, entry.Name()) - os.Remove(filepath.Join(directory, files.PartFileName)) - - createdAt := time.Now() - if info, err := entry.Info(); err == nil { - createdAt = info.ModTime() - } - - job := manager.newJob(entry.Name(), directory, createdAt) - job.status = StatusReady - job.fileName = firstEntryName(directory, entry.Name()) + job := manager.restoreJob(entry) manager.store(job) + if job.status == StatusQueued { + queuedJobs = append(queuedJobs, job) + } + } + + sort.Slice(queuedJobs, func(first, second int) bool { + return queuedJobs[first].createdAt.Before(queuedJobs[second].createdAt) + }) + for _, job := range queuedJobs { + job.workers.Add(1) + manager.launchDownload(job) } return nil } +func (manager *Manager) restoreJob(entry fs.DirEntry) *Job { + directory := filepath.Join(manager.settings.DownloadDir, entry.Name()) + createdAt := time.Now() + if info, err := entry.Info(); err == nil { + createdAt = info.ModTime() + } + job := manager.newJob(entry.Name(), directory, createdAt) + + record, found := readRecord(directory) + if found { + job.applyRecord(record) + } + if !found || record.Phase == phaseDone { + os.Remove(filepath.Join(directory, files.PartFileName)) + job.status = StatusReady + job.downloaded = true + job.fileName = firstEntryName(directory, entry.Name()) + return job + } + + job.downloadState = record.Download + if record.Download != nil { + job.bytesDone.Store(record.Download.DoneBytes()) + job.bytesTotal.Store(record.Download.TotalBytes) + } + switch record.Phase { + case phasePaused: + job.status = StatusPaused + case phaseFailed: + job.status = StatusFailed + job.message = record.Message + default: + job.status = StatusQueued + } + return job +} + func firstEntryName(directory, fallback string) string { entries, err := os.ReadDir(directory) if err != nil { return fallback } for _, entry := range entries { - if entry.Name() != files.PartFileName { + if !files.IsInternal(entry.Name()) { return entry.Name() } } diff --git a/internal/jobs/slots.go b/internal/jobs/slots.go index 70f5b30..cad604c 100644 --- a/internal/jobs/slots.go +++ b/internal/jobs/slots.go @@ -2,38 +2,57 @@ package jobs import ( "context" + "slices" "sync" "time" ) const slotWaitInterval = 500 * time.Millisecond +const ( + queueFirst = "first" + queueUp = "up" + queueDown = "down" +) + type slotLimiter struct { - mutex sync.Mutex - active int + mutex sync.Mutex + active int + waiting []string } -func (limiter *slotLimiter) acquire(ctx context.Context, limit func() int) bool { +func (limiter *slotLimiter) enqueue(id string) { + limiter.mutex.Lock() + defer limiter.mutex.Unlock() + if !slices.Contains(limiter.waiting, id) { + limiter.waiting = append(limiter.waiting, id) + } +} + +func (limiter *slotLimiter) acquire(ctx context.Context, id string, limit func() int) bool { + limiter.enqueue(id) ticker := time.NewTicker(slotWaitInterval) defer ticker.Stop() for { - if limiter.tryAcquire(limit()) { + if limiter.tryAcquire(id, limit()) { return true } select { case <-ctx.Done(): + limiter.remove(id) return false case <-ticker.C: } } } -func (limiter *slotLimiter) tryAcquire(limit int) bool { +func (limiter *slotLimiter) tryAcquire(id string, limit int) bool { limiter.mutex.Lock() defer limiter.mutex.Unlock() - if limiter.active >= limit { + if limiter.active >= limit || len(limiter.waiting) == 0 || limiter.waiting[0] != id { return false } + limiter.waiting = limiter.waiting[1:] limiter.active++ return true } @@ -43,3 +62,39 @@ func (limiter *slotLimiter) release() { defer limiter.mutex.Unlock() limiter.active-- } + +func (limiter *slotLimiter) remove(id string) { + limiter.mutex.Lock() + defer limiter.mutex.Unlock() + limiter.waiting = slices.DeleteFunc(limiter.waiting, func(waitingID string) bool { + return waitingID == id + }) +} + +func (limiter *slotLimiter) position(id string) int { + limiter.mutex.Lock() + defer limiter.mutex.Unlock() + return slices.Index(limiter.waiting, id) + 1 +} + +func (limiter *slotLimiter) reorder(id, direction string) bool { + limiter.mutex.Lock() + defer limiter.mutex.Unlock() + index := slices.Index(limiter.waiting, id) + if index < 0 { + return false + } + + target := index + switch direction { + case queueFirst: + target = 0 + case queueUp: + target = max(index-1, 0) + case queueDown: + target = min(index+1, len(limiter.waiting)-1) + } + limiter.waiting = slices.Delete(limiter.waiting, index, index+1) + limiter.waiting = slices.Insert(limiter.waiting, target, id) + return true +} diff --git a/internal/jobs/target.go b/internal/jobs/target.go new file mode 100644 index 0000000..437a337 --- /dev/null +++ b/internal/jobs/target.go @@ -0,0 +1,109 @@ +package jobs + +import ( + "fmt" + "os" + "path/filepath" + "strings" + + "ddc/internal/files" + "ddc/internal/libraries" + "ddc/internal/preferences" +) + +type moveTarget struct { + destination string + subfolder string +} + +func (target *moveTarget) displayPath() string { + return filepath.Join(target.destination, target.subfolder) +} + +func (manager *Manager) resolveTarget(destination, subfolder string) (*moveTarget, error) { + if strings.TrimSpace(destination) == "" { + return nil, nil + } + libraryPath, extraFolder, err := manager.splitDownloadPath(destination) + if err != nil { + return nil, invalid(err) + } + cleanSubfolder, err := files.SanitizeSubfolder(subfolder) + if err != nil { + return nil, invalid(err) + } + combinedSubfolder, err := files.SanitizeSubfolder(extraFolder + "/" + cleanSubfolder) + if err != nil { + return nil, invalid(err) + } + return &moveTarget{destination: libraryPath, subfolder: combinedSubfolder}, nil +} + +func (manager *Manager) splitDownloadPath(fullPath string) (string, string, error) { + cleaned := filepath.Clean(strings.TrimSpace(fullPath)) + for _, root := range manager.settings.LibraryRoots { + if !files.IsWithin(root, cleaned) { + continue + } + relative, err := filepath.Rel(root, cleaned) + if err != nil { + return "", "", err + } + libraryName, extraFolder, _ := strings.Cut(filepath.ToSlash(relative), "/") + library, err := libraries.Resolve(manager.settings.LibraryRoots, filepath.Join(root, libraryName)) + if err != nil { + return "", "", err + } + return library.Path, extraFolder, nil + } + return "", "", libraries.ErrUnknownDestination +} + +func (manager *Manager) normalizeDownloadPaths(downloadPaths []string) ([]string, error) { + normalized := make([]string, 0, len(downloadPaths)) + seenPaths := make(map[string]bool) + for _, downloadPath := range downloadPaths { + target, err := manager.resolveTarget(downloadPath, "") + if err != nil || target == nil { + return nil, fmt.Errorf("%w: %q is not inside a destination folder", preferences.ErrInvalid, downloadPath) + } + fullPath := target.displayPath() + if !seenPaths[fullPath] { + seenPaths[fullPath] = true + normalized = append(normalized, fullPath) + } + } + return normalized, nil +} + +func (manager *Manager) moveToTarget(job *Job) { + if job.target == nil { + return + } + names, err := topLevelNames(job.directory) + if err == nil { + err = manager.Move(job.id, MoveRequest{ + Paths: names, + Destination: job.target.destination, + Subfolder: job.target.subfolder, + Cleanup: true, + }) + } + if err != nil { + job.setStatus(StatusReady, "Move failed: "+err.Error()) + } +} + +func topLevelNames(directory string) ([]string, error) { + entries, err := os.ReadDir(directory) + if err != nil { + return nil, err + } + names := make([]string, 0, len(entries)) + for _, entry := range entries { + if !files.IsInternal(entry.Name()) { + names = append(names, entry.Name()) + } + } + return names, nil +} diff --git a/internal/libraries/libraries.go b/internal/libraries/libraries.go index 0d90925..4716b3c 100644 --- a/internal/libraries/libraries.go +++ b/internal/libraries/libraries.go @@ -5,14 +5,17 @@ import ( "os" "path/filepath" "strings" + + "ddc/internal/diskspace" ) var ErrUnknownDestination = errors.New("destination is not a first level folder of a library root") type Destination struct { - Name string `json:"name"` - Root string `json:"root"` - Path string `json:"path"` + Name string `json:"name"` + Root string `json:"root"` + Path string `json:"path"` + FreeBytes int64 `json:"freeBytes"` } func Discover(roots []string) []Destination { @@ -26,10 +29,16 @@ func Discover(roots []string) []Destination { if !entry.IsDir() || strings.HasPrefix(entry.Name(), ".") { continue } + path := filepath.Join(root, entry.Name()) + freeBytes, err := diskspace.Available(path) + if err != nil { + freeBytes = -1 + } destinations = append(destinations, Destination{ - Name: entry.Name(), - Root: root, - Path: filepath.Join(root, entry.Name()), + Name: entry.Name(), + Root: root, + Path: path, + FreeBytes: freeBytes, }) } } diff --git a/internal/preferences/preferences.go b/internal/preferences/preferences.go index f64daf3..dffcf00 100644 --- a/internal/preferences/preferences.go +++ b/internal/preferences/preferences.go @@ -3,22 +3,27 @@ package preferences import ( "errors" "fmt" + "path/filepath" ) const ( gigabyte = int64(1) << 30 maxConcurrentLimit = 10 + maxConnectionsLimit = 16 maxReservedGigabytes = 100000 + maxDownloadPaths = 50 ) var ErrInvalid = errors.New("invalid settings") type Preferences struct { MaxConcurrentDownloads int `json:"maxConcurrentDownloads"` + ConnectionsPerDownload int `json:"connectionsPerDownload"` MinFreeGB int `json:"minFreeGB"` AllowedPorts []uint16 `json:"allowedPorts"` AutoExtract bool `json:"autoExtract"` CleanupAfterMove bool `json:"cleanupAfterMove"` + DownloadPaths []string `json:"downloadPaths"` } func (preferences Preferences) MinFreeBytes() int64 { @@ -29,7 +34,10 @@ func (preferences Preferences) Validate() error { if preferences.MaxConcurrentDownloads < 1 || preferences.MaxConcurrentDownloads > maxConcurrentLimit { return fmt.Errorf("%w: parallel downloads must be between 1 and %d", ErrInvalid, maxConcurrentLimit) } - if preferences.MinFreeGB < 0 || preferences.MinFreeGB > maxReservedGigabytes { + if preferences.ConnectionsPerDownload < 1 || preferences.ConnectionsPerDownload > maxConnectionsLimit { + return fmt.Errorf("%w: connections per download must be between 1 and %d", ErrInvalid, maxConnectionsLimit) + } + if preferences.MinFreeGB < 0|| preferences.MinFreeGB > maxReservedGigabytes { return fmt.Errorf("%w: free space to keep must be between 0 and %d GB", ErrInvalid, maxReservedGigabytes) } for _, port := range preferences.AllowedPorts { @@ -37,5 +45,13 @@ func (preferences Preferences) Validate() error { return fmt.Errorf("%w: port 0 is not allowed", ErrInvalid) } } + if len(preferences.DownloadPaths) > maxDownloadPaths { + return fmt.Errorf("%w: at most %d download paths are allowed", ErrInvalid, maxDownloadPaths) + } + for _, downloadPath := range preferences.DownloadPaths { + if !filepath.IsAbs(downloadPath) { + return fmt.Errorf("%w: download path %q must be an absolute path", ErrInvalid, downloadPath) + } + } return nil } diff --git a/internal/preferences/store.go b/internal/preferences/store.go index 66fd572..c10213b 100644 --- a/internal/preferences/store.go +++ b/internal/preferences/store.go @@ -6,9 +6,10 @@ import ( "fmt" "io/fs" "os" - "path/filepath" "slices" "sync" + + "ddc/internal/files" ) type Store struct { @@ -47,6 +48,10 @@ func (store *Store) Get() Preferences { if current.AllowedPorts == nil { current.AllowedPorts = []uint16{} } + current.DownloadPaths = slices.Clone(current.DownloadPaths) + if current.DownloadPaths == nil { + current.DownloadPaths = []string{} + } return current } @@ -61,27 +66,9 @@ func (store *Store) Update(next Preferences) error { store.mutex.Lock() defer store.mutex.Unlock() - if err := writeAtomically(store.path, data); err != nil { + if err := files.WriteAtomically(store.path, data); err != nil { return err } store.current = next return nil } - -func writeAtomically(path string, data []byte) error { - temporary, err := os.CreateTemp(filepath.Dir(path), ".settings-*.tmp") - if err != nil { - return err - } - _, writeErr := temporary.Write(data) - closeErr := temporary.Close() - if writeErr != nil || closeErr != nil { - os.Remove(temporary.Name()) - return errors.Join(writeErr, closeErr) - } - if err := os.Rename(temporary.Name(), path); err != nil { - os.Remove(temporary.Name()) - return err - } - return nil -} diff --git a/internal/server/control_handlers.go b/internal/server/control_handlers.go new file mode 100644 index 0000000..befce4d --- /dev/null +++ b/internal/server/control_handlers.go @@ -0,0 +1,48 @@ +package server + +import "net/http" + +func (server *Server) handlePause(writer http.ResponseWriter, request *http.Request) { + if err := server.manager.Pause(request.PathValue("id")); err != nil { + writeFailure(writer, err) + return + } + writeJSON(writer, http.StatusAccepted, startedResponse{Started: true}) +} + +func (server *Server) handleResume(writer http.ResponseWriter, request *http.Request) { + if err := server.manager.Resume(request.PathValue("id")); err != nil { + writeFailure(writer, err) + return + } + writeJSON(writer, http.StatusAccepted, startedResponse{Started: true}) +} + +type reorderRequest struct { + Direction string `json:"direction"` +} + +func (server *Server) handleReorder(writer http.ResponseWriter, request *http.Request) { + var body reorderRequest + if err := decodeJSON(request, &body); err != nil { + writeError(writer, http.StatusBadRequest, "invalid request body") + return + } + if err := server.manager.Reorder(request.PathValue("id"), body.Direction); err != nil { + writeFailure(writer, err) + return + } + writeJSON(writer, http.StatusOK, startedResponse{Started: true}) +} + +func (server *Server) handleHistory(writer http.ResponseWriter, _ *http.Request) { + writeJSON(writer, http.StatusOK, server.manager.History()) +} + +func (server *Server) handleClearHistory(writer http.ResponseWriter, _ *http.Request) { + if err := server.manager.ClearHistory(); err != nil { + writeFailure(writer, err) + return + } + writer.WriteHeader(http.StatusNoContent) +} diff --git a/internal/server/job_handlers.go b/internal/server/job_handlers.go index 69f4c2b..1c839e2 100755 --- a/internal/server/job_handlers.go +++ b/internal/server/job_handlers.go @@ -6,11 +6,6 @@ import ( "ddc/internal/jobs" ) -type createJobRequest struct { - URL string `json:"url"` - AutoExtract bool `json:"autoExtract"` -} - type extractRequest struct { Path string `json:"path"` } @@ -42,12 +37,12 @@ func (server *Server) handleListJobs(writer http.ResponseWriter, _ *http.Request } func (server *Server) handleCreateJob(writer http.ResponseWriter, request *http.Request) { - var body createJobRequest + var body jobs.CreateRequest if err := decodeJSON(request, &body); err != nil { writeError(writer, http.StatusBadRequest, "invalid request body") return } - snapshot, err := server.manager.Create(body.URL, body.AutoExtract) + snapshot, err := server.manager.Create(body) if err != nil { writeFailure(writer, err) return diff --git a/internal/server/server.go b/internal/server/server.go index 12ffc0e..3c71dd0 100755 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -45,7 +45,12 @@ func (server *Server) routes() http.Handler { api.HandleFunc("GET /api/jobs/{id}/files", server.handleJobFiles) api.HandleFunc("POST /api/jobs/{id}/extract", server.handleExtract) api.HandleFunc("POST /api/jobs/{id}/move", server.handleMove) + api.HandleFunc("POST /api/jobs/{id}/pause", server.handlePause) + api.HandleFunc("POST /api/jobs/{id}/resume", server.handleResume) + api.HandleFunc("POST /api/jobs/{id}/queue", server.handleReorder) api.HandleFunc("DELETE /api/jobs/{id}", server.handleDeleteJob) + api.HandleFunc("GET /api/history", server.handleHistory) + api.HandleFunc("DELETE /api/history", server.handleClearHistory) root := http.NewServeMux() root.HandleFunc("POST /api/login", server.handleLogin) diff --git a/main.go b/main.go index 865cf91..17040f5 100755 --- a/main.go +++ b/main.go @@ -13,12 +13,16 @@ import ( "ddc/internal/auth" "ddc/internal/config" + "ddc/internal/history" "ddc/internal/jobs" "ddc/internal/preferences" "ddc/internal/server" ) -const settingsFileName = ".settings.json" +const ( + settingsFileName = ".settings.json" + historyFileName = ".history.json" +) func main() { settings, err := config.Load() @@ -43,7 +47,12 @@ func main() { log.Fatalf("settings error: %v", err) } - manager := jobs.NewManager(settings, store) + historyStore, err := history.Open(filepath.Join(settings.DownloadDir, historyFileName)) + if err != nil { + log.Fatalf("history error: %v", err) + } + + manager := jobs.NewManager(settings, store, historyStore) if err := manager.RestoreExisting(); err != nil { log.Printf("could not restore existing downloads: %v", err) } @@ -70,10 +79,9 @@ func main() { <-shutdownSignal.Done() log.Printf("shutting down") - manager.CancelAll() - shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() + manager.Shutdown(shutdownContext) if err := httpServer.Shutdown(shutdownContext); err != nil { log.Printf("shutdown error: %v", err) } diff --git a/web/static/icon.svg b/web/static/icon.svg new file mode 100644 index 0000000..a0b3469 --- /dev/null +++ b/web/static/icon.svg @@ -0,0 +1,5 @@ + + + + + diff --git a/web/static/index.html b/web/static/index.html index 125bd55..4f06b6b 100755 --- a/web/static/index.html +++ b/web/static/index.html @@ -5,7 +5,7 @@ DDC - + @@ -46,9 +46,25 @@

New download

- +
+
+ + + +
-

Downloads

+
+

Downloads

+
+ + +
+

No downloads yet. Paste a link above to start.

@@ -86,6 +115,15 @@ + +
+ + History + + +

Moved files will be listed here.

+ +
@@ -104,6 +142,10 @@ Parallel downloads + +
+ Download paths + +
+
Appearance +
diff --git a/web/static/js/api.js b/web/static/js/api.js index c967cc4..f53c072 100755 --- a/web/static/js/api.js +++ b/web/static/js/api.js @@ -53,9 +53,14 @@ export const api = { settings: () => request("GET", "/api/settings"), saveSettings: (settings) => request("POST", "/api/settings", settings), jobs: () => request("GET", "/api/jobs"), - createJob: (url, autoExtract) => request("POST", "/api/jobs", { url, autoExtract }), + createJob: (payload) => request("POST", "/api/jobs", payload), files: (id) => request("GET", jobPath(id, "/files")), extract: (id, path) => request("POST", jobPath(id, "/extract"), { path }), move: (id, payload) => request("POST", jobPath(id, "/move"), payload), + pause: (id) => request("POST", jobPath(id, "/pause"), {}), + resume: (id) => request("POST", jobPath(id, "/resume"), {}), + reorder: (id, direction) => request("POST", jobPath(id, "/queue"), { direction }), remove: (id) => request("DELETE", jobPath(id)), + history: () => request("GET", "/api/history"), + clearHistory: () => request("DELETE", "/api/history"), }; diff --git a/web/static/js/app.js b/web/static/js/app.js index 30080dc..8322579 100755 --- a/web/static/js/app.js +++ b/web/static/js/app.js @@ -1,12 +1,17 @@ import { api, onUnauthorized } from "./api.js"; import { element } from "./dom.js"; +import { destinationSelect, setAutoExtractDefault, setupDownloadForm, updateSubfolderState } from "./download-form.js"; import { formatBytes } from "./format.js"; -import { isActive, renderJobs, updateSpeeds } from "./jobs.js"; +import { recentDestinations, renderHistory } from "./history.js"; +import { isActive, matchesFilter, renderJobs, updateSpeeds } from "./jobs.js"; +import { fillLibrarySelect } from "./libraries.js"; import { setupLogin } from "./login.js"; +import { collectTransitions, notifyTransitions } from "./notify.js"; import { loadSettings, openSettings, setupSettings } from "./settings.js"; import { renderTree, topLevelSelection } from "./tree.js"; const pollIntervalMs = 1500; +const filterStorageKey = "ddc-job-filter"; const elements = { loginView: document.getElementById("login-view"), @@ -16,10 +21,8 @@ const elements = { logoutButton: document.getElementById("logout-button"), settingsButton: document.getElementById("settings-button"), diskStatus: document.getElementById("disk-status"), - downloadForm: document.getElementById("download-form"), - downloadUrl: document.getElementById("download-url"), - autoExtract: document.getElementById("auto-extract"), - downloadError: document.getElementById("download-error"), + jobFilter: document.getElementById("job-filter"), + clearCompletedButton: document.getElementById("clear-completed-button"), jobList: document.getElementById("job-list"), jobEmpty: document.getElementById("job-empty"), detail: document.getElementById("job-detail"), @@ -32,10 +35,19 @@ const elements = { cleanupCheckbox: document.getElementById("cleanup-checkbox"), moveButton: document.getElementById("move-button"), actionError: document.getElementById("action-error"), + history: { + title: document.getElementById("history-title"), + list: document.getElementById("history-list"), + empty: document.getElementById("history-empty"), + clearButton: document.getElementById("clear-history-button"), + }, }; const state = { jobs: [], + destinations: [], + downloadPaths: [], + historyEntries: [], selectedJobId: null, treeStatus: null, selection: new Set(), @@ -54,14 +66,16 @@ async function showApp() { elements.loginView.hidden = true; elements.appView.hidden = false; applyDefaults(await loadSettings()); - await loadLibraries(); + await Promise.all([loadLibraries(), loadHistory()]); await refresh(); startPolling(); } function applyDefaults(settings) { - elements.autoExtract.checked = settings.autoExtract; + setAutoExtractDefault(settings.autoExtract); elements.cleanupCheckbox.checked = settings.cleanupAfterMove; + state.downloadPaths = settings.downloadPaths || []; + renderDestinationSelects(); } async function showSettings() { @@ -83,26 +97,25 @@ function stopPolling() { } async function loadLibraries() { - const destinations = (await api.libraries()) || []; - const previousValue = elements.librarySelect.value; - const groups = new Map(); + state.destinations = (await api.libraries()) || []; + renderDestinationSelects(); +} - for (const destination of destinations) { - if (!groups.has(destination.root)) { - const group = document.createElement("optgroup"); - group.label = destination.root; - groups.set(destination.root, group); - } - const option = element("option", "", destination.name); - option.value = destination.path; - option.title = destination.path; - groups.get(destination.root).append(option); +async function loadHistory() { + try { + state.historyEntries = (await api.history()) || []; + renderHistory(elements.history, state.historyEntries); + renderDestinationSelects(); + } catch (error) { + if (error.status !== 401) console.error(error); } +} - elements.librarySelect.replaceChildren(...groups.values()); - if (destinations.some((destination) => destination.path === previousValue)) { - elements.librarySelect.value = previousValue; - } +function renderDestinationSelects() { + const extras = { recentPaths: recentDestinations(state.historyEntries), savedPaths: state.downloadPaths }; + fillLibrarySelect(elements.librarySelect, state.destinations, extras); + fillLibrarySelect(destinationSelect(), state.destinations, { ...extras, emptyLabel: "Temporary folder (move later)" }); + updateSubfolderState(); } async function refresh() { @@ -112,6 +125,7 @@ async function refresh() { const [jobs, disk] = await Promise.all([api.jobs(), api.disk()]); state.jobs = jobs || []; updateSpeeds(state.jobs); + handleTransitions(collectTransitions(state.jobs)); elements.diskStatus.textContent = `${formatBytes(disk.availableBytes)} free`; renderJobList(); await syncDetail(); @@ -122,14 +136,40 @@ async function refresh() { } } +function handleTransitions(transitions) { + notifyTransitions(transitions); + if (transitions.some((transition) => transition.previous === "moving")) loadHistory(); + if (transitions.length > 0) { + loadLibraries().catch((error) => console.error(error)); + } +} + function renderJobList() { - elements.jobEmpty.hidden = state.jobs.length > 0; - renderJobs(elements.jobList, state.jobs, state.selectedJobId, { + const filter = elements.jobFilter.value; + const visibleJobs = state.jobs.filter((job) => matchesFilter(job, filter)); + elements.jobEmpty.hidden = visibleJobs.length > 0; + elements.jobEmpty.textContent = state.jobs.length === 0 + ? "No downloads yet. Paste a link above to start." + : "No downloads match this filter."; + elements.clearCompletedButton.hidden = !state.jobs.some((job) => job.status === "completed"); + renderJobs(elements.jobList, visibleJobs, state.selectedJobId, { onBrowse: openDetail, + onPause: (job) => runJobAction(() => api.pause(job.id)), + onResume: (job) => runJobAction(() => api.resume(job.id)), + onReorder: (job, direction) => runJobAction(() => api.reorder(job.id, direction)), onDelete: deleteJob, }); } +async function runJobAction(action) { + try { + await action(); + await refresh(); + } catch (error) { + window.alert(error.message); + } +} + function selectedJob() { return state.jobs.find((job) => job.id === state.selectedJobId); } @@ -201,18 +241,6 @@ function updateSelectionSummary() { elements.moveButton.disabled = count === 0 || !job || job.status !== "ready"; } -async function submitDownload(event) { - event.preventDefault(); - elements.downloadError.textContent = ""; - try { - await api.createJob(elements.downloadUrl.value.trim(), elements.autoExtract.checked); - elements.downloadUrl.value = ""; - await refresh(); - } catch (error) { - elements.downloadError.textContent = error.message; - } -} - async function extractArchive(path) { elements.actionError.textContent = ""; try { @@ -257,6 +285,48 @@ async function deleteJob(job) { } } +async function clearCompleted() { + const completedJobs = state.jobs.filter((job) => job.status === "completed"); + for (const job of completedJobs) { + try { + await api.remove(job.id); + if (job.id === state.selectedJobId) closeDetail(); + } catch (error) { + console.error(error); + } + } + await refresh(); +} + +async function clearHistory() { + if (!window.confirm("Clear the move history?")) return; + try { + await api.clearHistory(); + await loadHistory(); + } catch (error) { + window.alert(error.message); + } +} + +function setupJobFilter() { + try { + const saved = localStorage.getItem(filterStorageKey); + if (saved && [...elements.jobFilter.options].some((option) => option.value === saved)) { + elements.jobFilter.value = saved; + } + } catch { + elements.jobFilter.value = "all"; + } + elements.jobFilter.addEventListener("change", () => { + try { + localStorage.setItem(filterStorageKey, elements.jobFilter.value); + } catch { + console.warn("Could not remember the download filter."); + } + renderJobList(); + }); +} + async function logout() { try { await api.logout(); @@ -268,7 +338,10 @@ async function logout() { async function init() { onUnauthorized(showLogin); setupLogin(elements.loginForm, elements.loginError, showApp); - elements.downloadForm.addEventListener("submit", submitDownload); + setupDownloadForm(refresh); + setupJobFilter(); + elements.clearCompletedButton.addEventListener("click", clearCompleted); + elements.history.clearButton.addEventListener("click", clearHistory); elements.moveButton.addEventListener("click", moveSelection); elements.closeDetailButton.addEventListener("click", () => { closeDetail(); diff --git a/web/static/js/download-form.js b/web/static/js/download-form.js new file mode 100644 index 0000000..7d39433 --- /dev/null +++ b/web/static/js/download-form.js @@ -0,0 +1,77 @@ +import { api } from "./api.js"; + +const fields = { + form: document.getElementById("download-form"), + urls: document.getElementById("download-urls"), + name: document.getElementById("download-name"), + autoExtract: document.getElementById("auto-extract"), + destination: document.getElementById("download-destination"), + subfolder: document.getElementById("download-subfolder"), + error: document.getElementById("download-error"), +}; + +export function setupDownloadForm(onCreated) { + fields.form.addEventListener("submit", (event) => submitDownloads(event, onCreated)); + fields.urls.addEventListener("keydown", (event) => { + if (event.key === "Enter" && (event.ctrlKey || event.metaKey)) { + event.preventDefault(); + fields.form.requestSubmit(); + } + }); + fields.destination.addEventListener("change", updateSubfolderState); +} + +export function setAutoExtractDefault(enabled) { + fields.autoExtract.checked = enabled; +} + +export function destinationSelect() { + return fields.destination; +} + +export function updateSubfolderState() { + fields.subfolder.disabled = fields.destination.value === ""; +} + +async function submitDownloads(event, onCreated) { + event.preventDefault(); + fields.error.textContent = ""; + + const urls = fields.urls.value.split("\n").map((line) => line.trim()).filter(Boolean); + const fileName = fields.name.value.trim(); + if (urls.length === 0) { + fields.error.textContent = "Paste at least one link."; + return; + } + if (fileName && urls.length > 1) { + fields.error.textContent = "A file name can only be set when downloading a single link."; + return; + } + + const failedUrls = []; + let firstError = ""; + for (const url of urls) { + try { + await api.createJob({ + url, + fileName, + autoExtract: fields.autoExtract.checked, + destination: fields.destination.value, + subfolder: fields.subfolder.value.trim(), + }); + } catch (error) { + failedUrls.push(url); + if (!firstError) firstError = error.message; + } + } + + fields.urls.value = failedUrls.join("\n"); + if (failedUrls.length === 0) { + fields.name.value = ""; + fields.subfolder.value = ""; + } else { + const prefix = urls.length > 1 ? `${failedUrls.length} of ${urls.length} links failed: ` : ""; + fields.error.textContent = prefix + firstError; + } + await onCreated(); +} diff --git a/web/static/js/format.js b/web/static/js/format.js index 1928efc..9ed57e2 100755 --- a/web/static/js/format.js +++ b/web/static/js/format.js @@ -3,6 +3,7 @@ const units = ["B", "KB", "MB", "GB", "TB"]; export const statusLabels = { queued: "Queued", downloading: "Downloading", + paused: "Paused", extracting: "Extracting", ready: "Ready", moving: "Moving", @@ -21,6 +22,21 @@ export function formatBytes(bytes) { return `${value.toFixed(unitIndex === 0 ? 0 : 1)} ${units[unitIndex]}`; } +export function formatDuration(seconds) { + const total = Math.max(0, Math.round(seconds)); + const hours = Math.floor(total / 3600); + const minutes = Math.floor((total % 3600) / 60); + const remainingSeconds = total % 60; + if (hours > 0) return `${hours}h ${minutes}m`; + if (minutes > 0) return `${minutes}m ${remainingSeconds}s`; + return `${remainingSeconds}s`; +} + +export function formatDate(value) { + const date = new Date(value); + return Number.isNaN(date.getTime()) ? "" : date.toLocaleString(); +} + export function percentOf(done, total) { if (!total || total <= 0) return null; return Math.min(100, (done / total) * 100); diff --git a/web/static/js/history.js b/web/static/js/history.js new file mode 100644 index 0000000..dc7a65e --- /dev/null +++ b/web/static/js/history.js @@ -0,0 +1,33 @@ +import { element } from "./dom.js"; +import { formatDate } from "./format.js"; + +const maxRecentPaths = 5; + +export function recentDestinations(entries) { + const paths = []; + for (const entry of entries) { + if (!paths.includes(entry.destination)) paths.push(entry.destination); + if (paths.length >= maxRecentPaths) break; + } + return paths; +} + +export function renderHistory(elements, entries) { + elements.title.textContent = entries.length > 0 ? `History (${entries.length})` : "History"; + elements.empty.hidden = entries.length > 0; + elements.clearButton.hidden = entries.length === 0; + elements.list.replaceChildren(...entries.map(renderEntry)); +} + +function renderEntry(entry) { + const item = element("li", "history-item"); + const items = entry.items || []; + const itemText = items.length === 1 ? items[0] : `${items.length} items`; + item.append( + element("div", "history-name", entry.fileName || itemText), + element("p", "history-meta muted", `${itemText} moved to ${entry.destination}`), + element("p", "history-meta muted", formatDate(entry.movedAt)), + ); + if (entry.sourceUrl) item.append(element("p", "history-meta muted", entry.sourceUrl)); + return item; +} diff --git a/web/static/js/jobs.js b/web/static/js/jobs.js index bfbf383..6f3f050 100755 --- a/web/static/js/jobs.js +++ b/web/static/js/jobs.js @@ -1,13 +1,21 @@ import { button, element } from "./dom.js"; -import { formatBytes, percentOf, statusLabels } from "./format.js"; +import { formatBytes, formatDuration, percentOf, statusLabels } from "./format.js"; const activeStatuses = new Set(["queued", "downloading", "extracting", "moving"]); +const pausableStatuses = new Set(["queued", "downloading"]); +const speedSmoothing = 0.3; const speedSamples = new Map(); export function isActive(job) { return activeStatuses.has(job.status); } +export function matchesFilter(job, filter) { + if (filter === "all") return true; + if (filter === "active") return isActive(job); + return job.status === filter; +} + export function updateSpeeds(jobs) { const now = performance.now(); const seenIds = new Set(); @@ -17,7 +25,8 @@ export function updateSpeeds(jobs) { const previous = speedSamples.get(job.id); let speed = 0; if (previous && job.bytesDone >= previous.bytes && now > previous.time) { - speed = ((job.bytesDone - previous.bytes) * 1000) / (now - previous.time); + const instantSpeed = ((job.bytesDone - previous.bytes) * 1000) / (now - previous.time); + speed = previous.speed > 0 ? previous.speed + speedSmoothing * (instantSpeed - previous.speed) : instantSpeed; } speedSamples.set(job.id, { bytes: job.bytesDone, time: now, speed }); } @@ -27,8 +36,25 @@ export function updateSpeeds(jobs) { } } +function displayRank(job) { + if (isActive(job) && !job.queuePosition) return 0; + if (job.queuePosition) return 1; + return 2; +} + +function sortForDisplay(jobs) { + return [...jobs].sort((first, second) => { + const rankDifference = displayRank(first) - displayRank(second); + if (rankDifference !== 0) return rankDifference; + if (first.queuePosition && second.queuePosition) return first.queuePosition - second.queuePosition; + return 0; + }); +} + export function renderJobs(container, jobs, selectedId, handlers) { - container.replaceChildren(...jobs.map((job) => renderJob(job, job.id === selectedId, handlers))); + const queueLength = jobs.filter((job) => job.queuePosition).length; + const context = { ...handlers, queueLength }; + container.replaceChildren(...sortForDisplay(jobs).map((job) => renderJob(job, job.id === selectedId, context))); } function renderJob(job, selected, handlers) { @@ -42,6 +68,12 @@ function renderJob(job, selected, handlers) { item.append(header); if (job.sourceUrl) item.append(element("p", "job-source muted", job.sourceUrl)); + if (job.moveTarget && job.status !== "completed") { + item.append(element("p", "job-source muted", `Download to: ${job.moveTarget}`)); + } + if (job.queuePosition) { + item.append(element("p", "job-source muted", `Queue position ${job.queuePosition} of ${handlers.queueLength}`)); + } if (job.status !== "completed") item.append(renderProgress(job)); if (job.message) { const messageClass = job.status === "failed" ? "job-message error" : "job-message muted"; @@ -79,7 +111,12 @@ function progressText(job, percent) { if (percent !== null) parts.push(`${percent.toFixed(1)}%`); const sample = speedSamples.get(job.id); - if (isActive(job) && sample && sample.speed > 0) parts.push(`${formatBytes(sample.speed)}/s`); + if (isActive(job) && sample && sample.speed > 0) { + parts.push(`${formatBytes(sample.speed)}/s`); + if (job.bytesTotal > job.bytesDone) { + parts.push(`${formatDuration((job.bytesTotal - job.bytesDone) / sample.speed)} left`); + } + } return parts.join(" | "); } @@ -88,6 +125,20 @@ function renderActions(job, handlers) { if (job.status === "ready") { actions.append(button("Browse files", "secondary small", () => handlers.onBrowse(job))); } + if (job.queuePosition > 1) { + actions.append(button("Start next", "secondary small", () => handlers.onReorder(job, "first"))); + actions.append(button("Move up", "secondary small", () => handlers.onReorder(job, "up"))); + } + if (job.queuePosition && job.queuePosition < handlers.queueLength) { + actions.append(button("Move down", "secondary small", () => handlers.onReorder(job, "down"))); + } + if (pausableStatuses.has(job.status)) { + actions.append(button("Pause", "secondary small", () => handlers.onPause(job))); + } + if (job.status === "paused" || job.status === "failed") { + const label = job.status === "paused" ? "Resume" : "Retry"; + actions.append(button(label, "secondary small", () => handlers.onResume(job))); + } if (job.status !== "moving") { actions.append(button(isActive(job) ? "Cancel" : "Remove", "danger small", () => handlers.onDelete(job))); } diff --git a/web/static/js/libraries.js b/web/static/js/libraries.js new file mode 100644 index 0000000..a542df1 --- /dev/null +++ b/web/static/js/libraries.js @@ -0,0 +1,61 @@ +import { element } from "./dom.js"; +import { formatBytes } from "./format.js"; + +function withFreeSpace(label, destination) { + if (!destination || destination.freeBytes < 0) return label; + return `${label} - ${formatBytes(destination.freeBytes)} free`; +} + +function destinationForPath(destinations, path) { + return destinations.find((destination) => path === destination.path || path.startsWith(`${destination.path}/`)); +} + +function libraryGroups(destinations) { + const groups = new Map(); + for (const destination of destinations) { + if (!groups.has(destination.root)) { + const group = document.createElement("optgroup"); + group.label = destination.root; + groups.set(destination.root, group); + } + const option = element("option", "", withFreeSpace(destination.name, destination)); + option.value = destination.path; + option.title = destination.path; + groups.get(destination.root).append(option); + } + return [...groups.values()]; +} + +function pathGroup(label, paths, destinations) { + const group = document.createElement("optgroup"); + group.label = label; + for (const path of paths) { + const option = element("option", "", withFreeSpace(path, destinationForPath(destinations, path))); + option.value = path; + option.title = path; + group.append(option); + } + return group; +} + +export function fillLibrarySelect(select, destinations, extras = {}) { + const previousValue = select.value; + const leadingOptions = []; + if (extras.emptyLabel) { + const emptyOption = element("option", "", extras.emptyLabel); + emptyOption.value = ""; + leadingOptions.push(emptyOption); + } + if (extras.recentPaths && extras.recentPaths.length > 0) { + leadingOptions.push(pathGroup("Recent", extras.recentPaths, destinations)); + } + if (extras.savedPaths && extras.savedPaths.length > 0) { + leadingOptions.push(pathGroup("Saved paths", extras.savedPaths, destinations)); + } + + select.replaceChildren(...leadingOptions, ...libraryGroups(destinations)); + const availableValues = [...select.options].map((option) => option.value); + if (availableValues.includes(previousValue)) { + select.value = previousValue; + } +} diff --git a/web/static/js/notify.js b/web/static/js/notify.js new file mode 100644 index 0000000..82647f5 --- /dev/null +++ b/web/static/js/notify.js @@ -0,0 +1,68 @@ +const storageKey = "ddc-notifications"; +const finishedTitles = { + ready: "Download ready", + completed: "Download moved", + failed: "Download failed", +}; +const previousStatuses = new Map(); + +function notificationsSupported() { + return "Notification" in window; +} + +function readPreference() { + try { + return localStorage.getItem(storageKey) === "on"; + } catch { + return false; + } +} + +function storePreference(enabled) { + try { + localStorage.setItem(storageKey, enabled ? "on" : "off"); + } catch { + return; + } +} + +export function notificationsEnabled() { + return notificationsSupported() && Notification.permission === "granted" && readPreference(); +} + +export async function setNotificationsEnabled(enabled) { + if (!enabled) { + storePreference(false); + return; + } + if (!notificationsSupported()) throw new Error("This browser does not support notifications."); + const permission = await Notification.requestPermission(); + if (permission !== "granted") throw new Error("Notifications are blocked for this site in the browser settings."); + storePreference(true); +} + +export function collectTransitions(jobs) { + const transitions = []; + const seenIds = new Set(); + for (const job of jobs) { + seenIds.add(job.id); + const previous = previousStatuses.get(job.id); + if (previous && previous !== job.status) transitions.push({ job, previous }); + previousStatuses.set(job.id, job.status); + } + for (const id of previousStatuses.keys()) { + if (!seenIds.has(id)) previousStatuses.delete(id); + } + return transitions; +} + +export function notifyTransitions(transitions) { + if (!notificationsEnabled()) return; + for (const { job, previous } of transitions) { + const title = finishedTitles[job.status]; + const wasWorking = previous === "queued" || previous === "downloading" || previous === "extracting" || previous === "moving"; + if (!title || !wasWorking) continue; + if (job.status === "ready" && previous === "moving") continue; + new Notification(title, { body: job.fileName || job.sourceUrl, tag: `ddc-${job.id}` }); + } +} diff --git a/web/static/js/settings.js b/web/static/js/settings.js index 7b16c99..0195f8f 100644 --- a/web/static/js/settings.js +++ b/web/static/js/settings.js @@ -1,5 +1,6 @@ import { api } from "./api.js"; import { element } from "./dom.js"; +import { notificationsEnabled, setNotificationsEnabled } from "./notify.js"; const themeStorageKey = "ddc-theme"; @@ -10,10 +11,13 @@ const fields = { cancelButton: document.getElementById("settings-cancel"), error: document.getElementById("settings-error"), concurrent: document.getElementById("setting-concurrent"), + connections: document.getElementById("setting-connections"), + notifications: document.getElementById("setting-notifications"), minFree: document.getElementById("setting-min-free"), ports: document.getElementById("setting-ports"), autoExtract: document.getElementById("setting-auto-extract"), cleanup: document.getElementById("setting-cleanup"), + downloadPaths: document.getElementById("setting-download-paths"), theme: document.getElementById("setting-theme"), downloadDir: document.getElementById("info-download-dir"), roots: document.getElementById("info-roots"), @@ -42,10 +46,13 @@ export async function openSettings() { function fillForm(data) { const settings = data.settings; fields.concurrent.value = settings.maxConcurrentDownloads; + fields.connections.value = settings.connectionsPerDownload; + fields.notifications.checked = notificationsEnabled(); fields.minFree.value = settings.minFreeGB; fields.ports.value = settings.allowedPorts.length > 0 ? settings.allowedPorts.join(", ") : "*"; fields.autoExtract.checked = settings.autoExtract; fields.cleanup.checked = settings.cleanupAfterMove; + fields.downloadPaths.value = (settings.downloadPaths || []).join("\n"); fields.theme.value = readTheme(); fields.downloadDir.textContent = data.downloadDir; fields.roots.replaceChildren(...data.libraryRoots.map((root) => element("div", "mono", root))); @@ -57,11 +64,14 @@ async function saveSettings(event, onSaved) { try { const payload = { maxConcurrentDownloads: Number(fields.concurrent.value), + connectionsPerDownload: Number(fields.connections.value), minFreeGB: Number(fields.minFree.value), allowedPorts: parsePorts(fields.ports.value), autoExtract: fields.autoExtract.checked, cleanupAfterMove: fields.cleanup.checked, + downloadPaths: parseDownloadPaths(fields.downloadPaths.value), }; + await setNotificationsEnabled(fields.notifications.checked); const data = await api.saveSettings(payload); storeTheme(fields.theme.value); fields.dialog.close(); @@ -80,6 +90,10 @@ function parsePorts(text) { return ports; } +function parseDownloadPaths(text) { + return text.split("\n").map((line) => line.trim()).filter(Boolean); +} + function readTheme() { try { return localStorage.getItem(themeStorageKey) || "system"; diff --git a/web/static/style.css b/web/static/style.css index 4bfdf46..91ba3f4 100755 --- a/web/static/style.css +++ b/web/static/style.css @@ -97,13 +97,15 @@ h2 { input, select, +textarea, button { font: inherit; color: inherit; } input:not([type="checkbox"]), -select { +select, +textarea { width: 100%; min-height: 42px; padding: 9px 12px; @@ -113,8 +115,13 @@ select { transition: border-color 0.15s ease, box-shadow 0.15s ease; } +textarea { + resize: vertical; +} + input:not([type="checkbox"]):focus, -select:focus { +select:focus, +textarea:focus { outline: none; border-color: var(--accent); box-shadow: 0 0 0 3px var(--accent-soft); @@ -724,3 +731,102 @@ button.block { display: none; } } + +.badge-paused { + color: var(--muted); + background: var(--border); +} + +.download-form { + align-items: flex-start; +} + +.download-form textarea { + flex: 1 1 auto; + min-height: 64px; +} + +.download-options { + display: grid; + grid-template-columns: repeat(3, 1fr); + gap: 12px; + margin-top: 14px; +} + +.list-toolbar { + display: flex; + justify-content: space-between; + align-items: center; + flex-wrap: wrap; + gap: 10px; + margin-bottom: 14px; +} + +.list-toolbar h2 { + margin: 0; +} + +.list-toolbar-actions { + display: flex; + align-items: center; + gap: 8px; +} + +.list-toolbar-actions select { + width: auto; + min-height: 36px; + padding: 6px 10px; +} + +.history-card summary { + cursor: pointer; + font-size: 1.05rem; + font-weight: 650; +} + +.history-list { + list-style: none; + padding: 0; + margin: 14px 0 12px; + display: flex; + flex-direction: column; + gap: 8px; + max-height: 420px; + overflow-y: auto; +} + +.history-list:empty { + display: none; +} + +.history-item { + border: 1px solid var(--border); + border-radius: var(--radius-small); + padding: 10px 12px; + background: var(--surface-alt); +} + +.history-name { + font-weight: 600; + overflow-wrap: anywhere; +} + +.history-meta { + font-size: 0.8rem; + margin: 4px 0 0; + overflow-wrap: anywhere; +} + +.history-card .empty-state { + margin: 14px 0 12px; +} + +@media (max-width: 640px) { + .download-form textarea { + width: 100%; + } + + .download-options { + grid-template-columns: 1fr; + } +}