feat(downloader): remove the torrent after a successful import (#2046)

internal/db/download_clients.go

@@ -17,7 +17,7 @@ type DownloadClientRepo struct {
 
 const downloadClientSelectColumns = `
 	id, name, type, host, port, api_key, use_ssl, url_base, username, password,
-	category, category_audiobook, path_remap, priority, enabled, created_at, updated_at`
+	category, category_audiobook, path_remap, priority, enabled, remove_on_import, created_at, updated_at`
 
 // torrentClientTypes and usenetClientTypes are the download-client types each
 // protocol can be served by. They back the protocol-scoped queries below; a new
@@ -142,13 +142,15 @@ func (r *DownloadClientRepo) List(ctx context.Context) ([]models.DownloadClient,
 	for rows.Next() {
 		var c models.DownloadClient
 		var enabled, useSSL int
+		var removeOnImport int
 		if err := rows.Scan(&c.ID, &c.Name, &c.Type, &c.Host, &c.Port, &c.APIKey,
 			&useSSL, &c.URLBase, &c.Username, &c.Password, &c.Category, &c.CategoryAudiobook, &c.PathRemap, &c.Priority,
-			&enabled, &c.CreatedAt, &c.UpdatedAt); err != nil {
+			&enabled, &removeOnImport, &c.CreatedAt, &c.UpdatedAt); err != nil {
 			return nil, err
 		}
 		c.Enabled = enabled == 1
 		c.UseSSL = useSSL == 1
+		c.RemoveOnImport = removeOnImport == 1
 		hydrateClientCredentials(&c)
 		clients = append(clients, c)
 	}
@@ -157,13 +159,13 @@ func (r *DownloadClientRepo) List(ctx context.Context) ([]models.DownloadClient,
 
 func (r *DownloadClientRepo) GetByID(ctx context.Context, id int64) (*models.DownloadClient, error) {
 	var c models.DownloadClient
-	var enabled, useSSL int
+	var enabled, useSSL, removeOnImport int
 	err := r.db.QueryRowContext(ctx, `
 		SELECT `+downloadClientSelectColumns+`
 		FROM download_clients WHERE id=?`, id).
 		Scan(&c.ID, &c.Name, &c.Type, &c.Host, &c.Port, &c.APIKey,
 			&useSSL, &c.URLBase, &c.Username, &c.Password, &c.Category, &c.CategoryAudiobook, &c.PathRemap, &c.Priority,
-			&enabled, &c.CreatedAt, &c.UpdatedAt)
+			&enabled, &removeOnImport, &c.CreatedAt, &c.UpdatedAt)
 	if errors.Is(err, sql.ErrNoRows) {
 		return nil, nil
 	}
@@ -172,6 +174,7 @@ func (r *DownloadClientRepo) GetByID(ctx context.Context, id int64) (*models.Dow
 	}
 	c.Enabled = enabled == 1
 	c.UseSSL = useSSL == 1
+	c.RemoveOnImport = removeOnImport == 1
 	hydrateClientCredentials(&c)
 	return &c, nil
 }
@@ -190,13 +193,15 @@ func (r *DownloadClientRepo) ListEnabled(ctx context.Context) ([]models.Download
 	for rows.Next() {
 		var c models.DownloadClient
 		var enabled, useSSL int
+		var removeOnImport int
 		if err := rows.Scan(&c.ID, &c.Name, &c.Type, &c.Host, &c.Port, &c.APIKey,
 			&useSSL, &c.URLBase, &c.Username, &c.Password, &c.Category, &c.CategoryAudiobook, &c.PathRemap, &c.Priority,
-			&enabled, &c.CreatedAt, &c.UpdatedAt); err != nil {
+			&enabled, &removeOnImport, &c.CreatedAt, &c.UpdatedAt); err != nil {
 			return nil, err
 		}
 		c.Enabled = enabled == 1
 		c.UseSSL = useSSL == 1
+		c.RemoveOnImport = removeOnImport == 1
 		hydrateClientCredentials(&c)
 		clients = append(clients, c)
 	}
@@ -205,13 +210,13 @@ func (r *DownloadClientRepo) ListEnabled(ctx context.Context) ([]models.Download
 
 func (r *DownloadClientRepo) GetFirstEnabled(ctx context.Context) (*models.DownloadClient, error) {
 	var c models.DownloadClient
-	var enabled, useSSL int
+	var enabled, useSSL, removeOnImport int
 	err := r.db.QueryRowContext(ctx, `
 		SELECT `+downloadClientSelectColumns+`
 		FROM download_clients WHERE enabled=1 ORDER BY priority LIMIT 1`).
 		Scan(&c.ID, &c.Name, &c.Type, &c.Host, &c.Port, &c.APIKey,
 			&useSSL, &c.URLBase, &c.Username, &c.Password, &c.Category, &c.CategoryAudiobook, &c.PathRemap, &c.Priority,
-			&enabled, &c.CreatedAt, &c.UpdatedAt)
+			&enabled, &removeOnImport, &c.CreatedAt, &c.UpdatedAt)
 	if errors.Is(err, sql.ErrNoRows) {
 		return nil, nil
 	}
@@ -220,6 +225,7 @@ func (r *DownloadClientRepo) GetFirstEnabled(ctx context.Context) (*models.Downl
 	}
 	c.Enabled = enabled == 1
 	c.UseSSL = useSSL == 1
+	c.RemoveOnImport = removeOnImport == 1
 	hydrateClientCredentials(&c)
 	return &c, nil
 }
@@ -229,14 +235,14 @@ func (r *DownloadClientRepo) GetFirstEnabled(ctx context.Context) (*models.Downl
 // Returns (nil, nil) if no matching client is configured.
 func (r *DownloadClientRepo) GetFirstEnabledByProtocol(ctx context.Context, protocol string) (*models.DownloadClient, error) {
 	var c models.DownloadClient
-	var enabled, useSSL int
+	var enabled, useSSL, removeOnImport int
 	placeholders, args := clientTypesForProtocol(protocol)
 	err := r.db.QueryRowContext(ctx, `
 		SELECT `+downloadClientSelectColumns+`
 		FROM download_clients WHERE enabled=1 AND type IN (`+placeholders+`) ORDER BY priority LIMIT 1`, args...).
 		Scan(&c.ID, &c.Name, &c.Type, &c.Host, &c.Port, &c.APIKey,
 			&useSSL, &c.URLBase, &c.Username, &c.Password, &c.Category, &c.CategoryAudiobook, &c.PathRemap, &c.Priority,
-			&enabled, &c.CreatedAt, &c.UpdatedAt)
+			&enabled, &removeOnImport, &c.CreatedAt, &c.UpdatedAt)
 	if err != nil && !errors.Is(err, sql.ErrNoRows) {
 		return nil, err
 	}
@@ -245,6 +251,7 @@ func (r *DownloadClientRepo) GetFirstEnabledByProtocol(ctx context.Context, prot
 	}
 	c.Enabled = enabled == 1
 	c.UseSSL = useSSL == 1
+	c.RemoveOnImport = removeOnImport == 1
 	hydrateClientCredentials(&c)
 	return &c, nil
 }
@@ -266,13 +273,15 @@ func (r *DownloadClientRepo) GetEnabledByProtocol(ctx context.Context, protocol
 	for rows.Next() {
 		var c models.DownloadClient
 		var enabled, useSSL int
+		var removeOnImport int
 		if err := rows.Scan(&c.ID, &c.Name, &c.Type, &c.Host, &c.Port, &c.APIKey,
 			&useSSL, &c.URLBase, &c.Username, &c.Password, &c.Category, &c.CategoryAudiobook, &c.PathRemap, &c.Priority,
-			&enabled, &c.CreatedAt, &c.UpdatedAt); err != nil {
+			&enabled, &removeOnImport, &c.CreatedAt, &c.UpdatedAt); err != nil {
 			return nil, err
 		}
 		c.Enabled = enabled == 1
 		c.UseSSL = useSSL == 1
+		c.RemoveOnImport = removeOnImport == 1
 		hydrateClientCredentials(&c)
 		clients = append(clients, c)
 	}
@@ -283,9 +292,9 @@ func (r *DownloadClientRepo) Create(ctx context.Context, c *models.DownloadClien
 	normalizeClientCredentialStorage(c)
 	now := time.Now().UTC()
 	result, err := r.db.ExecContext(ctx, `
-		INSERT INTO download_clients (name, type, host, port, api_key, use_ssl, url_base, username, password, category, category_audiobook, path_remap, priority, enabled, created_at, updated_at)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
-		c.Name, c.Type, c.Host, c.Port, c.APIKey, c.UseSSL, c.URLBase, c.Username, c.Password, c.Category, c.CategoryAudiobook, c.PathRemap, c.Priority, c.Enabled, now, now)
+		INSERT INTO download_clients (name, type, host, port, api_key, use_ssl, url_base, username, password, category, category_audiobook, path_remap, priority, enabled, remove_on_import, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		c.Name, c.Type, c.Host, c.Port, c.APIKey, c.UseSSL, c.URLBase, c.Username, c.Password, c.Category, c.CategoryAudiobook, c.PathRemap, c.Priority, c.Enabled, c.RemoveOnImport, now, now)
 	if err != nil {
 		return fmt.Errorf("create download client: %w", err)
 	}
@@ -304,9 +313,9 @@ func (r *DownloadClientRepo) Update(ctx context.Context, c *models.DownloadClien
 	now := time.Now().UTC()
 	_, err := r.db.ExecContext(ctx, `
 		UPDATE download_clients SET name=?, type=?, host=?, port=?, api_key=?, use_ssl=?,
-		                            url_base=?, username=?, password=?, category=?, category_audiobook=?, path_remap=?, priority=?, enabled=?, updated_at=?
+		                            url_base=?, username=?, password=?, category=?, category_audiobook=?, path_remap=?, priority=?, enabled=?, remove_on_import=?, updated_at=?
 		WHERE id=?`,
-		c.Name, c.Type, c.Host, c.Port, c.APIKey, c.UseSSL, c.URLBase, c.Username, c.Password, c.Category, c.CategoryAudiobook, c.PathRemap, c.Priority, c.Enabled, now, c.ID)
+		c.Name, c.Type, c.Host, c.Port, c.APIKey, c.UseSSL, c.URLBase, c.Username, c.Password, c.Category, c.CategoryAudiobook, c.PathRemap, c.Priority, c.Enabled, c.RemoveOnImport, now, c.ID)
 	return err
 }
 

