src/raw_audio.py

"""Optional encrypted recording retention. No public audio URLs or exports.

The main research receipt never implicitly authorizes recording retention.
This supplementary policy needs its own affirmative, server-recorded consent.
"""
from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import os
import secrets
from datetime import datetime, timedelta, timezone
from typing import Annotated, Literal

from fastapi import APIRouter, Header, HTTPException, Request, Response
from pydantic import Field

from src.collection_v2 import (
    IdempotencyKey, ParticipantToken, StrictModel, get_store, _new_id,
    _utc_now, _raise_api_error, _request_hash,
)
from src.consent_policy import get_current_consent_policy

router = APIRouter(prefix="/api/v2/raw-audio", tags=["private-raw-audio"])
MAX_BYTES = 8 * 1024 * 1024
MAX_PARTICIPANT_BYTES = 128 * 1024 * 1024
POLICY_VERSION = "private-recording-game-session-2026-10-06"
SESSION_HOURS = 12
ParticipantId = Annotated[str, Header(alias="X-Participant-ID", pattern=r"^participant_[0-9a-f]{32}$")]
AudioConsentId = Annotated[str, Header(alias="X-Audio-Consent-ID", pattern=r"^acn_[0-9a-f]{32}$")]
AudioSessionId = Annotated[str, Header(alias="X-Audio-Session-ID", pattern=r"^asn_[0-9a-f]{32}$")]
GameId = Annotated[str, Header(alias="X-Game-ID", min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")]


def configuration():
    try:
        enabled = os.getenv("RAW_AUDIO_COLLECTION_ENABLED", "").lower() == "true"
        days = int(os.getenv("RAW_AUDIO_RETENTION_DAYS", "0"))
        key = base64.b64decode(os.getenv("RAW_AUDIO_ENCRYPTION_KEY", ""), validate=True)
        from cryptography.hazmat.primitives.ciphers.aead import AESGCM
        if not enabled or not 1 <= days <= 3650 or len(key) != 32 or not os.getenv("COLLECTION_DB_PATH", "").strip():
            return None
        return days, AESGCM(key)
    except (ValueError, ImportError):
        return None


def policy():
    configured = configuration()
    days = configured[0] if configured else None
    notice = (
        f"I separately authorize private storage of recordings for speech, dialect, "
        f"and acoustic research for up to {days} days from upload, or earlier if I withdraw. "
        "Before each game and after returning from closing, minimizing, or backgrounding the app, "
        "I choose whether to store completed recordings automatically for that game session. "
        "This applies only to recordings I start with a game microphone; it does not turn on the microphone "
        "or authorize background storage. No per-recording Save confirmation follows. "
        f"Session permission expires after {SESSION_HOURS} hours at the latest and can be stopped sooner. "
        "I acknowledge that my voice may identify me or reveal sensitive traits. "
        "Only my own voice may be submitted through this option. Recordings are not public, "
        "are not covered by the text dataset's public-release permission, and will not be used "
        "to identify people, clone voices, or train models under this permission. "
        "I can decline without losing access to games or text contribution. "
        "Withdrawal removes active recording payloads; any backup erasure must be separately tracked."
    ) if configured else "Private recording storage is not configured. No recordings will be retained."
    canonical = json.dumps({"version": POLICY_VERSION, "retention_days": days, "notice": notice}, sort_keys=True, separators=(",", ":"))
    return {"enabled": configured is not None, "version": POLICY_VERSION, "retention_days": days,
            "notice": notice, "canonical_text": canonical,
            "policy_sha256": hashlib.sha256(canonical.encode()).hexdigest(), "max_bytes": MAX_BYTES}


def require_configuration():
    value = configuration()
    if value is None:
        raise HTTPException(503, "Private recording retention is not enabled and configured")
    return value


def authenticate(store, connection, participant_id, participant_token, *, active=True):
    store._authenticate_participant(connection, participant_id, participant_token)
    if active:
        store._ensure_active(connection, participant_id)


def require_audio_consent(store, connection, audio_consent_id, participant_id):
    consent = connection.execute("SELECT * FROM raw_audio_consents WHERE audio_consent_id = ? AND participant_id = ?",
                                 (audio_consent_id, participant_id)).fetchone()
    if consent is None or consent["revoked_at"]:
        raise HTTPException(403, "An active recording-storage consent receipt is required")
    if consent["policy_sha256"] != policy()["policy_sha256"]:
        raise HTTPException(403, "Review the current recording-storage policy before uploading")
    store._require_consent(connection, consent["consent_id"], participant_id, "speech", consent["app_id"])
    require_current_base_consent(connection, participant_id, consent["app_id"], consent["consent_id"])
    return consent


def require_current_base_consent(connection, participant_id, app_id, consent_id):
    latest = connection.execute("SELECT consent_id FROM consents WHERE participant_id = ? AND lower(app_id) = lower(?) "
                                "ORDER BY rowid DESC LIMIT 1", (participant_id, app_id)).fetchone()
    if latest is None or latest["consent_id"] != consent_id:
        raise HTTPException(403, "Research choices changed; renew recording permission under the latest receipt")


def purge_expired():
    store = get_store()
    store.initialize()
    with store._connect() as connection:
        connection.execute("PRAGMA secure_delete = ON")
        connection.execute("UPDATE raw_audio_recordings SET ciphertext = NULL, nonce = NULL, erased_at = ?, "
                           "erasure_reason = 'expired' WHERE erased_at IS NULL AND expires_at <= ?", (_utc_now(), _utc_now()))


def install_retention_worker(app):
    @app.on_event("startup")
    async def start_audio_retention():
        if not os.getenv("COLLECTION_DB_PATH", "").strip():
            return
        async def sweep():
            while True:
                try:
                    await asyncio.to_thread(purge_expired)
                except Exception:
                    # No recording content, participant identifiers, paths or secrets in logs.
                    print("Private recording expiry sweep failed; retry scheduled.")
                await asyncio.sleep(60)
        app.state.raw_audio_retention_task = asyncio.create_task(sweep())

    @app.on_event("shutdown")
    async def stop_audio_retention():
        task = getattr(app.state, "raw_audio_retention_task", None)
        if task:
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass


class AudioConsentCreate(StrictModel):
    participant_id: str = Field(pattern=r"^participant_[0-9a-f]{32}$")
    consent_id: str = Field(pattern=r"^cns_[0-9a-f]{32}$")
    app_id: str = Field(min_length=1, max_length=64)
    policy_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    authorize_recording_storage: Literal[True]
    voice_privacy_risks_acknowledged: Literal[True]


class AudioSessionCreate(StrictModel):
    audio_consent_id: str = Field(pattern=r"^acn_[0-9a-f]{32}$")
    game_id: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")
    automatic_storage_confirmed: Literal[True]
    own_voice_only_confirmed: Literal[True]


def require_supported_game(game_id):
    if game_id.upper() in {"HOME", "BRIDGE", "CONVO-BRIDGE", "CONVO_BRIDGE", "UNKNOWN"}:
        raise HTTPException(403, "Shared-device and unassigned recordings cannot be retained")


def require_audio_session(connection, session_id, participant_id, consent_id, game_id):
    row = connection.execute("SELECT * FROM raw_audio_sessions WHERE audio_session_id = ?", (session_id,)).fetchone()
    if (row is None or row["participant_id"] != participant_id or row["audio_consent_id"] != consent_id
            or row["game_id"] != game_id or row["ended_at"] or row["expires_at"] <= _utc_now()):
        raise HTTPException(403, "Confirm recording storage for this game session before uploading")
    return row


@router.post("/sessions", status_code=201)
def create_audio_session(payload: AudioSessionCreate, participant_id: ParticipantId,
                         participant_token: ParticipantToken, idempotency_key: IdempotencyKey):
    require_configuration()
    require_supported_game(payload.game_id)
    store = get_store(); store.initialize()
    try:
        with store._connect() as connection:
            store._begin(connection)
            authenticate(store, connection, participant_id, participant_token)
            require_audio_consent(store, connection, payload.audio_consent_id, participant_id)
            request_hash = _request_hash({**payload.model_dump(), "participant_id": participant_id})
            replay = store._idempotent_replay(connection, "audio_session", idempotency_key, request_hash)
            if replay:
                require_audio_session(connection, replay["audio_session_id"], participant_id, payload.audio_consent_id, payload.game_id)
                return replay
            now = datetime.now(timezone.utc)
            receipt = {"audio_session_id": _new_id("asn"), "participant_id": participant_id,
                       "audio_consent_id": payload.audio_consent_id, "game_id": payload.game_id,
                       "created_at": now.isoformat(timespec="milliseconds").replace("+00:00", "Z"),
                       "expires_at": (now + timedelta(hours=SESSION_HOURS)).isoformat(timespec="milliseconds").replace("+00:00", "Z")}
            connection.execute("INSERT INTO raw_audio_sessions VALUES (?, ?, ?, ?, ?, ?, ?, NULL)",
                               (receipt["audio_session_id"], participant_id, payload.audio_consent_id, payload.game_id,
                                json.dumps(payload.model_dump(), sort_keys=True), receipt["created_at"], receipt["expires_at"]))
            store._record_idempotency(connection, "audio_session", idempotency_key, request_hash,
                                      "audio_session", receipt["audio_session_id"], receipt, receipt["created_at"])
            connection.commit()
            return receipt
    except Exception as exc:
        _raise_api_error(exc)


@router.delete("/sessions/{audio_session_id}")
def end_audio_session(audio_session_id: str, participant_id: ParticipantId, participant_token: ParticipantToken):
    store = get_store(); store.initialize()
    try:
        with store._connect() as connection:
            store._begin(connection)
            authenticate(store, connection, participant_id, participant_token, active=False)
            row = connection.execute("SELECT 1 FROM raw_audio_sessions WHERE audio_session_id = ? AND participant_id = ?",
                                     (audio_session_id, participant_id)).fetchone()
            if row is None: raise HTTPException(404, "Recording session not found")
            connection.execute("UPDATE raw_audio_sessions SET ended_at = COALESCE(ended_at, ?) WHERE audio_session_id = ?",
                               (_utc_now(), audio_session_id))
            connection.commit()
        return {"audio_session_id": audio_session_id, "status": "ended"}
    except Exception as exc:
        _raise_api_error(exc)


@router.get("/policy")
def get_audio_policy():
    return policy()


@router.post("/consents", status_code=201)
def create_audio_consent(payload: AudioConsentCreate, participant_token: ParticipantToken, idempotency_key: IdempotencyKey):
    require_configuration()
    statement = policy()
    if payload.policy_sha256 != statement["policy_sha256"]:
        raise HTTPException(409, "Recording-storage policy changed; review it again")
    store = get_store()
    store.initialize()
    try:
        with store._connect() as connection:
            store._begin(connection)
            authenticate(store, connection, payload.participant_id, participant_token)
            base = store._require_consent(connection, payload.consent_id, payload.participant_id, "speech", payload.app_id)
            require_current_base_consent(connection, payload.participant_id, payload.app_id, payload.consent_id)
            if base["consent_version"] != get_current_consent_policy()["consent_version"]:
                raise HTTPException(403, "Renew the research consent under the current policy first")
            request_hash = _request_hash(payload.model_dump())
            replay = store._idempotent_replay(connection, "audio_consent", idempotency_key, request_hash)
            if replay:
                # Do not revive a revoked receipt through an idempotent retry.
                require_audio_consent(store, connection, replay["audio_consent_id"], payload.participant_id)
                return replay
            receipt = {"audio_consent_id": _new_id("acn"), "participant_id": payload.participant_id,
                       "consent_id": payload.consent_id, "policy_sha256": statement["policy_sha256"],
                       "retention_days": statement["retention_days"], "created_at": _utc_now()}
            connection.execute("INSERT INTO raw_audio_consents VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL)",
                               (receipt["audio_consent_id"], payload.participant_id, payload.consent_id, payload.app_id,
                                statement["policy_sha256"], statement["canonical_text"], statement["retention_days"], receipt["created_at"]))
            store._record_idempotency(connection, "audio_consent", idempotency_key, request_hash, "audio_consent",
                                      receipt["audio_consent_id"], receipt, receipt["created_at"])
            connection.commit()
            return receipt
    except Exception as exc:
        _raise_api_error(exc)


def detect_audio(data, media_type):
    media_type = media_type.split(";", 1)[0].strip().lower()
    valid = ((media_type in {"audio/wav", "audio/x-wav"} and len(data) >= 44 and data[:4] == b"RIFF" and data[8:12] == b"WAVE")
             or (media_type == "audio/webm" and data[:4] == b"\x1aE\xdf\xa3")
             or (media_type == "audio/ogg" and data[:4] == b"OggS")
             or (media_type == "audio/mp4" and len(data) >= 12 and data[4:8] == b"ftyp"))
    if not valid:
        raise HTTPException(415, "A supported WAV, WebM, Ogg or MP4 audio recording is required")
    return media_type


@router.post("/recordings", status_code=201)
async def upload_recording(request: Request, participant_id: ParticipantId, participant_token: ParticipantToken,
                           audio_consent_id: AudioConsentId, game_id: GameId, idempotency_key: IdempotencyKey,
                           audio_session_id: AudioSessionId):
    days, cipher = require_configuration()
    require_supported_game(game_id)
    store = get_store()
    store.initialize()
    try:
        with store._connect() as connection:
            authenticate(store, connection, participant_id, participant_token)
            require_audio_consent(store, connection, audio_consent_id, participant_id)
            require_audio_session(connection, audio_session_id, participant_id, audio_consent_id, game_id)
        data = bytearray()
        async for chunk in request.stream():
            if len(data) + len(chunk) > MAX_BYTES:
                raise HTTPException(413, "Recording exceeds the size limit")
            data.extend(chunk)
        media_type = detect_audio(data, request.headers.get("content-type", ""))
        content_hash = hashlib.sha256(data).hexdigest()
        request_hash = _request_hash({"participant_id": participant_id, "audio_consent_id": audio_consent_id,
                                     "audio_session_id": audio_session_id, "game_id": game_id,
                                     "media_type": media_type, "content_hash": content_hash})
        with store._connect() as connection:
            store._begin(connection)
            authenticate(store, connection, participant_id, participant_token)
            require_audio_consent(store, connection, audio_consent_id, participant_id)
            require_audio_session(connection, audio_session_id, participant_id, audio_consent_id, game_id)
            replay = store._idempotent_replay(connection, "audio_upload", idempotency_key, request_hash)
            if replay:
                row = connection.execute("SELECT erased_at, expires_at FROM raw_audio_recordings WHERE audio_id = ?", (replay["audio_id"],)).fetchone()
                if row is None or row["erased_at"] or row["expires_at"] <= _utc_now():
                    raise HTTPException(410, "This recording has been erased and cannot be replayed")
                return replay
            used = connection.execute("SELECT COALESCE(SUM(LENGTH(ciphertext)), 0) FROM raw_audio_recordings WHERE participant_id = ?", (participant_id,)).fetchone()[0]
            if used + len(data) + 16 > MAX_PARTICIPANT_BYTES:
                raise HTTPException(413, "Participant recording storage limit reached")
            audio_id, nonce = _new_id("aud"), secrets.token_bytes(12)
            now = datetime.now(timezone.utc)
            expires = (now + timedelta(days=days)).isoformat(timespec="milliseconds").replace("+00:00", "Z")
            ciphertext = cipher.encrypt(nonce, bytes(data), (audio_id + ":" + participant_id).encode())
            connection.execute("INSERT INTO raw_audio_recordings VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL)",
                               (audio_id, participant_id, audio_consent_id, game_id, media_type, nonce, ciphertext, now.isoformat(timespec="milliseconds").replace("+00:00", "Z"), expires))
            connection.execute("INSERT INTO raw_audio_recording_sessions VALUES (?, ?)", (audio_id, audio_session_id))
            result = {"audio_id": audio_id, "audio_session_id": audio_session_id, "expires_at": expires,
                      "status": "privately_stored", "public_release": False}
            store._record_idempotency(connection, "audio_upload", idempotency_key, request_hash, "audio_recording", audio_id, result, now.isoformat(timespec="milliseconds"))
            connection.commit()
            return result
    except Exception as exc:
        _raise_api_error(exc)


@router.get("/recordings")
def list_recordings(participant_id: ParticipantId, participant_token: ParticipantToken):
    store = get_store(); store.initialize()
    try:
        with store._connect() as connection:
            authenticate(store, connection, participant_id, participant_token, active=False)
            rows = connection.execute("SELECT audio_id, game_id, created_at, expires_at, erased_at, erasure_reason "
                                      "FROM raw_audio_recordings WHERE participant_id = ? ORDER BY created_at DESC LIMIT 200", (participant_id,)).fetchall()
            return {"recordings": [dict(row) for row in rows]}
    except Exception as exc:
        _raise_api_error(exc)


@router.delete("/recordings/{audio_id}")
def delete_recording(audio_id: str, participant_id: ParticipantId, participant_token: ParticipantToken):
    store = get_store(); store.initialize()
    try:
        with store._connect() as connection:
            store._begin(connection)
            authenticate(store, connection, participant_id, participant_token, active=False)
            connection.execute("PRAGMA secure_delete = ON")
            row = connection.execute("SELECT 1 FROM raw_audio_recordings WHERE audio_id = ? AND participant_id = ?", (audio_id, participant_id)).fetchone()
            if row is None: raise HTTPException(404, "Recording not found")
            connection.execute("UPDATE raw_audio_recordings SET ciphertext = NULL, nonce = NULL, erased_at = COALESCE(erased_at, ?), "
                               "erasure_reason = COALESCE(erasure_reason, 'participant_request') WHERE audio_id = ?", (_utc_now(), audio_id))
            connection.commit()
        return {"audio_id": audio_id, "active_payload": "erased", "backup_erasure": "not_attested"}
    except Exception as exc:
        _raise_api_error(exc)


@router.delete("/consents/{audio_consent_id}")
def revoke_audio_consent(audio_consent_id: str, participant_id: ParticipantId, participant_token: ParticipantToken):
    store = get_store(); store.initialize()
    try:
        with store._connect() as connection:
            store._begin(connection)
            authenticate(store, connection, participant_id, participant_token, active=False)
            row = connection.execute("SELECT app_id FROM raw_audio_consents WHERE audio_consent_id = ? AND participant_id = ?", (audio_consent_id, participant_id)).fetchone()
            if row is None: raise HTTPException(404, "Recording-storage consent not found")
            connection.execute("PRAGMA secure_delete = ON")
            # Revoke every receipt for this app, including retries and other tabs.
            connection.execute("UPDATE raw_audio_sessions SET ended_at = COALESCE(ended_at, ?) WHERE audio_consent_id IN "
                               "(SELECT audio_consent_id FROM raw_audio_consents WHERE participant_id = ? AND app_id = ?)",
                               (_utc_now(), participant_id, row["app_id"]))
            connection.execute("UPDATE raw_audio_consents SET revoked_at = COALESCE(revoked_at, ?) WHERE participant_id = ? AND app_id = ?", (_utc_now(), participant_id, row["app_id"]))
            connection.execute("UPDATE raw_audio_recordings SET ciphertext = NULL, nonce = NULL, erased_at = COALESCE(erased_at, ?), "
                               "erasure_reason = COALESCE(erasure_reason, 'audio_consent_revoked') WHERE audio_consent_id IN "
                               "(SELECT audio_consent_id FROM raw_audio_consents WHERE participant_id = ? AND app_id = ?)", (_utc_now(), participant_id, row["app_id"]))
            connection.commit()
        return {"audio_consent_id": audio_consent_id, "status": "revoked", "active_payloads": "erased", "backup_erasure": "not_attested"}
    except Exception as exc:
        _raise_api_error(exc)