"""Push notifications behind the Task Designer's Firebase nodes, through one
"firebase" Connector.

Web and mobile go through Firebase Cloud Messaging's HTTP v1 API. FCM has no
native Windows client, so Windows is reached two ways: a browser, installed
PWA or Electron app on Windows has an ordinary FCM web token; a native Windows
app has a WNS channel URI, sent through Microsoft's Windows Push Notification
Services with credentials on the same connector.

A channel URI is only ever posted to when its host is notify.windows.com —
Microsoft's own rule, and what stops a planted URL receiving our bearer token.
"""

import asyncio
import hashlib
import json
import re
import time
from typing import Any
from urllib.parse import quote, urlsplit
from xml.sax.saxutils import escape, quoteattr

import httpx

from app.crud import google_auth
from app.crud.google_auth import GoogleError, error_reason

FCM_SCOPE = ("https://www.googleapis.com/auth/firebase.messaging",)
FCM_BASE = "https://fcm.googleapis.com/v1/projects"
WNS_LIVE_TOKEN_URL = "https://login.live.com/accesstoken.srf"
ENTRA_TOKEN_URL = "https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token"
WNS_ENTRA_SCOPE = "https://wns.windows.com/.default"
WNS_LIVE_SCOPE = "notify.windows.com"

MAX_TARGETS = 1000
_CONCURRENCY = 20
_MAX_REPORTED_ERRORS = 10
_TOPIC = re.compile(r"^[a-zA-Z0-9\-_.~%]{1,900}$")
_GONE_CODES = {"UNREGISTERED", "NOT_FOUND"}

_wns_tokens: dict[str, tuple[str, float]] = {}


def project_id(config: dict) -> str:
    project = str(config.get("project_id") or "").strip()
    if not project:
        project = str(google_auth.service_account(config).get("project_id") or "").strip()
    if not project:
        raise GoogleError("the connector needs a Firebase project id")
    return project


def targets(value: Any) -> list[str]:
    """One token, a list, a comma/line separated text, or a list of records
    with a token/channel field — duplicates and blanks removed."""
    if value is None or value == "":
        return []
    if isinstance(value, str):
        text = value.strip()
        if text.startswith("["):
            try:
                value = json.loads(text)
            except ValueError as exc:
                raise GoogleError("the list of tokens isn't valid JSON") from exc
        else:
            value = re.split(r"[,\s]+", text)
    if not isinstance(value, (list, tuple)):
        value = [value]
    found: list[str] = []
    for item in value:
        if isinstance(item, dict):
            item = next((item[k] for k in ("token", "fcm_token", "device_token", "channel", "channel_uri") if item.get(k)), "")
        item = str(item or "").strip()
        if item and item not in found:
            found.append(item)
    return found


def _flag(settings: dict, key: str, default: bool = False) -> bool:
    value = settings.get(key)
    if value is None or value == "":
        return default
    if isinstance(value, bool):
        return value
    return str(value).strip().lower() in ("yes", "true", "1", "on")


def _text(settings: dict, key: str) -> str:
    value = settings.get(key)
    return "" if value is None else str(value).strip()


def _prune(value: Any) -> Any:
    """Removes what's unset, so a platform block only exists when it says something."""
    if isinstance(value, dict):
        kept = {k: _prune(v) for k, v in value.items()}
        return {k: v for k, v in kept.items() if v not in (None, "", {}, [])}
    return value


def _int(settings: dict, key: str, label: str, low: int = 0, high: int = 2_419_200) -> int | None:
    text = _text(settings, key)
    if not text:
        return None
    try:
        number = int(float(text))
    except ValueError as exc:
        raise GoogleError(f"{label} must be a number") from exc
    return max(low, min(high, number))


def data_payload(settings: dict) -> dict[str, str]:
    """FCM only takes string values in "data"."""
    value = settings.get("data")
    if isinstance(value, str):
        text = value.strip()
        if not text:
            return {}
        try:
            value = json.loads(text)
        except ValueError as exc:
            raise GoogleError('the data isn\'t valid JSON — use an object like {"order_id": "{{id}}"}') from exc
    if not value:
        return {}
    if not isinstance(value, dict):
        raise GoogleError("the data must be a JSON object")
    return {str(k): (v if isinstance(v, str) else json.dumps(v, default=str)) for k, v in value.items()}


