if campaign_id:
        q = q.where(EmailEvent.campaign_id == campaign_id)
    if event_type:
        q = q.where(EmailEvent.event_type == event_type)
    total = (await db.execute(
        select(func.count()).select_from(q.subquery()))).scalar() or 0
    rows = (await db.execute(
        q.order_by(EmailEvent.id.desc()).offset((page - 1) * page_size).limit(page_size)
    )).all()
    out = []
    for e, cr in rows:
        out.append({
            "id": e.id,
            "campaign_id": e.campaign_id,
            "email": cr.email,
            "event_type": e.event_type,
            "detail": (e.detail or "")[:300],
            "message_id": e.message_id,
            "created_at": e.created_at.isoformat() if e.created_at else None,
        })
    return {"events": out, "total": total, "page": page, "page_size": page_size}


@router.get("/audit")
async def list_audit(action: 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)):
    """Audit log (own actions only). Secrets are redacted at write time."""
    q = select(AuditLog).where(AuditLog.user_id == user.id)
    if action:
        q = q.where(AuditLog.action == action)
    total = (await db.execute(
        select(func.count()).select_from(q.subquery()))).scalar() or 0
    rows = (await db.execute(
        q.order_by(AuditLog.id.desc()).offset((page - 1) * page_size).limit(page_size)
    )).scalars().all()
    out = []
    for r in rows:
        out.append({
            "id": r.id,
            "action": r.action,
            "entity": r.entity,
            "entity_id": r.entity_id,
            "detail": r.detail if isinstance(r.detail, dict) else {},
            "ip_address": r.ip_address,
            "created_at": r.created_at.isoformat() if r.created_at else None,
        })
    return {"audit": out, "total": total, "page": page, "page_size": page_size}


@router.get("/dlq")
async def list_dlq(page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200),
                   user: User = Depends(get_current_user)):
    """Dead-letter queue (Redis list dlq:send): jobs that exhausted retries."""
    import redis as redis_lib
    from app.core.config import settings
    r = redis_lib.Redis.from_url(settings.REDIS_URL, decode_responses=True)
    try:
        items = r.lrange("dlq:send", 0, -1)
    except Exception as e:  # Redis down — don't 500 the whole logs page
        return {"dlq": [], "total": 0, "error": f"Redis unavailable: {e}"}
    import json as _json
    out = []
    for raw in items:
        try:
            out.append(_json.loads(raw))
        except Exception:
            out.append({"raw": raw[:200]})
    total = len(out)
    start = (page - 1) * page_size
    return {"dlq": out[start:start + page_size], "total": total,
            "page": page, "page_size": page_size}


@router.get("/export")
async def export_logs(campaign_id: int = Query(0),
                      outcome: str = Query(""),
                      db: AsyncSession = Depends(get_db),
                      user: User = Depends(get_current_user)):
    """CSV export of send attempts for a campaign (all pages)."""
    q = (select(SendAttempt, CampaignRecipient)
         .join(CampaignRecipient, SendAttempt.campaign_recipient_id == CampaignRecipient.id)
         .join(Campaign, Campaign.id == SendAttempt.campaign_id)
         .where(Campaign.user_id == user.id))
    if campaign_id:
        q = q.where(SendAttempt.campaign_id == campaign_id)
    if outcome:
        q = q.where(SendAttempt.outcome == outcome)
    rows = (await db.execute(q.order_by(SendAttempt.id.asc()))).all()

    def generate():
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(["attempt_id", "campaign_id", "email", "attempt_number", "outcome",
                    "error_class", "error_message", "message_id", "provider_response", "timestamp"])
        yield buf.getvalue()
        buf.seek(0); buf.truncate(0)
        for a, cr in rows:
            w.writerow([a.id, a.campaign_id, cr.email, a.attempt_number, a.outcome,
                        a.error_class, (a.error_message or "")[:500], a.message_id,
                        (a.provider_response or "")[:500],
                        a.created_at.isoformat() if a.created_at else ""])
            yield buf.getvalue()
            buf.seek(0); buf.truncate(0)

    return StreamingResponse(generate(), media_type="text/csv",
                             headers={"Content-Disposition":
                                      f"attachment; filename=send_logs{('_' + str(campaign_id)) if campaign_id else 's'}.csv"})


@router.get("/recipient/{cr_id}/history")
async def recipient_history(cr_id: int, db: AsyncSession = Depends(get_db),
                            user: User = Depends(get_current_user)):
    """Full attempt history for one campaign recipient (modal in Logs page)."""
    cr = await db.get(CampaignRecipient, cr_id)
    if not cr:
        raise HTTPException(status_code=404, detail="Recipient not found")
    camp = await db.get(Campaign, cr.campaign_id)
    if not camp or camp.user_id != user.id: