fix(importer): keep failed imports out of completed history

internal/api/sabnzbd_failed_history_test.go

@@ -0,0 +1,94 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/gofiber/fiber/v2"
+	"github.com/kipsilabs/altmount/internal/config"
+	"github.com/kipsilabs/altmount/internal/database"
+	"github.com/kipsilabs/altmount/internal/importer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSABnzbdHistory_FailedVerificationCannotImportAnotherJob(t *testing.T) {
+	db, err := database.NewDB(database.Config{Type: "sqlite", DatabasePath: filepath.Join(t.TempDir(), "history.db")})
+	require.NoError(t, err)
+	defer db.Close()
+	repo := database.NewRepository(db.Connection(), db.Dialect())
+	s := &Server{queueRepo: repo, importerService: &importer.Service{}}
+	root := t.TempDir()
+	s.configManager = &mockConfigManager{cfg: &config.Config{
+		MountPath: root,
+		Import:    config.ImportConfig{ImportStrategy: config.ImportStrategySYMLINK, ImportDir: &root},
+		SABnzbd:   config.SABnzbdConfig{CompleteDir: "/", Categories: []config.SABnzbdCategory{{Name: "tv", Dir: "tv"}}},
+	}}
+	ctx := context.Background()
+	failedID, nextID, category := "failed-download", "next-download", "tv"
+	failed := &database.ImportQueueItem{NzbPath: "Lucky.S01E04.nzb", DownloadID: &failedID, Category: &category, Status: database.QueueStatusFailed, Priority: database.QueuePriorityNormal}
+	require.NoError(t, repo.AddToQueue(ctx, failed))
+	require.NoError(t, repo.AddImportHistory(ctx, &database.ImportHistory{NzbID: &failed.ID, DownloadID: &failedID, NzbName: failed.NzbPath, FileName: "Lucky.mkv", VirtualPath: "/tv/Lucky/Lucky.mkv", Category: &category}))
+	nextPath := "/tv/Sugar/Sugar.S02E07.mkv"
+	require.NoError(t, os.MkdirAll(filepath.Join(root, "tv/Sugar"), 0755))
+	require.NoError(t, os.WriteFile(filepath.Join(root, "tv/Sugar/Sugar.S02E07.mkv"), []byte("media"), 0644))
+	next := &database.ImportQueueItem{NzbPath: "Sugar.S02E07.nzb", DownloadID: &nextID, Category: &category, Status: database.QueueStatusCompleted, Priority: database.QueuePriorityNormal, StoragePath: &nextPath}
+	require.NoError(t, repo.AddToQueue(ctx, next))
+	require.NoError(t, db.Repository.AddStoragePath(ctx, next.ID, nextPath))
+	app := fiber.New()
+	app.Get("/history", s.handleSABnzbdHistory)
+	for _, query := range []string{"", "?nzo_ids=failed-download,next-download", "?cat=tv&limit=1", "?nzo_ids=failed-download"} {
+		t.Run(query, func(t *testing.T) {
+			resp, err := app.Test(httptest.NewRequest("GET", "/history"+query, nil))
+			require.NoError(t, err)
+			defer resp.Body.Close()
+			var result SABnzbdCompleteHistoryResponse
+			require.NoError(t, json.NewDecoder(resp.Body).Decode(&result))
+			seen := map[string]bool{}
+			for _, slot := range result.History.Slots {
+				assert.False(t, seen[slot.NzoID], "duplicate download ID")
+				seen[slot.NzoID] = true
+				if slot.NzoID == failedID {
+					assert.Equal(t, "Failed", slot.Status)
+					assert.Empty(t, slot.Storage, "failed job must not expose the import root")
+					assert.Empty(t, slot.Path)
+				} else {
+					assert.Equal(t, nextID, slot.NzoID)
+					assert.Equal(t, "Completed", slot.Status)
+					assert.Equal(t, filepath.Join(root, "tv/Sugar/Sugar.S02E07.mkv"), slot.Storage)
+				}
+			}
+			if query != "?cat=tv&limit=1" {
+				assert.True(t, seen[failedID])
+			}
+			if query == "" {
+				assert.Equal(t, 2, result.History.Noofslots)
+				assert.Len(t, result.History.Slots, 2)
+			}
+		})
+	}
+}
+
+func TestSABnzbdHistory_MissingStorageCannotExposeImportRoot(t *testing.T) {
+	s, repo := newSABnzbdDeleteTestServer(t)
+	root := t.TempDir()
+	s.configManager = &mockConfigManager{cfg: &config.Config{MountPath: root, Import: config.ImportConfig{ImportStrategy: config.ImportStrategySYMLINK, ImportDir: &root}}}
+	ctx := context.Background()
+	item := &database.ImportQueueItem{NzbPath: "incomplete.nzb", Status: database.QueueStatusCompleted, Priority: database.QueuePriorityNormal}
+	require.NoError(t, repo.AddToQueue(ctx, item))
+	app := fiber.New()
+	app.Get("/history", s.handleSABnzbdHistory)
+	resp, err := app.Test(httptest.NewRequest("GET", "/history", nil))
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	var result SABnzbdCompleteHistoryResponse
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&result))
+	require.Len(t, result.History.Slots, 1)
+	assert.Equal(t, "Failed", result.History.Slots[0].Status)
+	assert.Empty(t, result.History.Slots[0].Storage)
+	assert.Empty(t, result.History.Slots[0].Path)
+}