def build_message(settings: dict) -> dict:
    """The FCM message minus its target. Each platform block is optional and
    only applies on that platform."""
    title, body = _text(settings, "title"), _text(settings, "body")
    data = data_payload(settings)
    silent = _flag(settings, "silent")
    if not silent and not title and not body:
        raise GoogleError("set a title or a message")
    if silent and not data:
        raise GoogleError("a silent notification needs data to deliver")
    link = _text(settings, "web_link")
    if link and not link.lower().startswith("https://"):
        raise GoogleError("the web link must start with https://")
    ttl = _int(settings, "ttl_seconds", "time to live")
    urgency = _text(settings, "urgency") or "high"
    message: dict[str, Any] = {
        "notification": None if silent else {"title": title, "body": body, "image": _text(settings, "image")},
        "data": data,
        "android": {
            "priority": "HIGH" if urgency == "high" else "NORMAL",
            "ttl": f"{ttl}s" if ttl is not None else None,
            "collapse_key": _text(settings, "collapse_key"),
            "notification": None if silent else {
                "channel_id": _text(settings, "android_channel"), "sound": _text(settings, "sound"), "icon": _text(settings, "android_icon"),
                "color": _text(settings, "android_color"), "tag": _text(settings, "tag"), "click_action": _text(settings, "android_click_action"),
            },
        },
        "apns": {
            "headers": {
                "apns-priority": "5" if silent or urgency != "high" else "10",
                "apns-push-type": "background" if silent else None,
                "apns-expiration": str(int(time.time()) + ttl) if ttl else None,
                "apns-collapse-id": _text(settings, "collapse_key"),
            },
            "payload": {"aps": {
                "sound": None if silent else _text(settings, "sound"),
                "badge": _int(settings, "ios_badge", "badge", 0, 100_000),
                "thread-id": _text(settings, "ios_thread"),
                "content-available": 1 if silent else None,
            }},
        },
        "webpush": {
            "headers": {"TTL": str(ttl) if ttl is not None else None, "Urgency": "high" if urgency == "high" else "normal", "Topic": _text(settings, "collapse_key")[:32] if re.fullmatch(r"[A-Za-z0-9_-]+", _text(settings, "collapse_key")) else None},
            "notification": None if silent else {
                "icon": _text(settings, "web_icon"), "badge": _text(settings, "web_badge"), "tag": _text(settings, "tag"),
                "requireInteraction": True if _flag(settings, "web_require_interaction") else None,
            },
            "fcm_options": {"link": link},
        },
    }
    pruned = _prune(message)
    apns = pruned.get("apns") or {}
    # APNs needs an "aps" even when nothing was set on it.
    if apns and "payload" not in apns:
        apns["payload"] = {"aps": {}}
    return pruned


def toast_xml(settings: dict) -> str:
    title, body = _text(settings, "title"), _text(settings, "body")
    image = _text(settings, "image")
    launch = _text(settings, "web_link") or _text(settings, "windows_launch")
    launch_attr = f" launch={quoteattr(launch)}" if launch else ""
    lines = "".join(f"<text>{escape(t)}</text>" for t in (title, body) if t)
    picture = f'<image placement="hero" src={quoteattr(image)}/>' if image else ""
    return f'<toast{launch_attr}><visual><binding template="ToastGeneric">{lines}{picture}</binding></visual></toast>'


def valid_channel(uri: str) -> bool:
    try:
        parts = urlsplit(uri)
        host = (parts.hostname or "").lower()
    except ValueError:
        return False
    return parts.scheme == "https" and (host == "notify.windows.com" or host.endswith(".notify.windows.com"))


def _short(target: str) -> str:
    """A token is a credential to reach someone's device — never quoted whole."""
    return f"…{target[-6:]}" if len(target) > 6 else "…"


async def _send_fcm(config: dict, project: str, message: dict, validate_only: bool) -> tuple[bool, str, str]:
    """(ok, message name or error code, error words)."""
    url = f"{FCM_BASE}/{quote(project, safe='')}/messages:send"
    payload: dict[str, Any] = {"message": message}
    if validate_only:
        payload["validate_only"] = True
    try:
        resp = await google_auth.request(config, FCM_SCOPE, "POST", url, json=payload)
    except httpx.HTTPError as exc:
        return False, "NETWORK", f"couldn't reach Firebase: {exc}"
    if resp.status_code < 400:
        return True, str(resp.json().get("name") or ""), ""
    code, words = error_reason(resp)
    return False, code, words


