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(""),
                              search: str = Query(""),
                              page: int = Query(1, ge=1),
                              page_size: int = Query(50, ge=1, le=200),
                              db: AsyncSession = Depends(get_db),
                              user: User = Depends(get_current_user)):
    """Monitor table: per-recipient status, assigned account, subject/template, errors."""
    c = await _get_own(db, user, cid)
    q = select(CampaignRecipient).where(CampaignRecipient.campaign_id == cid)
    if status:
        q = q.where(CampaignRecipient.status == status)
    if search:
        q = q.where(CampaignRecipient.email.ilike(f"%{search}%"))
    total = (await db.execute(select(func.count()).select_from(q.subquery()))).scalar() or 0
    rows = (await db.execute(q.order_by(CampaignRecipient.seq)
                             .offset((page - 1) * page_size).limit(page_size))).scalars().all()

    async def names():
        acc_map, subj_map, tpl_map = {}, {}, {}
        for cr in rows:
            if cr.assigned_account_id and cr.assigned_account_id not in acc_map:
                a = await db.get(GmailAccount, cr.assigned_account_id)
                acc_map[cr.assigned_account_id] = a.email if a else "?"
            if cr.generated_subject_id and cr.generated_subject_id not in subj_map:
                s = await db.get(SubjectLine, cr.generated_subject_id)
                subj_map[cr.generated_subject_id] = (s.subject[:80] if s else "?")
            if cr.generated_template_id and cr.generated_template_id not in tpl_map:
                t = await db.get(EmailTemplate, cr.generated_template_id)
                tpl_map[cr.generated_template_id] = (t.name if t else "?")
        return acc_map, subj_map, tpl_map

    acc_map, subj_map, tpl_map = await names()
    out = []
    for cr in rows:
        out.append({
            "id": cr.id, "seq": cr.seq, "email": cr.email, "status": cr.status,
            "account": acc_map.get(cr.assigned_account_id, ""),
            "subject": subj_map.get(cr.generated_subject_id, ""),
            "template": tpl_map.get(cr.generated_template_id, ""),
            "invoice_number": cr.generated_invoice_number,
            "amount": cr.generated_amount, "shipping": cr.generated_shipping,
            "attempts": cr.attempt_count, "message_id": cr.msg_id,
            "thread_id": cr.thread_id,
            "sent_at": cr.sent_at.isoformat() if cr.sent_at else None,
            "last_error": (cr.last_error or "")[:200],
        })
    return {"recipients": out, "total": total, "page": page, "page_size": page_size}


@router.get("/{cid}/export")
async def export_campaign(cid: int, db: AsyncSession = Depends(get_db),
                          user: User = Depends(get_current_user)):
    """CSV export of campaign results (one row per campaign recipient)."""
    c = await _get_own(db, user, cid)
    rows = (await db.execute(
        select(CampaignRecipient).where(CampaignRecipient.campaign_id == cid)
        .order_by(CampaignRecipient.seq))).scalars().all()

    def generate():
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(["seq", "email", "status", "account", "subject", "template",
                    "invoice_number", "amount", "shipping", "attempts",
                    "message_id", "thread_id", "sent_at", "last_error"])
        yield buf.getvalue()
        buf.seek(0); buf.truncate(0)
        for cr in rows:
            w.writerow([cr.seq, cr.email, cr.status, cr.assigned_account_id,
                        cr.generated_subject_id, cr.generated_template_id,
                        cr.generated_invoice_number, cr.generated_amount or "",
                        cr.generated_shipping or "", cr.attempt_count,
                        cr.msg_id, cr.thread_id,
                        cr.sent_at.isoformat() if cr.sent_at else "",
                        (cr.last_error or "")[:300]])
            yield buf.getvalue()
            buf.seek(0); buf.truncate(0)

    fname = re.sub(r"[^A-Za-z0-9_-]+", "_", c.name)[:60] or f"campaign_{cid}"
    return StreamingResponse(generate(), media_type="text/csv",