# Monitor: progress, SSE stream, recipient table
# ---------------------------------------------------------------------------

async def _progress_payload(db: AsyncSession, c: Campaign) -> dict:
    """Live counters for the monitor: campaign fields + per-status breakdown."""
    breakdown = (await db.execute(
        select(CampaignRecipient.status, func.count())
        .where(CampaignRecipient.campaign_id == c.id)
        .group_by(CampaignRecipient.status))).all()
    counts = {s: int(n) for s, n in breakdown}
    jobs_agg = (await db.execute(
        select(func.count()).select_from(SendJob).where(SendJob.campaign_id == c.id)
    )).scalar() or 0
    done = counts.get("sent", 0) + counts.get("perm_failed", 0) + counts.get("suppressed", 0) + counts.get("skipped", 0)
    total = c.total_recipients or 0
    return {
        "campaign_id": c.id, "status": c.status, "total": total,
        "sent": counts.get("sent", 0), "temp_failed": counts.get("temp_failed", 0),
        "perm_failed": counts.get("perm_failed", 0),
        "suppressed": counts.get("suppressed", 0), "skipped": counts.get("skipped", 0),
        "pending": counts.get("pending", 0), "queued": counts.get("queued", 0),
        "processing": counts.get("processing", 0),
        "jobs_total": int(jobs_agg), "completed": done >= total and total > 0 and c.status in ("completed",),
        "progress_pct": round(100.0 * done / total, 1) if total else 0.0,
        "started_at": c.started_at.isoformat() if c.started_at else None,
        "finished_at": c.finished_at.isoformat() if c.finished_at else None,
    }


@router.get("/{cid}/progress")
async def get_progress(cid: int, user: User = Depends(get_current_user),
                       db: AsyncSession = Depends(get_db)):
    c = await _get_own(db, user, cid)
    return await _progress_payload(db, c)


@router.get("/{cid}/stream")
async def campaign_stream(cid: int, request: Request,
                          user: User = Depends(get_current_user),
                          db: AsyncSession = Depends(get_db)):
    """SSE live progress stream. Manual implementation (no sse-starlette): a Redis
    pub/sub subscription on campaign:{id}:progress plus periodic DB polling fallback.
    Emits event: progress with counters; event: done when campaign finishes."""
    c = await _get_own(db, user, cid)

    import redis.asyncio as aioredis
    from app.db.session import AsyncSessionLocal

    async def event_gen():
        r = aioredis.Redis.from_url(cfg.REDIS_URL, decode_responses=True)
        pubsub = r.pubsub()
        channel = f"campaign:{cid}:progress"
        await pubsub.subscribe(channel)
        heartbeat = 2.0
        poll_interval = 5.0
        last_payload = None
        last_poll = 0.0
        import time
        try:
            yield "event: connected\ndata: {\"ok\": true}\n\n"
            while True:
                if await request.is_disconnected():
                    break
                # check for pub/sub tick (non-blocking)
                msg = await pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1)
                now = time.monotonic()
                payload = None
                if msg is not None or now - last_poll >= poll_interval:
                    # refresh counters from DB with a fresh session
                    async with AsyncSessionLocal() as s:
                        camp = await s.get(Campaign, cid)
                        if camp:
                            payload = await _progress_payload(s, camp)
                    last_poll = now
                if payload is None and msg is None and now - last_poll < poll_interval:
                    await asyncio.sleep(heartbeat)
                    yield ": keepalive\n\n"
                    continue
                if payload:
                    data = json.dumps(payload)
                    if data != last_payload or msg is not None:
                        yield f"event: progress\ndata: {data}\n\n"
                        last_payload = data
                    if payload.get("status") in ("completed", "cancelled", "failed"):
                        yield f"event: done\ndata: {json.dumps(payload)}\n\n"
                        break
                await asyncio.sleep(0.2)
        finally:
            try:
                await pubsub.unsubscribe(channel)
                await pubsub.close()
                await r.aclose()
            except Exception:
                pass

    return StreamingResponse(event_gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache",
                                      "X-Accel-Buffering": "no",
                                      "Connection": "keep-alive"})


@router.get("/{cid}/recipients")
async def campaign_recipients(cid: int, status: str = Query(""),