3cx-telefonie-middleware/app/call_context_repository.py

from __future__ import annotations

import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


DB_PATH = Path(
    "/opt/3cx-middleware/3cx-telefonie-middleware/data/telephony.sqlite3"
)


class CallContextRepository:

    def __init__(self, db_path: str | Path = DB_PATH):
        self.db_path = str(db_path)
        self._ensure_schema()

    def _connect(self):
        con = sqlite3.connect(
            self.db_path,
            timeout=10,
        )
        con.row_factory = sqlite3.Row
        return con

    def _ensure_schema(self):
        with self._connect() as con:
            con.execute(
                """
                CREATE TABLE IF NOT EXISTS call_context (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    callid TEXT NOT NULL,
                    legid TEXT,
                    phone_e164 TEXT NOT NULL,
                    context_json TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    UNIQUE(callid, legid)
                )
                """
            )

            con.execute(
                """
                CREATE INDEX IF NOT EXISTS
                idx_call_context_phone
                ON call_context(phone_e164)
                """
            )

            con.commit()

    def upsert(
        self,
        *,
        callid: str,
        legid: str | None,
        phone_e164: str,
        context: dict[str, Any],
    ) -> None:

        now = datetime.now(
            timezone.utc
        ).isoformat()

        payload = json.dumps(
            context,
            ensure_ascii=False,
            separators=(",", ":"),
            default=str,
        )

        with self._connect() as con:
            con.execute(
                """
                INSERT INTO call_context (
                    callid,
                    legid,
                    phone_e164,
                    context_json,
                    created_at,
                    updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(callid, legid)
                DO UPDATE SET
                    phone_e164 = excluded.phone_e164,
                    context_json = excluded.context_json,
                    updated_at = excluded.updated_at
                """,
                (
                    str(callid),
                    str(legid) if legid is not None else None,
                    phone_e164,
                    payload,
                    now,
                    now,
                ),
            )
            con.commit()

    def get(
        self,
        *,
        callid: str,
        legid: str | None = None,
    ) -> dict[str, Any] | None:

        with self._connect() as con:
            if legid is None:
                row = con.execute(
                    """
                    SELECT *
                    FROM call_context
                    WHERE callid = ?
                    ORDER BY id DESC
                    LIMIT 1
                    """,
                    (str(callid),),
                ).fetchone()
            else:
                row = con.execute(
                    """
                    SELECT *
                    FROM call_context
                    WHERE callid = ?
                      AND legid = ?
                    LIMIT 1
                    """,
                    (str(callid), str(legid)),
                ).fetchone()

        if row is None:
            return None

        return {
            "callid": row["callid"],
            "legid": row["legid"],
            "phone_e164": row["phone_e164"],
            "context": json.loads(row["context_json"]),
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
        }