Store queue in SQLite with pagination

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.2.0 - 2026-05-11
+
+- Replaced JSON queue storage with SQLite-backed queue persistence.
+- Added automatic migration from the previous `queue.json` queue file.
+- Limited the queue view to 10 jobs per page with previous and next controls.
+- Added bulk queue actions to remove finished jobs and retry all failed jobs.
+
 ## 0.1.2 - 2026-05-11
 
 - Updated finalized download layout to save media under `ANI_CLI_DOWNLOAD_DIR/anime_name/Season <season>/`.

README.md

@@ -2,7 +2,7 @@
 
 A small local web UI for a system-wide `ani-cli` install.
 
-Current version: `0.1.2`
+Current version: `0.2.0`
 
 ## Project Layout
 
@@ -29,6 +29,8 @@ http://127.0.0.1:8421/
 
 The app searches anime, queues downloads, supports original or dubbed mode, quality, episode ranges, editable library names, season numbers, and a default download folder.
 
+The queue is stored in SQLite, shown 10 jobs at a time, and includes bulk actions for removing finished jobs and retrying failed jobs.
+
 Downloads are staged in the web app state directory first. After `ani-cli` finishes, the wrapper moves them into a Plex/Jellyfin-friendly layout:
 
 ```text
@@ -44,6 +46,6 @@ ANI_CLI_DOWNLOAD_DIR/anime_name/Season <season>/anime_name - S01E01.mp4
 - `ANI_CLI_WEB_HOST`: server host, default `127.0.0.1`.
 - `ANI_CLI_WEB_PORT`: server port, default `8421`.
 
-Settings and the queue are stored under XDG config/state directories when writable, with a local `.ani-cli-web/` fallback inside this project.
+Settings, queue data, and staging files are stored under XDG config/state directories when writable, with a local `.ani-cli-web/` fallback inside this project. Existing `queue.json` data is migrated into SQLite automatically.
 
 Runtime folders such as `.ani-cli-web/`, `downloads/`, and Python bytecode cache directories at any depth are ignored by git.

VERSION

@@ -1 +1 @@
-0.1.2
+0.2.0

app.py

@@ -4,6 +4,7 @@ import os
 import re
 import shutil
 import signal
+import sqlite3
 import subprocess
 import threading
 import time
@@ -20,7 +21,7 @@ from uuid import uuid4
 PROJECT_ROOT = Path(__file__).resolve().parent
 ANI_CLI = os.environ.get("ANI_CLI_BIN") or shutil.which("ani-cli") or "ani-cli"
 APP_NAME = "ani-cli-web"
-VERSION = "0.1.2"
+VERSION = "0.2.0"
 ALLANIME_BASE = "allanime.day"
 ALLANIME_API = f"https://api.{ALLANIME_BASE}"
 ALLANIME_REFERER = "https://allmanga.to"
@@ -71,6 +72,7 @@ def app_dir(kind):
 
 CONFIG_PATH = app_dir("config") / "config.json"
 QUEUE_PATH = app_dir("state") / "queue.json"
+QUEUE_DB_PATH = app_dir("state") / "queue.sqlite3"
 STAGING_ROOT = app_dir("state") / "staging"
 
 
@@ -202,41 +204,114 @@ class DownloadQueue:
     def __init__(self):
         self.lock = threading.RLock()
         self.wakeup = threading.Event()
-        self.jobs = load_json(QUEUE_PATH, [])
         self.current_process = None
         self.current_job_id = None
+        self._init_db()
+        self._migrate_json_queue()
         self._restore_interrupted_jobs()
         self.worker = threading.Thread(target=self._run, daemon=True)
         self.worker.start()
 
+    def _connect(self):
+        conn = sqlite3.connect(QUEUE_DB_PATH, timeout=30)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _init_db(self):
+        QUEUE_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+        with self._connect() as conn:
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS jobs (
+                    id TEXT PRIMARY KEY,
+                    status TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL,
+                    payload TEXT NOT NULL
+                )
+                """
+            )
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_created_at ON jobs(created_at)")
+
+    def _migrate_json_queue(self):
+        if not QUEUE_PATH.exists():
+            return
+        jobs = load_json(QUEUE_PATH, [])
+        if not jobs:
+            QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated"))
+            return
+        with self.lock, self._connect() as conn:
+            existing = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"]
+            if existing:
+                return
+            for job in jobs:
+                if isinstance(job, dict) and job.get("id"):
+                    self._upsert_job_conn(conn, job)
+        QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated"))
+
     def _restore_interrupted_jobs(self):
-        changed = False
-        for job in self.jobs:
-            if job.get("status") == "running":
+        with self.lock, self._connect() as conn:
+            rows = conn.execute("SELECT payload FROM jobs WHERE status = 'running'").fetchall()
+            for row in rows:
+                job = self._job_from_row(row)
                 job["status"] = "pending"
                 job["pid"] = None
                 job.setdefault("log", []).append("Restarted after web app restart.")
-                changed = True
-        if changed:
-            self._save_locked()
+                self._upsert_job_conn(conn, job)
 
-    def _save_locked(self):
-        clean_jobs = []
-        for job in self.jobs:
+    def _job_from_row(self, row):
+        payload = row["payload"] if isinstance(row, sqlite3.Row) else row[0]
+        return json.loads(payload)
+
+    def _upsert_job_conn(self, conn, job):
         clean = dict(job)
         clean.pop("process", None)
-            clean_jobs.append(clean)
-        write_json(QUEUE_PATH, clean_jobs)
+        conn.execute(
+            """
+            INSERT INTO jobs (id, status, created_at, updated_at, payload)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT(id) DO UPDATE SET
+                status = excluded.status,
+                updated_at = excluded.updated_at,
+                payload = excluded.payload
+            """,
+            (
+                clean["id"],
+                clean.get("status", "pending"),
+                clean.get("created_at") or now_iso(),
+                clean.get("updated_at") or now_iso(),
+                json.dumps(clean, sort_keys=True),
+            ),
+        )
 
-    def list(self):
-        with self.lock:
-            return [dict(job) for job in self.jobs]
+    def _save_job_locked(self, job):
+        with self._connect() as conn:
+            self._upsert_job_conn(conn, job)
+
+    def list(self, page=1, per_page=10):
+        page = max(1, int(page or 1))
+        per_page = min(50, max(1, int(per_page or 10)))
+        offset = (page - 1) * per_page
+        with self.lock, self._connect() as conn:
+            total = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"]
+            rows = conn.execute(
+                "SELECT payload FROM jobs ORDER BY created_at DESC LIMIT ? OFFSET ?",
+                (per_page, offset),
+            ).fetchall()
+        pages = max(1, (total + per_page - 1) // per_page)
+        return {
+            "jobs": [self._job_from_row(row) for row in rows],
+            "page": page,
+            "per_page": per_page,
+            "total": total,
+            "pages": pages,
+        }
 
     def add(self, payload):
         job = build_job(payload)
         with self.lock:
-            self.jobs.append(job)
-            self._save_locked()
+            self._save_job_locked(job)
         self.wakeup.set()
         return job
 
@@ -252,17 +327,36 @@ class DownloadQueue:
             job["finished_at"] = None
             job["updated_at"] = now_iso()
             job["log"] = []
-            self._save_locked()
+            self._save_job_locked(job)
         self.wakeup.set()
         return job
 
+    def retry_all_failed(self):
+        count = 0
+        with self.lock, self._connect() as conn:
+            rows = conn.execute("SELECT payload FROM jobs WHERE status = 'failed'").fetchall()
+            for row in rows:
+                job = self._job_from_row(row)
+                job["status"] = "pending"
+                job["exit_code"] = None
+                job["pid"] = None
+                job["started_at"] = None
+                job["finished_at"] = None
+                job["updated_at"] = now_iso()
+                job["log"] = []
+                self._upsert_job_conn(conn, job)
+                count += 1
+        if count:
+            self.wakeup.set()
+        return {"ok": True, "count": count}
+
     def cancel(self, job_id):
         with self.lock:
             job = self._find(job_id)
             if job["status"] == "pending":
                 job["status"] = "canceled"
                 job["updated_at"] = now_iso()
-                self._save_locked()
+                self._save_job_locked(job)
                 return job
             if job["status"] != "running":
                 return job
@@ -273,7 +367,7 @@ class DownloadQueue:
                     self.current_process.terminate()
                 job["cancel_requested"] = True
                 job["updated_at"] = now_iso()
-                self._save_locked()
+                self._save_job_locked(job)
                 return job
             raise ValueError("Could not find the running process")
 
@@ -282,27 +376,30 @@ class DownloadQueue:
             job = self._find(job_id)
             if job["status"] == "running":
                 raise ValueError("Running jobs must be canceled before removing")
-            self.jobs = [item for item in self.jobs if item["id"] != job_id]
-            self._save_locked()
+            with self._connect() as conn:
+                conn.execute("DELETE FROM jobs WHERE id = ?", (job_id,))
             return {"ok": True}
 
     def clear_finished(self):
-        with self.lock:
-            self.jobs = [job for job in self.jobs if job.get("status") in {"pending", "running"}]
-            self._save_locked()
-            return {"ok": True}
+        with self.lock, self._connect() as conn:
+            cursor = conn.execute("DELETE FROM jobs WHERE status IN ('done', 'canceled')")
+            return {"ok": True, "count": cursor.rowcount}
 
     def _find(self, job_id):
-        for job in self.jobs:
-            if job.get("id") == job_id:
-                return job
+        with self._connect() as conn:
+            row = conn.execute("SELECT payload FROM jobs WHERE id = ?", (job_id,)).fetchone()
+        if row:
+            return self._job_from_row(row)
         raise KeyError("Job not found")
 
     def _next_pending(self):
         with self.lock:
-            for job in self.jobs:
-                if job.get("status") == "pending":
-                    return job
+            with self._connect() as conn:
+                row = conn.execute(
+                    "SELECT payload FROM jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1"
+                ).fetchone()
+            if row:
+                return self._job_from_row(row)
         return None
 
     def _append_log(self, job, line):
@@ -313,7 +410,7 @@ class DownloadQueue:
             job.setdefault("log", []).append(text)
             job["log"] = job["log"][-MAX_LOG_LINES:]
             job["updated_at"] = now_iso()
-            self._save_locked()
+            self._save_job_locked(job)
 
     def _run(self):
         while True:
@@ -353,7 +450,7 @@ class DownloadQueue:
                 f"Final folder: {job['target_dir']}",
             ]
             job["cancel_requested"] = False
-            self._save_locked()
+            self._save_job_locked(job)
 
         process = subprocess.Popen(
             command,
@@ -369,7 +466,7 @@ class DownloadQueue:
             self.current_process = process
             self.current_job_id = job["id"]
             job["pid"] = process.pid
-            self._save_locked()
+            self._save_job_locked(job)
 
         assert process.stdout is not None
         for line in process.stdout:
@@ -405,7 +502,7 @@ class DownloadQueue:
                 job.setdefault("log", []).append(f"Download failed with exit code {exit_code}.")
             self.current_process = None
             self.current_job_id = None
-            self._save_locked()
+            self._save_job_locked(job)
 
 
 def strip_control(value):
@@ -622,7 +719,10 @@ class Handler(BaseHTTPRequestHandler):
                     raise ValueError("Mode must be sub or dub")
                 self.json({"episodes": episode_list(show_id, mode)})
             elif parsed.path == "/api/queue":
-                self.json({"jobs": DOWNLOAD_QUEUE.list()})
+                params = parse_qs(parsed.query)
+                page = (params.get("page") or ["1"])[0]
+                per_page = (params.get("per_page") or ["10"])[0]
+                self.json(DOWNLOAD_QUEUE.list(page=page, per_page=per_page))
             else:
                 self.error(HTTPStatus.NOT_FOUND, "Not found")
         except Exception as exc:
@@ -637,6 +737,8 @@ class Handler(BaseHTTPRequestHandler):
                 self.json(DOWNLOAD_QUEUE.add(self.body_json()), HTTPStatus.CREATED)
             elif parsed.path == "/api/queue/clear-finished":
                 self.json(DOWNLOAD_QUEUE.clear_finished())
+            elif parsed.path == "/api/queue/retry-failed":
+                self.json(DOWNLOAD_QUEUE.retry_all_failed())
             elif parsed.path.startswith("/api/queue/"):
                 parts = parsed.path.strip("/").split("/")
                 if len(parts) != 4:
@@ -1032,9 +1134,13 @@ INDEX_HTML = r"""<!doctype html>
       <section>
         <div class="toolbar">
           <h2>Download queue</h2>
-          <button class="ghost" id="clearBtn" type="button">Clear finished</button>
+          <div class="row" style="flex:0 0 auto">
+            <button class="ghost" id="retryFailedBtn" type="button">Retry failed</button>
+            <button class="ghost" id="removeFinishedBtn" type="button">Remove finished</button>
+          </div>
         </div>
         <div class="queue" id="queue"></div>
+        <div class="toolbar" id="queuePager"></div>
       </section>
     </main>
   </div>
@@ -1044,6 +1150,9 @@ INDEX_HTML = r"""<!doctype html>
       config: { mode: "sub", quality: "best", download_dir: "" },
       selected: null,
       episodes: [],
+      queuePage: 1,
+      queuePages: 1,
+      queueTotal: 0,
       queueTimer: null
     };
 
@@ -1167,14 +1276,19 @@ INDEX_HTML = r"""<!doctype html>
       }
     }
 
-    function renderQueue(jobs) {
+    function renderQueue(data) {
+      const jobs = data.jobs || [];
+      state.queuePage = data.page || 1;
+      state.queuePages = data.pages || 1;
+      state.queueTotal = data.total || 0;
       const el = $("queue");
       el.innerHTML = "";
       if (!jobs.length) {
         el.innerHTML = '<div class="empty">Queue is empty.</div>';
+        renderQueuePager();
         return;
       }
-      for (const job of jobs.slice().reverse()) {
+      for (const job of jobs) {
         const item = document.createElement("article");
         item.className = "job";
         const log = (job.log || []).slice(-28).join("\n");
@@ -1215,6 +1329,29 @@ INDEX_HTML = r"""<!doctype html>
         }
         el.appendChild(item);
       }
+      renderQueuePager();
+    }
+
+    function renderQueuePager() {
+      const el = $("queuePager");
+      const start = state.queueTotal ? ((state.queuePage - 1) * 10) + 1 : 0;
+      const end = Math.min(state.queuePage * 10, state.queueTotal);
+      el.innerHTML = `
+        <span class="muted"></span>
+        <div class="row" style="flex:0 0 auto"></div>
+      `;
+      el.querySelector(".muted").textContent =
+        `${start}-${end} of ${state.queueTotal} · page ${state.queuePage} of ${state.queuePages}`;
+      const controls = el.querySelector(".row");
+      const prev = actionButton("Previous", () => {
+        if (state.queuePage > 1) loadQueue(state.queuePage - 1);
+      });
+      const next = actionButton("Next", () => {
+        if (state.queuePage < state.queuePages) loadQueue(state.queuePage + 1);
+      });
+      prev.disabled = state.queuePage <= 1;
+      next.disabled = state.queuePage >= state.queuePages;
+      controls.append(prev, next);
     }
 
     function actionButton(text, handler, extra = "") {
@@ -1229,15 +1366,26 @@ INDEX_HTML = r"""<!doctype html>
     async function queueAction(id, action) {
       try {
         await api(`/api/queue/${id}/${action}`, { method: "POST", body: "{}" });
-        await loadQueue();
+        await loadQueue(state.queuePage);
       } catch (error) {
         setNotice(error.message);
       }
     }
 
-    async function loadQueue() {
-      const data = await api("/api/queue");
-      renderQueue(data.jobs);
+    async function queueBulkAction(path, message) {
+      try {
+        const data = await api(path, { method: "POST", body: "{}" });
+        setNotice(`${message}: ${data.count || 0}`);
+        await loadQueue(state.queuePage);
+      } catch (error) {
+        setNotice(error.message);
+      }
+    }
+
+    async function loadQueue(page = state.queuePage || 1) {
+      const data = await api(`/api/queue?page=${page}&per_page=10`);
+      if (data.page > data.pages && data.pages > 0) return loadQueue(data.pages);
+      renderQueue(data);
     }
 
     async function loadConfig() {
@@ -1289,10 +1437,8 @@ INDEX_HTML = r"""<!doctype html>
     });
     $("addBtn").addEventListener("click", addSelected);
     $("saveConfigBtn").addEventListener("click", saveConfig);
-    $("clearBtn").addEventListener("click", async () => {
-      await api("/api/queue/clear-finished", { method: "POST", body: "{}" });
-      await loadQueue();
-    });
+    $("removeFinishedBtn").addEventListener("click", () => queueBulkAction("/api/queue/clear-finished", "Removed finished jobs"));
+    $("retryFailedBtn").addEventListener("click", () => queueBulkAction("/api/queue/retry-failed", "Retried failed jobs"));
     $("quality").addEventListener("change", () => { state.config.quality = $("quality").value; });
     $("firstEpBtn").addEventListener("click", () => { $("episodesInput").value = state.episodes[0] || ""; });
     $("latestEpBtn").addEventListener("click", () => {
@@ -1340,7 +1486,7 @@ def main():
     install_signal_handlers(httpd)
     print(f"ani-cli web is running at http://{host}:{port}/")
     print(f"Config: {CONFIG_PATH}")
-    print(f"Queue:  {QUEUE_PATH}")
+    print(f"Queue:  {QUEUE_DB_PATH}")
     httpd.serve_forever()