feat(downloads): unlink imported files once *arr confirms the import (#79)

* feat(downloads): unlink imported files once *arr confirms the import

Adds `downloads.unlinkImportedFiles` (default false). When enabled, the import
watcher drops jack's copy in completedPath right after a row flips to `imported`
— only on a confirmed import (the destination reports the hash in its history, or
the manual-import command reports completed), never on a queued or failed row.

It is a plain unlink of that one file: if *arr hardlinked into the library the
data lives on through the library's link; if it copied, only jack's copy goes.

The path guards from DownloadsService.delete move into a shared
`unlinkDownloadArtifact` (inside completedPath, still owned by the live row, not
shared with a sibling, ENOENT-tolerant) and both call sites use it.

Configurable from the management UI (Settings -> Downloads) via a new
GET/PATCH /config/downloads; the PATCH merges into the stored block and the
watcher reads the flag per import, so the toggle applies without a restart.

* fix(downloads): let the last row sharing a destination clean it up

The sibling guard blocked an unlink whenever any other row referenced the path,
including rows that were themselves already imported. Two rows can legitimately
share a destination — the in-flight duplicate guard only covers active transfers,
so a re-grab that lands while the first row is still import_queued gets its own
row on the same path. Both would then import, each defer to the other, and since
`imported` is terminal and never revisited, the file outlived both rows despite
unlinkImportedFiles being on.

A sibling now only blocks while it still has a use for the file: downloading into
it, awaiting import, or holding a failed import that a retry would re-trigger
against that path. `imported` is the one status that doesn't qualify, so the row
that imports last does the cleanup.

This also applies to the delete path, which shares the helper: deleting a row now
removes a shared file that only already-imported rows still point at.

Reported by Greptile on #79.

Author: roziscoding

apps/backend/src/__tests__/config-management.test.ts

@@ -586,3 +586,75 @@ describe('Management API jack config', () => {
     expect(res.status).toBe(404)
   })
 })
+
+describe('Management API downloads config', () => {
+  // Seed a config file that already carries a downloads block, so the partial-patch
+  // behaviour (merge onto what's stored) has something to merge onto.
+  async function withDownloadsBlock(app: ReturnType<typeof mgmtApp>, path: string) {
+    const res = await app.request('/config/downloads', {
+      method: 'PATCH',
+      headers: { ...KEY, 'Content-Type': 'application/json' },
+      body: JSON.stringify({ completedPath: '/downloads', maxConcurrentDownloads: 5 }),
+    })
+    expect(res.status).toBe(200)
+    return path
+  }
+
+  test('GET /config/downloads returns null when no downloads block is configured', async () => {
+    const { app } = await makeMutableApp()
+    const res = await app.request('/config/downloads', { headers: KEY })
+    expect(res.status).toBe(200)
+    expect(await res.json()).toBeNull()
+  })
+
+  test('PATCH merges into the stored block instead of replacing it', async () => {
+    const { app, path } = await makeMutableApp()
+    await withDownloadsBlock(app, path)
+
+    const res = await app.request('/config/downloads', {
+      method: 'PATCH',
+      headers: { ...KEY, 'Content-Type': 'application/json' },
+      body: JSON.stringify({ unlinkImportedFiles: true }),
+    })
+    expect(res.status).toBe(200)
+
+    // The unrelated knobs survive, and only what was set is written — defaults are
+    // never baked into the file.
+    const onDisk = jsonc.parse(await Bun.file(path).text()) as { downloads: Record<string, unknown> }
+    expect(onDisk.downloads).toEqual({ completedPath: '/downloads', maxConcurrentDownloads: 5, unlinkImportedFiles: true })
+
+    const get = await app.request('/config/downloads', { headers: KEY })
+    expect(await get.json()).toMatchObject({ unlinkImportedFiles: true, maxConcurrentDownloads: 5 })
+  })
+
+  test('PATCH rejects a first patch that leaves the merged block without completedPath', async () => {
+    const { app } = await makeMutableApp()
+    const res = await app.request('/config/downloads', {
+      method: 'PATCH',
+      headers: { ...KEY, 'Content-Type': 'application/json' },
+      body: JSON.stringify({ unlinkImportedFiles: true }),
+    })
+    expect(res.status).toBe(400)
+  })
+
+  test('PATCH with a non-boolean unlinkImportedFiles returns 400', async () => {
+    const { app, path } = await makeMutableApp()
+    await withDownloadsBlock(app, path)
+
+    const res = await app.request('/config/downloads', {
+      method: 'PATCH',
+      headers: { ...KEY, 'Content-Type': 'application/json' },
+      body: JSON.stringify({ unlinkImportedFiles: 'yes' }),
+    })
+    expect(res.status).toBe(400)
+  })
+
+  test('PATCH /config/downloads is unregistered (404) without a ConfigService', async () => {
+    const res = await mgmtApp().request('/config/downloads', {
+      method: 'PATCH',
+      headers: { ...KEY, 'Content-Type': 'application/json' },
+      body: JSON.stringify({ unlinkImportedFiles: true }),
+    })
+    expect(res.status).toBe(404)
+  })
+})

