feat(backend): wire playlist upsert in SyncService before scanning;

publish events and persist DB on rename/delete/download;
worker: retry failed downloads with simple backoff;
add compute/apply CLI with per-playlist summary;
record last_sync after applying actions

src/app/cli.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+from pathlib import Path
+
+from .config.settings import Settings
+from .core.database.db import Database
+from .core.sync.service import SyncService
+from .core.sync.executor import ActionExecutor
+from .core.events.event_bus import EventBus
+from .core.utils.yt import extract_playlist_id
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="YouTube Playlist Sync — compute/apply actions")
+    parser.add_argument("--apply", action="store_true", help="Apply actions (otherwise compute-only)")
+    parser.add_argument("--db", type=Path, default=Path("app/data/app.db"), help="Path to SQLite database")
+    parser.add_argument("--playlist", type=int, default=None, help="Only run for a specific playlist index (0-based)")
+    args = parser.parse_args(argv)
+
+    settings = Settings()
+    db = Database(args.db.resolve())
+    service = SyncService(db)
+    bus = EventBus()
+    executor = ActionExecutor(db, event_bus=bus)
+
+    async def log_event(payload):
+        # Placeholder subscriber for future GUI/log integration
+        print(f"EVENT: {payload}")
+
+    # Subscribe to key events
+    bus.subscribe("DownloadStarted", log_event)
+    bus.subscribe("DownloadCompleted", log_event)
+    bus.subscribe("DownloadFailed", log_event)
+    bus.subscribe("RenameApplied", log_event)
+    bus.subscribe("FileRecycled", log_event)
+
+    playlists = settings.playlists
+    if args.playlist is not None:
+        playlists = [playlists[args.playlist]] if 0 <= args.playlist < len(playlists) else []
+
+    for pl in playlists:
+        url = pl.get("url")
+        pid = extract_playlist_id(url) or (url or "")
+        actions = service.sync_from_config(pl)
+        counts: dict[str, int] = {}
+        for a in actions:
+            counts[a.type.name] = counts.get(a.type.name, 0) + 1
+        summary = ", ".join(f"{k}:{v}" for k, v in sorted(counts.items()))
+        print(f"Playlist {pid}: {len(actions)} actions → {summary}")
+        if args.apply and actions:
+            asyncio.run(executor.execute(actions, pl))
+            db.set_playlist_last_sync(pid)
+            print(f"Applied actions for {pid}.")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/app/core/download/workers.py

@@ -1,9 +1,17 @@
 from __future__ import annotations
 
+import asyncio
 from .downloader import Downloader
-from .queue_manager import DownloadJob
+from .queue_manager import DownloadJob, JobState
 
 
-async def default_worker(job: DownloadJob):
+async def default_worker(job: DownloadJob, *, max_retries: int = 2, delay_seconds: float = 1.5):
     dl = Downloader()
-    await dl.handle_job(job)
+    attempt = 0
+    while attempt <= max_retries:
+        await dl.handle_job(job)
+        if job.state == JobState.COMPLETED:
+            return
+        attempt += 1
+        if attempt <= max_retries:
+            await asyncio.sleep(delay_seconds)

src/app/core/sync/executor.py

@@ -11,12 +11,14 @@ from ..models import SyncAction, SyncActionType
 from ..sync.reorder import safe_multi_rename
 from ..database.db import Database
 from ..utils.yt import extract_playlist_id
+from ..events.event_bus import EventBus
 
 
 class ActionExecutor:
-    def __init__(self, db: Database, concurrency: int = 2) -> None:
+    def __init__(self, db: Database, concurrency: int = 2, event_bus: EventBus | None = None) -> None:
         self.concurrency = max(1, concurrency)
         self.db = db
+        self.bus = event_bus
 
     async def execute(self, actions: Iterable[SyncAction], playlist_cfg: dict) -> None:
         save_path = Path(playlist_cfg.get("save_path", "./downloads")).resolve()
