"""Kanban subscriptions + delivery-queue notifications.

Plan B's small notification surface. Operators (or agents via the HTTP /
MCP layer when allowed) can subscribe to terminal lifecycle events on a
particular task; on transition we enqueue a row on the existing
``utils.delivery_queue`` (``KIND_NOTIFICATION``), which the runner-owned
delivery worker handles (thread post / push / future webhook).

Why subscriptions live next to kanban storage and not in delivery_queue
itself: subscriptions are per-task lifecycle hooks, not deliveries. They
fan into delivery_queue when the lifecycle fires.

Invariants:

- Targets are stored encrypted (``target_ciphertext``) because they may
  carry thread ids, user-mention identifiers, or webhook URLs.
- ``done`` notifications fire **only** for root tasks (``parent_ids = []``)
  to prevent fan-out spam on a multi-task DAG. Children completing is
  observable on the activity bus / SSE board; only the root is interesting
  enough for a separate alert.
- ``gave_up`` notifications fire on every terminal failure (root or child),
  because operators want to see ANY task that exhausts retries.
- The hook runs after ``_emit_lifecycle_event`` in
  :mod:`kanban_storage`; failures here MUST NOT break the storage path.
"""

from __future__ import annotations

import json
import logging
from typing import Iterable

_log = logging.getLogger(__name__)
_dispatch_enqueue_failures = 0

VALID_CHANNELS = frozenset({"thread", "notification", "webhook", "discord", "slack"})
DEFAULT_FIRE_ON: tuple[str, ...] = ("gave_up", "done")
MAX_TARGET_CHARS = 1024
MAX_SUBSCRIPTIONS_PER_TASK = 10


def _new_id() -> str:
    import uuid
    return "ksub_" + uuid.uuid4().hex[:14]


def _now_iso() -> str:
    from datetime import datetime, timezone
    return datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _validate_webhook_target(channel: str, target: str) -> None:
    """Validate webhook URL at subscribe-time rather than delivery-time."""
    import re
    from urllib.parse import urlparse

    if channel == "discord":
        if not re.match(
            r"^https://(?:ptb\.|canary\.)?discord(?:app)?\.com/api/webhooks/\d+/.+$",
            target,
        ):
            raise ValueError(f"invalid Discord webhook URL: {target[:80]}")
    elif channel == "slack":
        if not re.match(r"^https://hooks\.slack\.com/services/.+$", target):
            raise ValueError(f"invalid Slack webhook URL: {target[:80]}")
    elif channel == "webhook":
        parsed = urlparse(target)
        if parsed.scheme not in ("https", "http"):
            raise ValueError(f"webhook target must be http(s): {target[:80]}")
        if not parsed.hostname:
            raise ValueError(f"webhook target missing hostname: {target[:80]}")
        if parsed.scheme == "http":
            from core.config import get_kanban_config

            hostname = (parsed.hostname or "").lower()
            loopback = hostname in ("127.0.0.1", "::1", "localhost")
            if not loopback and not bool(get_kanban_config().get("allow_http_webhooks", False)):
                raise ValueError(
                    "http webhooks require HTTPS for remote hosts or "
                    "[kanban] allow_http_webhooks = true"
                )

    if channel in ("discord", "slack", "webhook"):
        try:
            parsed = urlparse(target)
            target_json = json.dumps({"webhook_url": target})
            if len(target_json) > MAX_TARGET_CHARS:
                raise ValueError("target too large")
        except ValueError:
            raise


def _normalize_fire_on(fire_on: Iterable[str]) -> tuple[str, ...]:
    out: list[str] = []
    for entry in fire_on or ():
        if not isinstance(entry, str):
            continue
        s = entry.strip().lower()
        if not s:
            continue
        if s not in {"gave_up", "done"}:
            raise ValueError(
                f"unknown fire_on event {s!r}; expected one of gave_up,done"
            )
        if s not in out:
            out.append(s)
    if not out:
        out = list(DEFAULT_FIRE_ON)
    return tuple(out)


def kanban_notify_subscribe(
    task_id: str,
    *,
    channel: str,
    target: str,
    fire_on: Iterable[str] = DEFAULT_FIRE_ON,
) -> str:
    """Persist a subscription. Returns the new subscription_id.

    *task_id*: the kanban task id to watch (must exist).
    *channel*: one of ``thread`` (thread message via delivery_queue),
       ``notification`` (system notification banner), ``webhook`` (URL).
    *target*: the channel-specific target (thread id, banner subject,
       URL). Stored encrypted.
    *fire_on*: subset of (``gave_up``, ``done``).
    """
    from chat_storage import _encrypt, _get_connection
    import kanban_storage as ks

    if not isinstance(task_id, str) or not task_id.strip():
        raise ValueError("task_id is required")
    if channel not in VALID_CHANNELS:
        raise ValueError(f"unknown channel {channel!r}; expected {sorted(VALID_CHANNELS)}")
    if not isinstance(target, str) or not target.strip():
        raise ValueError("target is required")
    if len(target) > MAX_TARGET_CHARS:
        raise ValueError(f"target exceeds {MAX_TARGET_CHARS}-char limit")

    if channel in ("discord", "slack", "webhook"):
        _validate_webhook_target(channel, target.strip())

    fire_on_t = _normalize_fire_on(fire_on)

    task = ks.get_task(task_id)
    if task is None:
        raise ValueError(f"task {task_id} not found")

    sub_id = _new_id()
    now = _now_iso()
    conn = _get_connection()
    try:
        # Cap check + insert in one IMMEDIATE transaction so concurrent
        # subscribes cannot both pass the count and exceed the cap.
        conn.execute("BEGIN IMMEDIATE")
        count = conn.execute(
            "SELECT COUNT(*) FROM kanban_subscriptions WHERE task_id=?",
            (task_id,),
        ).fetchone()[0]
        if count >= MAX_SUBSCRIPTIONS_PER_TASK:
            conn.rollback()
            raise ValueError(
                f"task {task_id} already has {count} subscriptions "
                f"(max {MAX_SUBSCRIPTIONS_PER_TASK})"
            )
        conn.execute(
            "INSERT INTO kanban_subscriptions "
            "(id, task_id, channel, target_ciphertext, fire_on, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (sub_id, task_id, channel, _encrypt(target), ",".join(fire_on_t), now),
        )
        conn.commit()
    except ValueError:
        raise
    except Exception as exc:
        _log.error(
            "subscription write failed for task %s; refusing new subscription",
            task_id,
            exc_info=True,
        )
        raise ValueError(
            f"could not record subscription for task {task_id}: {exc}"
        ) from exc
    finally:
        conn.close()
    return sub_id


def kanban_notify_subscribe_if_absent(
    task_id: str,
    *,
    channel: str,
    target: str,
    fire_on: Iterable[str] = DEFAULT_FIRE_ON,
) -> str:
    """Subscribe only when no matching (task_id, channel, target) row exists."""
    target = (target or "").strip()
    if not target:
        raise ValueError("target is required")
    for sub in list_subscriptions_for_task(task_id):
        if sub.get("channel") == channel and (sub.get("target") or "").strip() == target:
            return str(sub.get("id") or "")
    return kanban_notify_subscribe(
        task_id,
        channel=channel,
        target=target,
        fire_on=fire_on,
    )


def kanban_notify_unsubscribe(subscription_id: str) -> bool:
    """Remove a subscription. Returns True when a row was deleted."""
    if not isinstance(subscription_id, str) or not subscription_id.strip():
        return False
    from chat_storage import _get_connection
    conn = _get_connection()
    try:
        cur = conn.execute(
            "DELETE FROM kanban_subscriptions WHERE id=?", (subscription_id,)
        )
        conn.commit()
        return bool(cur.rowcount)
    finally:
        conn.close()


def list_subscriptions_for_task(task_id: str) -> list[dict]:
    """Return all subscriptions on *task_id* (decrypted target). Used by the
    lifecycle hook + (later) the SwiftUI drawer."""
    if not isinstance(task_id, str) or not task_id.strip():
        return []
    import sqlite3
    from chat_storage import _decrypt, _get_connection
    out: list[dict] = []
    conn = _get_connection()
    conn.row_factory = sqlite3.Row
    try:
        rows = conn.execute(
            "SELECT id, task_id, channel, target_ciphertext, fire_on, created_at "
            "FROM kanban_subscriptions WHERE task_id=?",
            (task_id,),
        ).fetchall()
    finally:
        conn.close()
    for row in rows:
        decrypt_failed = False
        try:
            target = _decrypt(row["target_ciphertext"])
        except Exception:
            _log.warning(
                "subscription target decrypt failed sub=%s task=%s",
                row["id"],
                task_id,
                exc_info=True,
            )
            target = ""
            decrypt_failed = True
        fire_on = (row["fire_on"] or "").split(",")
        out.append({
            "id": row["id"],
            "task_id": row["task_id"],
            "channel": row["channel"],
            "target": target,
            "target_decrypt_failed": decrypt_failed,
            "fire_on": [s for s in (s.strip() for s in fire_on) if s],
            "created_at": row["created_at"],
        })
    return out