async def _wns_token(config: dict) -> str:
    mode = _text(config, "wns_auth")
    client_id, secret = _text(config, "wns_client_id"), _text(config, "wns_client_secret")
    if mode not in ("entra", "package") or not client_id or not secret:
        raise GoogleError("to reach native Windows apps, set the Windows (WNS) credentials on the Firebase connector")
    key = hashlib.sha256(json.dumps([mode, client_id, secret, _text(config, "wns_tenant_id")]).encode()).hexdigest()
    cached = _wns_tokens.get(key)
    if cached and cached[1] > time.time() + 60:
        return cached[0]
    if mode == "entra":
        tenant = _text(config, "wns_tenant_id")
        if not re.fullmatch(r"[A-Za-z0-9.\-]{1,100}", tenant):
            raise GoogleError("the connector needs the Microsoft Entra directory (tenant) ID")
        url, scope = ENTRA_TOKEN_URL.format(tenant=tenant), WNS_ENTRA_SCOPE
    else:
        url, scope = WNS_LIVE_TOKEN_URL, WNS_LIVE_SCOPE
    async with google_auth._client() as client:
        resp = await client.post(url, data={"grant_type": "client_credentials", "client_id": client_id, "client_secret": secret, "scope": scope})
    if resp.status_code >= 400:
        raise GoogleError(f"Microsoft refused the Windows push credentials (HTTP {resp.status_code})")
    body = resp.json()
    token = body.get("access_token")
    if not token:
        raise GoogleError("Microsoft returned no Windows push token")
    if len(_wns_tokens) >= 256:
        _wns_tokens.clear()
    _wns_tokens[key] = (token, time.time() + float(body.get("expires_in") or 3600))
    return token


async def _send_wns(config: dict, channel: str, settings: dict) -> tuple[bool, str, str]:
    if not valid_channel(channel):
        return False, "BAD_CHANNEL", "not a notify.windows.com channel URI"
    if _text(settings, "windows_type") == "raw":
        content, headers = json.dumps({"title": _text(settings, "title"), "body": _text(settings, "body"), "data": data_payload(settings)}).encode(), {"Content-Type": "application/octet-stream", "X-WNS-Type": "wns/raw"}
    else:
        content, headers = toast_xml(settings).encode(), {"Content-Type": "text/xml", "X-WNS-Type": "wns/toast"}
    ttl = _int(settings, "ttl_seconds", "time to live")
    if ttl is not None:
        headers["X-WNS-TTL"] = str(ttl)
    tag = _text(settings, "tag")
    if tag and headers["X-WNS-Type"] == "wns/toast":
        headers["X-WNS-Tag"] = tag[:16]
    for attempt in (0, 1):
        token = await _wns_token(config)
        try:
            async with google_auth._client() as client:
                resp = await client.post(channel, content=content, headers={**headers, "Authorization": f"Bearer {token}"})
        except httpx.HTTPError as exc:
            return False, "NETWORK", f"couldn't reach Windows push: {exc}"
        if resp.status_code == 401 and attempt == 0:
            _wns_tokens.clear()
            continue
        if resp.status_code < 400:
            return True, resp.headers.get("X-WNS-Msg-ID", ""), ""
        if resp.status_code in (404, 410):
            return False, "UNREGISTERED", "this channel has expired"
        return False, f"HTTP_{resp.status_code}", resp.headers.get("X-WNS-Error-Description", "") or f"Windows push said HTTP {resp.status_code}"
    return False, "AUTH", "Windows push refused the credentials"


def _summarise(outcomes: list[tuple[str, str, tuple[bool, str, str]]], dry_run: bool) -> tuple[str, dict]:
    """outcomes: (platform, target, (ok, id or code, words))."""
    sent = [o for o in outcomes if o[2][0]]
    failed = [o for o in outcomes if not o[2][0]]
    gone = [o[1] for o in failed if o[2][1] in _GONE_CODES]
    errors = [f"{o[2][1]}: {o[2][2]} ({_short(o[1])})" for o in failed if o[2][1] not in _GONE_CODES][:_MAX_REPORTED_ERRORS]
    verb = "validated" if dry_run else "sent"
    detail = f"{verb} {len(sent)} of {len(outcomes)}"
    if gone:
        detail += f", {len(gone)} no longer registered"
    if errors:
        detail += f" — {errors[0]}"
    ids = [o[2][1] for o in sent if o[2][1]]
    return detail, {
        "sent": len(sent), "failed": len(failed), "total": len(outcomes),
        "fcm_sent": sum(1 for o in sent if o[0] == "fcm"), "windows_sent": sum(1 for o in sent if o[0] == "wns"),
        "message_id": ids[0] if ids else "", "message_ids": ids,
        "invalid_tokens": [o[1] for o in failed if o[2][1] in _GONE_CODES and o[0] == "fcm"],
        "expired_channels": [o[1] for o in failed if o[2][1] in _GONE_CODES and o[0] == "wns"],
        "errors": errors, "dry_run": dry_run,
    }


async def send(config: dict, settings: dict) -> tuple[bool, str, dict]:
    """(ok, detail, result). Fails only when something was targeted and
    nothing got through (unless stop_on_failure is off) — one stale token among
    many shouldn't stop a task; invalid_tokens lets it clean them up."""
    target_kind = _text(settings, "target") or "tokens"
    dry_run = _flag(settings, "dry_run")
    tokens: list[str] = []
    topic = condition = ""
    if target_kind == "tokens":
        tokens = targets(settings.get("tokens"))
    elif target_kind == "topic":
        topic = _text(settings, "topic").removeprefix("/topics/")
        if not _TOPIC.match(topic):
            raise GoogleError("set a topic name — letters, numbers and - _ . ~ % only")
    elif target_kind == "condition":
        condition = _text(settings, "condition")
        if not condition:
            raise GoogleError("set the topic condition, like 'news' in topics && 'india' in topics")
    elif target_kind != "none":  # "none": native Windows apps only
        raise GoogleError(f"unknown target {target_kind}")
    channels = targets(settings.get("windows_channels"))
    if len(tokens) + len(channels) > MAX_TARGETS:
        raise GoogleError(f"at most {MAX_TARGETS} devices per step — use a topic, or a Loop over batches")
    if not tokens and not topic and not condition and not channels:
        return True, "no devices to notify — nothing sent", _summarise([], dry_run)[1]

    gate = asyncio.Semaphore(_CONCURRENCY)
    outcomes: list[tuple[str, str, tuple[bool, str, str]]] = []
    fcm_jobs: list[tuple[str, dict]] = []
    if tokens or topic or condition:
        base = build_message(settings)
        project = project_id(config)
        fcm_jobs = [(t, {**base, "token": t}) for t in tokens]
        if topic:
            fcm_jobs.append((topic, {**base, "topic": topic}))
        if condition:
            fcm_jobs.append(("condition", {**base, "condition": condition}))

    async def fcm_one(target: str, message: dict) -> None:
        async with gate:
            outcomes.append(("fcm", target, await _send_fcm(config, project, message, dry_run)))

    async def wns_one(channel: str) -> None:
        async with gate:
            if dry_run:
                outcomes.append(("wns", channel, (valid_channel(channel), "" if valid_channel(channel) else "BAD_CHANNEL", "" if valid_channel(channel) else "not a notify.windows.com channel URI")))
            else:
                outcomes.append(("wns", channel, await _send_wns(config, channel, settings)))

    await asyncio.gather(*(fcm_one(t, m) for t, m in fcm_jobs), *(wns_one(c) for c in channels))
    detail, result = _summarise(outcomes, dry_run)
    ok = result["sent"] > 0 or not _flag(settings, "stop_on_failure", True)
    return ok, detail, result


async def manage_topic(config: dict, settings: dict) -> tuple[bool, str, dict]:
    """Subscribe / unsubscribe / list — FCM's topicSubscriptions REST API, one
    device at a time. Already subscribed counts as done."""
    operation = _text(settings, "operation") or "subscribe"
    tokens = targets(settings.get("tokens"))
    if not tokens:
        return True, "no devices — nothing to do", {"done": 0, "failed": 0, "invalid_tokens": [], "topics": []}
    if len(tokens) > MAX_TARGETS:
        raise GoogleError(f"at most {MAX_TARGETS} devices per step")
    base = f"{FCM_BASE}/{quote(project_id(config), safe='')}/registrations"

    if operation == "list":
        resp = await google_auth.request(config, FCM_SCOPE, "GET", f"{base}/{quote(tokens[0], safe='')}/topicSubscriptions")
        if resp.status_code >= 400:
            code, words = error_reason(resp)
            raise GoogleError(f"Firebase said {code}: {words}")
        subs = resp.json().get("topic_subscriptions") or []
        names = [s.get("topic_name", "").rsplit("/", 1)[-1] for s in subs]
        return True, f"{len(names)} topic(s)", {"topics": names, "count": len(names)}

    topic = _text(settings, "topic").removeprefix("/topics/")
    if not _TOPIC.match(topic):
        raise GoogleError("set a topic name — letters, numbers and - _ . ~ % only")
    if operation not in ("subscribe", "unsubscribe"):
        raise GoogleError(f"unknown operation {operation}")
    gate = asyncio.Semaphore(_CONCURRENCY)
    done = 0
    gone: list[str] = []
    errors: list[str] = []

    async def one(token: str) -> None:
        nonlocal done
        reg = quote(token, safe="")
        async with gate:
            if operation == "subscribe":
                resp = await google_auth.request(config, FCM_SCOPE, "POST", f"{base}/{reg}/topicSubscriptions", params={"topic_name": topic}, json={})
            else:
                resp = await google_auth.request(config, FCM_SCOPE, "DELETE", f"{base}/{reg}/topicSubscriptions/{quote(topic, safe='')}", json={})
        if resp.status_code < 400 or resp.status_code == 409:
            done += 1
            return
        code, words = error_reason(resp)
        if resp.status_code == 404 or code in _GONE_CODES:
            if operation == "unsubscribe":
                done += 1  # it isn't subscribed, which is what was asked for
            else:
                gone.append(token)
            return
        errors.append(f"{code}: {words} ({_short(token)})")

    await asyncio.gather(*(one(t) for t in tokens))
    verb = "subscribed" if operation == "subscribe" else "unsubscribed"
    detail = f"{verb} {done} of {len(tokens)} to {topic}" + (f" — {errors[0]}" if errors else "")
    result = {"done": done, "failed": len(gone) + len(errors), "invalid_tokens": gone, "errors": errors[:_MAX_REPORTED_ERRORS], "topic": topic}
    return done > 0 or not errors, detail, result


async def test_credentials(config: dict) -> tuple[bool, str]:
    """A validate-only send to a topic, so nothing reaches any device."""
    try:
        project = project_id(config)
        ok, code, words = await _send_fcm(config, project, {"topic": "flowfabric-connection-test", "notification": {"title": "Connection test"}}, True)
        if not ok:
            return False, f"Firebase said {code}: {words}"
        detail = f"Connected to Firebase project {project}"
        if _text(config, "wns_auth") in ("entra", "package"):
            await _wns_token(config)
            detail += "; Windows push credentials accepted"
        return True, detail
    except GoogleError as exc:
        return False, str(exc)
    except httpx.HTTPError as exc:
        return False, f"couldn't reach the push service: {exc}"


def _f(key: str, label: str, *, secret: bool = False, required: bool = True, placeholder: str = "", help: str = "", options: list | None = None, advanced: bool = False) -> dict:
    return {"key": key, "label": label, "secret": secret, "required": required, "placeholder": placeholder, "help": help, "options": options, "advanced": advanced, "multiline": False}


PROVIDERS: list[dict] = [
    {
        "id": "fcm", "name": "Firebase Cloud Messaging", "tagline": "Web, Android and iOS — with Windows through WNS", "environments": [],
        "docs_url": "https://firebase.google.com/docs/cloud-messaging", "capabilities": [],
        "notes": "Your apps register with Firebase and give you their device tokens — keep them in a datasource and pass them to the node. A browser, PWA or Electron app on Windows has a normal web token; only native Windows apps need the WNS settings.",
        "fields": [
            _f("service_account_json", "Firebase service account key (JSON)", secret=True, placeholder='{ "type": "service_account", … }', help="Firebase console › Project settings › Service accounts › Generate new private key. Paste the whole file, or save it under Settings › Secrets and pick it here."),
            _f("project_id", "Project ID", required=False, advanced=True, help="Taken from the key file unless you set it here."),
            _f("wns_auth", "Native Windows apps (WNS)", required=False, advanced=True, options=[["none", "Not used"], ["entra", "Windows App SDK (Microsoft Entra ID)"], ["package", "UWP (Package SID, legacy)"]], help="Only for native Windows apps, which get a channel URI instead of a Firebase token."),
            _f("wns_client_id", "WNS client ID", required=False, advanced=True, help="Entra: the Application (client) ID. UWP: ms-app://<Package SID>."),
            _f("wns_client_secret", "WNS client secret", secret=True, required=False, advanced=True),
            _f("wns_tenant_id", "Entra directory (tenant) ID", required=False, advanced=True, help="Entra only."),
        ],
    },
]