@@ -63,6 +65,8 @@ class ActionExecutor:
                     self.db.update_local_filename(playlist_id, a.item.video_id, a.to_name)
                 except Exception:
                     pass
+                if self.bus:
+                    await self.bus.publish("RenameApplied", {"playlist_id": playlist_id, "video_id": a.item.video_id, "to": a.to_name})
 
     def _apply_deletions(self, actions: Iterable[SyncAction], audio_root: Path, video_root: Path, playlist_cfg: dict) -> None:
         playlist_id = extract_playlist_id(playlist_cfg.get("url", "")) or playlist_cfg.get("url", "")
@@ -97,12 +101,16 @@ class ActionExecutor:
                     self.db.clear_file_state(playlist_id, a.item.video_id)
                 except Exception:
                     pass
+                if self.bus:
+                    asyncio.create_task(self.bus.publish("FileRecycled", {"playlist_id": playlist_id, "video_id": a.item.video_id, "name": a.from_name}))
 
     async def _apply_downloads(self, actions: Iterable[SyncAction], mode: str, audio_root: Path, video_root: Path, playlist_cfg: dict) -> None:
         playlist_id = extract_playlist_id(playlist_cfg.get("url", "")) or playlist_cfg.get("url", "")
         queue = QueueManager(concurrency=self.concurrency)
 
         async def worker(job: DownloadJob):
+            if self.bus and job.item:
+                await self.bus.publish("DownloadStarted", {"playlist_id": playlist_id, "video_id": job.item.video_id, "target": str(job.output_path)})
             await default_worker(job)
 
         await queue.start(worker)
@@ -130,8 +138,12 @@ class ActionExecutor:
                     if job.state.name == "COMPLETED":
                         self.db.update_local_filename(playlist_id, job.item.video_id, job.output_path.name)
                         self.db.mark_downloaded(playlist_id, job.item.video_id, True)
+                        if self.bus:
+                            await self.bus.publish("DownloadCompleted", {"playlist_id": playlist_id, "video_id": job.item.video_id, "target": str(job.output_path)})
                     else:
                         # Ensure not marked as downloaded if failed
                         self.db.mark_downloaded(playlist_id, job.item.video_id, False)
+                        if self.bus:
+                            await self.bus.publish("DownloadFailed", {"playlist_id": playlist_id, "video_id": job.item.video_id, "error": job.error or "unknown"})
                 except Exception:
                     pass

src/app/core/sync/service.py

@@ -35,6 +35,16 @@ class SyncService:
         save_path.mkdir(parents=True, exist_ok=True)
 
         playlist_id = extract_playlist_id(url) or url
+        # Ensure playlist row exists/updated
+        self.db.upsert_playlist(
+            id=playlist_id,
+            name=playlist_cfg.get("name"),
+            url=url,
+            path=str(save_path),
+            mode=mode,
+            auto_sync=int(bool(playlist_cfg.get("auto_sync", False))),
+            sync_interval_minutes=int(playlist_cfg.get("sync_interval_minutes", 0) or 0),
+        )
         items = self.scanner.scan(url, playlist_id)
 
         sanitized: List[PlaylistItem] = []

src/app/main.py

@@ -13,6 +13,7 @@ from .core.database.db import Database
 from .core.sync.service import SyncService
 from .core.sync.executor import ActionExecutor
 from .core.models import SyncActionType
+from .core.utils.yt import extract_playlist_id
 
 
 def bootstrap(db_path: Path | None = None) -> None:
@@ -38,6 +39,8 @@ def bootstrap(db_path: Path | None = None) -> None:
                 import asyncio
                 asyncio.run(executor.execute(actions, pl))
                 # Post summary (no DB readback yet)
+                pid = extract_playlist_id(pl.get('url', '')) or pl.get('url', '')
+                db.set_playlist_last_sync(pid)
                 print("Applied actions.")
             else:
                 print(f"No actions needed for: {pl.get('url')}")