call-center/services/recording_service/app.py

from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path
import shutil
import tempfile

from fastapi import Depends, FastAPI, File, Form, HTTPException, Query, UploadFile
from fastapi.responses import FileResponse
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from services.shared.core import Role, new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.event_bus import append_outbox_event, event_bus_enabled
from services.shared.models import HealthResponse, RecordingOut, RecordingRegisterIn
from services.shared.security import get_actor, require_roles
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import CallRecordingRow, VoiceEventRow

app = FastAPI(title="recording-service", version="1.0.0")

init_sql_schema()

ALLOWED_AUDIO_TYPES = {
    ".wav": "audio/wav",
    ".mp3": "audio/mpeg",
    ".ogg": "audio/ogg",
}


def _bool_env(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def _trusted_import_subjects() -> set[str]:
    raw = os.getenv("RECORDING_IMPORT_TRUSTED_SERVICE_SUBJECTS", "svc:asterisk-bridge")
    return {item.strip() for item in raw.split(",") if item.strip()}


def _require_import_upload_actor(actor: dict = Depends(get_actor)) -> dict:
    if actor.get("auth_source") == "service":
        if str(actor.get("sub") or "").strip() in _trusted_import_subjects():
            return actor
        raise HTTPException(status_code=403, detail="Untrusted service subject")

    if _bool_env("RECORDING_IMPORT_ALLOW_ADMIN", True) and actor.get("role") == Role.ADMIN.value:
        return actor

    raise HTTPException(status_code=403, detail="Insufficient role")


def _recordings_root() -> Path:
    base_dir = os.getenv("CC_RECORDINGS_DIR")
    if base_dir:
        root = Path(base_dir).resolve()
    else:
        data_dir = Path(os.getenv("CC_DATA_DIR", ".data")).resolve()
        root = data_dir / "recordings"
    root.mkdir(parents=True, exist_ok=True)
    return root


def _max_bytes() -> int:
    raw = os.getenv("RECORDING_MAX_BYTES", "26214400").strip()
    try:
        value = int(raw)
    except ValueError:
        value = 26214400
    return max(value, 1024)


def _safe_file_name(value: str | None, source_path: Path) -> str:
    base_name = (value or source_path.name or "recording.wav").strip()
    cleaned = Path(base_name).name.replace(" ", "-")
    return cleaned or "recording.wav"


def _normalize_audio_type(file_name: str, mime_type: str | None) -> tuple[str, str]:
    suffix = Path(file_name).suffix.lower()
    canonical = ALLOWED_AUDIO_TYPES.get(suffix)
    if not canonical:
        raise HTTPException(status_code=415, detail="Unsupported audio format")
    return canonical, suffix


def _sha256_for(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _copy_into_storage(
    *,
    source_path: Path,
    recording_id: str,
    file_name: str,
) -> tuple[Path, int, str]:
    if not source_path.exists() or not source_path.is_file():
        raise HTTPException(status_code=404, detail="Source recording not found")

    size_bytes = source_path.stat().st_size
    if size_bytes > _max_bytes():
        raise HTTPException(status_code=413, detail="Recording exceeds max size")

    day_prefix = utc_now_iso().split("T", 1)[0].split("-")
    target_dir = _recordings_root().joinpath(*day_prefix)
    target_dir.mkdir(parents=True, exist_ok=True)
    target_path = (target_dir / f"{recording_id}_{file_name}").resolve()
    shutil.copy2(source_path, target_path)
    checksum = _sha256_for(target_path)
    return target_path, size_bytes, checksum


def _to_out(row: CallRecordingRow) -> RecordingOut:
    return RecordingOut(
        recording_id=row.recording_id,
        channel="voice",
        call_id=row.call_id,
        interaction_id=row.interaction_id,
        source_event_id=row.source_event_id,
        file_name=row.file_name,
        mime_type=row.mime_type,
        size_bytes=row.size_bytes,
        duration_seconds=row.duration_seconds,
        storage_backend="local_fs",
        status=row.status,
        recorded_at=row.recorded_at,
        created_at=row.created_at,
        updated_at=row.updated_at,
        archived_at=row.archived_at,
    )


def _load_existing_by_source_event(session, source_event_id: str | None) -> CallRecordingRow | None:
    if not source_event_id:
        return None
    return session.execute(
        select(CallRecordingRow).where(CallRecordingRow.source_event_id == source_event_id)
    ).scalar_one_or_none()


def _register_recording(
    session,
    *,
    call_id: str,
    interaction_id: str | None,
    source_path: Path,
    file_name: str | None,
    mime_type: str | None,
    duration_seconds: int | None,
    recorded_at: str | None,
    source_event_id: str | None,
) -> CallRecordingRow:
    existing = _load_existing_by_source_event(session, source_event_id)
    if existing:
        return existing

    recording_id = new_id("rec")
    safe_name = _safe_file_name(file_name, source_path)
    normalized_mime, _ = _normalize_audio_type(safe_name, mime_type)
    target_path: Path | None = None
    target_path, size_bytes, checksum = _copy_into_storage(
        source_path=source_path,
        recording_id=recording_id,
        file_name=safe_name,
    )
    now = utc_now_iso()
    row = CallRecordingRow(
        recording_id=recording_id,
        channel="voice",
        call_id=call_id,
        interaction_id=interaction_id,
        source_event_id=source_event_id,
        file_name=safe_name,
        storage_backend="local_fs",
        storage_path=str(target_path),
        mime_type=normalized_mime,
        size_bytes=size_bytes,
        duration_seconds=duration_seconds,
        checksum_sha256=checksum,
        status="ready",
        recorded_at=recorded_at or now,
        created_at=now,
        updated_at=now,
        archived_at=None,
    )
    session.add(row)
    if event_bus_enabled():
        append_outbox_event(
            session,
            event_type="call.recording.ready",
            producer_service="recording-service",
            entity_type="recording",
            entity_id=recording_id,
            correlation_id=interaction_id or call_id,
            payload={
                "event": "call.recording.ready",
                "recording_id": recording_id,
                "call_id": call_id,
                "interaction_id": interaction_id,
                "source_event_id": source_event_id,
                "file_name": safe_name,
                "mime_type": normalized_mime,
                "duration_seconds": duration_seconds,
                "status": "ready",
                "recorded_at": recorded_at or now,
            },
        )
    try:
        session.commit()
    except IntegrityError:
        session.rollback()
        if target_path is not None and target_path.exists():
            target_path.unlink(missing_ok=True)
        existing = _load_existing_by_source_event(session, source_event_id)
        if existing is not None:
            return existing
        raise
    session.refresh(row)
    return row


def _get_recording_row(session, recording_id: str) -> CallRecordingRow:
    row = session.execute(
        select(CallRecordingRow).where(CallRecordingRow.recording_id == recording_id)
    ).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Recording not found")
    return row


@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
    return HealthResponse(status="ok", service="recording-service")


@app.post(
    "/recordings/import-from-voice-event/{event_id}",
    response_model=RecordingOut,
)
def import_from_voice_event(
    event_id: str,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> RecordingOut:
    session = get_session()
    try:
        voice_event = session.execute(
            select(VoiceEventRow).where(VoiceEventRow.event_id == event_id)
        ).scalar_one_or_none()
        if not voice_event:
            raise HTTPException(status_code=404, detail="Voice event not found")
        if voice_event.event_type != "recording.ready":
            raise HTTPException(status_code=400, detail="Voice event is not recording.ready")
        payload = json.loads(voice_event.payload_json or "{}")
        source_path_raw = str(payload.get("source_path") or "").strip()
        if not source_path_raw:
            raise HTTPException(status_code=400, detail="Recording source_path is required")

        row = _register_recording(
            session,
            call_id=voice_event.call_id,
            interaction_id=voice_event.interaction_id,
            source_path=Path(source_path_raw).resolve(),
            file_name=payload.get("file_name"),
            mime_type=payload.get("mime_type"),
            duration_seconds=payload.get("duration_seconds"),
            recorded_at=payload.get("recorded_at"),
            source_event_id=voice_event.event_id,
        )
        return _to_out(row)
    finally:
        session.close()


@app.post("/recordings/register", response_model=RecordingOut)
def register_recording(
    payload: RecordingRegisterIn,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> RecordingOut:
    session = get_session()
    try:
        row = _register_recording(
            session,
            call_id=payload.call_id,
            interaction_id=payload.interaction_id,
            source_path=Path(payload.source_path).resolve(),
            file_name=payload.file_name,
            mime_type=payload.mime_type,
            duration_seconds=payload.duration_seconds,
            recorded_at=payload.recorded_at,
            source_event_id=payload.source_event_id,
        )
        return _to_out(row)
    finally:
        session.close()


@app.post("/recordings/import-upload", response_model=RecordingOut)
def import_uploaded_recording(
    file: UploadFile = File(...),
    call_id: str = Form(...),
    interaction_id: str | None = Form(default=None),
    source_event_id: str | None = Form(default=None),
    file_name: str | None = Form(default=None),
    mime_type: str | None = Form(default=None),
    duration_seconds: int | None = Form(default=None),
    recorded_at: str | None = Form(default=None),
    _: dict = Depends(_require_import_upload_actor),
) -> RecordingOut:
    suffix = Path(file.filename or file_name or "recording.wav").suffix or ".wav"
    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(prefix="mvpcc-rec-", suffix=suffix, delete=False) as handle:
            shutil.copyfileobj(file.file, handle)
            temp_path = Path(handle.name)

        session = get_session()
        try:
            row = _register_recording(
                session,
                call_id=call_id,
                interaction_id=interaction_id,
                source_path=temp_path.resolve(),
                file_name=file_name or file.filename,
                mime_type=mime_type or file.content_type,
                duration_seconds=duration_seconds,
                recorded_at=recorded_at,
                source_event_id=source_event_id,
            )
            return _to_out(row)
        finally:
            session.close()
    finally:
        try:
            file.file.close()
        except Exception:  # noqa: BLE001
            pass
        if temp_path and temp_path.exists():
            temp_path.unlink(missing_ok=True)


@app.get("/recordings", response_model=list[RecordingOut])
def list_recordings(
    call_id: str | None = None,
    interaction_id: str | None = None,
    status: str | None = None,
    limit: int = Query(default=100, ge=1, le=500),
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> list[RecordingOut]:
    session = get_session()
    try:
        stmt = select(CallRecordingRow).order_by(CallRecordingRow.id.desc())
        if call_id:
            stmt = stmt.where(CallRecordingRow.call_id == call_id)
        if interaction_id:
            stmt = stmt.where(CallRecordingRow.interaction_id == interaction_id)
        if status:
            stmt = stmt.where(CallRecordingRow.status == status)
        rows = session.execute(stmt.limit(max(limit, 1))).scalars().all()
        return [_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/recordings/{recording_id}", response_model=RecordingOut)
def get_recording(
    recording_id: str,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> RecordingOut:
    session = get_session()
    try:
        return _to_out(_get_recording_row(session, recording_id))
    finally:
        session.close()


@app.get("/recordings/{recording_id}/content")
def get_recording_content(
    recording_id: str,
    disposition: str = Query(default="inline"),
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> FileResponse:
    if disposition not in {"inline", "attachment"}:
        raise HTTPException(status_code=400, detail="Invalid disposition")

    session = get_session()
    try:
        row = _get_recording_row(session, recording_id)
        path = Path(row.storage_path)
        if not path.exists() or not path.is_file():
            row.status = "missing"
            row.updated_at = utc_now_iso()
            session.commit()
            raise HTTPException(status_code=410, detail="Recording file is missing")
        return FileResponse(
            path=path,
            media_type=row.mime_type,
            filename=row.file_name,
            content_disposition_type=disposition,
        )
    finally:
        session.close()


@app.post("/recordings/{recording_id}/archive", response_model=RecordingOut)
def archive_recording(
    recording_id: str,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> RecordingOut:
    session = get_session()
    try:
        row = _get_recording_row(session, recording_id)
        now = utc_now_iso()
        row.status = "archived"
        row.archived_at = now
        row.updated_at = now
        session.commit()
        session.refresh(row)
        return _to_out(row)
    finally:
        session.close()