"""Celery tasks.

Tasks open their own database session (via the configured factory) since they run
outside the request lifecycle.
"""

from __future__ import annotations

import asyncio
import uuid
from datetime import UTC, datetime

from legate.db.session import get_session_factory
from legate.rag.service import ingest_now
from legate.workers.celery_app import celery_app
from legate.workflow.service import run_due_workflows


@celery_app.task(name="legate.rag.ingest_document")
def ingest_document_task(document_id: str) -> str:
    """Ingest a document by id. Returns the document's terminal status."""
    ingest_now(uuid.UUID(document_id), get_session_factory())
    return "done"


@celery_app.task(name="legate.workflow.scheduler")
def run_scheduled_workflows_task() -> int:
    """Beat task: run every scheduled workflow whose cron is due."""
    started = asyncio.run(run_due_workflows(get_session_factory(), datetime.now(UTC)))
    return len(started)
