feat: add download tracking endpoints (sections 4.1, 4.2, 4.3, 4.4)

internal/api/handlers.go

@@ -577,6 +577,224 @@ func parseUUID(s string) (uuid.UUID, error) {
 	return uuid.Parse(s)
 }
 
+func (h *Handlers) ListQueue(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	var status *string
+	if s := r.URL.Query().Get("status"); s != "" {
+		status = &s
+	}
+
+	items, err := h.DB.ListDownloadQueue(r.Context(), status)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]any{
+		"items": items,
+		"total": len(items),
+	})
+}
+
+func (h *Handlers) GetQueueItem(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	idStr := chi.URLParam(r, "id")
+	id, err := parseUUID(idStr)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "invalid ID")
+		return
+	}
+
+	item, err := h.DB.GetDownloadQueueItem(r.Context(), id)
+	if err != nil {
+		writeError(w, http.StatusNotFound, "queue item not found")
+		return
+	}
+
+	writeJSON(w, http.StatusOK, item)
+}
+
+func (h *Handlers) AddToQueue(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	var req struct {
+		Title       string  `json:"title"`
+		TorrentHash *string `json:"torrent_hash"`
+		Size        int64   `json:"size"`
+		Indexer     *string `json:"indexer"`
+		AlbumID     *string `json:"album_id"`
+		ArtistID    *string `json:"artist_id"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body")
+		return
+	}
+
+	var albumID, artistID *uuid.UUID
+	if req.AlbumID != nil {
+		if id, err := parseUUID(*req.AlbumID); err == nil {
+			albumID = &id
+		}
+	}
+	if req.ArtistID != nil {
+		if id, err := parseUUID(*req.ArtistID); err == nil {
+			artistID = &id
+		}
+	}
+
+	id, err := h.DB.AddToDownloadQueue(r.Context(), req.Title, req.Size, req.TorrentHash, req.Indexer, albumID, artistID)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	item, _ := h.DB.GetDownloadQueueItem(r.Context(), id)
+	writeJSON(w, http.StatusOK, item)
+}
+
+func (h *Handlers) UpdateQueueItem(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	idStr := chi.URLParam(r, "id")
+	id, err := parseUUID(idStr)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "invalid ID")
+		return
+	}
+
+	var req struct {
+		Status       *string `json:"status"`
+		ErrorMessage *string `json:"error_message"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid request body")
+		return
+	}
+
+	if req.Status != nil {
+		if *req.Status == "failed" && req.ErrorMessage != nil {
+			services.HandleFailedDownload(r.Context(), h.DB, id, *req.ErrorMessage)
+		} else {
+			if err := h.DB.UpdateDownloadQueueStatus(r.Context(), id, *req.Status, req.ErrorMessage); err != nil {
+				writeError(w, http.StatusInternalServerError, err.Error())
+				return
+			}
+
+			if *req.Status == "completed" {
+				item, _ := h.DB.GetDownloadQueueItem(r.Context(), id)
+				if item != nil && item.AlbumID != nil {
+					h.DB.RemoveFromWantedAlbums(r.Context(), *item.AlbumID)
+				}
+			}
+		}
+	}
+
+	item, err := h.DB.GetDownloadQueueItem(r.Context(), id)
+	if err != nil {
+		writeError(w, http.StatusNotFound, "queue item not found")
+		return
+	}
+
+	writeJSON(w, http.StatusOK, item)
+}
+
+func (h *Handlers) DeleteQueueItem(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	idStr := chi.URLParam(r, "id")
+	id, err := parseUUID(idStr)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "invalid ID")
+		return
+	}
+
+	item, err := h.DB.GetDownloadQueueItem(r.Context(), id)
+	if err != nil {
+		writeError(w, http.StatusNotFound, "queue item not found")
+		return
+	}
+
+	if item.TorrentHash != nil && h.TorrentService.IsConfigured() {
+		h.TorrentService.RemoveTorrent(r.Context(), *item.TorrentHash, false)
+	}
+
+	if err := h.DB.DeleteDownloadQueueItem(r.Context(), id); err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]any{"deleted": true})
+}
+
+func (h *Handlers) SyncQueue(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	result, err := services.SyncDownloadQueue(r.Context(), h.DB, h.TorrentService)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, result)
+}
+
+func (h *Handlers) BlocklistQueueItem(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	idStr := chi.URLParam(r, "id")
+	id, err := parseUUID(idStr)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "invalid ID")
+		return
+	}
+
+	result, err := services.BlocklistAndRemove(r.Context(), h.DB, h.TorrentService, id)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, result)
+}
+
+func (h *Handlers) QueueStats(w http.ResponseWriter, r *http.Request) {
+	if h.DB == nil {
+		writeError(w, http.StatusServiceUnavailable, "database not connected")
+		return
+	}
+
+	stats, err := h.DB.GetDownloadQueueStats(r.Context())
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, stats)
+}
+
 func writeJSON(w http.ResponseWriter, status int, v any) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(status)

