app/api/ingest.py

"""Author RAG — Ingestion API Router.

Routes (per implementation plan):
  POST /api/admin/{author_slug}/books/upload          — Upload a book file
  GET  /api/admin/{author_slug}/books/upload/progress — SSE ingestion progress

Saves uploaded file to disk first, then passes the path to ingestion_service.
"""

import json

from fastapi import APIRouter, Depends, File, Query, UploadFile, HTTPException
from fastapi.responses import StreamingResponse

from app.config import get_settings
from app.dependencies import get_db, get_current_author_scoped, get_redis
from app.services.ingestion_service import start_file_upload

router = APIRouter()
cfg = get_settings()


@router.post("/{author_slug}/books/upload", status_code=202)
async def upload_book(
    author_slug: str,
    file: UploadFile = File(...),
    title: str = "",
    genre: str = "",
    buy_url: str = "",
    force_replace: bool = Query(False),
    current_user=Depends(get_current_author_scoped),
    db=Depends(get_db),
    redis=Depends(get_redis),
):
    """Upload a book file (PDF/EPUB/DOCX/TXT) and start async ingestion.

    MINOR-7 fix: removed 'price' param — start_file_upload() never accepted it
    (it was silently dropped on every request).
    """
    MAX_UPLOAD_BYTES = cfg.UPLOAD_MAX_FILE_SIZE_MB * 1024 * 1024
    contents = bytearray()
    while chunk := await file.read(64 * 1024):
        contents.extend(chunk)
        if len(contents) > MAX_UPLOAD_BYTES:
            raise HTTPException(413, f"File too large. Maximum: {cfg.UPLOAD_MAX_FILE_SIZE_MB}MB")
    filename = file.filename or "upload"

    return await start_file_upload(
        db=db,
        author_id=current_user.id,
        contents=bytes(contents),
        filename=filename,
        title=title,
        genre=genre,
        buy_url=buy_url,
        redis=redis,
        force_replace=force_replace,
    )


@router.get("/{author_slug}/books/upload/progress")
async def ingestion_progress(
    author_slug: str,
    book_id: str,
    current_user=Depends(get_current_author_scoped),
    redis=Depends(get_redis),
):
    """SSE stream for real-time ingestion progress."""
    async def event_stream():
        channel = f"ingestion:{current_user.id}:{book_id}"
        pubsub = redis.pubsub()
        await pubsub.subscribe(channel)
        try:
            import time as _time
            deadline = _time.monotonic() + 300
            heartbeat_interval = 30
            last_heartbeat = _time.monotonic()
            async for message in pubsub.listen():
                now = _time.monotonic()
                if now >= deadline:
                    yield 'data: {"stage": "timeout"}\n\n'
                    break
                if now - last_heartbeat > heartbeat_interval:
                    yield ": heartbeat\n\n"
                    last_heartbeat = now
                if message["type"] == "message":
                    yield f"data: {message['data']}\n\n"
                    data = json.loads(message["data"])
                    if data.get("stage") in ("complete", "error"):
                        break
        finally:
            await pubsub.unsubscribe(channel)

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )