"""Task event bus: persist events to DB and publish to Redis for live streaming."""
from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Any

import redis.asyncio as aioredis

from app.config import settings
from app.database import SessionLocal
from app.models.task import TaskEvent

_redis: aioredis.Redis | None = None

# Redis keys
TASK_QUEUE = "agentic:task_queue"

# High-frequency live-only event kinds: streamed to the UI but not written to
# Postgres, to avoid a separate transaction per progress/tool-start tick.
EPHEMERAL_KINDS = frozenset({"progress", "tool_start"})


def get_redis() -> aioredis.Redis:
    global _redis
    if _redis is None:
        _redis = aioredis.from_url(
            settings.redis_url, decode_responses=True, max_connections=32
        )
    return _redis


def _channel(task_id: int) -> str:
    return f"agentic:task:{task_id}:events"


async def dedupe_task_queue(task_id: int) -> int:
    """Remove stale duplicate entries for a task id from the Redis queue."""
    redis = get_redis()
    needle = str(task_id)
    items = await redis.lrange(TASK_QUEUE, 0, -1)
    removed = sum(1 for x in items if x == needle)
    if removed:
        await redis.delete(TASK_QUEUE)
        for item in items:
            if item != needle:
                await redis.rpush(TASK_QUEUE, item)
    return removed


async def enqueue_task(task_id: int) -> None:
    await dedupe_task_queue(task_id)
    await get_redis().rpush(TASK_QUEUE, str(task_id))


async def publish_event(
    task_id: int,
    kind: str,
    message: str | None = None,
    payload: dict[str, Any] | None = None,
    persist: bool | None = None,
) -> None:
    """Persist an event and publish it to the task's Redis channel.

    When ``persist`` is None, high-frequency ephemeral kinds (progress,
    tool_start) are streamed live but not written to the DB, while all other
    kinds are persisted. Pass an explicit bool to override.
    """
    if persist is None:
        persist = kind not in EPHEMERAL_KINDS
    if persist:
        async with SessionLocal() as db:
            db.add(TaskEvent(task_id=task_id, kind=kind, message=message, payload=payload))
            await db.commit()

    event = {
        "task_id": task_id,
        "kind": kind,
        "message": message,
        "payload": payload,
        "ts": datetime.now(timezone.utc).isoformat(),
    }
    await get_redis().publish(_channel(task_id), json.dumps(event))


async def subscribe_events(task_id: int):
    """Async generator yielding decoded events for a task."""
    pubsub = get_redis().pubsub()
    await pubsub.subscribe(_channel(task_id))
    try:
        async for message in pubsub.listen():
            if message["type"] == "message":
                yield json.loads(message["data"])
    finally:
        await pubsub.unsubscribe(_channel(task_id))
        await pubsub.close()