internal/api/router.go

@@ -60,6 +60,17 @@ func NewRouter(h *Handlers) *chi.Mux {
 
 		r.Post("/blocklist", h.AddToBlocklist)
 
+		r.Route("/queue", func(r chi.Router) {
+			r.Get("/", h.ListQueue)
+			r.Post("/", h.AddToQueue)
+			r.Post("/sync", h.SyncQueue)
+			r.Get("/stats", h.QueueStats)
+			r.Get("/{id}", h.GetQueueItem)
+			r.Put("/{id}", h.UpdateQueueItem)
+			r.Delete("/{id}", h.DeleteQueueItem)
+			r.Post("/{id}/blocklist", h.BlocklistQueueItem)
+		})
+
 		r.Route("/library", func(r chi.Router) {
 			r.Get("/artists", h.ListLibraryArtists)
 			r.Get("/albums", h.ListLibraryAlbums)

internal/database/db.go

@@ -607,3 +607,155 @@ func (db *DB) GetArtistIDByAlbum(ctx context.Context, albumID uuid.UUID) (*uuid.
 	}
 	return &artistID, nil
 }
+
+type DownloadQueueRow struct {
+	ID             uuid.UUID  `json:"id"`
+	ArtistID       *uuid.UUID `json:"artist_id"`
+	AlbumID        *uuid.UUID `json:"album_id"`
+	DownloadID     *string    `json:"download_id"`
+	Title          string     `json:"title"`
+	Size           int64      `json:"size"`
+	SizeLeft       int64      `json:"size_left"`
+	Status         string     `json:"status"`
+	Progress       float32    `json:"progress"`
+	ErrorMessage   *string    `json:"error_message"`
+	Protocol       string     `json:"protocol"`
+	Indexer        *string    `json:"indexer"`
+	DownloadClient *string    `json:"download_client"`
+	TorrentHash    *string    `json:"torrent_hash"`
+	OutputPath     *string    `json:"output_path"`
+	AddedAt        time.Time  `json:"added_at"`
+	CompletedAt    *time.Time `json:"completed_at"`
+}
+
+func (db *DB) AddToDownloadQueue(ctx context.Context, title string, size int64, torrentHash, indexer *string, albumID, artistID *uuid.UUID) (uuid.UUID, error) {
+	var id uuid.UUID
+	err := db.pool.QueryRow(ctx, `
+		INSERT INTO download_queue (title, size, torrent_hash, indexer, album_id, artist_id, status)
+		VALUES ($1, $2, $3, $4, $5, $6, 'queued')
+		RETURNING id
+	`, title, size, torrentHash, indexer, albumID, artistID).Scan(&id)
+	return id, err
+}
+
+func (db *DB) GetDownloadQueueItem(ctx context.Context, id uuid.UUID) (*DownloadQueueRow, error) {
+	var row DownloadQueueRow
+	err := db.pool.QueryRow(ctx, `
+		SELECT id, artist_id, album_id, download_id, title, size, size_left, status, progress,
+		       error_message, protocol, indexer, download_client, torrent_hash, output_path, added_at, completed_at
+		FROM download_queue WHERE id = $1
+	`, id).Scan(&row.ID, &row.ArtistID, &row.AlbumID, &row.DownloadID, &row.Title, &row.Size,
+		&row.SizeLeft, &row.Status, &row.Progress, &row.ErrorMessage, &row.Protocol, &row.Indexer,
+		&row.DownloadClient, &row.TorrentHash, &row.OutputPath, &row.AddedAt, &row.CompletedAt)
+	if err != nil {
+		return nil, err
+	}
+	return &row, nil
+}
+
+func (db *DB) ListDownloadQueue(ctx context.Context, status *string) ([]DownloadQueueRow, error) {
+	var rows []DownloadQueueRow
+	var query string
+	var args []any
+
+	if status != nil {
+		query = `
+			SELECT id, artist_id, album_id, download_id, title, size, size_left, status, progress,
+			       error_message, protocol, indexer, download_client, torrent_hash, output_path, added_at, completed_at
+			FROM download_queue WHERE status = $1 ORDER BY added_at DESC
+		`
+		args = []any{*status}
+	} else {
+		query = `
+			SELECT id, artist_id, album_id, download_id, title, size, size_left, status, progress,
+			       error_message, protocol, indexer, download_client, torrent_hash, output_path, added_at, completed_at
+			FROM download_queue ORDER BY added_at DESC
+		`
+	}
+
+	dbRows, err := db.pool.Query(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer dbRows.Close()
+
+	for dbRows.Next() {
+		var row DownloadQueueRow
+		err := dbRows.Scan(&row.ID, &row.ArtistID, &row.AlbumID, &row.DownloadID, &row.Title, &row.Size,
+			&row.SizeLeft, &row.Status, &row.Progress, &row.ErrorMessage, &row.Protocol, &row.Indexer,
+			&row.DownloadClient, &row.TorrentHash, &row.OutputPath, &row.AddedAt, &row.CompletedAt)
+		if err != nil {
+			return nil, err
+		}
+		rows = append(rows, row)
+	}
+	return rows, nil
+}
+
+func (db *DB) UpdateDownloadQueueStatus(ctx context.Context, id uuid.UUID, status string, errorMessage *string) error {
+	if status == "completed" {
+		_, err := db.pool.Exec(ctx, `
+			UPDATE download_queue SET status = $1, completed_at = NOW() WHERE id = $2
+		`, status, id)
+		return err
+	}
+	if errorMessage != nil {
+		_, err := db.pool.Exec(ctx, `
+			UPDATE download_queue SET status = $1, error_message = $2 WHERE id = $3
+		`, status, *errorMessage, id)
+		return err
+	}
+	_, err := db.pool.Exec(ctx, `
+		UPDATE download_queue SET status = $1 WHERE id = $2
+	`, status, id)
+	return err
+}
+
+func (db *DB) UpdateDownloadQueueProgress(ctx context.Context, id uuid.UUID, progress float32, sizeLeft int64, status string) error {
+	_, err := db.pool.Exec(ctx, `
+		UPDATE download_queue SET progress = $1, size_left = $2, status = $3 WHERE id = $4
+	`, progress, sizeLeft, status, id)
+	return err
+}
+
+func (db *DB) DeleteDownloadQueueItem(ctx context.Context, id uuid.UUID) error {
+	_, err := db.pool.Exec(ctx, `DELETE FROM download_queue WHERE id = $1`, id)
+	return err
+}
+
+func (db *DB) GetDownloadQueueByTorrentHash(ctx context.Context, hash string) (*DownloadQueueRow, error) {
+	var row DownloadQueueRow
+	err := db.pool.QueryRow(ctx, `
+		SELECT id, artist_id, album_id, download_id, title, size, size_left, status, progress,
+		       error_message, protocol, indexer, download_client, torrent_hash, output_path, added_at, completed_at
+		FROM download_queue WHERE torrent_hash = $1
+	`, hash).Scan(&row.ID, &row.ArtistID, &row.AlbumID, &row.DownloadID, &row.Title, &row.Size,
+		&row.SizeLeft, &row.Status, &row.Progress, &row.ErrorMessage, &row.Protocol, &row.Indexer,
+		&row.DownloadClient, &row.TorrentHash, &row.OutputPath, &row.AddedAt, &row.CompletedAt)
+	if err != nil {
+		return nil, err
+	}
+	return &row, nil
+}
+
+type DownloadQueueStats struct {
+	Total       int64 `json:"total"`
+	Downloading int64 `json:"downloading"`
+	Queued      int64 `json:"queued"`
+	Completed   int64 `json:"completed"`
+	Failed      int64 `json:"failed"`
+}
+
+func (db *DB) GetDownloadQueueStats(ctx context.Context) (*DownloadQueueStats, error) {
+	var stats DownloadQueueStats
+	err := db.pool.QueryRow(ctx, `
+		SELECT 
+			COUNT(*) as total,
+			COUNT(*) FILTER (WHERE status = 'downloading') as downloading,
+			COUNT(*) FILTER (WHERE status = 'queued') as queued,
+			COUNT(*) FILTER (WHERE status = 'completed') as completed,
+			COUNT(*) FILTER (WHERE status = 'failed') as failed
+		FROM download_queue
+	`).Scan(&stats.Total, &stats.Downloading, &stats.Queued, &stats.Completed, &stats.Failed)
+	return &stats, err
+}

internal/services/queue.go

@@ -0,0 +1,147 @@
+package services
+
+import (
+	"context"
+
+	"github.com/fujin/music-agregator/internal/database"
+	"github.com/fujin/music-agregator/internal/torrent"
+	"github.com/google/uuid"
+)
+
+type QueueSyncResult struct {
+	Synced  int `json:"synced"`
+	Updated int `json:"updated"`
+}
+
+func SyncDownloadQueue(ctx context.Context, db *database.DB, torrentService *TorrentService) (*QueueSyncResult, error) {
+	if !torrentService.IsConfigured() {
+		return &QueueSyncResult{}, nil
+	}
+
+	torrents, err := torrentService.ListTorrents(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	torrentMap := make(map[string]torrent.TorrentInfo)
+	for _, t := range torrents {
+		torrentMap[t.Hash] = t
+	}
+
+	queueItems, err := db.ListDownloadQueue(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var synced, updated int
+	for _, item := range queueItems {
+		if item.TorrentHash == nil {
+			continue
+		}
+
+		synced++
+		t, exists := torrentMap[*item.TorrentHash]
+		if !exists {
+			continue
+		}
+
+		newStatus := mapTorrentState(t.State)
+		sizeLeft := int64(float64(item.Size) * (1 - t.Progress))
+
+		if newStatus != item.Status || item.Progress != float32(t.Progress) {
+			if err := db.UpdateDownloadQueueProgress(ctx, item.ID, float32(t.Progress), sizeLeft, newStatus); err != nil {
+				continue
+			}
+			updated++
+
+			if newStatus == "completed" && item.AlbumID != nil {
+				db.RemoveFromWantedAlbums(ctx, *item.AlbumID)
+			}
+		}
+	}
+
+	return &QueueSyncResult{Synced: synced, Updated: updated}, nil
+}
+
+func mapTorrentState(state torrent.TorrentState) string {
+	switch state {
+	case torrent.StateDownloading:
+		return "downloading"
+	case torrent.StateSeeding:
+		return "completed"
+	case torrent.StatePaused:
+		return "paused"
+	case torrent.StateQueued:
+		return "queued"
+	case torrent.StateChecking:
+		return "checking"
+	case torrent.StateError:
+		return "failed"
+	default:
+		return "queued"
+	}
+}
+
+func HandleFailedDownload(ctx context.Context, db *database.DB, queueID uuid.UUID, errorMessage string) error {
+	item, err := db.GetDownloadQueueItem(ctx, queueID)
+	if err != nil {
+		return err
+	}
+
+	if err := db.UpdateDownloadQueueStatus(ctx, queueID, "failed", &errorMessage); err != nil {
+		return err
+	}
+
+	if item.ArtistID != nil && item.AlbumID != nil {
+		if err := db.AddToBlocklist(ctx, *item.ArtistID, *item.AlbumID, item.Title, item.TorrentHash, item.Indexer); err != nil {
+			return err
+		}
+	}
+
+	if item.AlbumID != nil {
+		if err := db.AddToWantedAlbums(ctx, *item.AlbumID); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+type BlocklistResult struct {
+	Blocklisted bool `json:"blocklisted"`
+	Removed     bool `json:"removed"`
+}
+
+func BlocklistAndRemove(ctx context.Context, db *database.DB, torrentService *TorrentService, queueID uuid.UUID) (*BlocklistResult, error) {
+	item, err := db.GetDownloadQueueItem(ctx, queueID)
+	if err != nil {
+		return nil, err
+	}
+
+	result := &BlocklistResult{}
+
+	if item.ArtistID != nil {
+		albumID := item.AlbumID
+		if albumID == nil {
+			albumID = &uuid.Nil
+		}
+		if err := db.AddToBlocklist(ctx, *item.ArtistID, *albumID, item.Title, item.TorrentHash, item.Indexer); err == nil {
+			result.Blocklisted = true
+		}
+	}
+
+	if item.TorrentHash != nil && torrentService.IsConfigured() {
+		torrentService.RemoveTorrent(ctx, *item.TorrentHash, true)
+	}
+
+	if err := db.DeleteDownloadQueueItem(ctx, queueID); err != nil {
+		return nil, err
+	}
+	result.Removed = true
+
+	if item.AlbumID != nil {
+		db.AddToWantedAlbums(ctx, *item.AlbumID)
+	}
+
+	return result, nil
+}

testing/e2e/download_test.go

@@ -0,0 +1,458 @@
+// Package e2e contains end-to-end tests for the music aggregator.
+//
+// This file covers Section 4 of FLOWS.md: Download Tracking
+//   - 4.1 Track Active Downloads
+//   - 4.2 Completed Download Handling
+//   - 4.3 Failed Download Handling
+//   - 4.4 Download Queue Management
+package e2e
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/fujin/music-agregator/testing/e2e/testutil"
+)
+
+// TestTrackActiveDownloads_Flow covers section 4.1 of FLOWS.md:
+//  1. Poll torrent client for status of all active downloads
+//  2. Match against download_queue entries by torrent_hash
+//  3. Update: progress, size_left, status
+//  4. Detect state transitions: queued → downloading → seeding → completed
+func TestTrackActiveDownloads_Flow(t *testing.T) {
+	env := testutil.NewTestEnv(t)
+	defer env.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	if err := env.CleanupDownloadQueue(ctx); err != nil {
+		t.Fatalf("cleanup failed: %v", err)
+	}
+
+	t.Cleanup(func() {
+		env.CleanupDownloadQueue(context.Background())
+	})
+
+	t.Run("Step1_ListDownloadQueue", func(t *testing.T) {
+		resp, err := env.GET("/api/queue")
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			Items []struct {
+				ID          string  `json:"id"`
+				Title       string  `json:"title"`
+				Status      string  `json:"status"`
+				Progress    float64 `json:"progress"`
+				TorrentHash *string `json:"torrent_hash"`
+			} `json:"items"`
+			Total int `json:"total"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		t.Logf("queue has %d items", result.Total)
+	})
+
+	t.Run("Step2_AddToQueue", func(t *testing.T) {
+		resp, err := env.POST("/api/queue", map[string]any{
+			"title":        "Test Album - FLAC",
+			"torrent_hash": "abc123def456",
+			"size":         500000000,
+			"indexer":      "test-indexer",
+		})
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			ID          string `json:"id"`
+			Title       string `json:"title"`
+			Status      string `json:"status"`
+			TorrentHash string `json:"torrent_hash"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		if result.ID == "" {
+			t.Error("expected non-empty ID")
+		}
+		if result.Status != "queued" {
+			t.Errorf("expected status=queued, got %s", result.Status)
+		}
+	})
+
+	t.Run("Step3_SyncQueueWithTorrentClient", func(t *testing.T) {
+		resp, err := env.POST("/api/queue/sync", nil)
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			Synced  int `json:"synced"`
+			Updated int `json:"updated"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		t.Logf("synced %d items, updated %d", result.Synced, result.Updated)
+	})
+
+	t.Run("Step4_GetQueueItem", func(t *testing.T) {
+		listResp, err := env.GET("/api/queue")
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+
+		var listResult struct {
+			Items []struct {
+				ID string `json:"id"`
+			} `json:"items"`
+		}
+		if err := listResp.DecodeJSON(&listResult); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		if len(listResult.Items) == 0 {
+			t.Skip("no items in queue")
+		}
+
+		itemID := listResult.Items[0].ID
+		resp, err := env.GET("/api/queue/" + itemID)
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var item struct {
+			ID       string  `json:"id"`
+			Title    string  `json:"title"`
+			Status   string  `json:"status"`
+			Progress float64 `json:"progress"`
+			Size     int64   `json:"size"`
+			SizeLeft int64   `json:"size_left"`
+		}
+		if err := resp.DecodeJSON(&item); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		if item.ID != itemID {
+			t.Errorf("expected id=%s, got %s", itemID, item.ID)
+		}
+	})
+}
+
+// TestCompletedDownloadHandling_Flow covers section 4.2 of FLOWS.md:
+//  1. Detect download_queue entry where torrent reports completed/seeding
+//  2. Mark download_queue.status = completed, set completed_at
+//  3. Remove from wanted_albums
+func TestCompletedDownloadHandling_Flow(t *testing.T) {
+	env := testutil.NewTestEnv(t)
+	defer env.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	if err := env.CleanupDownloadQueue(ctx); err != nil {
+		t.Fatalf("cleanup failed: %v", err)
+	}
+	if err := env.CleanupWantedAlbums(ctx); err != nil {
+		t.Fatalf("cleanup wanted_albums failed: %v", err)
+	}
+
+	t.Cleanup(func() {
+		env.CleanupDownloadQueue(context.Background())
+		env.CleanupWantedAlbums(context.Background())
+	})
+
+	t.Run("Step1_MarkDownloadCompleted", func(t *testing.T) {
+		addResp, err := env.POST("/api/queue", map[string]any{
+			"title":        "Completed Album - FLAC",
+			"torrent_hash": "completed123",
+			"size":         100000000,
+		})
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		addResp.AssertStatus(t, 200)
+
+		var addResult struct {
+			ID string `json:"id"`
+		}
+		if err := addResp.DecodeJSON(&addResult); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		resp, err := env.PUT("/api/queue/"+addResult.ID, map[string]any{
+			"status": "completed",
+		})
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			Status      string  `json:"status"`
+			CompletedAt *string `json:"completed_at"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		if result.Status != "completed" {
+			t.Errorf("expected status=completed, got %s", result.Status)
+		}
+		if result.CompletedAt == nil {
+			t.Error("expected completed_at to be set")
+		}
+	})
+}
+
+// TestFailedDownloadHandling_Flow covers section 4.3 of FLOWS.md:
+//  1. Detect download failure (torrent client reports error)
+//  2. Mark download_queue.status = failed, set error_message
+//  3. Add release to blocklist (source_title, torrent_hash, indexer, quality)
+//  4. Re-add album to wanted_albums for retry search
+func TestFailedDownloadHandling_Flow(t *testing.T) {
+	env := testutil.NewTestEnv(t)
+	defer env.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	if err := env.CleanupDownloadQueue(ctx); err != nil {
+		t.Fatalf("cleanup failed: %v", err)
+	}
+	if err := env.CleanupBlocklist(ctx); err != nil {
+		t.Fatalf("cleanup blocklist failed: %v", err)
+	}
+
+	t.Cleanup(func() {
+		env.CleanupDownloadQueue(context.Background())
+		env.CleanupBlocklist(context.Background())
+	})
+
+	t.Run("Step1_MarkDownloadFailed", func(t *testing.T) {
+		addResp, err := env.POST("/api/queue", map[string]any{
+			"title":        "Failed Album - FLAC",
+			"torrent_hash": "failed456",
+			"size":         100000000,
+			"indexer":      "test-indexer",
+		})
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		addResp.AssertStatus(t, 200)
+
+		var addResult struct {
+			ID string `json:"id"`
+		}
+		if err := addResp.DecodeJSON(&addResult); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		resp, err := env.PUT("/api/queue/"+addResult.ID, map[string]any{
+			"status":        "failed",
+			"error_message": "Tracker returned error: torrent not found",
+		})
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			Status       string `json:"status"`
+			ErrorMessage string `json:"error_message"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		if result.Status != "failed" {
+			t.Errorf("expected status=failed, got %s", result.Status)
+		}
+		if result.ErrorMessage == "" {
+			t.Error("expected error_message to be set")
+		}
+	})
+
+	t.Run("Step2_BlocklistAndRemove", func(t *testing.T) {
+		addResp, err := env.POST("/api/queue", map[string]any{
+			"title":        "To Blocklist Album",
+			"torrent_hash": "blocklist789",
+			"size":         100000000,
+			"indexer":      "test-indexer",
+		})
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		addResp.AssertStatus(t, 200)
+
+		var addResult struct {
+			ID string `json:"id"`
+		}
+		if err := addResp.DecodeJSON(&addResult); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		resp, err := env.POST("/api/queue/"+addResult.ID+"/blocklist", nil)
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			Blocklisted bool `json:"blocklisted"`
+			Removed     bool `json:"removed"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		if !result.Removed {
+			t.Error("expected item to be removed from queue")
+		}
+
+		getResp, err := env.GET("/api/queue/" + addResult.ID)
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		if getResp.StatusCode != 404 {
+			t.Error("expected item to return 404 after removal")
+		}
+	})
+}
+
+// TestDownloadQueueManagement_Flow covers section 4.4 of FLOWS.md:
+//  1. List all download_queue entries with status and progress
+//  2. Remove entry (cancel download in torrent client)
+//  3. Blocklist and remove (add to blocklist, cancel, re-search)
+func TestDownloadQueueManagement_Flow(t *testing.T) {
+	env := testutil.NewTestEnv(t)
+	defer env.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	if err := env.CleanupDownloadQueue(ctx); err != nil {
+		t.Fatalf("cleanup failed: %v", err)
+	}
+
+	t.Cleanup(func() {
+		env.CleanupDownloadQueue(context.Background())
+	})
+
+	t.Run("Step1_AddMultipleItems", func(t *testing.T) {
+		for i := 0; i < 3; i++ {
+			resp, err := env.POST("/api/queue", map[string]any{
+				"title":        "Queue Item " + string(rune('A'+i)),
+				"torrent_hash": "hash" + string(rune('a'+i)),
+				"size":         100000000 * (i + 1),
+			})
+			if err != nil {
+				t.Fatalf("request failed: %v", err)
+			}
+			resp.AssertStatus(t, 200)
+		}
+	})
+
+	t.Run("Step2_ListWithFilters", func(t *testing.T) {
+		resp, err := env.GET("/api/queue?status=queued")
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			Items []struct {
+				Status string `json:"status"`
+			} `json:"items"`
+			Total int `json:"total"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		for _, item := range result.Items {
+			if item.Status != "queued" {
+				t.Errorf("expected all items to have status=queued, got %s", item.Status)
+			}
+		}
+	})
+
+	t.Run("Step3_RemoveFromQueue", func(t *testing.T) {
+		listResp, err := env.GET("/api/queue")
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+
+		var listResult struct {
+			Items []struct {
+				ID string `json:"id"`
+			} `json:"items"`
+		}
+		if err := listResp.DecodeJSON(&listResult); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		if len(listResult.Items) == 0 {
+			t.Skip("no items in queue")
+		}
+
+		itemID := listResult.Items[0].ID
+		countBefore, _ := env.CountDownloadQueue(ctx)
+
+		resp, err := env.DELETE("/api/queue/" + itemID)
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		countAfter, _ := env.CountDownloadQueue(ctx)
+		if countAfter >= countBefore {
+			t.Error("expected queue count to decrease")
+		}
+	})
+
+	t.Run("Step4_QueueStats", func(t *testing.T) {
+		resp, err := env.GET("/api/queue/stats")
+		if err != nil {
+			t.Fatalf("request failed: %v", err)
+		}
+		resp.AssertStatus(t, 200)
+
+		var result struct {
+			Total       int `json:"total"`
+			Downloading int `json:"downloading"`
+			Queued      int `json:"queued"`
+			Completed   int `json:"completed"`
+			Failed      int `json:"failed"`
+		}
+		if err := resp.DecodeJSON(&result); err != nil {
+			t.Fatalf("failed to decode response: %v", err)
+		}
+
+		t.Logf("queue stats: total=%d, downloading=%d, queued=%d, completed=%d, failed=%d",
+			result.Total, result.Downloading, result.Queued, result.Completed, result.Failed)
+	})
+}
+
+func TestDownloadQueue_NotFound(t *testing.T) {
+	env := testutil.NewTestEnv(t)
+	defer env.Close()
+
+	resp, err := env.GET("/api/queue/00000000-0000-0000-0000-000000000000")
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	resp.AssertStatus(t, 404)
+}

testing/e2e/testutil/testutil.go

@@ -361,6 +361,19 @@ func (e *TestEnv) CleanupBlocklist(ctx context.Context) error {
 	return err
 }
 
+// CleanupDownloadQueue removes all download_queue entries (for test cleanup).
+func (e *TestEnv) CleanupDownloadQueue(ctx context.Context) error {
+	_, err := e.DB.Exec(ctx, "DELETE FROM download_queue")
+	return err
+}
+
+// CountDownloadQueue returns the number of entries in download_queue.
+func (e *TestEnv) CountDownloadQueue(ctx context.Context) (int64, error) {
+	var count int64
+	err := e.DB.QueryRow(ctx, "SELECT COUNT(*) FROM download_queue").Scan(&count)
+	return count, err
+}
+
 // GetAlbumsByArtistForeignID retrieves albums for an artist by foreign artist ID.
 func (e *TestEnv) GetAlbumsByArtistForeignID(ctx context.Context, foreignArtistID string) ([]map[string]any, error) {
 	rows, err := e.DB.Query(ctx, `