Bugfix/user celery jobs

README.md

@@ -152,20 +152,20 @@ Controls Docker behavior.
 
 ```bash
 # Backend tests
-docker-compose exec api-dev poetry run pytest
+docker-compose exec api poetry run pytest
 
 # With coverage
-docker-compose exec api-dev poetry run pytest --cov=app
+docker-compose exec api poetry run pytest --cov=app
 ```
 
 ## Database Migrations
 
 ```bash
 # Create migration
-docker-compose exec api-dev poetry run alembic revision --autogenerate -m "description"
+docker-compose exec api poetry run alembic revision --autogenerate -m "description"
 
 # Apply migrations
-docker-compose exec api-dev poetry run alembic upgrade head
+docker-compose exec api poetry run alembic upgrade head
 ```
 
 ## License

backend/Dockerfile

@@ -12,6 +12,8 @@ FROM base AS dev
 RUN poetry install --no-interaction --no-ansi
 COPY ./app ./app
 COPY alembic.ini ./
+COPY ./scripts ./scripts
+RUN chmod +x /backend/scripts/start.sh
 
 ENV HOME=/tmp
 EXPOSE 8000
@@ -22,12 +24,13 @@ RUN addgroup --system appgroup && adduser --system --group appuser
 RUN poetry install --without dev --no-interaction --no-ansi
 COPY ./app ./app
 COPY alembic.ini ./
+COPY ./scripts ./scripts
 
-RUN printf '#!/bin/bash\nset -e\necho "Running database migrations..."\npoetry run alembic upgrade head\necho "Starting server..."\nexec "$@"\n' > /backend/entrypoint.sh && chmod +x /backend/entrypoint.sh
+RUN chmod +x /backend/scripts/start.sh
 RUN chown -R appuser:appgroup /backend
 # Set HOME for appuser and ensure Poetry doesn't create virtualenvs
 ENV HOME=/tmp
 ENV POETRY_VIRTUALENVS_CREATE=false
 USER appuser
 EXPOSE 8000
-ENTRYPOINT ["/backend/entrypoint.sh"]
+ENTRYPOINT ["/backend/scripts/start.sh"]

backend/app/alembic/versions/4830093f7fc4_add_celery_job_structure.py