apps/backend/src/__tests__/downloads-service.test.ts

@@ -43,6 +43,7 @@ function downloadsConfig(overrides: Partial<Record<string, number>> = {}) {
     maxManualImportAttempts: 6,
     manualImportBackoffBaseMs: 60_000,
     manualImportBackoffMaxMs: 1_800_000,
+    unlinkImportedFiles: false,
     ...overrides,
   }
 }

apps/backend/src/__tests__/import-watcher.test.ts

@@ -1,8 +1,10 @@
 import type { ArrServerConnector, ManualImportParams } from '../lib/servers/arr/base'
 import type { DownloadsRepository } from '../modules/downloads/downloads.repository'
-import { dirname } from 'node:path'
+import { mkdir, mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { dirname, join } from 'node:path'
 import { Database } from 'bun:sqlite'
-import { describe, expect, mock, test } from 'bun:test'
+import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test'
 import { drizzle } from 'drizzle-orm/bun-sqlite'
 import { runMigrations } from '../database/connection'
 import * as schema from '../database/schema'
@@ -414,3 +416,241 @@ describe('ImportWatcher tracked manual imports', () => {
     expect(repo.get(row.id)).toMatchObject({ status: 'failed', error: 'episode ids could not be resolved' })
   })
 })
+
+// `downloads.unlinkImportedFiles`: once *arr confirms the import, jack's own copy in
+// completedPath is dead weight (*arr hardlinked or copied it into the library), so the
+// watcher drops jack's link. Only ever on a confirmed import — never on a queued,
+// failed, or still-pending row.
+describe('ImportWatcher imported-file cleanup', () => {
+  let completedPath: string
+
+  beforeEach(async () => {
+    completedPath = await mkdtemp(join(tmpdir(), 'jack-unlink-'))
+  })
+
+  afterEach(async () => {
+    await rm(completedPath, { recursive: true, force: true })
+  })
+
+  // A queued row whose destPath is a real file inside the temp completedPath.
+  async function queuedRowWithFile(repo: DownloadsRepository, opts: { name?: string, mode?: 'jack_manual' } = {}) {
+    const destPath = join(completedPath, opts.name ?? 'x.mkv')
+    await Bun.write(destPath, 'payload')
+    const row = repo.create({
+      torrentFilename: 't.torrent',
+      peerId: 'peer-1',
+      peerName: 'Friend',
+      itemId: 'movie:1',
+      filename: release.filename,
+      destPath,
+      partPath: `${destPath}.part`,
+      releaseSize: release.size,
+      release,
+      qbSourceServer: 'My Radarr',
+      sourceServerId: 'radarr-1',
+      ...(opts.mode === 'jack_manual'
+        ? { importMode: 'jack_manual' as const, importTarget: { kind: 'movie' as const, movieId: 42 } }
+        : {}),
+    })
+    repo.markImportQueued(row.id)
+    return row
+  }
+
+  function cleanupWatcher(repo: DownloadsRepository, servers: ImportWatcherServer[], enabled: () => boolean) {
+    return new ImportWatcher(repo, { servers: servers as unknown as ArrServerConnector[] }, 1000, undefined, undefined, {
+      enabled,
+      completedPath,
+    })
+  }
+
+  test('unlinks the file once *arr history confirms the import', async () => {
+    const repo = makeRepo()
+    const row = await queuedRowWithFile(repo)
+    const watcher = cleanupWatcher(repo, [{ ...fakeServer('My Radarr', [HASH]), id: 'radarr-1' }], () => true)
+
+    expect(await watcher.tick()).toBe(1)
+    expect(repo.get(row.id)?.status).toBe('imported')
+    expect(await Bun.file(row.destPath).exists()).toBe(false)
+  })
+
+  test('unlinks the file once a tracked manual import command completes', async () => {
+    const repo = makeRepo()
+    const row = await queuedRowWithFile(repo, { mode: 'jack_manual' })
+    const server: ImportWatcherServer = {
+      id: 'radarr-1',
+      name: 'My Radarr',
+      isInitialized: true,
+      recentlyImportedDownloadIds: async () => new Set(),
+      manualImport: async () => 44,
+      manualImportCommandStatus: async () => ({ state: 'completed' }),
+    }
+    const watcher = cleanupWatcher(repo, [server], () => true)
+
+    // First tick pushes the manual import, second sees the command completed.
+    await watcher.tick()
+    expect(await Bun.file(row.destPath).exists()).toBe(true)
+    await watcher.tick()
+
+    expect(repo.get(row.id)?.status).toBe('imported')
+    expect(await Bun.file(row.destPath).exists()).toBe(false)
+  })
+
+  test('keeps the file while the row is still queued, and when the import fails', async () => {
+    const repo = makeRepo()
+    const queued = await queuedRowWithFile(repo, { name: 'queued.mkv', mode: 'jack_manual' })
+    const failing = await queuedRowWithFile(repo, { name: 'failing.mkv', mode: 'jack_manual' })
+    const server: ImportWatcherServer = {
+      id: 'radarr-1',
+      name: 'My Radarr',
+      isInitialized: true,
+      recentlyImportedDownloadIds: async () => new Set(),
+      manualImport: async params => (params.paths[0] === failing.destPath ? 46 : 45),
+      manualImportCommandStatus: async commandId =>
+        (commandId === 46 ? { state: 'failed', error: 'arr rejected it' } : { state: 'pending' }),
+    }
+    const watcher = cleanupWatcher(repo, [server], () => true)
+
+    await watcher.tick()
+    await watcher.tick()
+
+    expect(repo.get(queued.id)?.status).toBe('import_queued')
+    expect(repo.get(failing.id)?.status).toBe('failed')
+    expect(await Bun.file(queued.destPath).exists()).toBe(true)
+    expect(await Bun.file(failing.destPath).exists()).toBe(true)
+  })
+
+  test('keeps the file when the option is off', async () => {
+    const repo = makeRepo()
+    const row = await queuedRowWithFile(repo)
+    const watcher = cleanupWatcher(repo, [{ ...fakeServer('My Radarr', [HASH]), id: 'radarr-1' }], () => false)
+
+    expect(await watcher.tick()).toBe(1)
+    expect(repo.get(row.id)?.status).toBe('imported')
+    expect(await Bun.file(row.destPath).exists()).toBe(true)
+  })
+
+  test('reads the option per import, so toggling it applies without a restart', async () => {
+    const repo = makeRepo()
+    const kept = await queuedRowWithFile(repo, { name: 'kept.mkv' })
+    let enabled = false
+    const watcher = cleanupWatcher(repo, [{ ...fakeServer('My Radarr', [HASH]), id: 'radarr-1' }], () => enabled)
+
+    await watcher.tick()
+    expect(await Bun.file(kept.destPath).exists()).toBe(true)
+
+    enabled = true
+    const unlinked = await queuedRowWithFile(repo, { name: 'unlinked.mkv' })
+    await watcher.tick()
+
+    expect(await Bun.file(unlinked.destPath).exists()).toBe(false)
+  })
+
+  test('leaves the row imported when the unlink fails', async () => {
+    const repo = makeRepo()
+    const row = await queuedRowWithFile(repo)
+    // A directory where the file is expected: unlink(2) fails with EISDIR/EPERM.
+    await rm(row.destPath)
+    await mkdir(row.destPath)
+    const watcher = cleanupWatcher(repo, [{ ...fakeServer('My Radarr', [HASH]), id: 'radarr-1' }], () => true)
+
+    expect(await watcher.tick()).toBe(1)
+    expect(repo.get(row.id)?.status).toBe('imported')
+  })
+
+  // Two rows can share a destination: the in-flight duplicate guard only covers active
+  // transfers, so a re-grab that lands while the first row is still import_queued gets
+  // its own row on the same path. Whichever imports last has to do the cleanup —
+  // otherwise both defer to each other and the file outlives both rows.
+  test('defers cleanup to the last of two rows sharing a destination', async () => {
+    const repo = makeRepo()
+    const first = await queuedRowWithFile(repo, { name: 'shared.mkv' })
+    const second = repo.create({
+      torrentFilename: 't2.torrent',
+      peerId: 'peer-1',
+      peerName: 'Friend',
+      itemId: 'movie:1',
+      filename: release.filename,
+      destPath: first.destPath,
+      partPath: first.partPath,
+      releaseSize: release.size,
+      release,
+      qbSourceServer: 'My Radarr',
+      sourceServerId: 'radarr-1',
+    })
+    repo.markImportQueued(second.id)
+
+    // Only the first row's hash is in *arr history to begin with, so it imports alone.
+    const importedHashes = [HASH]
+    const server = {
+      id: 'radarr-1',
+      name: 'My Radarr',
+      isInitialized: true,
+      recentlyImportedDownloadIds: async () => new Set(importedHashes.map(h => h.toLowerCase())),
+    } as any
+    const watcher = cleanupWatcher(repo, [server], () => true)
+
+    await watcher.tick()
+
+    // Both rows flip on the same hash, so drive them one at a time by re-reading state.
+    const imported = [first, second].filter(row => repo.get(row.id)?.status === 'imported')
+    const stillQueued = [first, second].filter(row => repo.get(row.id)?.status === 'import_queued')
+    expect(imported).toHaveLength(2)
+    expect(stillQueued).toHaveLength(0)
+    // The row that imported second saw an already-imported sibling and cleaned up.
+    expect(await Bun.file(first.destPath).exists()).toBe(false)
+  })
+
+  test('keeps the file while a row sharing the destination is still awaiting import', async () => {
+    const repo = makeRepo()
+    const importing = await queuedRowWithFile(repo, { name: 'shared.mkv' })
+    // A second row on the same path that *arr has not imported — it still needs the file.
+    const waiting = repo.create({
+      torrentFilename: 't2.torrent',
+      peerId: 'peer-1',
+      peerName: 'Friend',
+      itemId: 'movie:2',
+      filename: release.filename,
+      destPath: importing.destPath,
+      partPath: importing.partPath,
+      releaseSize: release.size,
+      release,
+      qbSourceServer: 'Other Radarr',
+      sourceServerId: 'radarr-2',
+    })
+    repo.markImportQueued(waiting.id)
+
+    // Only radarr-1 is connected, so only `importing` can flip to imported this tick.
+    const watcher = cleanupWatcher(repo, [{ ...fakeServer('My Radarr', [HASH]), id: 'radarr-1' }], () => true)
+
+    expect(await watcher.tick()).toBe(1)
+    expect(repo.get(importing.id)?.status).toBe('imported')
+    expect(repo.get(waiting.id)?.status).toBe('import_queued')
+    expect(await Bun.file(importing.destPath).exists()).toBe(true)
+  })
+
+  test('never unlinks a path outside completedPath', async () => {
+    const repo = makeRepo()
+    const outside = await mkdtemp(join(tmpdir(), 'jack-outside-'))
+    const destPath = join(outside, 'x.mkv')
+    await Bun.write(destPath, 'payload')
+    const row = repo.create({
+      torrentFilename: 't.torrent',
+      peerId: 'peer-1',
+      peerName: 'Friend',
+      itemId: 'movie:1',
+      filename: release.filename,
+      destPath,
+      partPath: `${destPath}.part`,
+      releaseSize: release.size,
+      release,
+      qbSourceServer: 'My Radarr',
+      sourceServerId: 'radarr-1',
+    })
+    repo.markImportQueued(row.id)
+    const watcher = cleanupWatcher(repo, [{ ...fakeServer('My Radarr', [HASH]), id: 'radarr-1' }], () => true)
+
+    expect(await watcher.tick()).toBe(1)
+    expect(await Bun.file(destPath).exists()).toBe(true)
+    await rm(outside, { recursive: true, force: true })
+  })
+})

