diff --git a/README.md b/README.md index 8e4cd91..5f4211c 100644 --- a/README.md +++ b/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 diff --git a/backend/Dockerfile b/backend/Dockerfile index 0476340..3f5dd10 100644 --- a/backend/Dockerfile +++ b/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"] \ No newline at end of file +ENTRYPOINT ["/backend/scripts/start.sh"] \ No newline at end of file diff --git a/backend/app/alembic/versions/4830093f7fc4_add_celery_job_structure.py b/backend/app/alembic/versions/4830093f7fc4_add_celery_job_structure.py new file mode 100644 index 0000000..a4be7e8 --- /dev/null +++ b/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 ### diff --git a/backend/app/api/routers/tasks.py b/backend/app/api/routers/tasks.py index 38c8f09..d18c37a 100644 --- a/backend/app/api/routers/tasks.py +++ b/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 {}, - } diff --git a/backend/app/crud/celery_job_crud.py b/backend/app/crud/celery_job_crud.py new file mode 100644 index 0000000..2efd817 --- /dev/null +++ b/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 diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 3f2c977..178ba26 100644 --- a/backend/app/models/__init__.py +++ b/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", ] diff --git a/backend/app/models/celery_job.py b/backend/app/models/celery_job.py new file mode 100644 index 0000000..2345cf6 --- /dev/null +++ b/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) diff --git a/backend/scripts/start.sh b/backend/scripts/start.sh old mode 100644 new mode 100755 index 5ce1bf9..be35574 --- a/backend/scripts/start.sh +++ b/backend/scripts/start.sh @@ -1,6 +1,8 @@ #!/bin/bash set -e +chmod +x "$0" + echo "Running database migrations..." poetry run alembic upgrade head diff --git a/docker-compose.override.yml b/docker-compose.override.yml index bbd6dd4..396b1f5 100644 --- a/docker-compose.override.yml +++ b/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 diff --git a/frontend/src/context/job-context.tsx b/frontend/src/context/job-context.tsx index 10e3dd2..a351ca9 100644 --- a/frontend/src/context/job-context.tsx +++ b/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(undefined); export function JobProvider({ children }: { children: React.ReactNode }) { const [isHydrated, setIsHydrated] = useState(false); const toastShownRef = useRef>(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(() => 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); + } } }) );