feat: database-backed jobs for robust uploads and exports (#337)

* feat: database-backed jobs for robust uploads and exports

* fix: add missing job system files

* refactor: use R2ProgressTracker utility for upload/export progress

* fix: resolve linting errors (undefined Any, nested with)

* fix: address PR comments (SSE newlines, migration default)

* fix: merge job results to prevent data loss

backend/alembic/versions/2025_11_24_003101_63b5345e1707_add_jobs_table.py

···
+"""add jobs table
+
+Revision ID: 63b5345e1707
+Revises: 7f3d3a0f6c1a
+Create Date: 2025-11-24 00:31:01.929398
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "63b5345e1707"
+down_revision: str | Sequence[str] | None = "7f3d3a0f6c1a"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_table(
+        "jobs",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("type", sa.String(), nullable=False),
+        sa.Column("status", sa.String(), nullable=False),
+        sa.Column("owner_did", sa.String(), nullable=False),
+        sa.Column("progress_pct", sa.Float(), nullable=False, server_default="0.0"),
+        sa.Column("message", sa.String(), nullable=True),
+        sa.Column("phase", sa.String(), nullable=True),
+        sa.Column(
+            "result", sa.dialects.postgresql.JSONB(astext_type=sa.Text()), nullable=True
+        ),
+        sa.Column("error", sa.String(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("idx_jobs_owner", "jobs", ["owner_did"], unique=False)
+    op.create_index("idx_jobs_updated_at", "jobs", ["updated_at"], unique=False)
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_index("idx_jobs_updated_at", table_name="jobs")
+    op.drop_index("idx_jobs_owner", table_name="jobs")
+    op.drop_table("jobs")

backend/src/backend/_internal/exports.py

···
-"""async export tracking and background processing."""
-
-import asyncio
-import contextlib
-import logging
-from dataclasses import dataclass
-from datetime import UTC, datetime
-from enum import Enum
-from typing import Any
-from uuid import uuid4
-
-logger = logging.getLogger(__name__)
-
-# maximum number of buffered progress updates per export listener
-MAX_PROGRESS_QUEUE_SIZE = 10
-
-
-class ExportStatus(str, Enum):
-    """export status enum."""
-
-    PENDING = "pending"
-    PROCESSING = "processing"
-    COMPLETED = "completed"
-    FAILED = "failed"
-
-
-@dataclass
-class ExportProgress:
-    """export progress state."""
-
-    export_id: str
-    status: ExportStatus
-    message: str
-    track_count: int | None = None
-    processed_count: int = 0
-    error: str | None = None
-    created_at: datetime | None = None
-    completed_at: datetime | None = None
-    download_url: str | None = None
-
-    def to_dict(self) -> dict[str, Any]:
-        """serialize to dict."""
-        return {
-            "export_id": self.export_id,
-            "status": self.status.value,
-            "message": self.message,
-            "track_count": self.track_count,
-            "processed_count": self.processed_count,
-            "error": self.error,
-            "created_at": self.created_at.isoformat() if self.created_at else None,
-            "completed_at": self.completed_at.isoformat()
-            if self.completed_at
-            else None,
-            "download_url": self.download_url,
-        }
-
-
-class ExportTracker:
-    """tracks export progress in memory."""
-
-    def __init__(self):
-        self._exports: dict[str, ExportProgress] = {}
-        self._listeners: dict[str, list[asyncio.Queue]] = {}
-        self._export_data: dict[str, bytes] = {}  # store zip data temporarily
-
-    def create_export(self, track_count: int) -> str:
-        """create a new export and return its ID."""
-        export_id = str(uuid4())
-        self._exports[export_id] = ExportProgress(
-            export_id=export_id,
-            status=ExportStatus.PENDING,
-            message="export queued",
-            track_count=track_count,
-            created_at=datetime.now(UTC),
-        )
-        self._listeners[export_id] = []
-        return export_id
-
-    def update_status(
-        self,
-        export_id: str,
-        status: ExportStatus,
-        message: str,
-        processed_count: int | None = None,
-        error: str | None = None,
-    ) -> None:
-        """update export status and notify listeners."""
-        if export_id not in self._exports:
-            logger.warning(f"attempted to update unknown export: {export_id}")
-            return
-
-        export = self._exports[export_id]
-        export.status = status
-        export.message = message
-        if processed_count is not None:
-            export.processed_count = processed_count
-        export.error = error
-
-        if status in (ExportStatus.COMPLETED, ExportStatus.FAILED):
-            export.completed_at = datetime.now(UTC)
-
-        # notify all listeners
-        if export_id in self._listeners:
-            for queue in self._listeners[export_id]:
-                try:
-                    queue.put_nowait(export.to_dict())
-                except asyncio.QueueFull:
-                    logger.warning(f"listener queue full for export {export_id}")
-
-    def store_export_data(self, export_id: str, data: bytes) -> None:
-        """store the export zip data temporarily."""
-        self._export_data[export_id] = data
-        if export_id in self._exports:
-            self._exports[export_id].download_url = f"/exports/{export_id}/download"
-
-    def get_export_data(self, export_id: str) -> bytes | None:
-        """get the stored export data."""
-        return self._export_data.get(export_id)
-
-    def get_status(self, export_id: str) -> ExportProgress | None:
-        """get current export status."""
-        return self._exports.get(export_id)
-
-    async def subscribe(self, export_id: str) -> asyncio.Queue:
-        """subscribe to export progress updates."""
-        if export_id not in self._listeners:
-            self._listeners[export_id] = []
-
-        queue: asyncio.Queue = asyncio.Queue(maxsize=MAX_PROGRESS_QUEUE_SIZE)
-        self._listeners[export_id].append(queue)
-
-        # send current status immediately
-        if export_id in self._exports:
-            await queue.put(self._exports[export_id].to_dict())
-
-        return queue
-
-    def unsubscribe(self, export_id: str, queue: asyncio.Queue) -> None:
-        """unsubscribe from export progress."""
-        if export_id in self._listeners:
-            with contextlib.suppress(ValueError):
-                self._listeners[export_id].remove(queue)
-
-    def cleanup_old_exports(self, max_age_seconds: int = 3600) -> None:
-        """remove exports older than max_age_seconds."""
-        now = datetime.now(UTC)
-        to_remove = []
-
-        for export_id, export in self._exports.items():
-            if export.completed_at:
-                age = (now - export.completed_at).total_seconds()
-                if age > max_age_seconds:
-                    to_remove.append(export_id)
-
-        for export_id in to_remove:
-            del self._exports[export_id]
-            if export_id in self._listeners:
-                del self._listeners[export_id]
-            if export_id in self._export_data:
-                del self._export_data[export_id]
-
-
-# global tracker instance
-export_tracker = ExportTracker()

backend/src/backend/_internal/jobs.py

···
+"""Database-backed job tracking service."""
+
+import logging
+from datetime import UTC, datetime
+from typing import Any
+
+import logfire
+from sqlalchemy import select
+
+from backend.models.job import Job, JobStatus, JobType
+from backend.utilities.database import db_session
+
+logger = logging.getLogger(__name__)
+
+
+class JobService:
+    """Service for managing database-backed jobs."""
+
+    async def create_job(
+        self,
+        job_type: JobType,
+        owner_did: str,
+        initial_message: str = "job created",
+    ) -> str:
+        """Create a new job and return its ID."""
+        async with db_session() as db:
+            job = Job(
+                type=job_type.value,
+                owner_did=owner_did,
+                status=JobStatus.PENDING.value,
+                message=initial_message,
+                progress_pct=0.0,
+            )
+            db.add(job)
+            await db.commit()
+            await db.refresh(job)
+            return job.id
+
+    async def update_progress(
+        self,
+        job_id: str,
+        status: JobStatus,
+        message: str,
+        progress_pct: float | None = None,
+        phase: str | None = None,
+        result: dict[str, Any] | None = None,
+        error: str | None = None,
+    ) -> None:
+        """Update job progress."""
+        async with db_session() as db:
+            stmt = select(Job).where(Job.id == job_id)
+            result_db = await db.execute(stmt)
+            job = result_db.scalar_one_or_none()
+
+            if not job:
+                logger.warning(f"attempted to update unknown job: {job_id}")
+                return
+
+            job.status = status.value
+            job.message = message
+            if progress_pct is not None:
+                job.progress_pct = progress_pct
+            if phase:
+                job.phase = phase
+            if result:
+                job.result = {**(job.result or {}), **result}
+            if error:
+                job.error = error
+
+            if status in (JobStatus.COMPLETED, JobStatus.FAILED):
+                job.completed_at = datetime.now(UTC)
+
+            await db.commit()
+
+            # log significant updates
+            if status in (JobStatus.COMPLETED, JobStatus.FAILED) or (
+                progress_pct and int(progress_pct) % 25 == 0
+            ):
+                logfire.info(
+                    "job updated",
+                    job_id=job_id,
+                    status=status.value,
+                    progress=progress_pct,
+                )
+
+    async def get_job(self, job_id: str) -> Job | None:
+        """Get job by ID."""
+        async with db_session() as db:
+            stmt = select(Job).where(Job.id == job_id)
+            result = await db.execute(stmt)
+            return result.scalar_one_or_none()
+
+
+job_service = JobService()

backend/src/backend/_internal/uploads.py

···
-"""async upload tracking and background processing."""
-
-import asyncio
-import contextlib
-import logging
-from dataclasses import dataclass
-from datetime import UTC, datetime
-from enum import Enum
-from typing import Any
-from uuid import uuid4
-
-logger = logging.getLogger(__name__)
-
-# maximum number of buffered progress updates per upload listener
-# this prevents memory buildup if a client disconnects without properly closing the SSE connection
-MAX_PROGRESS_QUEUE_SIZE = 10
-
-
-class UploadStatus(str, Enum):
-    """upload status enum."""
-
-    PENDING = "pending"
-    PROCESSING = "processing"
-    COMPLETED = "completed"
-    FAILED = "failed"
-
-
-@dataclass
-class UploadProgress:
-    """upload progress state."""
-
-    upload_id: str
-    status: UploadStatus
-    message: str
-    track_id: int | None = None
-    error: str | None = None
-    created_at: datetime | None = None
-    completed_at: datetime | None = None
-    server_progress_pct: float | None = None
-    phase: str | None = None
-
-    def to_dict(self) -> dict[str, Any]:
-        """serialize to dict."""
-        return {
-            "upload_id": self.upload_id,
-            "status": self.status.value,
-            "message": self.message,
-            "track_id": self.track_id,
-            "error": self.error,
-            "created_at": self.created_at.isoformat() if self.created_at else None,
-            "completed_at": self.completed_at.isoformat()
-            if self.completed_at
-            else None,
-            "server_progress_pct": self.server_progress_pct,
-            "phase": self.phase,
-        }
-
-
-class UploadTracker:
-    """tracks upload progress in memory."""
-
-    def __init__(self):
-        self._uploads: dict[str, UploadProgress] = {}
-        self._listeners: dict[str, list[asyncio.Queue]] = {}
-
-    def create_upload(self) -> str:
-        """create a new upload and return its ID."""
-        upload_id = str(uuid4())
-        self._uploads[upload_id] = UploadProgress(
-            upload_id=upload_id,
-            status=UploadStatus.PENDING,
-            message="upload queued",
-            created_at=datetime.now(UTC),
-        )
-        self._listeners[upload_id] = []
-        return upload_id
-
-    def update_status(
-        self,
-        upload_id: str,
-        status: UploadStatus,
-        message: str,
-        track_id: int | None = None,
-        error: str | None = None,
-        server_progress_pct: float | None = None,
-        phase: str | None = None,
-    ) -> None:
-        """update upload status and notify listeners."""
-        if upload_id not in self._uploads:
-            logger.warning(f"attempted to update unknown upload: {upload_id}")
-            return
-
-        upload = self._uploads[upload_id]
-        upload.status = status
-        upload.message = message
-        upload.track_id = track_id
-        upload.error = error
-        upload.server_progress_pct = server_progress_pct
-        upload.phase = phase
-
-        if status in (UploadStatus.COMPLETED, UploadStatus.FAILED):
-            upload.completed_at = datetime.now(UTC)
-
-        # notify all listeners
-        if upload_id in self._listeners:
-            for queue in self._listeners[upload_id]:
-                try:
-                    queue.put_nowait(upload.to_dict())
-                except asyncio.QueueFull:
-                    logger.warning(f"listener queue full for upload {upload_id}")
-
-    def get_status(self, upload_id: str) -> UploadProgress | None:
-        """get current upload status."""
-        return self._uploads.get(upload_id)
-
-    async def subscribe(self, upload_id: str) -> asyncio.Queue:
-        """subscribe to upload progress updates."""
-        if upload_id not in self._listeners:
-            self._listeners[upload_id] = []
-
-        queue: asyncio.Queue = asyncio.Queue(maxsize=MAX_PROGRESS_QUEUE_SIZE)
-        self._listeners[upload_id].append(queue)
-
-        # send current status immediately
-        if upload_id in self._uploads:
-            await queue.put(self._uploads[upload_id].to_dict())
-
-        return queue
-
-    def unsubscribe(self, upload_id: str, queue: asyncio.Queue) -> None:
-        """unsubscribe from upload progress."""
-        if upload_id in self._listeners:
-            with contextlib.suppress(ValueError):
-                self._listeners[upload_id].remove(queue)
-
-    def cleanup_old_uploads(self, max_age_seconds: int = 3600) -> None:
-        """remove uploads older than max_age_seconds."""
-        now = datetime.now(UTC)
-        to_remove = []
-
-        for upload_id, upload in self._uploads.items():
-            if upload.completed_at:
-                age = (now - upload.completed_at).total_seconds()
-                if age > max_age_seconds:
-                    to_remove.append(upload_id)
-
-        for upload_id in to_remove:
-            del self._uploads[upload_id]
-            if upload_id in self._listeners:
-                del self._listeners[upload_id]
-
-
-# global tracker instance
-upload_tracker = UploadTracker()

backend/src/backend/api/exports.py

···
 import asyncio
 import io
 import json
+import logging
 import zipfile
 from typing import Annotated
 
 import aioboto3
 import logfire
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
-from fastapi.responses import StreamingResponse
+from fastapi.responses import RedirectResponse, StreamingResponse
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend._internal import Session, require_auth
-from backend._internal.exports import ExportStatus, export_tracker
+from backend._internal.jobs import job_service
 from backend.config import settings
 from backend.models import Track, get_db
+from backend.models.job import JobStatus, JobType
 from backend.utilities.database import db_session
+from backend.utilities.progress import R2ProgressTracker
 
 router = APIRouter(prefix="/exports", tags=["exports"])
+logger = logging.getLogger(__name__)
 
 
 async def _process_export_background(export_id: str, artist_did: str) -> None:
     """background task to process export."""
     try:
-        export_tracker.update_status(
-            export_id, ExportStatus.PROCESSING, "fetching tracks..."
+        await job_service.update_progress(
+            export_id, JobStatus.PROCESSING, "fetching tracks..."
         )
 
         # query all tracks for the user
···
             tracks = result.scalars().all()
 
         if not tracks:
-            export_tracker.update_status(
+            await job_service.update_progress(
                 export_id,
-                ExportStatus.FAILED,
+                JobStatus.FAILED,
                 "export failed",
                 error="no tracks found to export",
             )
···
                 # track counter for duplicate titles
                 title_counts: dict[str, int] = {}
                 processed = 0
+                total = len(tracks)
 
                 for track in tracks:
                     if not track.file_id or not track.file_type:
···
 
                     try:
                         # update progress
-                        export_tracker.update_status(
+                        pct = (processed / total) * 100
+                        await job_service.update_progress(
                             export_id,
-                            ExportStatus.PROCESSING,
+                            JobStatus.PROCESSING,
                             f"downloading {track.title}...",
-                            processed_count=processed,
+                            progress_pct=pct,
+                            result={"processed_count": processed, "total_count": total},
                         )
 
                         # download file from R2
···
                         )
                         # continue with other tracks instead of failing entire export
 
-        # store the zip data
+        # store the zip data to R2
         zip_buffer.seek(0)
-        export_tracker.store_export_data(export_id, zip_buffer.getvalue())
+        zip_filename = f"{export_id}.zip"
+        key = f"exports/{zip_filename}"
+
+        await job_service.update_progress(
+            export_id,
+            JobStatus.PROCESSING,
+            "uploading export to storage...",
+        )
+
+        # Upload using aioboto3 directly
+        try:
+            async_session = aioboto3.Session()
+            zip_size = zip_buffer.getbuffer().nbytes
+
+            async with (
+                R2ProgressTracker(
+                    job_id=export_id,
+                    total_size=zip_size,
+                    message="uploading export to storage...",
+                    phase="upload",
+                ) as tracker,
+                async_session.client(
+                    "s3",
+                    endpoint_url=settings.storage.r2_endpoint_url,
+                    aws_access_key_id=settings.storage.aws_access_key_id,
+                    aws_secret_access_key=settings.storage.aws_secret_access_key,
+                ) as s3_client,
+            ):
+                await s3_client.upload_fileobj(
+                    zip_buffer,
+                    settings.storage.r2_bucket,
+                    key,
+                    ExtraArgs={"ContentType": "application/zip"},
+                    Callback=tracker.on_progress,
+                )
+
+            # Final 100% update
+            await job_service.update_progress(
+                export_id,
+                JobStatus.PROCESSING,
+                "uploading export to storage...",
+                phase="upload",
+                progress_pct=100.0,
+            )
+
+        except Exception as e:
+            logfire.error("failed to upload export zip", error=str(e))
+            raise
+
+        # get download URL
+        download_url = f"{settings.storage.r2_public_bucket_url}/{key}"
 
         # mark as completed
-        export_tracker.update_status(
+        await job_service.update_progress(
             export_id,
-            ExportStatus.COMPLETED,
+            JobStatus.COMPLETED,
             f"export completed - {processed} tracks ready",
-            processed_count=processed,
+            result={
+                "processed_count": processed,
+                "total_count": len(tracks),
+                "download_url": download_url,
+            },
         )
 
     except Exception as e:
···
             "export failed with unexpected error",
             export_id=export_id,
         )
-        export_tracker.update_status(
+        await job_service.update_progress(
             export_id,
-            ExportStatus.FAILED,
+            JobStatus.FAILED,
             "export failed",
             error=f"unexpected error: {e!s}",
         )
···
         raise HTTPException(status_code=404, detail="no tracks found to export")
 
     # create export tracking
-    export_id = export_tracker.create_export(track_count=len(tracks))
+    export_id = await job_service.create_job(
+        JobType.EXPORT, session.did, "export queued for processing"
+    )
 
     # schedule background processing
     background_tasks.add_task(_process_export_background, export_id, session.did)
···
 
     async def event_stream():
         """generate SSE events for export progress."""
-        queue = await export_tracker.subscribe(export_id)
+        # Polling loop
         try:
             while True:
-                try:
-                    # wait for next update with timeout
-                    update = await asyncio.wait_for(queue.get(), timeout=30.0)
-                    yield f"data: {json.dumps(update)}\n\n"
+                job = await job_service.get_job(export_id)
+                if not job:
+                    yield f"data: {json.dumps({'status': 'failed', 'message': 'export job not found', 'error': 'job lost'})}\n\n"
+                    break
+
+                # Construct payload
+                payload = {
+                    "export_id": job.id,
+                    "status": job.status,
+                    "message": job.message,
+                    "error": job.error,
+                    "processed_count": job.result.get("processed_count")
+                    if job.result
+                    else 0,
+                    "total_count": job.result.get("total_count") if job.result else 0,
+                }
+                if job.result and "download_url" in job.result:
+                    payload["download_url"] = job.result["download_url"]
+
+                yield f"data: {json.dumps(payload)}\n\n"
 
-                    # if export completed or failed, close stream
-                    if update["status"] in ("completed", "failed"):
-                        break
+                if job.status in (JobStatus.COMPLETED.value, JobStatus.FAILED.value):
+                    break
 
-                except TimeoutError:
-                    # send keepalive
-                    yield ": keepalive\n\n"
+                await asyncio.sleep(1.0)
 
-        finally:
-            export_tracker.unsubscribe(export_id, queue)
+        except Exception as e:
+            logger.error(f"error in export progress stream: {e}")
+            yield f"data: {json.dumps({'status': 'failed', 'message': 'connection error', 'error': str(e)})}\n\n"
 
     return StreamingResponse(
         event_stream(),
···
 async def download_export(
     export_id: str,
     session: Annotated[Session, Depends(require_auth)],
-) -> StreamingResponse:
+) -> RedirectResponse:
     """download the completed export zip file."""
-    export_data = export_tracker.get_export_data(export_id)
+    job = await job_service.get_job(export_id)
 
-    if not export_data:
-        raise HTTPException(status_code=404, detail="export not found or expired")
+    if not job:
+        raise HTTPException(status_code=404, detail="export not found")
 
-    return StreamingResponse(
-        iter([export_data]),
-        media_type="application/zip",
-        headers={
-            "Content-Disposition": 'attachment; filename="plyr-tracks.zip"',
-        },
-    )
+    if job.owner_did != session.did:
+        raise HTTPException(status_code=403, detail="not authorized")
+
+    if job.status != JobStatus.COMPLETED.value:
+        raise HTTPException(status_code=400, detail="export not ready")
+
+    if not job.result or "download_url" not in job.result:
+        raise HTTPException(status_code=500, detail="export url not found")
+
+    return RedirectResponse(url=job.result["download_url"])

backend/src/backend/api/tracks/uploads.py

···
 from backend._internal.atproto.handles import resolve_handle
 from backend._internal.audio import AudioFormat
 from backend._internal.image import ImageFormat
-from backend._internal.uploads import UploadStatus, upload_tracker
+from backend._internal.jobs import job_service
 from backend.config import settings
 from backend.models import Artist, Track
+from backend.models.job import JobStatus, JobType
 from backend.storage import storage
 from backend.utilities.database import db_session
 from backend.utilities.hashing import CHUNK_SIZE
+from backend.utilities.progress import R2ProgressTracker
 from backend.utilities.rate_limit import limiter
 
 from .router import router
···
         "process upload background", upload_id=upload_id, filename=filename
     ):
         try:
-            upload_tracker.update_status(
-                upload_id, UploadStatus.PROCESSING, "processing upload..."
+            await job_service.update_progress(
+                upload_id, JobStatus.PROCESSING, "processing upload..."
             )
 
             # validate file type
             ext = Path(filename).suffix.lower()
             audio_format = AudioFormat.from_extension(ext)
             if not audio_format:
-                upload_tracker.update_status(
+                await job_service.update_progress(
                     upload_id,
-                    UploadStatus.FAILED,
+                    JobStatus.FAILED,
                     "upload failed",
                     error=f"unsupported file type: {ext}",
                 )
                 return
 
             # save audio file
-            upload_tracker.update_status(
+            await job_service.update_progress(
                 upload_id,
-                UploadStatus.PROCESSING,
+                JobStatus.PROCESSING,
                 "uploading to storage...",
                 phase="upload",
+                progress_pct=0,
             )
             try:
                 logfire.info("preparing to save audio file", filename=filename)
 
-                # define progress callback for storage upload
-                def on_upload_progress(progress_pct: float) -> None:
-                    """callback invoked during R2 upload with progress percentage."""
-                    upload_tracker.update_status(
-                        upload_id,
-                        UploadStatus.PROCESSING,
-                        f"uploading to storage... {int(progress_pct)}%",
-                        server_progress_pct=progress_pct,
-                        phase="upload",
-                    )
+                file_size = Path(file_path).stat().st_size
+
+                async with R2ProgressTracker(
+                    job_id=upload_id,
+                    total_size=file_size,
+                    message="uploading to storage...",
+                    phase="upload",
+                ) as tracker:
+                    with open(file_path, "rb") as file_obj:
+                        logfire.info("calling storage.save")
+                        file_id = await storage.save(
+                            file_obj, filename, progress_callback=tracker.on_progress
+                        )
+
+                # Final 100% update
+                await job_service.update_progress(
+                    upload_id,
+                    JobStatus.PROCESSING,
+                    "uploading to storage...",
+                    phase="upload",
+                    progress_pct=100.0,
+                )
+
+                logfire.info("storage.save completed", file_id=file_id)
 
-                with open(file_path, "rb") as file_obj:
-                    logfire.info("calling storage.save")
-                    file_id = await storage.save(
-                        file_obj, filename, progress_callback=on_upload_progress
-                    )
-                    logfire.info("storage.save completed", file_id=file_id)
             except ValueError as e:
                 logfire.error("ValueError during storage.save", error=str(e))
-                upload_tracker.update_status(
-                    upload_id, UploadStatus.FAILED, "upload failed", error=str(e)
+                await job_service.update_progress(
+                    upload_id, JobStatus.FAILED, "upload failed", error=str(e)
                 )
                 return
             except Exception as e:
···
                     error=str(e),
                     exc_info=True,
                 )
-                upload_tracker.update_status(
-                    upload_id, UploadStatus.FAILED, "upload failed", error=str(e)
+                await job_service.update_progress(
+                    upload_id, JobStatus.FAILED, "upload failed", error=str(e)
                 )
                 return
 
···
                         existing_track_id=existing_track.id,
                         artist_did=artist_did,
                     )
-                    upload_tracker.update_status(
+                    await job_service.update_progress(
                         upload_id,
-                        UploadStatus.FAILED,
+                        JobStatus.FAILED,
                         "upload failed",
                         error=f"duplicate upload: track already exists (id: {existing_track.id})",
                     )
···
             image_id = None
             image_url = None
             if image_path and image_filename:
-                upload_tracker.update_status(
+                await job_service.update_progress(
                     upload_id,
-                    UploadStatus.PROCESSING,
+                    JobStatus.PROCESSING,
                     "saving image...",
                     phase="image",
                 )
···
                 )
                 artist = result.scalar_one_or_none()
                 if not artist:
-                    upload_tracker.update_status(
+                    await job_service.update_progress(
                         upload_id,
-                        UploadStatus.FAILED,
+                        JobStatus.FAILED,
                         "upload failed",
                         error="artist profile not found",
                     )
···
                 # resolve featured artist handles
                 featured_artists = []
                 if features:
-                    upload_tracker.update_status(
+                    await job_service.update_progress(
                         upload_id,
-                        UploadStatus.PROCESSING,
+                        JobStatus.PROCESSING,
                         "resolving featured artists...",
                         phase="metadata",
                     )
···
                             "reason": reason,
                         },
                     )
-                    upload_tracker.update_status(
+                    await job_service.update_progress(
                         upload_id,
-                        UploadStatus.FAILED,
+                        JobStatus.FAILED,
                         "upload failed",
                         error=f"failed to sync track to ATProto: {reason}",
                         phase="atproto",
···
                 atproto_uri = None
                 atproto_cid = None
                 if r2_url:
-                    upload_tracker.update_status(
+                    await job_service.update_progress(
                         upload_id,
-                        UploadStatus.PROCESSING,
+                        JobStatus.PROCESSING,
                         "creating atproto record...",
                         phase="atproto",
                     )
···
                     return
 
                 # create track record
-                upload_tracker.update_status(
+                await job_service.update_progress(
                     upload_id,
-                    UploadStatus.PROCESSING,
+                    JobStatus.PROCESSING,
                     "saving track metadata...",
                     phase="database",
                 )
···
                             f"failed to send notification for track {track.id}: {e}"
                         )
 
-                    upload_tracker.update_status(
+                    await job_service.update_progress(
                         upload_id,
-                        UploadStatus.COMPLETED,
+                        JobStatus.COMPLETED,
                         "upload completed successfully",
-                        track_id=track.id,
+                        result={"track_id": track.id},
                     )
 
                 except IntegrityError as e:
                     await db.rollback()
                     # integrity errors now only occur for foreign key violations or other constraints
                     error_msg = f"database constraint violation: {e!s}"
-                    upload_tracker.update_status(
-                        upload_id, UploadStatus.FAILED, "upload failed", error=error_msg
+                    await job_service.update_progress(
+                        upload_id,
+                        JobStatus.FAILED,
+                        "upload failed",
+                        error=error_msg,
                     )
                     # cleanup: delete uploaded file
                     with contextlib.suppress(Exception):
···
 
         except Exception as e:
             logger.exception(f"upload {upload_id} failed with unexpected error")
-            upload_tracker.update_status(
+            await job_service.update_progress(
                 upload_id,
-                UploadStatus.FAILED,
+                JobStatus.FAILED,
                 "upload failed",
                 error=f"unexpected error: {e!s}",
             )
···
                         )
                     tmp_image.write(chunk)
 
-        # create upload tracking
-        upload_id = upload_tracker.create_upload()
+        # create upload tracking via JobService
+        upload_id = await job_service.create_job(
+            JobType.UPLOAD, auth_session.did, "upload queued for processing"
+        )
 
         # schedule background processing once response is sent
         background_tasks.add_task(
···
 
     async def event_stream():
         """Generate SSE events for upload progress."""
-        queue = await upload_tracker.subscribe(upload_id)
+        # Polling loop
         try:
             while True:
-                try:
-                    # wait for next update with timeout
-                    update = await asyncio.wait_for(queue.get(), timeout=30.0)
-                    yield f"data: {json.dumps(update)}\n\n"
+                job = await job_service.get_job(upload_id)
+                if not job:
+                    # Job not found or lost
+                    yield f"data: {json.dumps({'status': 'failed', 'message': 'upload job not found', 'error': 'job lost'})}\n\n"
+                    break
 
-                    # if upload completed or failed, close stream
-                    if update["status"] in ("completed", "failed"):
-                        break
+                # Construct payload matching old UploadProgress.to_dict()
+                payload = {
+                    "upload_id": job.id,
+                    "status": job.status,
+                    "message": job.message,
+                    "error": job.error,
+                    "phase": job.phase,
+                    "server_progress_pct": job.progress_pct,
+                    "created_at": job.created_at.isoformat()
+                    if job.created_at
+                    else None,
+                    "completed_at": job.completed_at.isoformat()
+                    if job.completed_at
+                    else None,
+                }
+                if job.result and "track_id" in job.result:
+                    payload["track_id"] = job.result["track_id"]
 
-                except TimeoutError:
-                    # send keepalive
-                    yield ": keepalive\n\n"
+                yield f"data: {json.dumps(payload)}\n\n"
 
-        finally:
-            upload_tracker.unsubscribe(upload_id, queue)
+                if job.status in (JobStatus.COMPLETED.value, JobStatus.FAILED.value):
+                    break
+
+                await asyncio.sleep(1.0)
+
+        except Exception as e:
+            logger.error(f"error in upload progress stream: {e}")
+            yield f"data: {json.dumps({'status': 'failed', 'message': 'connection error', 'error': str(e)})}\n\n"
 
     return StreamingResponse(
         event_stream(),

backend/src/backend/models/__init__.py

···
 from backend.models.artist import Artist
 from backend.models.database import Base
 from backend.models.exchange_token import ExchangeToken
+from backend.models.job import Job
 from backend.models.oauth_state import OAuthStateModel
 from backend.models.preferences import UserPreferences
 from backend.models.queue import QueueState
···
     "Artist",
     "Base",
     "ExchangeToken",
+    "Job",
     "OAuthStateModel",
     "QueueState",
     "Track",

backend/src/backend/models/job.py

···
+"""Job models for tracking long-running tasks."""
+
+from datetime import UTC, datetime
+from enum import Enum
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy import DateTime, Float, Index, String
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.models.database import Base
+
+
+class JobStatus(str, Enum):
+    """Status of a job."""
+
+    PENDING = "pending"
+    PROCESSING = "processing"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class JobType(str, Enum):
+    """Type of job."""
+
+    UPLOAD = "upload"
+    EXPORT = "export"
+
+
+class Job(Base):
+    """Job for tracking long-running tasks."""
+
+    __tablename__ = "jobs"
+
+    id: Mapped[str] = mapped_column(
+        String, primary_key=True, default=lambda: str(uuid4())
+    )
+    type: Mapped[str] = mapped_column(String, nullable=False)  # JobType enum
+    status: Mapped[str] = mapped_column(
+        String, nullable=False, default=JobStatus.PENDING.value
+    )  # JobStatus enum
+    owner_did: Mapped[str] = mapped_column(String, nullable=False)
+
+    # Progress
+    progress_pct: Mapped[float] = mapped_column(Float, default=0.0)
+    message: Mapped[str | None] = mapped_column(String, nullable=True)
+    phase: Mapped[str | None] = mapped_column(String, nullable=True)
+
+    # Result/Error
+    result: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
+    error: Mapped[str | None] = mapped_column(String, nullable=True)
+
+    # Metadata
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=lambda: datetime.now(UTC)
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+    )
+    completed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+
+    __table_args__ = (
+        Index("idx_jobs_owner", "owner_did"),
+        Index("idx_jobs_updated_at", "updated_at"),
+    )

backend/src/backend/utilities/progress.py

···
+"""Progress tracking utilities for long-running operations."""
+
+import asyncio
+import contextlib
+import logging
+from typing import Any
+
+from backend._internal.jobs import job_service
+from backend.models.job import JobStatus
+
+logger = logging.getLogger(__name__)
+
+
+class R2ProgressTracker:
+    """Tracks R2 upload progress and updates the job service.
+
+    Bridges the gap between boto3's synchronous callback and our async
+    database job service.
+    """
+
+    def __init__(
+        self,
+        job_id: str,
+        total_size: int,
+        message: str = "uploading to storage...",
+        phase: str = "upload",
+        update_interval: float = 1.0,
+    ):
+        self.job_id = job_id
+        self.total_size = total_size
+        self.message = message
+        self.phase = phase
+        self.update_interval = update_interval
+        self._bytes_transferred = 0
+        self._reporter_task: asyncio.Task | None = None
+
+    def on_progress(self, bytes_amount: int) -> None:
+        """Synchronous callback for boto3."""
+        self._bytes_transferred += bytes_amount
+
+    async def __aenter__(self) -> "R2ProgressTracker":
+        await self.start()
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        await self.stop()
+
+    async def start(self) -> None:
+        """Start the background reporting task."""
+        self._reporter_task = asyncio.create_task(self._report_loop())
+
+    async def stop(self) -> None:
+        """Stop the reporting task and ensure final update."""
+        if self._reporter_task:
+            self._reporter_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._reporter_task
+            self._reporter_task = None
+
+    async def _report_loop(self) -> None:
+        """Periodic reporting loop."""
+        while True:
+            pct = self.percentage
+
+            # Don't report 100% until explicitly finished by the caller
+            # via final job update, or let it sit at 99.9%
+            report_pct = min(pct, 99.9)
+
+            await job_service.update_progress(
+                self.job_id,
+                JobStatus.PROCESSING,
+                self.message,
+                phase=self.phase,
+                progress_pct=report_pct,
+            )
+
+            if pct >= 100.0:
+                break
+
+            await asyncio.sleep(self.update_interval)
+
+    @property
+    def percentage(self) -> float:
+        """Calculate current percentage."""
+        if self.total_size <= 0:
+            return 0.0 if self._bytes_transferred == 0 else 100.0
+        return (self._bytes_transferred / self.total_size) * 100.0

frontend/src/lib/uploader.svelte.ts

···
 		const uploadMessage = fileSizeMB > 10
 			? 'uploading track... (large file, this may take a moment)'
 			: 'uploading track...';
-		const toastId = toast.info(uploadMessage, 30000);
+		// 0 means infinite/persist until dismissed
+		const toastId = toast.info(uploadMessage, 0);
 
 		if (!browser) return;
 		const formData = new FormData();
···
 
 						// show backend processing messages
 						if (update.message && update.status === 'processing') {
-							// for upload phase with progress, show determinate progress
-							// for other phases, just show the message
-							toast.update(task.toastId, update.message);
+							// if we have server-side progress, show it
+							const serverProgress = update.server_progress_pct;
+							if (serverProgress !== undefined && serverProgress !== null) {
+								toast.update(task.toastId, `${update.message} (${Math.round(serverProgress)}%)`);
+							} else {
+								toast.update(task.toastId, update.message);
+							}
 						}
 
 						if (update.status === 'completed') {

frontend/src/routes/portal/+page.svelte

···
 		if (exportingMedia) return;
 
 		const trackCount = tracks.length;
-		const toastId = toast.info(`preparing export of ${trackCount} ${trackCount === 1 ? 'track' : 'tracks'}...`, 30000);
+		// 0 means infinite/persist until dismissed
+		const toastId = toast.info(`preparing export of ${trackCount} ${trackCount === 1 ? 'track' : 'tracks'}...`, 0);
 
 		exportingMedia = true;
 		try {
···
 
 				// show progress messages
 				if (update.message && update.status === 'processing') {
-					const progressInfo = update.track_count
-						? ` (${update.processed_count}/${update.track_count})`
+					const progressInfo = update.total_count
+						? ` (${update.processed_count}/${update.total_count})`
 						: '';
 					toast.update(toastId, `${update.message}${progressInfo}`);
 				}
···
 					exportingMedia = false;
 
 					// update toast to show download is starting
-					toast.update(toastId, 'downloading export...');
-
-					// trigger download
-					try {
-						const downloadResponse = await fetch(`${API_URL}/exports/${exportId}/download`, {
-							credentials: 'include'
-						});
-
-						if (downloadResponse.ok) {
-							const blob = await downloadResponse.blob();
-							const url = window.URL.createObjectURL(blob);
-							const a = document.createElement('a');
-							a.href = url;
-							a.download = `plyr-tracks-${new Date().toISOString().split('T')[0]}.zip`;
-							document.body.appendChild(a);
-							a.click();
-							window.URL.revokeObjectURL(url);
-							document.body.removeChild(a);
-
-							toast.dismiss(toastId);
-							toast.success(`${update.processed_count || trackCount} ${trackCount === 1 ? 'track' : 'tracks'} exported successfully`);
-						} else {
-							toast.dismiss(toastId);
-							toast.error('failed to download export');
-						}
-					} catch (e) {
-						console.error('download failed:', e);
+					toast.update(toastId, 'download starting...');
+					
+					if (update.download_url) {
+						// Trigger download directly from R2
+						const a = document.createElement('a');
+						a.href = update.download_url;
+						a.download = `plyr-tracks-${new Date().toISOString().split('T')[0]}.zip`;
+						document.body.appendChild(a);
+						a.click();
+						document.body.removeChild(a);
+						
+						toast.dismiss(toastId);
+						toast.success(`${update.processed_count || trackCount} ${trackCount === 1 ? 'track' : 'tracks'} exported successfully`);
+					} else {
 						toast.dismiss(toastId);
-						toast.error('failed to download export');
+						toast.error('export completed but download url missing');
 					}
 				}