ScaiWave connector client

"""Async client for the ScaiDial ↔ ScaiWave connector.

Dependency: ``httpx``. Semantics: https://www.scailabs.ai/docs/scaigrid/scaidial/integrations/scaiwave-connector

    async with ScaiDialWaveClient("https://scaigrid.scailabs.ai", token) as dial:
        me = await dial.me("usr_…")
"""

from __future__ import annotations

import hashlib
import hmac
import json
import time
from collections.abc import AsyncIterator, Mapping
from typing import Any

import httpx

API_PREFIX = "/v1/public/scaidial/wave"


class ScaiDialWaveError(Exception):
    """Non-2xx response from the connector."""

    def __init__(self, status: int, code: str, message: str) -> None:
        super().__init__(f"{status} {code}: {message}")
        self.status = status
        self.code = code
        self.message = message


class WebhookVerificationError(Exception):
    """Webhook signature missing/invalid or timestamp outside the window."""


def verify_webhook(
    body: bytes,
    headers: Mapping[str, str],
    secret: str,
    tolerance_s: int = 300,
) -> dict[str, Any]:
    """Check ``X-ScaiDial-Signature`` over the raw body; return the event."""
    lower = {k.lower(): v for k, v in headers.items()}
    ts_raw = lower.get("x-scaidial-timestamp", "")
    sig = lower.get("x-scaidial-signature", "")
    try:
        ts = int(ts_raw)
    except ValueError as exc:
        raise WebhookVerificationError("missing or invalid timestamp") from exc
    if abs(time.time() - ts) > tolerance_s:
        raise WebhookVerificationError("timestamp outside tolerance")
    expected = (
        "v1=" + hmac.new(secret.encode(), f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    )
    if not sig or not hmac.compare_digest(sig.encode(), expected.encode()):
        raise WebhookVerificationError("signature mismatch")
    return json.loads(body)


class ScaiDialWaveClient:
    """One method per connector endpoint. ``user`` is the ScaiKey id
    (``usr_…``) or ScaiGrid user id the call is made on behalf of."""

    def __init__(self, base_url: str, token: str, *, timeout: float = 15.0) -> None:
        self._http = httpx.AsyncClient(
            base_url=base_url.rstrip("/") + API_PREFIX,
            headers={"Authorization": f"Bearer {token}"},
            timeout=timeout,
        )

    async def __aenter__(self) -> ScaiDialWaveClient:
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.aclose()

    async def aclose(self) -> None:
        await self._http.aclose()

    # -- plumbing --------------------------------------------------------

    async def _request(
        self,
        method: str,
        path: str,
        *,
        user: str | None = None,
        json_body: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
        raw: bool = False,
        files: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
    ) -> Any:  # noqa: ANN401
        headers = {"X-ScaiDial-User": user} if user else {}
        if params:
            params = {k: v for k, v in params.items() if v is not None}
        resp = await self._http.request(
            method, path, headers=headers, json=json_body, params=params, files=files, data=data
        )
        if resp.status_code >= 400:
            code, message = "http_error", resp.text[:300]
            try:
                payload = resp.json()
            except ValueError:
                payload = {}
            if isinstance(payload.get("error"), dict):
                code = payload["error"].get("code", code)
                message = payload["error"].get("message", message)
            elif "detail" in payload:  # request validation (FastAPI shape)
                code, message = "validation_error", json.dumps(payload["detail"])[:500]
            raise ScaiDialWaveError(resp.status_code, code, message)
        if raw:
            return resp.content
        if resp.status_code == 204 or not resp.content:
            return None
        return resp.json().get("data")

    # -- events + rings --------------------------------------------------

    async def list_events(self, after: str = "0", limit: int = 100) -> dict[str, Any]:
        return await self._request("GET", "/events", params={"after": after, "limit": limit})

    async def iter_events(
        self, after: str = "0", page_size: int = 500
    ) -> AsyncIterator[dict[str, Any]]:
        """Yield every event after ``after`` (oldest first) until caught up."""
        while True:
            page = await self.list_events(after, page_size)
            for event in page["items"]:
                yield event
            if len(page["items"]) < page_size:
                return
            after = page["last_id"]

    async def list_rings(
        self, user: str | None = None, tenant_id: str | None = None
    ) -> list[dict[str, Any]]:
        data = await self._request("GET", "/rings", user=user, params={"tenant_id": tenant_id})
        return data["items"]

    async def accept_ring(
        self, user: str, leg_id: str, device: str | None = None
    ) -> dict[str, Any]:
        return await self._request(
            "POST", f"/rings/{leg_id}/accept", user=user, json_body={"device": device}
        )

    async def decline_ring(
        self,
        user: str,
        leg_id: str,
        fallback: str = "voicemail",
        forward_to: str | None = None,
    ) -> dict[str, Any]:
        return await self._request(
            "POST",
            f"/rings/{leg_id}/decline",
            user=user,
            json_body={"fallback": fallback, "forward_to": forward_to},
        )

    # -- user + extensions -----------------------------------------------

    async def me(self, user: str) -> dict[str, Any]:
        return await self._request("GET", "/me", user=user)

    async def get_extension(self, user: str, extension_id: str) -> dict[str, Any]:
        return await self._request("GET", f"/extensions/{extension_id}", user=user)

    async def update_extension(self, user: str, extension_id: str, **fields: Any) -> dict[str, Any]:  # noqa: ANN401
        """PATCH settings, e.g. ``dnd=True``, ``ring_timeout_s=20``."""
        return await self._request(
            "PATCH", f"/extensions/{extension_id}", user=user, json_body=fields
        )

    async def list_forward_rules(self, user: str, extension_id: str) -> list[dict[str, Any]]:
        data = await self._request("GET", f"/extensions/{extension_id}/forward-rules", user=user)
        return data["items"]

    async def create_forward_rule(
        self,
        user: str,
        extension_id: str,
        condition: str,
        forward_to: str,
        priority: int = 100,
        enabled: bool = True,
    ) -> dict[str, Any]:
        return await self._request(
            "POST",
            f"/extensions/{extension_id}/forward-rules",
            user=user,
            json_body={
                "condition": condition,
                "forward_to": forward_to,
                "priority": priority,
                "enabled": enabled,
            },
        )

    async def delete_forward_rule(self, user: str, extension_id: str, rule_id: str) -> None:
        await self._request(
            "DELETE", f"/extensions/{extension_id}/forward-rules/{rule_id}", user=user
        )

    async def directory(self, user: str, q: str = "", limit: int = 20) -> list[dict[str, Any]]:
        data = await self._request("GET", "/directory", user=user, params={"q": q, "limit": limit})
        return data["items"]

    # -- calls -----------------------------------------------------------

    async def place_call(self, user: str, from_extension_id: str, to: str) -> dict[str, Any]:
        return await self._request(
            "POST",
            "/calls",
            user=user,
            json_body={"from_extension_id": from_extension_id, "to": to},
        )

    async def active_calls(self, user: str) -> list[dict[str, Any]]:
        return (await self._request("GET", "/calls/active", user=user))["items"]

    async def call_history(
        self, user: str, limit: int = 50, before: str | None = None
    ) -> list[dict[str, Any]]:
        data = await self._request(
            "GET", "/calls/history", user=user, params={"limit": limit, "before": before}
        )
        return data["items"]

    async def rejoin_token(self, user: str, call_id: str) -> dict[str, Any]:
        return await self._request("POST", f"/calls/{call_id}/token", user=user)

    async def hangup(self, user: str, leg_id: str) -> None:
        await self._request("POST", f"/legs/{leg_id}/hangup", user=user)

    async def hold(self, user: str, leg_id: str) -> None:
        await self._request("POST", f"/legs/{leg_id}/hold", user=user)

    async def unhold(self, user: str, leg_id: str) -> None:
        await self._request("POST", f"/legs/{leg_id}/unhold", user=user)

    async def transfer(self, user: str, leg_id: str, to: str) -> dict[str, Any]:
        return await self._request(
            "POST", f"/legs/{leg_id}/transfer", user=user, json_body={"to": to, "mode": "blind"}
        )

    async def merge_calls(self, user: str, keep_leg_id: str, merge_leg_id: str) -> dict[str, Any]:
        """Three-way call: ``merge_leg_id``'s call joins ``keep_leg_id``'s."""
        return await self._request(
            "POST",
            "/calls/merge",
            user=user,
            json_body={"keep_leg_id": keep_leg_id, "merge_leg_id": merge_leg_id},
        )

    async def conference(self, user: str, leg_id: str) -> dict[str, Any]:
        return await self._request("GET", f"/legs/{leg_id}/conference", user=user)

    async def conference_focus(self, user: str, leg_id: str, focus: str) -> dict[str, Any]:
        """``all`` (join), ``kept`` (split), ``<bridge_id>`` (swap), ``none`` (hold)."""
        return await self._request(
            "POST", f"/legs/{leg_id}/conference/focus", user=user, json_body={"focus": focus}
        )

    async def conference_drop(self, user: str, leg_id: str, bridge_id: str) -> dict[str, Any]:
        return await self._request(
            "POST", f"/legs/{leg_id}/conference/drop", user=user, json_body={"bridge_id": bridge_id}
        )

    async def moh_playlists(self, user: str) -> dict[str, Any]:
        return await self._request("GET", "/moh/playlists", user=user)

    async def moh_create_playlist(
        self, user: str, name: str, shuffle: bool = True
    ) -> dict[str, Any]:
        return await self._request(
            "POST", "/moh/playlists", user=user, json_body={"name": name, "shuffle": shuffle}
        )

    async def moh_upload(
        self,
        user: str,
        playlist_id: str,
        filename: str,
        data: bytes,
        *,
        title: str | None = None,
        artist: str | None = None,
    ) -> dict[str, Any]:
        """Upload a song; the caller states the user confirmed the rights."""
        form = {"rights_confirmed": "true"}
        if title:
            form["title"] = title
        if artist:
            form["artist"] = artist
        return await self._request(
            "POST",
            f"/moh/playlists/{playlist_id}/tracks",
            user=user,
            files={"file": (filename, data)},
            data=form,
        )

    async def start_attended_transfer(self, user: str, leg_id: str, to: str) -> dict[str, Any]:
        """Consult ``to`` first; returns ``{consult_id, state, ...}``."""
        return await self._request(
            "POST", f"/legs/{leg_id}/transfer", user=user, json_body={"to": to, "mode": "attended"}
        )

    async def transfer_status(self, user: str, consult_id: str) -> dict[str, Any]:
        return await self._request("GET", f"/transfers/{consult_id}", user=user)

    async def complete_transfer(self, user: str, consult_id: str) -> dict[str, Any]:
        return await self._request("POST", f"/transfers/{consult_id}/complete", user=user)

    async def cancel_transfer(self, user: str, consult_id: str) -> dict[str, Any]:
        return await self._request("POST", f"/transfers/{consult_id}/cancel", user=user)

    # -- voicemail -------------------------------------------------------

    async def list_voicemail(
        self,
        user: str,
        extension_id: str | None = None,
        unheard: bool = False,
        limit: int = 50,
    ) -> list[dict[str, Any]]:
        data = await self._request(
            "GET",
            "/voicemail",
            user=user,
            params={"extension_id": extension_id, "unheard": str(unheard).lower(), "limit": limit},
        )
        return data["items"]

    async def get_voicemail(self, user: str, voicemail_id: str) -> dict[str, Any]:
        return await self._request("GET", f"/voicemail/{voicemail_id}", user=user)

    async def voicemail_audio(self, user: str, voicemail_id: str) -> bytes:
        """WAV, 16 kHz mono."""
        return await self._request("GET", f"/voicemail/{voicemail_id}/audio", user=user, raw=True)

    async def mark_voicemail_listened(self, user: str, voicemail_id: str) -> None:
        await self._request("POST", f"/voicemail/{voicemail_id}/listened", user=user)

    async def delete_voicemail(self, user: str, voicemail_id: str) -> None:
        await self._request("DELETE", f"/voicemail/{voicemail_id}", user=user)

    # -- call recording, transcripts, live captions ------------------------

    async def recording_control(self, user: str, call_id: str, action: str) -> dict[str, Any]:
        """``action``: start | pause | resume | stop. ``start`` returns once
        the announcement has played. Returns the recording object."""
        return await self._request("POST", f"/calls/{call_id}/recording/{action}", user=user)

    async def captions(self, user: str, call_id: str, on: bool) -> dict[str, Any]:
        """Live captions on/off; they arrive in the LiveKit room as text
        streams on topic ``lk.transcription``."""
        action = "start" if on else "stop"
        return await self._request("POST", f"/calls/{call_id}/captions/{action}", user=user)

    async def list_recordings(
        self,
        user: str,
        call_id: str | None = None,
        extension_id: str | None = None,
        before: str | None = None,
        limit: int = 50,
    ) -> list[dict[str, Any]]:
        data = await self._request(
            "GET",
            "/recordings",
            user=user,
            params={
                "call_id": call_id,
                "extension_id": extension_id,
                "before": before,
                "limit": limit,
            },
        )
        return list(data["items"])

    async def get_recording(self, user: str, recording_id: str) -> dict[str, Any]:
        return await self._request("GET", f"/recordings/{recording_id}", user=user)

    async def recording_audio(self, user: str, recording_id: str) -> bytes:
        """Ogg/Opus, 16 kHz mono. For playback in a client, proxy the
        endpoint and forward the ``Range`` header instead."""
        return await self._request("GET", f"/recordings/{recording_id}/audio", user=user, raw=True)

    async def recording_transcript(self, user: str, recording_id: str) -> dict[str, Any]:
        """``{transcript_state, language, summary, text, turns}``."""
        return await self._request("GET", f"/recordings/{recording_id}/transcript", user=user)

    async def delete_recording(self, user: str, recording_id: str) -> None:
        await self._request("DELETE", f"/recordings/{recording_id}", user=user)