def maybe_dispatch_lifecycle(
    *,
    event_type: str,
    task_id: str,
    assignee: str,
    tenant: str,
    run_id: str,
    extra: dict | None,
) -> None:
    """Lifecycle hook called from ``kanban_storage._emit_lifecycle_event``.

    Fires for ``gave_up`` always and for ``completed`` only on root tasks
    (no parents). Each matching subscription enqueues a
    :data:`utils.delivery_queue.KIND_NOTIFICATION` row with the task /
    event payload. Errors here are swallowed so storage never breaks.
    """
    try:
        if event_type == "completed":
            kind = "done"
        elif event_type == "gave_up":
            kind = "gave_up"
        else:
            return

        import kanban_storage as ks
        task = ks.get_task(task_id)
        if not task:
            return

        task_meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
        from utils.kanban_coordinator import COORDINATOR_BATCH_ID_META

        coordinator_batch = bool(task_meta.get(COORDINATOR_BATCH_ID_META))

        if kind == "done":
            from utils.kanban_coordinator import _task_is_root

            if not _task_is_root(task):
                # Not a root task; don't fan out.
                return
            if coordinator_batch:
                # Coordinator batches wake the parent thread once at batch terminal
                # (refresh_coordinator_state_for_task), not when an intermediate root
                # finishes while downstream tasks are still open.
                return

        subs = list_subscriptions_for_task(task_id)
        if not subs:
            return

        from utils.delivery_queue import (
            KIND_DISCORD,
            KIND_NOTIFICATION,
            KIND_SLACK,
            KIND_THREAD,
            KIND_WEBHOOK,
            enqueue,
        )
        title = (task.get("title") or task_id)
        if isinstance(title, str) and len(title) > 200:
            title = title[:200] + "…"
        verb = "completed" if kind == "done" else "gave up"
        banner = f"Kanban: {title} {verb}"
        body = (
            f"Task {task_id} {verb}. "
            f"assignee={assignee or '-'} tenant={tenant or 'default'} "
            f"run={run_id or '-'}"
        )
        for sub in subs:
            if kind not in (sub.get("fire_on") or ()):
                continue
            if sub.get("target_decrypt_failed"):
                _log.warning(
                    "skipping kanban subscription %s for task %s: target decrypt failed",
                    sub.get("id"),
                    task_id,
                )
                continue
            target = (sub.get("target") or "").strip()
            if not target:
                continue
            channel = (sub.get("channel") or "notification").strip().lower()
            if coordinator_batch and channel == "thread" and kind in ("done", "gave_up"):
                continue
            try:
                if channel == "thread":
                    enqueue(
                        KIND_THREAD,
                        {
                            "thread_id": target,
                            "result": body,
                            "banner": banner,
                            "kanban_event": kind,
                            "task_id": task_id,
                            "subscription_id": sub.get("id") or "",
                            "coordinator_resume": coordinator_batch,
                        },
                        run_id=run_id or task_id,
                        origin_thread_id=target,
                    )
                elif channel == "webhook":
                    enqueue(
                        KIND_WEBHOOK,
                        {
                            "url": target,
                            "job_id": run_id or task_id,
                            "prompt": banner,
                            "result": body,
                            "kanban_event": kind,
                            "task_id": task_id,
                            "subscription_id": sub.get("id") or "",
                        },
                        run_id=run_id or task_id,
                        origin_thread_id=task.get("origin_thread_id") or None,
                    )
                elif channel in ("discord", "slack"):
                    enqueue(
                        KIND_DISCORD if channel == "discord" else KIND_SLACK,
                        {
                            "url": target,
                            "banner": banner,
                            "result": body,
                            "body": body,
                            "kanban_event": kind,
                            "task_id": task_id,
                            "subscription_id": sub.get("id") or "",
                        },
                        run_id=run_id or task_id,
                        origin_thread_id=task.get("origin_thread_id") or None,
                    )
                else:
                    enqueue(
                        KIND_NOTIFICATION,
                        {
                            "prompt": banner,
                            "result": body,
                            "kanban_event": kind,
                            "task_id": task_id,
                            "subscription_id": sub.get("id") or "",
                        },
                        run_id=run_id or task_id,
                        origin_thread_id=task.get("origin_thread_id") or None,
                    )
            except Exception as exc:
                global _dispatch_enqueue_failures
                _dispatch_enqueue_failures += 1
                _log.warning(
                    "kanban_notifications enqueue failed for sub=%s task=%s channel=%s",
                    sub.get("id"),
                    task_id,
                    channel,
                    exc_info=True,
                )
                try:
                    from activity_bus import emit_activity_enriched

                    emit_activity_enriched(
                        {
                            "source": "kanban_notifications",
                            "state": "error",
                            "summary": f"Kanban notify enqueue failed ({channel})",
                            "meta": {
                                "task_id": task_id,
                                "subscription_id": sub.get("id") or "",
                                "error": str(exc)[:200],
                            },
                        },
                        tags=["kanban", "delivery"],
                    )
                except Exception:
                    _log.debug("kanban_notifications activity emit failed", exc_info=True)
    except Exception:
        _log.warning("kanban_notifications.maybe_dispatch_lifecycle raised", exc_info=True)
