forecasting/backend/api/sync.py

"""
Data Sync API endpoints
"""
import uuid
import logging
from datetime import date, timedelta
from dateutil.relativedelta import relativedelta
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Query, BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
from pydantic import BaseModel

from database import get_db, SyncSessionLocal
from auth import get_current_user

logger = logging.getLogger(__name__)
router = APIRouter()


class BackfillRequest(BaseModel):
    """Request model for backfill job"""
    source: str  # 'newbook', 'resos', or 'all'
    from_date: date
    to_date: date
    chunk_months: int = 1  # Process in monthly chunks to avoid timeouts


@router.get("/status")
async def get_sync_status(
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Get last sync times and status for all data sources and sync types.

    Returns dict keyed by "{source}_{sync_type}" (e.g., "newbook_bookings", "newbook_earned_revenue")
    """
    query = """
        SELECT DISTINCT ON (source, sync_type)
            source,
            sync_type,
            completed_at as last_sync,
            status as last_status,
            records_fetched as last_records,
            records_created as last_created,
            date_from,
            date_to
        FROM sync_log
        ORDER BY source, sync_type, completed_at DESC
    """

    result = await db.execute(text(query))
    rows = result.fetchall()

    response = {}
    for row in rows:
        # Create key like "newbook_bookings" or "newbook_earned_revenue"
        key = f"{row.source}_{row.sync_type}" if row.sync_type else row.source
        response[key] = {
            "source": row.source,
            "sync_type": row.sync_type,
            "last_sync": row.last_sync,
            "status": row.last_status,
            "records_fetched": row.last_records,
            "records_created": row.last_created,
            "date_from": row.date_from,
            "date_to": row.date_to
        }

    return response


@router.post("/newbook")
async def trigger_newbook_sync(
    background_tasks: BackgroundTasks,
    full_sync: bool = Query(False, description="If True, fetches all bookings. If False, only fetches since last sync."),
    from_date: Optional[date] = Query(None, description="Start date for stay period (filters by arrival/stay dates)"),
    to_date: Optional[date] = Query(None, description="End date for stay period (filters by arrival/stay dates)"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Trigger manual Newbook data sync.
    Runs in background to avoid timeout.

    - full_sync=False (default): Incremental sync - only bookings modified since last successful sync
    - full_sync=True: Full sync - fetches entire booking database
    - from_date/to_date: If provided, fetches bookings staying during this period (overrides full_sync)
    """
    from jobs.data_sync import sync_newbook_data

    # Queue background task
    background_tasks.add_task(
        sync_newbook_data,
        full_sync=full_sync,
        from_date=from_date,
        to_date=to_date,
        triggered_by=f"user:{current_user['username']}"
    )

    msg = "Newbook sync started in background"
    if from_date and to_date:
        msg = f"Newbook sync for {from_date} to {to_date} started in background"
    elif full_sync:
        msg = "Newbook full sync started in background"
    else:
        msg = "Newbook incremental sync started in background"

    return {
        "status": "started",
        "source": "newbook",
        "full_sync": full_sync,
        "from_date": from_date,
        "to_date": to_date,
        "message": msg
    }


@router.post("/resos")
async def trigger_resos_sync(
    background_tasks: BackgroundTasks,
    from_date: Optional[date] = Query(None, description="Start date for sync"),
    to_date: Optional[date] = Query(None, description="End date for sync"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Trigger manual Resos data sync.
    Runs in background to avoid timeout.
    """
    from jobs.data_sync import sync_resos_data

    if from_date is None:
        from_date = date.today() - timedelta(days=7)
    if to_date is None:
        to_date = date.today() + timedelta(days=365)

    # Queue background task
    background_tasks.add_task(
        sync_resos_data,
        from_date=from_date,
        to_date=to_date,
        triggered_by=f"user:{current_user['username']}"
    )

    return {
        "status": "started",
        "source": "resos",
        "from_date": from_date,
        "to_date": to_date,
        "message": "Resos sync started in background"
    }


@router.post("/newbook/occupancy-report")
async def trigger_occupancy_report_sync(
    background_tasks: BackgroundTasks,
    from_date: Optional[date] = Query(None, description="Start date for report"),
    to_date: Optional[date] = Query(None, description="End date for report"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Trigger Newbook occupancy report sync.

    This fetches Newbook's official occupancy report which provides:
    - Available rooms (accounting for maintenance/offline)
    - Official occupied room counts
    - Maintenance/offline room counts
    - Official revenue figures (gross and net)

    Use this to ensure accurate occupancy % calculations when rooms
    have been taken offline for maintenance.
    """
    from jobs.data_sync import sync_newbook_occupancy_report

    if from_date is None:
        from_date = date.today() - timedelta(days=90)
    if to_date is None:
        to_date = date.today() + timedelta(days=30)

    background_tasks.add_task(
        sync_newbook_occupancy_report,
        from_date=from_date,
        to_date=to_date,
        triggered_by=f"user:{current_user['username']}"
    )

    return {
        "status": "started",
        "source": "newbook_occupancy_report",
        "from_date": from_date,
        "to_date": to_date,
        "message": f"Newbook occupancy report sync started for {from_date} to {to_date}"
    }


@router.post("/newbook/earned-revenue")
async def trigger_earned_revenue_sync(
    background_tasks: BackgroundTasks,
    from_date: Optional[date] = Query(None, description="Start date for revenue"),
    to_date: Optional[date] = Query(None, description="End date for revenue"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Trigger Newbook earned revenue sync.

    This fetches official financial figures by GL account from Newbook's
    report_earned_revenue endpoint. Uses accommodation_gl_codes config
    to identify which GL accounts are room revenue.

    Defaults to last 7 days if no dates specified (catches adjustments).
    For historical backfill, specify a wider date range.
    """
    from jobs.data_sync import sync_newbook_earned_revenue

    if from_date is None:
        from_date = date.today() - timedelta(days=7)
    if to_date is None:
        to_date = date.today()

    background_tasks.add_task(
        sync_newbook_earned_revenue,
        from_date=from_date,
        to_date=to_date,
        triggered_by=f"user:{current_user['username']}"
    )

    return {
        "status": "started",
        "source": "newbook_earned_revenue",
        "from_date": from_date,
        "to_date": to_date,
        "message": f"Newbook earned revenue sync started for {from_date} to {to_date}"
    }


@router.post("/full")
async def trigger_full_sync(
    background_tasks: BackgroundTasks,
    full_sync: bool = Query(False, description="If True, fetches all bookings from both sources"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Trigger full sync from all sources (Newbook bookings, Newbook occupancy report, Resos).
    """
    from jobs.data_sync import run_data_sync

    background_tasks.add_task(
        run_data_sync,
        full_sync=full_sync,
        triggered_by=f"user:{current_user['username']}"
    )

    return {
        "status": "started",
        "sources": ["newbook", "newbook_occupancy_report", "resos"],
        "full_sync": full_sync,
        "message": f"Full {'complete' if full_sync else 'incremental'} sync started in background"
    }


@router.post("/aggregate")
async def trigger_aggregation(
    background_tasks: BackgroundTasks,
    source: Optional[str] = Query(None, description="Filter by source: newbook, resos. Leave empty for all."),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Trigger manual aggregation of pending dates.
    Processes the aggregation queue and updates daily_occupancy/daily_covers tables.
    """
    from jobs.aggregation import run_aggregation

    if source and source not in ['newbook', 'resos']:
        raise HTTPException(status_code=400, detail="Source must be 'newbook', 'resos', or omitted for all")

    background_tasks.add_task(
        run_aggregation,
        source=source
    )

    return {
        "status": "started",
        "source": source or "all",
        "message": f"Aggregation started for {source or 'all sources'}"
    }


@router.post("/aggregate/requeue")
async def requeue_for_aggregation(
    background_tasks: BackgroundTasks,
    source: str = Query(..., description="Source to requeue: newbook or resos"),
    from_date: Optional[date] = Query(None, description="Start date (optional, defaults to all)"),
    to_date: Optional[date] = Query(None, description="End date (optional, defaults to all)"),
    run_aggregation_after: bool = Query(True, description="Automatically run aggregation after queuing"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Re-queue dates from raw data for aggregation.

    Useful when you want to re-aggregate existing data (e.g., after changing
    mappings or fixing bugs in aggregation logic).
    """
    if source not in ['newbook', 'resos']:
        raise HTTPException(status_code=400, detail="Source must be 'newbook' or 'resos'")

    # Get distinct dates from raw data
    if source == 'resos':
        date_query = "SELECT DISTINCT booking_date as date FROM resos_bookings WHERE 1=1"
    else:  # newbook
        date_query = """
            SELECT DISTINCT stay_date as date
            FROM newbook_booking_nights bn
            JOIN newbook_bookings b ON bn.booking_id = b.id
            WHERE 1=1
        """

    params = {}
    if from_date:
        if source == 'resos':
            date_query += " AND booking_date >= :from_date"
        else:
            date_query += " AND stay_date >= :from_date"
        params["from_date"] = from_date

    if to_date:
        if source == 'resos':
            date_query += " AND booking_date <= :to_date"
        else:
            date_query += " AND stay_date <= :to_date"
        params["to_date"] = to_date

    date_query += " ORDER BY date"

    result = await db.execute(text(date_query), params)
    dates = [row.date for row in result.fetchall()]

    if not dates:
        return {
            "status": "no_data",
            "message": f"No dates found in {source} raw data for the specified range"
        }

    # Insert dates into queue (delete existing pending entries first, then insert)
    # Clear any existing pending entries for these dates
    await db.execute(
        text("""
            DELETE FROM aggregation_queue
            WHERE source = :source
              AND aggregated_at IS NULL
        """),
        {"source": source}
    )

    # Insert all dates
    for d in dates:
        await db.execute(
            text("""
                INSERT INTO aggregation_queue (date, source, reason, queued_at)
                VALUES (:date, :source, 'manual_requeue', NOW())
            """),
            {"date": d, "source": source}
        )

    await db.commit()

    # Optionally trigger aggregation
    if run_aggregation_after:
        from jobs.aggregation import run_aggregation as do_aggregation
        background_tasks.add_task(do_aggregation, source=source)

    return {
        "status": "queued",
        "source": source,
        "dates_queued": len(dates),
        "date_range": f"{min(dates)} to {max(dates)}",
        "aggregation_started": run_aggregation_after,
        "message": f"Queued {len(dates)} dates for {source} aggregation"
    }


@router.get("/aggregate/status")
async def get_aggregation_status(
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Get aggregation status summary including pending counts and totals.
    """
    # Get pending counts by source
    pending_query = """
        SELECT
            source,
            COUNT(*) as pending_count,
            MIN(date) as earliest_date,
            MAX(date) as latest_date
        FROM aggregation_queue
        WHERE aggregated_at IS NULL
        GROUP BY source
    """
    result = await db.execute(text(pending_query))
    pending_rows = result.fetchall()

    pending_by_source = {
        row.source: {
            "count": row.pending_count,
            "earliest": row.earliest_date,
            "latest": row.latest_date
        }
        for row in pending_rows
    }

    # Get total pending count
    total_pending_result = await db.execute(
        text("SELECT COUNT(*) as total FROM aggregation_queue WHERE aggregated_at IS NULL")
    )
    total_pending = total_pending_result.fetchone().total

    # Get aggregated totals
    occupancy_result = await db.execute(
        text("SELECT COUNT(*) as count, MIN(date) as earliest, MAX(date) as latest FROM daily_occupancy")
    )
    occupancy_row = occupancy_result.fetchone()

    covers_result = await db.execute(
        text("SELECT COUNT(*) as count, MIN(date) as earliest, MAX(date) as latest FROM daily_covers")
    )
    covers_row = covers_result.fetchone()

    # Get last aggregation timestamp (from most recent processed queue entry)
    last_agg_result = await db.execute(
        text("SELECT MAX(aggregated_at) as last_run FROM aggregation_queue WHERE aggregated_at IS NOT NULL")
    )
    last_agg_row = last_agg_result.fetchone()

    return {
        "pending": {
            "total": total_pending,
            "by_source": pending_by_source
        },
        "aggregated": {
            "daily_occupancy": {
                "count": occupancy_row.count if occupancy_row else 0,
                "earliest": occupancy_row.earliest if occupancy_row else None,
                "latest": occupancy_row.latest if occupancy_row else None
            },
            "daily_covers": {
                "count": covers_row.count if covers_row else 0,
                "earliest": covers_row.earliest if covers_row else None,
                "latest": covers_row.latest if covers_row else None
            }
        },
        "last_aggregation": last_agg_row.last_run if last_agg_row else None
    }


@router.get("/aggregate/queue")
async def get_aggregation_queue(
    source: Optional[str] = Query(None, description="Filter by source"),
    pending_only: bool = Query(True, description="Only show pending (un-aggregated) entries"),
    limit: int = Query(100, description="Max entries to return"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    View the aggregation queue status.
    """
    query = """
        SELECT date, source, reason, booking_id, queued_at, aggregated_at
        FROM aggregation_queue
        WHERE 1=1
    """
    params = {"limit": limit}

    if source:
        query += " AND source = :source"
        params["source"] = source

    if pending_only:
        query += " AND aggregated_at IS NULL"

    query += " ORDER BY date, source LIMIT :limit"

    result = await db.execute(text(query), params)
    rows = result.fetchall()

    return {
        "count": len(rows),
        "entries": [
            {
                "date": row.date,
                "source": row.source,
                "reason": row.reason,
                "booking_id": row.booking_id,
                "queued_at": row.queued_at,
                "aggregated_at": row.aggregated_at
            }
            for row in rows
        ]
    }


@router.get("/logs")
async def get_sync_logs(
    source: Optional[str] = Query(None, description="Filter by source: newbook, resos"),
    limit: int = Query(20, description="Number of logs to return"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Get recent sync logs for monitoring.
    """
    query = """
        SELECT
            id,
            sync_type,
            source,
            started_at,
            completed_at,
            status,
            records_fetched,
            records_created,
            records_updated,
            date_from,
            date_to,
            error_message,
            triggered_by
        FROM sync_log
    """
    params = {"limit": limit}

    if source:
        query += " WHERE source = :source"
        params["source"] = source

    query += " ORDER BY started_at DESC LIMIT :limit"

    result = await db.execute(text(query), params)
    rows = result.fetchall()

    return [
        {
            "id": row.id,
            "sync_type": row.sync_type,
            "source": row.source,
            "started_at": row.started_at,
            "completed_at": row.completed_at,
            "status": row.status,
            "records_fetched": row.records_fetched,
            "records_created": row.records_created,
            "records_updated": row.records_updated,
            "date_range": f"{row.date_from} to {row.date_to}" if row.date_from else None,
            "error_message": row.error_message,
            "triggered_by": row.triggered_by
        }
        for row in rows
    ]


# ============================================
# HISTORICAL BACKFILL ENDPOINTS
# ============================================

@router.post("/backfill")
async def start_backfill(
    request: BackfillRequest,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Start a historical data backfill job.

    Processes data in monthly chunks to avoid API timeouts and rate limits.
    Progress can be monitored via GET /sync/backfill/status/{job_id}
    """
    if request.source not in ['newbook', 'resos', 'all']:
        raise HTTPException(status_code=400, detail="Source must be 'newbook', 'resos', or 'all'")

    if request.from_date >= request.to_date:
        raise HTTPException(status_code=400, detail="from_date must be before to_date")

    # Calculate number of chunks
    total_months = (request.to_date.year - request.from_date.year) * 12 + \
                   (request.to_date.month - request.from_date.month) + 1
    chunks_total = (total_months + request.chunk_months - 1) // request.chunk_months

    # Create backfill job record
    job_id = str(uuid.uuid4())
    await db.execute(
        text("""
            INSERT INTO backfill_jobs (
                job_id, source, from_date, to_date, chunk_months,
                status, chunks_total, triggered_by, created_at
            ) VALUES (
                :job_id, :source, :from_date, :to_date, :chunk_months,
                'pending', :chunks_total, :triggered_by, NOW()
            )
        """),
        {
            "job_id": job_id,
            "source": request.source,
            "from_date": request.from_date,
            "to_date": request.to_date,
            "chunk_months": request.chunk_months,
            "chunks_total": chunks_total,
            "triggered_by": f"user:{current_user['username']}"
        }
    )
    await db.commit()

    # Queue background task
    background_tasks.add_task(
        run_backfill_job,
        job_id=job_id,
        source=request.source,
        from_date=request.from_date,
        to_date=request.to_date,
        chunk_months=request.chunk_months
    )

    return {
        "status": "started",
        "job_id": job_id,
        "source": request.source,
        "from_date": request.from_date,
        "to_date": request.to_date,
        "chunk_months": request.chunk_months,
        "chunks_total": chunks_total,
        "message": f"Backfill job started. Monitor progress at /sync/backfill/status/{job_id}"
    }


@router.get("/backfill/status/{job_id}")
async def get_backfill_status(
    job_id: str,
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    Get status of a backfill job.
    """
    result = await db.execute(
        text("""
            SELECT
                job_id, source, from_date, to_date, chunk_months,
                status, current_chunk_start, current_chunk_end,
                chunks_total, chunks_completed, records_total,
                error_message, started_at, completed_at, triggered_by
            FROM backfill_jobs
            WHERE job_id = :job_id
        """),
        {"job_id": job_id}
    )
    row = result.fetchone()

    if not row:
        raise HTTPException(status_code=404, detail="Backfill job not found")

    progress_pct = 0
    if row.chunks_total and row.chunks_total > 0:
        progress_pct = round((row.chunks_completed / row.chunks_total) * 100, 1)

    return {
        "job_id": row.job_id,
        "source": row.source,
        "date_range": f"{row.from_date} to {row.to_date}",
        "chunk_months": row.chunk_months,
        "status": row.status,
        "progress": {
            "current_chunk": f"{row.current_chunk_start} to {row.current_chunk_end}" if row.current_chunk_start else None,
            "chunks_completed": row.chunks_completed,
            "chunks_total": row.chunks_total,
            "percent_complete": progress_pct,
            "records_synced": row.records_total
        },
        "error_message": row.error_message,
        "started_at": row.started_at,
        "completed_at": row.completed_at,
        "triggered_by": row.triggered_by
    }


@router.get("/backfill/jobs")
async def list_backfill_jobs(
    status: Optional[str] = Query(None, description="Filter by status: pending, running, completed, failed"),
    limit: int = Query(20, description="Number of jobs to return"),
    db: AsyncSession = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """
    List recent backfill jobs.
    """
    query = """
        SELECT
            job_id, source, from_date, to_date, status,
            chunks_completed, chunks_total, records_total,
            started_at, completed_at
        FROM backfill_jobs
    """
    params = {"limit": limit}

    if status:
        query += " WHERE status = :status"
        params["status"] = status

    query += " ORDER BY created_at DESC LIMIT :limit"

    result = await db.execute(text(query), params)
    rows = result.fetchall()

    return [
        {
            "job_id": row.job_id,
            "source": row.source,
            "date_range": f"{row.from_date} to {row.to_date}",
            "status": row.status,
            "progress": f"{row.chunks_completed}/{row.chunks_total} chunks",
            "records_synced": row.records_total,
            "started_at": row.started_at,
            "completed_at": row.completed_at
        }
        for row in rows
    ]


async def run_backfill_job(
    job_id: str,
    source: str,
    from_date: date,
    to_date: date,
    chunk_months: int
):
    """
    Background task to run backfill in chunks.

    For Newbook: Uses modified_since/modified_until to backfill booking data by modification date.
    For Resos: Uses from_date/to_date to backfill by booking date.
    """
    from jobs.data_sync import sync_resos_data
    from services.newbook_client import NewbookClient
    import json

    db = SyncSessionLocal()

    try:
        # Mark job as running
        db.execute(
            text("""
                UPDATE backfill_jobs
                SET status = 'running', started_at = NOW()
                WHERE job_id = :job_id
            """),
            {"job_id": job_id}
        )
        db.commit()

        # Process in chunks
        current_start = from_date
        chunks_completed = 0
        total_records = 0

        while current_start <= to_date:
            # Calculate chunk end date
            current_end = current_start + relativedelta(months=chunk_months) - timedelta(days=1)
            if current_end > to_date:
                current_end = to_date

            logger.info(f"Backfill {job_id}: Processing {current_start} to {current_end}")

            # Update current chunk in job
            db.execute(
                text("""
                    UPDATE backfill_jobs
                    SET current_chunk_start = :start, current_chunk_end = :end
                    WHERE job_id = :job_id
                """),
                {"job_id": job_id, "start": current_start, "end": current_end}
            )
            db.commit()

            # Sync data for this chunk
            try:
                if source in ['newbook', 'all']:
                    # For backfill, do a full sync (no modified_since filter)
                    # This pulls all bookings - the sync job handles deduplication via upserts
                    from jobs.data_sync import sync_newbook_data, sync_newbook_occupancy_report
                    await sync_newbook_data(
                        full_sync=True,
                        triggered_by=f"backfill:{job_id}"
                    )
                    # Also backfill occupancy report for this chunk
                    # This provides available rooms, maintenance, official occupancy figures
                    await sync_newbook_occupancy_report(
                        from_date=current_start,
                        to_date=current_end,
                        triggered_by=f"backfill:{job_id}"
                    )

                    # Backfill earned revenue for this chunk (historical dates only)
                    # This provides official financial figures by GL account
                    from jobs.data_sync import sync_newbook_earned_revenue
                    # Only sync earned revenue for historical dates (not future)
                    earned_rev_end = min(current_end, date.today())
                    if current_start <= earned_rev_end:
                        await sync_newbook_earned_revenue(
                            from_date=current_start,
                            to_date=earned_rev_end,
                            triggered_by=f"backfill:{job_id}"
                        )

                if source in ['resos', 'all']:
                    await sync_resos_data(current_start, current_end, f"backfill:{job_id}")

            except Exception as chunk_error:
                logger.error(f"Backfill chunk error: {chunk_error}")
                # Continue with next chunk instead of failing entire job

            chunks_completed += 1

            # Get records synced for this chunk from sync_log
            result = db.execute(
                text("""
                    SELECT COALESCE(SUM(records_fetched), 0) as total
                    FROM sync_log
                    WHERE triggered_by = :triggered_by
                """),
                {"triggered_by": f"backfill:{job_id}"}
            )
            row = result.fetchone()
            total_records = row.total if row else 0

            # Update progress
            db.execute(
                text("""
                    UPDATE backfill_jobs
                    SET chunks_completed = :completed, records_total = :records
                    WHERE job_id = :job_id
                """),
                {"job_id": job_id, "completed": chunks_completed, "records": total_records}
            )
            db.commit()

            # Move to next chunk
            current_start = current_end + timedelta(days=1)

            # For Newbook full sync, we only need to run once (not chunked)
            if source == 'newbook':
                break

        # Run aggregation after backfill
        from jobs.aggregation import run_aggregation
        await run_aggregation()

        # Mark job as completed
        db.execute(
            text("""
                UPDATE backfill_jobs
                SET status = 'completed', completed_at = NOW()
                WHERE job_id = :job_id
            """),
            {"job_id": job_id}
        )
        db.commit()

        logger.info(f"Backfill {job_id} completed: {total_records} records synced")

    except Exception as e:
        logger.error(f"Backfill {job_id} failed: {e}")
        db.execute(
            text("""
                UPDATE backfill_jobs
                SET status = 'failed', error_message = :error, completed_at = NOW()
                WHERE job_id = :job_id
            """),
            {"job_id": job_id, "error": str(e)}
        )
        db.commit()
        raise
    finally:
        db.close()