internal/api/sabnzbd_handlers.go

@@ -1472,7 +1472,15 @@ func (s *Server) calculateItemBasePath() string {
 // stop Sonarr/Radarr from looping on FileNotFoundException for items whose
 // symlink under Import.ImportDir was never created (issue #596).
 func (s *Server) calculateHistoryStoragePath(item *database.ImportQueueItem, basePath string) (string, bool) {
-	if s.configManager == nil || item.StoragePath == nil || *item.StoragePath == "" {
+	// Failed jobs have no importable content, even when a retry or failed
+	// post-processing left a storage path behind. Never expose the import root.
+	if item.Status == database.QueueStatusFailed {
+		return "", false
+	}
+	if item.StoragePath == nil || *item.StoragePath == "" {
+		return "", false
+	}
+	if s.configManager == nil {
 		return basePath, true
 	}
 

internal/database/queue_repository.go

@@ -53,12 +53,13 @@ func NewQueueRepository(db *sql.DB, d Dialect) *QueueRepository {
 // reach.
 func (r *QueueRepository) RemoveFromQueue(ctx context.Context, id int64) error {
 	return r.withQueueTransaction(ctx, func(txRepo *QueueRepository) error {
-		if _, err := txRepo.db.ExecContext(ctx, `DELETE FROM import_queue WHERE id = ?`, id); err != nil {
-			return fmt.Errorf("failed to remove from queue: %w", err)
-		}
+		// Delete history first: deleting the queue first sets history.nzb_id to NULL.
 		if _, err := txRepo.db.ExecContext(ctx, `DELETE FROM import_history WHERE nzb_id = ?`, id); err != nil {
 			return fmt.Errorf("failed to remove import history for queue item %d: %w", id, err)
 		}
+		if _, err := txRepo.db.ExecContext(ctx, `DELETE FROM import_queue WHERE id = ?`, id); err != nil {
+			return fmt.Errorf("failed to remove from queue: %w", err)
+		}
 		return nil
 	})
 }
@@ -503,6 +504,15 @@ func (r *QueueRepository) GetImportHistory(ctx context.Context, days int) ([]*Im
 	return r.GetImportDailyStats(ctx, days)
 }
 
+// DeleteImportHistoryByNzbID rolls back only the history produced by this queue
+// item, preserving successful imports that share the same virtual directory.
+func (r *QueueRepository) DeleteImportHistoryByNzbID(ctx context.Context, nzbID int64) error {
+	if _, err := r.db.ExecContext(ctx, `DELETE FROM import_history WHERE nzb_id = ?`, nzbID); err != nil {
+		return fmt.Errorf("failed to delete import history for queue item %d: %w", nzbID, err)
+	}
+	return nil
+}
+
 // AddImportHistory records a successful file import in the persistent history table
 func (r *QueueRepository) AddImportHistory(ctx context.Context, history *ImportHistory) error {
 	virtualPath := normalizeHealthPath(history.VirtualPath)
@@ -885,6 +895,14 @@ func (r *QueueRepository) DeleteFailedItemsOlderThan(ctx context.Context, olderT
 			return nil
 		}
 
+		// Delete provisional history before its owning queue row: the foreign
+		// key may otherwise null nzb_id and resurrect an old failed import as
+		// Completed. Keep both deletions in this transaction.
+		if _, err := txRepo.db.ExecContext(ctx, `DELETE FROM import_history WHERE nzb_id IN
+			(SELECT id FROM import_queue WHERE status = 'failed' AND updated_at < ?)`, olderThan); err != nil {
+			return fmt.Errorf("failed to delete history for old failed queue items: %w", err)
+		}
+
 		// Single ranged DELETE — uses idx_queue_status_updated.
 		const deleteQuery = `DELETE FROM import_queue WHERE status = 'failed' AND updated_at < ?`
 		if _, err := txRepo.db.ExecContext(ctx, deleteQuery, olderThan); err != nil {

internal/database/queue_repository_history_test.go

@@ -79,5 +79,8 @@ func TestQueueRepository_RemoveFromQueueBulk_SkipsProcessingAndItsHistory(t *tes
 
 	rows, err := repo.ListSABnzbdHistory(ctx, "", 100, 0)
 	require.NoError(t, err)
-	assert.Len(t, rows, 1, "a refused delete must leave the history row in place")
+	assert.Empty(t, rows, "provisional history stays hidden while the item is processing")
+	var count int
+	require.NoError(t, db.QueryRow("SELECT COUNT(*) FROM import_history WHERE nzb_id = 8").Scan(&count))
+	assert.Equal(t, 1, count, "a refused delete must leave the stored history row in place")
 }

internal/database/repository.go

@@ -504,6 +504,10 @@ func (r *Repository) RemoveFromQueue(ctx context.Context, id int64) error {
 	// (issue #586). Deleting the queue row alone would also be unrecoverable —
 	// a retry can no longer find the item to clean up after.
 	return r.WithTransaction(ctx, func(tx *Repository) error {
+		// Delete history first: deleting the queue first sets history.nzb_id to NULL.
+		if _, err := tx.db.ExecContext(ctx, `DELETE FROM import_history WHERE nzb_id = ?`, id); err != nil {
+			return fmt.Errorf("failed to remove import history for queue item %d: %w", id, err)
+		}
 		result, err := tx.db.ExecContext(ctx, `DELETE FROM import_queue WHERE id = ?`, id)
 		if err != nil {
 			return fmt.Errorf("failed to remove from queue: %w", err)
@@ -518,10 +522,6 @@ func (r *Repository) RemoveFromQueue(ctx context.Context, id int64) error {
 			return sql.ErrNoRows
 		}
 
-		if _, err := tx.db.ExecContext(ctx, `DELETE FROM import_history WHERE nzb_id = ?`, id); err != nil {
-			return fmt.Errorf("failed to remove import history for queue item %d: %w", id, err)
-		}
-
 		return nil
 	})
 }
@@ -1339,11 +1339,12 @@ type SABnzbdHistoryRow struct {
 
 // sabnzbdHistoryUnion builds the deduplicated UNION-ALL body shared by
 // ListSABnzbdHistory and CountSABnzbdHistory, plus its bound args (the category
-// filter, repeated once per arm). A completed import can exist in both
-// import_queue (status=completed, never auto-pruned) and import_history; the
-// anti-join drops the history copy when the live completed queue row is present,
-// so each import is counted once. Failed items live only in import_queue and are
-// unaffected. Portable across SQLite + Postgres (no window functions, no
+// filter, repeated once per arm). The live queue owns the job's status until
+// it is removed: processors write provisional history before verification and
+// post-processing finish. Suppress that history for every live queue state,
+// including failed, retried and notification-suppressed jobs. This also hides
+// stale history left by older versions without deleting successful archives.
+// Portable across SQLite + Postgres (no window functions, no
 // basename/regexp); ? placeholders and FALSE are rewritten by the dialect layer.
 func sabnzbdHistoryUnion(category string) (string, []any) {
 	body := `
@@ -1359,7 +1360,7 @@ func sabnzbdHistoryUnion(category string) (string, []any) {
 		       h.completed_at
 		  FROM import_history h
 		  LEFT JOIN import_queue q
-		    ON q.id = h.nzb_id AND q.status = 'completed' AND q.skip_arr_notification = FALSE
+		    ON q.id = h.nzb_id
 		 WHERE q.id IS NULL
 		   AND (? = '' OR LOWER(h.category) = LOWER(?))
 		UNION ALL

internal/database/sabnzbd_history_test.go

@@ -148,3 +148,92 @@ func TestSABnzbdHistory_PaginationAndCategory(t *testing.T) {
 	require.NoError(t, err)
 	assert.Empty(t, empty)
 }
+
+func TestSABnzbdHistory_LiveQueueOwnsProvisionalHistory(t *testing.T) {
+	for _, status := range []string{"failed", "pending", "processing", "paused", "completed"} {
+		for _, skip := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/skip=%v", status, skip), func(t *testing.T) {
+				repo, db := newSABHistoryRepo(t)
+				insertQ(t, db, 5, status, "tv", time.Now(), skip)
+				insertH(t, db, 100, 5, "tv", time.Now())
+				insertH(t, db, 101, 5, "tv", time.Now()) // multi-file job
+				want := 0
+				if !skip && (status == "failed" || status == "completed") {
+					want = 1
+				}
+				total, err := repo.CountSABnzbdHistory(context.Background(), "tv")
+				require.NoError(t, err)
+				assert.Equal(t, want, total)
+				rows, err := repo.ListSABnzbdHistory(context.Background(), "tv", 100, 0)
+				require.NoError(t, err)
+				assert.Len(t, rows, want)
+			})
+		}
+	}
+}
+
+func TestFailedRetention_DoesNotResurrectLegacyHistory(t *testing.T) {
+	db, err := NewDB(Config{Type: "sqlite", DatabasePath: t.TempDir() + "/retention.db"})
+	require.NoError(t, err)
+	defer db.Close()
+	conn := db.Connection()
+	old := time.Now().Add(-48 * time.Hour)
+	insertQ(t, conn, 5, "failed", "tv", old, false)
+	insertH(t, conn, 100, 5, "tv", old)
+	insertH(t, conn, 101, 5, "tv", old)
+	insertQ(t, conn, 6, "completed", "tv", old, false)
+	insertH(t, conn, 102, 6, "tv", old)
+	insertQ(t, conn, 7, "failed", "tv", time.Now(), false)
+	insertH(t, conn, 103, 7, "tv", time.Now())
+	deleted, err := db.Repository.DeleteFailedItemsOlderThan(context.Background(), time.Now().Add(-24*time.Hour))
+	require.NoError(t, err)
+	require.Len(t, deleted, 1)
+	var count int
+	require.NoError(t, conn.QueryRow("SELECT COUNT(*) FROM import_history WHERE nzb_name IN ('h100.nzb', 'h101.nzb')").Scan(&count))
+	assert.Zero(t, count)
+	require.NoError(t, conn.QueryRow("SELECT COUNT(*) FROM import_history WHERE nzb_id IN (6, 7)").Scan(&count))
+	assert.Equal(t, 2, count, "completed and recent history must survive retention")
+	rows, err := NewRepository(conn, db.Dialect()).ListSABnzbdHistory(context.Background(), "", 100, 0)
+	require.NoError(t, err)
+	assert.Len(t, rows, 2, "old failed history must not reappear as Completed")
+}
+
+func TestRemoveFromQueue_DeletesHistoryBeforeForeignKeyUnlinksIt(t *testing.T) {
+	for _, kind := range []string{"queue", "repository"} {
+		t.Run(kind, func(t *testing.T) {
+			db, err := NewDB(Config{Type: "sqlite", DatabasePath: t.TempDir() + "/remove.db"})
+			require.NoError(t, err)
+			defer db.Close()
+			conn := db.Connection()
+			insertQ(t, conn, 5, "failed", "tv", time.Now(), false)
+			insertH(t, conn, 100, 5, "tv", time.Now())
+			insertH(t, conn, 101, 5, "tv", time.Now())
+			repo := NewRepository(conn, db.Dialect())
+			remove := func() error {
+				if kind == "queue" {
+					return db.Repository.RemoveFromQueue(context.Background(), 5)
+				}
+				return repo.RemoveFromQueue(context.Background(), 5)
+			}
+			_, err = conn.Exec("CREATE TRIGGER fail_history_delete BEFORE DELETE ON import_history BEGIN SELECT RAISE(ABORT, 'history unavailable'); END")
+			require.NoError(t, err)
+			require.Error(t, remove(), "history deletion failure must keep the owning queue row")
+			var count int
+			require.NoError(t, conn.QueryRow("SELECT COUNT(*) FROM import_queue WHERE id = 5").Scan(&count))
+			assert.Equal(t, 1, count)
+			require.NoError(t, conn.QueryRow("SELECT COUNT(*) FROM import_history WHERE nzb_id = 5").Scan(&count))
+			assert.Equal(t, 2, count)
+			_, err = conn.Exec("DROP TRIGGER fail_history_delete")
+			require.NoError(t, err)
+			if kind == "queue" {
+				err = db.Repository.RemoveFromQueue(context.Background(), 5)
+			} else {
+				err = repo.RemoveFromQueue(context.Background(), 5)
+			}
+			require.NoError(t, err)
+			rows, err := repo.ListSABnzbdHistory(context.Background(), "", 100, 0)
+			require.NoError(t, err)
+			assert.Empty(t, rows, "queue removal must not orphan Completed histories via ON DELETE SET NULL")
+		})
+	}
+}

internal/importer/handle_failure_test.go

@@ -64,6 +64,7 @@ func newMoveToFailedTestService(t *testing.T) *Service {
 			UNIQUE(nzb_path)
 		);
 		CREATE INDEX IF NOT EXISTS idx_queue_nzb_path ON import_queue(nzb_path);
+ CREATE TABLE import_history (id INTEGER PRIMARY KEY, nzb_id INTEGER);
 	`)
 	require.NoError(t, err)
 

internal/importer/history_failure_test.go

@@ -0,0 +1,78 @@
+package importer
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"path/filepath"
+	"testing"
+
+	"github.com/kipsilabs/altmount/internal/config"
+	"github.com/kipsilabs/altmount/internal/database"
+	"github.com/kipsilabs/altmount/internal/importer/postprocessor"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestHandleFailure_RollsBackHistoryBeforeRetry(t *testing.T) {
+	for _, kind := range []string{"content", "inconclusive", "repair", "cancelled"} {
+		t.Run(kind, func(t *testing.T) {
+			db, err := database.NewDB(database.Config{Type: "sqlite", DatabasePath: filepath.Join(t.TempDir(), "import.db")})
+			require.NoError(t, err)
+			defer db.Close()
+			cfg := &config.Config{}
+			getter := func() *config.Config { return cfg }
+			svc := &Service{database: db, configGetter: getter, log: slog.Default(), postProcessor: postprocessor.NewCoordinator(postprocessor.Config{ConfigGetter: getter})}
+			repo := database.NewRepository(db.Connection(), db.Dialect())
+			ctx := context.Background()
+			download := "failed"
+			item := &database.ImportQueueItem{NzbPath: "bad.nzb", DownloadID: &download, Status: database.QueueStatusProcessing, Priority: database.QueuePriorityNormal}
+			require.NoError(t, db.Repository.AddToQueue(ctx, item))
+			good := &database.ImportQueueItem{NzbPath: "good.nzb", Status: database.QueueStatusCompleted, Priority: database.QueuePriorityNormal}
+			require.NoError(t, db.Repository.AddToQueue(ctx, good))
+			goodID := good.ID
+			require.NoError(t, repo.AddImportHistory(ctx, &database.ImportHistory{NzbID: &goodID, NzbName: "good.nzb", FileName: "good.mkv", VirtualPath: "/same/good.mkv"}))
+			record := func(name string) {
+				require.NoError(t, svc.AddImportHistory(ctx, &database.ImportHistory{NzbID: &item.ID, DownloadID: &download, NzbName: item.NzbPath, FileName: name, VirtualPath: "/same/" + name}))
+			}
+			record("bad.mkv")
+			record("bad.srt")
+			enabled := true
+			cfg.Import.VerifyContent = &enabled
+			svc.contentVerifyFS = &fakeContentOpener{data: make([]byte, 512)}
+			failure := svc.verifyWrittenContent(ctx, []string{"bad.mkv"})
+			require.Error(t, failure)
+			switch kind {
+			case "inconclusive":
+				failure = fmt.Errorf("probe unavailable: %w", ErrContentProbeInconclusive)
+			case "repair":
+				failure = ErrDeferredForRepair
+			case "cancelled":
+				failure = context.Canceled
+			}
+			failureCtx := ctx
+			if kind == "cancelled" {
+				cancelled, cancel := context.WithCancel(ctx)
+				cancel()
+				failureCtx = cancelled
+			}
+			svc.HandleFailure(failureCtx, item, failure)
+			var count int
+			require.NoError(t, db.Connection().QueryRow("SELECT COUNT(*) FROM import_history WHERE nzb_id = ?", item.ID).Scan(&count))
+			assert.Zero(t, count, "all files from the failed attempt must be rolled back")
+			require.NoError(t, db.Connection().QueryRow("SELECT COUNT(*) FROM import_history WHERE nzb_id = ?", goodID).Scan(&count))
+			assert.Equal(t, 1, count, "successful sibling history must survive")
+			// A retry produces fresh history and becomes visible only on completion.
+			require.NoError(t, db.Repository.UpdateQueueItemStatus(ctx, item.ID, database.QueueStatusPending, nil))
+			record("retry.mkv")
+			rows, err := repo.ListSABnzbdHistory(ctx, "", 100, 0)
+			require.NoError(t, err)
+			require.Len(t, rows, 1)
+			assert.Equal(t, goodID, rows[0].ID)
+			require.NoError(t, db.Repository.UpdateQueueItemStatus(ctx, item.ID, database.QueueStatusCompleted, nil))
+			rows, err = repo.ListSABnzbdHistory(ctx, "", 100, 0)
+			require.NoError(t, err)
+			assert.Len(t, rows, 2)
+		})
+	}
+}