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

import aiosqlite
from .phone import normalize
from datetime import datetime, timezone, timedelta


def now_iso():
    return datetime.now(timezone.utc).isoformat()


class Repository:
    def __init__(self, path):
        self.path = path

    async def init(self):
        async with aiosqlite.connect(self.path) as db:
            await db.executescript("""
            CREATE TABLE IF NOT EXISTS calls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                callid INTEGER NOT NULL,
                legid INTEGER,
                phone TEXT,
                queue_dn TEXT NOT NULL,
                first_seen_at TEXT NOT NULL,
                last_seen_at TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'active',
                answered_by_dn TEXT,
                UNIQUE(queue_dn, callid)
            );

            CREATE TABLE IF NOT EXISTS missed_call_groups (
                phone TEXT PRIMARY KEY,
                first_call_at TEXT NOT NULL,
                last_call_at TEXT NOT NULL,
                attempt_count INTEGER NOT NULL DEFAULT 0,
                status TEXT NOT NULL DEFAULT 'open',
                customer_id TEXT,
                customer_label TEXT,
                resolved_until TEXT,
                resolved_by TEXT,
                resolved_at TEXT,
                resolved_note TEXT
            );

            CREATE TABLE IF NOT EXISTS idempotency_keys (
                idem_key TEXT PRIMARY KEY,
                operation TEXT NOT NULL,
                response_json TEXT NOT NULL,
                created_at TEXT NOT NULL
            );
            """)
            await db.commit()

    def normalize_phone(self, raw):
        """
        Zentrale Telefonnummern-Normalisierung.

        Die eigentliche Logik liegt in app.phone.normalize().
        Dadurch verwendet die Repository-/Queue-Verarbeitung
        dieselbe Normalisierung wie die /phone/normalize API.
        """
        result = normalize(raw, "DE")

        if not result.get("valid"):
            return None

        return result.get("e164")

    async def observe_queue(self, participants, queue_dn, agent_dns=None):
        active = set()

        async with aiosqlite.connect(self.path) as db:
            for p in participants:
                if p.get("callid") is None:
                    continue

                callid = int(p["callid"])
                active.add(callid)

                raw = p.get("party_caller_id")
                is_external = p.get("party_dn_type") == "Wexternalline"
                e164 = self.normalize_phone(raw) if is_external else None
                now = now_iso()

                status_raw = str(p.get("status") or "").lower()

                participant_dn = str(
                    p.get("partyDn") or p.get("party_dn") or ""
                ).strip()

                agent_dns_set = {
                    str(x).strip()
                    for x in (agent_dns or [])
                    if str(x).strip()
                }

                if (
                    status_raw in ("connected", "talking", "answered")
                    and participant_dn in agent_dns_set
                ):
                    current_status = "answered"
                else:
                    current_status = "active"

                await db.execute("""
                    INSERT INTO calls
                    (callid, legid, phone, raw, queue_dn,
                     first_seen_at, last_seen_at, status,
                     direction, started_at, answered_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    ON CONFLICT(queue_dn, callid) DO UPDATE SET
                      legid=excluded.legid,
                      phone=excluded.phone,
                      raw=excluded.raw,
                      last_seen_at=excluded.last_seen_at,
                      status=excluded.status,
                      answered_at=CASE
                        WHEN calls.answered_at IS NOT NULL
                          THEN calls.answered_at
                        WHEN excluded.status='answered'
                          THEN excluded.last_seen_at
                        ELSE NULL
                      END
                """, (
                    callid,
                    p.get("legid"),
                    e164,
                    raw,
                    queue_dn,
                    now,
                    now,
                    current_status,
                    "inbound" if is_external else "internal",
                    now,
                    now if current_status == "answered" else None,
                ))

            await db.commit()

        return active

    async def finalize_disappeared(self, queue_dn, active):
        async with aiosqlite.connect(self.path) as db:
            cur = await db.execute("""
                SELECT id, callid, phone, raw, last_seen_at
                FROM calls
                WHERE queue_dn=? AND status IN ('active', 'answered')
            """, (queue_dn,))

            for row_id, callid, phone, raw, last_seen_at in await cur.fetchall():
                if callid in active:
                    continue

                last_seen = datetime.fromisoformat(last_seen_at)

                if datetime.now(timezone.utc) - last_seen < timedelta(seconds=10):
                    continue

                now = now_iso()

                cur_state = await db.execute(
                    "SELECT answered_at, started_at FROM calls WHERE id=?",
                    (row_id,),
                )
                state_row = await cur_state.fetchone()

                answered_at = state_row[0] if state_row else None
                started_at = state_row[1] if state_row else None

                final_status = "ended" if answered_at else "missed"

                duration_seconds = None
                if started_at:
                    started_dt = datetime.fromisoformat(started_at)
                    ended_dt = datetime.fromisoformat(now)
                    duration_seconds = max(
                        0,
                        int((ended_dt - started_dt).total_seconds())
                    )

                await db.execute(
                    """
                    UPDATE calls
                    SET status=?,
                        ended_at=?,
                        last_seen_at=?,
                        duration_seconds=?
                    WHERE id=?
                    """,
                    (
                        final_status,
                        now,
                        now,
                        duration_seconds,
                        row_id,
                    ),
                )

                if final_status == "missed" and phone:
                    await db.execute("""
                        INSERT INTO missed_call_groups
                        (phone, raw, first_call_at, last_call_at,
                         attempt_count, status)
                        VALUES (?, ?, ?, ?, 1, 'open')
                        ON CONFLICT(phone) DO UPDATE SET
                          raw=excluded.raw,
                          last_call_at=excluded.last_call_at,
                          attempt_count=missed_call_groups.attempt_count+1
                    """, (phone, raw, last_seen_at, now))

            await db.commit()




    async def get_outbound_idempotency(self, key):
        async with aiosqlite.connect(self.path) as db:
            db.row_factory = aiosqlite.Row

            cur = await db.execute("""
                SELECT idempotency_key, request_hash, status, response_json, created_at
                FROM outbound_idempotency
                WHERE idempotency_key=?
            """, (key,))

            row = await cur.fetchone()
            return dict(row) if row else None

    async def claim_outbound_idempotency(self, key, request_hash, created_at):
        async with aiosqlite.connect(self.path) as db:
            try:
                await db.execute("BEGIN IMMEDIATE")

                cur = await db.execute("""
                    INSERT OR IGNORE INTO outbound_idempotency
                    (idempotency_key, request_hash, status, response_json, created_at)
                    VALUES (?, ?, 'processing', NULL, ?)
                """, (key, request_hash, created_at))

                inserted = cur.rowcount == 1
                await db.commit()
                return inserted

            except Exception:
                await db.rollback()
                raise

    async def finish_outbound_idempotency(self, key, response_json):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                UPDATE outbound_idempotency
                SET status='completed', response_json=?
                WHERE idempotency_key=?
            """, (response_json, key))
            await db.commit()

    async def release_outbound_idempotency(self, key):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                DELETE FROM outbound_idempotency
                WHERE idempotency_key=? AND status='processing'
            """, (key,))
            await db.commit()


    async def insert_outbound_call(
        self,
        callid,
        legid,
        source_dn,
        destination,
        phone,
        raw,
        started_at,
    ):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                INSERT INTO calls (
                    callid,
                    legid,
                    phone,
                    queue_dn,
                    first_seen_at,
                    last_seen_at,
                    status,
                    raw,
                    direction,
                    started_at,
                    source_dn,
                    destination
                )
                VALUES (?, ?, ?, NULL, ?, ?, 'dialing', ?, 'outbound', ?, ?, ?)
            """, (
                callid,
                legid,
                phone,
                started_at,
                started_at,
                raw,
                started_at,
                source_dn,
                destination,
            ))

            await db.commit()


    async def list_active_outbound_calls(self):
        async with aiosqlite.connect(self.path) as db:
            db.row_factory = aiosqlite.Row

            cur = await db.execute("""
                SELECT
                    id,
                    callid,
                    legid,
                    source_dn,
                    destination,
                    status,
                    started_at,
                    answered_at,
                    ended_at,
                    duration_seconds
                FROM calls
                WHERE direction='outbound'
                  AND ended_at IS NULL
                ORDER BY id
            """)

            return [dict(row) for row in await cur.fetchall()]

    async def update_outbound_connected(
        self,
        call_id,
        answered_at,
        last_seen_at,
    ):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                UPDATE calls
                SET
                    status='connected',
                    answered_at=COALESCE(answered_at, ?),
                    last_seen_at=?
                WHERE id=?
                  AND direction='outbound'
                  AND ended_at IS NULL
            """, (
                answered_at,
                last_seen_at,
                call_id,
            ))

            await db.commit()

    async def update_outbound_seen(
        self,
        call_id,
        status,
        last_seen_at,
    ):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                UPDATE calls
                SET
                    status=?,
                    last_seen_at=?
                WHERE id=?
                  AND direction='outbound'
                  AND ended_at IS NULL
            """, (
                status,
                last_seen_at,
                call_id,
            ))

            await db.commit()

    async def finalize_outbound_call(
        self,
        call_id,
        ended_at,
        status,
        duration_seconds,
    ):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                UPDATE calls
                SET
                    status=?,
                    ended_at=?,
                    duration_seconds=?,
                    last_seen_at=?
                WHERE id=?
                  AND direction='outbound'
                  AND ended_at IS NULL
            """, (
                status,
                ended_at,
                duration_seconds,
                ended_at,
                call_id,
            ))

            await db.commit()

    async def get_call(self, call_id):
        async with aiosqlite.connect(self.path) as db:
            db.row_factory = aiosqlite.Row

            cur = await db.execute("""
                SELECT
                    id,
                    callid,
                    legid,
                    phone,
                    raw,
                    queue_dn,
                    status,
                    direction,
                    first_seen_at,
                    last_seen_at,
                    started_at,
                    answered_at,
                    ended_at,
                    duration_seconds,
                    agent_dn,
                    agent_name
                FROM calls
                WHERE id=?
                LIMIT 1
            """, (call_id,))

            row = await cur.fetchone()
            return dict(row) if row else None

    async def list_calls(self, limit=100, offset=0, status=None):
        async with aiosqlite.connect(self.path) as db:
            db.row_factory = aiosqlite.Row

            where = ""
            params = []

            if status:
                where = "WHERE status=?"
                params.append(status)

            cur = await db.execute(
                f"""
                SELECT
                    id,
                    callid,
                    legid,
                    phone,
                    raw,
                    queue_dn,
                    status,
                    direction,
                    first_seen_at,
                    last_seen_at,
                    started_at,
                    answered_at,
                    ended_at,
                    duration_seconds,
                    agent_dn,
                    agent_name
                FROM calls
                {where}
                ORDER BY COALESCE(started_at, first_seen_at) DESC
                LIMIT ? OFFSET ?
                """,
                (*params, limit, offset),
            )

            rows = [dict(row) for row in await cur.fetchall()]

            cur = await db.execute(
                f"SELECT COUNT(*) FROM calls {where}",
                tuple(params),
            )
            total = (await cur.fetchone())[0]

            return rows, total

    def _decorate(self, row):
        last_call = row["last_call_at"]
        resolved_until = row["resolved_until"]
        row["e164"] = row["phone"]
        row["status"] = (
            "resolved"
            if resolved_until and resolved_until >= last_call
            else "open"
        )
        return row

    async def list_groups(self, open_only=False):
        async with aiosqlite.connect(self.path) as db:
            db.row_factory = aiosqlite.Row
            cur = await db.execute(
                "SELECT * FROM missed_call_groups ORDER BY last_call_at DESC"
            )
            rows = [dict(x) for x in await cur.fetchall()]

        rows = [self._decorate(x) for x in rows]

        if open_only:
            rows = [x for x in rows if x["status"] == "open"]

        return rows

    async def get_group(self, e164):
        async with aiosqlite.connect(self.path) as db:
            db.row_factory = aiosqlite.Row
            cur = await db.execute(
                "SELECT * FROM missed_call_groups WHERE phone=?",
                (e164,),
            )
            row = await cur.fetchone()

        return self._decorate(dict(row)) if row else None

    async def get_idempotency(self, key, operation):
        async with aiosqlite.connect(self.path) as db:
            cur = await db.execute(
                """SELECT response_json
                   FROM idempotency_keys
                   WHERE idem_key=? AND operation=?""",
                (key, operation),
            )
            row = await cur.fetchone()
            return row[0] if row else None

    async def save_idempotency(self, key, operation, response_json):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                INSERT OR IGNORE INTO idempotency_keys
                (idem_key, operation, response_json, created_at)
                VALUES (?, ?, ?, ?)
            """, (key, operation, response_json, now_iso()))
            await db.commit()

    async def resolve(self, e164, until, by, note):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                UPDATE missed_call_groups
                SET resolved_until=?,
                    resolved_by=?,
                    resolved_at=?,
                    resolved_note=?
                WHERE phone=?
            """, (until, by, now_iso(), note, e164))
            await db.commit()

        return await self.get_group(e164)

    async def reopen(self, e164, by):
        async with aiosqlite.connect(self.path) as db:
            await db.execute("""
                UPDATE missed_call_groups
                SET resolved_until=NULL,
                    resolved_by=?,
                    resolved_at=?,
                    resolved_note=NULL
                WHERE phone=?
            """, (by, now_iso(), e164))
            await db.commit()

        return await self.get_group(e164)