apps/backend/src/index.ts

@@ -69,11 +69,18 @@ logger.info({
 
 // Detect *arr imports of finished downloads and flip them import_queued → imported.
 // Construct this before the management app so failed imports can be retried there.
-const importWatcher = config.downloads
-  ? new ImportWatcher(downloadsRepository, connectorManager, config.downloads.importPollIntervalMs, {
-      maxAttempts: config.downloads.maxManualImportAttempts,
-      backoffBaseMs: config.downloads.manualImportBackoffBaseMs,
-      backoffMaxMs: config.downloads.manualImportBackoffMaxMs,
+const downloadsConfig = config.downloads
+const importWatcher = downloadsConfig
+  ? new ImportWatcher(downloadsRepository, connectorManager, downloadsConfig.importPollIntervalMs, {
+      maxAttempts: downloadsConfig.maxManualImportAttempts,
+      backoffBaseMs: downloadsConfig.manualImportBackoffBaseMs,
+      backoffMaxMs: downloadsConfig.manualImportBackoffMaxMs,
+    }, undefined, {
+      // Read per import, from the file when the management API can rewrite it, so
+      // toggling it in the UI applies without a restart. completedPath still comes
+      // from the boot config — it's the root every unlink is confined to.
+      enabled: () => configService?.getRawDownloads()?.unlinkImportedFiles ?? downloadsConfig.unlinkImportedFiles,
+      completedPath: downloadsConfig.completedPath,
     })
   : undefined
 

apps/backend/src/lib/config.ts

@@ -189,10 +189,34 @@ export const DownloadsConfig = z.object({
   maxManualImportAttempts: z.number().int().min(1).default(6),
   manualImportBackoffBaseMs: z.number().int().min(0).default(60_000),
   manualImportBackoffMaxMs: z.number().int().min(0).default(1_800_000),
+  // Once *arr has imported a download, jack's copy in completedPath is dead weight:
+  // *arr either hardlinked it into the library (so the data lives on through the
+  // library's link) or copied it. Enabling this unlinks jack's directory entry —
+  // never a recursive delete — right after a row flips to `imported`.
+  unlinkImportedFiles: z.boolean().default(false),
 })
 
 export type DownloadsConfig = z.infer<typeof DownloadsConfig>
 
+// Raw downloads for persistence: a PATCH body merged onto the stored block, so every
+// field is optional and the merged result is validated through DownloadsConfig before
+// it reaches the file (see ConfigService.updateDownloads).
+export const RawDownloadsConfig = z.object({
+  completedPath: z.string().min(1).optional(),
+  maxConcurrentDownloads: z.number().int().min(1).optional(),
+  maxDownloadAttempts: z.number().int().min(1).optional(),
+  retryBaseDelayMs: z.number().int().min(0).optional(),
+  retryMaxDelayMs: z.number().int().min(0).optional(),
+  idleTimeoutMs: z.number().int().min(1000).optional(),
+  importPollIntervalMs: z.number().int().min(1000).optional(),
+  maxManualImportAttempts: z.number().int().min(1).optional(),
+  manualImportBackoffBaseMs: z.number().int().min(0).optional(),
+  manualImportBackoffMaxMs: z.number().int().min(0).optional(),
+  unlinkImportedFiles: z.boolean().optional(),
+})
+
+export type RawDownloadsConfig = z.infer<typeof RawDownloadsConfig>
+
 export const AppConfig = z.object({
   version: z.number(),
   jack: JackConfig,