@@ -0,0 +1,43 @@
+"""add celery job structure
+
+Revision ID: 4830093f7fc4
+Revises: fb4f1d257e34
+Create Date: 2026-01-05 21:34:02.734533
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+# revision identifiers, used by Alembic.
+revision: str = '4830093f7fc4'
+down_revision: Union[str, None] = 'fb4f1d257e34'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('celery_jobs',
+    sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+    sa.Column('task_name', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+    sa.Column('user_id', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_celery_jobs_task_name'), 'celery_jobs', ['task_name'], unique=False)
+    op.create_index(op.f('ix_celery_jobs_user_id'), 'celery_jobs', ['user_id'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_celery_jobs_user_id'), table_name='celery_jobs')
+    op.drop_index(op.f('ix_celery_jobs_task_name'), table_name='celery_jobs')
+    op.drop_table('celery_jobs')
+    # ### end Alembic commands ###

backend/app/api/routers/tasks.py

@@ -7,8 +7,9 @@
 from app.celery_app import celery_app
 from app.core.auth import get_current_user_id
 from app.core.db import get_db
-from app.crud import task_crud, temp_upload_crud
+from app.crud import celery_job_crud, task_crud, temp_upload_crud
 from app.crud.setting_crud import get_user_setting, get_user_timezone
+from app.models.celery_job import CeleryJob
 from app.models.task import Task
 from app.models.temp_upload import TempUpload
 from app.schemas.job import IngestTaskJob, JobStatus
@@ -63,6 +64,10 @@ async def ingest_file(
         user_id=user_id,
     )
 
+    celery_job_crud.create_celery_job(
+        CeleryJob(id=str(job.id), task_name="ingest_file", user_id=user_id), session
+    )
+
     return JobResponse(job_id=str(job.id))
 
 
@@ -79,6 +84,10 @@ async def ingest_text(
         text=text_request.text, language=language, user_id=user_id
     )
 
+    celery_job_crud.create_celery_job(
+        CeleryJob(id=str(job.id), task_name="ingest_text", user_id=user_id), session
+    )
+
     return JobResponse(job_id=str(job.id))
 
 
@@ -205,24 +214,22 @@ async def deschedule_tasks(
 @router.get("/jobs/{job_id}", status_code=status.HTTP_200_OK)
 async def get_job_status(
     job_id: str,
-    _user_id: int = Depends(get_current_user_id),
+    user_id: int = Depends(get_current_user_id),
+    session: Session = Depends(get_db),
 ) -> IngestTaskJob:
     """Get the status of a Celery job."""
+    _celery_job = celery_job_crud.get_celery_job(job_id, user_id, session)
     task_result: AsyncResult[dict[str, Any]] = AsyncResult(job_id, app=celery_app)
-
     # Map Celery states to our JobStatus enum
-    celery_state = task_result.state
-    if celery_state == "PENDING":
-        status_enum = JobStatus.PENDING
-    elif celery_state in ("STARTED", "RETRY"):
-        status_enum = JobStatus.RUNNING
-    elif celery_state == "SUCCESS":
-        status_enum = JobStatus.SUCCESS
-    elif celery_state in ("FAILURE", "REVOKED"):
-        status_enum = JobStatus.FAILED
-    else:
-        status_enum = JobStatus.PENDING
-
+    state_mapping: dict[str, JobStatus] = {
+        "PENDING": JobStatus.PENDING,
+        "STARTED": JobStatus.RUNNING,
+        "RETRY": JobStatus.RUNNING,
+        "SUCCESS": JobStatus.SUCCESS,
+        "FAILURE": JobStatus.FAILED,
+        "REVOKED": JobStatus.FAILED,
+    }
+    status_enum = state_mapping.get(task_result.state, JobStatus.PENDING)
     result: IngestTaskResponse | None = None
     error: str | None = None
 
@@ -242,22 +249,3 @@ async def get_job_status(
         result=result,
         error=error,
     )
-
-
-@router.get("/jobs", status_code=status.HTTP_200_OK)
-async def get_active_jobs(
-    _user_id: int = Depends(get_current_user_id),
-) -> dict[str, Any]:
-    """Get information about active Celery tasks (monitoring endpoint)."""
-    inspect = celery_app.control.inspect()
-
-    active = inspect.active() or {}
-    scheduled = inspect.scheduled() or {}
-    reserved = inspect.reserved() or {}
-
-    return {
-        "active": active,
-        "scheduled": scheduled,
-        "reserved": reserved,
-        "stats": inspect.stats() or {},
-    }

backend/app/crud/celery_job_crud.py

@@ -0,0 +1,22 @@
+from sqlmodel import Session, select
+
+from app.core.exceptions import NotFoundError
+from app.models.celery_job import CeleryJob
+
+
+def create_celery_job(celery_job: CeleryJob, session: Session) -> CeleryJob:
+    session.add(celery_job)
+    session.commit()
+    session.refresh(celery_job)
+    return celery_job
+
+
+def get_celery_job(job_id: str, user_id: int, session: Session) -> CeleryJob:
+    job = session.exec(
+        select(CeleryJob)
+        .where(CeleryJob.id == job_id)
+        .where(CeleryJob.user_id == user_id)
+    ).first()
+    if not job:
+        raise NotFoundError(f"Celery job with id {job_id} not found")
+    return job

backend/app/models/__init__.py

@@ -1,4 +1,5 @@
 from .availability import DailyWindowModel, WeeklyAvailability
+from .celery_job import CeleryJob
 from .schedule_item import ScheduleItem
 from .task import Task
 from .temp_upload import TempUpload
@@ -13,4 +14,5 @@
     "ScheduleItem",
     "UserSetting",
     "TempUpload",
+    "CeleryJob",
 ]

backend/app/models/celery_job.py

@@ -0,0 +1,14 @@
+import datetime as dt
+
+from sqlmodel import Field, SQLModel
+
+from app.core.timezone import now_utc
+
+
+class CeleryJob(SQLModel, table=True):
+    __tablename__ = "celery_jobs"  # type: ignore[assignment]
+
+    id: str = Field(primary_key=True)
+    task_name: str = Field(index=True)
+    user_id: int = Field(index=True)
+    created_at: dt.datetime = Field(default_factory=now_utc)

backend/scripts/start.sh

@@ -1,6 +1,8 @@
 #!/bin/bash
 set -e
 
+chmod +x "$0"
+
 echo "Running database migrations..."
 poetry run alembic upgrade head
 

docker-compose.override.yml

@@ -5,10 +5,9 @@ services:
     build:
       target: dev
     user: "${UID:-1000}:${GID:-1000}"
+    entrypoint: ["/backend/scripts/start.sh"]
     command:
       [
-        "/bin/bash",
-        "./scripts/start.sh",
         "uvicorn",
         "app.main:app",
         "--host",
@@ -37,6 +36,7 @@ services:
       ["poetry", "run", "celery", "-A", "app.celery_app", "worker", "--loglevel=info"]
     environment:
       HOME: /tmp
+      POETRY_VIRTUALENVS_CREATE: "false"
     volumes:
       - ./backend/app:/backend/app
       - ./backend/pyproject.toml:/backend/pyproject.toml

frontend/src/context/job-context.tsx

@@ -10,7 +10,8 @@ import {
   useRef,
 } from "react";
 import { z } from "zod";
-import { apiRequest } from "@/lib/chrono-client";
+import { apiRequest, ApiError } from "@/lib/chrono-client";
+import { useAuth } from "./auth-context";
 
 const jobResponseSchema = z.object({
   job_id: z.string(),
@@ -53,9 +54,10 @@ export const JobContext = createContext<JobContextType | undefined>(undefined);
 export function JobProvider({ children }: { children: React.ReactNode }) {
   const [isHydrated, setIsHydrated] = useState(false);
   const toastShownRef = useRef<Set<string>>(new Set());
+  const { isAuthenticated } = useAuth();
 
   const getInitialJobs = (): TrackedJob[] => {
-    if (typeof window === "undefined") {
+    if (typeof window === "undefined" || !isAuthenticated) {
       return [];
     }
     const storedJobs = localStorage.getItem("background-jobs");
@@ -65,14 +67,21 @@ export function JobProvider({ children }: { children: React.ReactNode }) {
     setIsHydrated(true);
     return [];
   };
-
   const [jobs, setJobs] = useState<TrackedJob[]>(() => getInitialJobs());
+
   useEffect(() => {
     if (isHydrated && typeof window !== "undefined") {
       localStorage.setItem("background-jobs", JSON.stringify(jobs));
     }
   }, [jobs, isHydrated]);
 
+  useEffect(() => {
+    if (!isAuthenticated) {
+      localStorage.removeItem("background-jobs");
+      toastShownRef.current.clear();
+    }
+  }, [isAuthenticated]);
+
   const dismissJob = useCallback((jobId: string) => {
     setJobs((prevJobs) => prevJobs.filter((job) => job.id !== jobId));
     // Clean up toast tracking when job is dismissed
@@ -143,8 +152,15 @@ export function JobProvider({ children }: { children: React.ReactNode }) {
               updateJobStatus(job, jobStatus);
             }
           } catch (error) {
-            console.error("Failed to fetch job status:", error);
-            // Don't throw - just log the error
+            // If job doesn't belong to current user (403) or doesn't exist (404), remove it
+            if (
+              error instanceof ApiError &&
+              (error.status === 403 || error.status === 404)
+            ) {
+              setJobs((prevJobs) => prevJobs.filter((j) => j.id !== job.id));
+            } else {
+              console.error("Failed to fetch job status:", error);
+            }
           }
         })
       );