internal/db/migrations/090_download_client_remove_on_import.sql

@@ -0,0 +1 @@
+ALTER TABLE download_clients ADD COLUMN remove_on_import INTEGER NOT NULL DEFAULT 0;

internal/downloader/adapter.go

@@ -12,6 +12,7 @@ import (
 	"strings"
 
 	"github.com/vavallee/bindery/internal/downloader/nzbget"
+	"github.com/vavallee/bindery/internal/downloader/transmission"
 	"github.com/vavallee/bindery/internal/models"
 	"github.com/vavallee/bindery/internal/pathmap"
 )
@@ -128,14 +129,25 @@ func SendDownload(ctx context.Context, client *models.DownloadClient, sourceURL,
 		if !strings.HasPrefix(transDL, "/") {
 			transDL = ""
 		}
-		torrentID, err := trans.AddTorrent(ctx, sourceURL, transDL, opts.SeedRatio)
+		added, err := trans.AddTorrentDetailed(ctx, sourceURL, transDL, opts.SeedRatio)
 		if err != nil {
 			return nil, err
 		}
-		if torrentID == 0 {
+		if added.ID == 0 {
 			return nil, fmt.Errorf("downloader accepted request but did not return a trackable torrent ID")
 		}
-		result.RemoteID = strconv.FormatInt(torrentID, 10)
+		// Persist the info hash, not the numeric id. Transmission ids are
+		// session-scoped and are renumbered on every daemon restart, so a
+		// stored id stops matching the torrent it was grabbed for — the
+		// download then sits at "downloading" forever, and any later action
+		// keyed on that id (import, removal) lands on whichever torrent
+		// inherited the number. Fall back to the id only if the daemon gave
+		// us no hash at all, which keeps a grab trackable in the same session.
+		if hash := strings.ToLower(strings.TrimSpace(added.HashString)); hash != "" {
+			result.RemoteID = hash
+		} else {
+			result.RemoteID = strconv.FormatInt(added.ID, 10)
+		}
 		return result, nil
 	case "qbittorrent":
 		qb := QbittorrentFor(client)
@@ -219,6 +231,24 @@ func torrentSavePath(client *models.DownloadClient, opts SendOptions) string {
 	return pathmap.Parse(client.PathRemap).ApplyInverse(localPath)
 }
 
+// RemoveTransmissionTorrent removes a torrent identified by whatever Bindery
+// persisted for it. Anything grabbed since hashes were persisted stores an
+// info hash, which Transmission accepts anywhere an id is taken; rows written
+// before that store the session-scoped numeric id and are removed by id for
+// compatibility. The poller rewrites those rows to the hash as soon as it can
+// confirm the torrent, so the numeric branch is only reached for a download
+// the poller has not yet reconciled.
+func RemoveTransmissionTorrent(ctx context.Context, trans *transmission.Client, ref string, deleteFiles bool) error {
+	ref = strings.TrimSpace(ref)
+	if ref == "" {
+		return nil
+	}
+	if torrentID, err := strconv.ParseInt(ref, 10, 64); err == nil {
+		return trans.RemoveTorrent(ctx, torrentID, deleteFiles)
+	}
+	return trans.RemoveTorrentByHash(ctx, ref, deleteFiles)
+}
+
 // RemoveDownload removes a download from its client, optionally taking the data
 // with it.
 //
@@ -234,12 +264,7 @@ func RemoveDownload(ctx context.Context, client *models.DownloadClient, dl *mode
 		if dl.TorrentID == nil || *dl.TorrentID == "" {
 			return nil
 		}
-		torrentID, err := strconv.ParseInt(*dl.TorrentID, 10, 64)
-		if err != nil {
-			return fmt.Errorf("invalid transmission torrent id %q: %w", *dl.TorrentID, err)
-		}
-		trans := TransmissionFor(client)
-		return trans.RemoveTorrent(ctx, torrentID, deleteFiles)
+		return RemoveTransmissionTorrent(ctx, TransmissionFor(client), *dl.TorrentID, deleteFiles)
 	case "qbittorrent":
 		if dl.TorrentID == nil || *dl.TorrentID == "" {
 			return nil
@@ -316,7 +341,13 @@ func GetStalledIDs(ctx context.Context, client *models.DownloadClient) (map[stri
 		for _, t := range torrents {
 			// status 0 = stopped; treat stopped+error as stalled
 			if t.Status == 0 && strings.TrimSpace(t.ErrorString) != "" {
-				out[strconv.FormatInt(t.ID, 10)] = true
+				// Keyed by info hash, which is what a download stores. The
+				// numeric id is deliberately not offered as a fallback: the
+				// caller removes what it matches here, and a session id that
+				// has been renumbered would remove the wrong torrent.
+				if hash := strings.ToLower(strings.TrimSpace(t.HashString)); hash != "" {
+					out[hash] = true
+				}
 			}
 		}
 		return out, true, nil
@@ -427,18 +458,27 @@ func getTorrentLiveStatuses(ctx context.Context, client *models.DownloadClient)
 		out := make(map[string]LiveStatus, len(torrents))
 		for _, t := range torrents {
 			id := strconv.FormatInt(t.ID, 10)
+			hash := strings.ToLower(strings.TrimSpace(t.HashString))
 			status := strconv.Itoa(t.Status)
 			if errString := strings.TrimSpace(t.ErrorString); errString != "" {
 				status = "error: " + errString
 			}
-			out[id] = LiveStatus{
+			live := LiveStatus{
 				Percentage: fmt.Sprintf("%.1f", t.PercentDone*100),
 				TimeLeft:   etaToTimeLeft(t.ETA),
 				Speed:      bytesPerSecondToString(t.DownloadRate),
 				Size:       t.TotalSize,
 				SizeLeft:   t.LeftUntilDone,
 				Status:     status,
 			}
+			// Hash is what a download stores; the numeric id is kept as an
+			// alias so a row the poller has not reconciled yet still shows
+			// progress. This overlay is read-only, so a stale id can only
+			// mislabel a queue row, never act on a torrent.
+			if hash != "" {
+				out[hash] = live
+			}
+			out[id] = live
 		}
 		return out, nil
 	}

internal/downloader/adapter_stall_test.go

@@ -87,7 +87,9 @@ func TestGetStalledIDs_QBittorrent_EmptyList(t *testing.T) {
 
 // TestGetStalledIDs_Transmission_StoppedWithError verifies that Transmission
 // torrents in status 0 (stopped) with a non-empty errorString are reported
-// as stalled, while other states are not.
+// as stalled, while other states are not. Entries are keyed by info hash,
+// which is what a download stores: the caller removes what it matches here,
+// and a renumbered session id would remove the wrong torrent.
 func TestGetStalledIDs_Transmission_StoppedWithError(t *testing.T) {
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path != "/transmission/rpc" {
@@ -96,10 +98,10 @@ func TestGetStalledIDs_Transmission_StoppedWithError(t *testing.T) {
 		_ = json.NewEncoder(w).Encode(map[string]any{
 			"arguments": map[string]any{
 				"torrents": []map[string]any{
-					{"id": 1, "status": 0, "errorString": "tracker error"},
-					{"id": 2, "status": 0, "errorString": ""},
-					{"id": 3, "status": 2, "errorString": "some error"},
-					{"id": 4, "status": 0, "errorString": "   "},
+					{"id": 1, "hashString": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", "status": 0, "errorString": "tracker error"},
+					{"id": 2, "hashString": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "status": 0, "errorString": ""},
+					{"id": 3, "hashString": "cccccccccccccccccccccccccccccccccccccccc", "status": 2, "errorString": "some error"},
+					{"id": 4, "hashString": "dddddddddddddddddddddddddddddddddddddddd", "status": 0, "errorString": "   "},
 				},
 			},
 			"result": "success",
@@ -120,8 +122,10 @@ func TestGetStalledIDs_Transmission_StoppedWithError(t *testing.T) {
 	if len(stalled) != 1 {
 		t.Fatalf("expected 1 stalled entry, got %d: %v", len(stalled), stalled)
 	}
-	if !stalled["1"] {
-		t.Error("expected transmission id '1' to be stalled")
+	// Lower-cased: downloads store the hash lower-cased, and the caller looks
+	// it up that way.
+	if !stalled["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"] {
+		t.Errorf("expected the stopped-with-error torrent to be stalled, got %v", stalled)
 	}
 }