"""Send task: real Gmail API/SMTP send with retry, backoff, DLQ, pause/cancel, rate limits.

Status model (no fake success):
  queued -> processing -> sent (Gmail accepted, message ID recorded)
  temporary errors -> retry with exponential backoff + jitter, dead-letter after max attempts
  permanent errors -> perm_failed (no retry)
  suppressed / skipped -> dropped pre-send
"""
import logging
import random
import re as _re
import time
from datetime import date, datetime, timedelta, timezone

import googleapiclient.errors
import redis as redis_lib
from celery import Task
from sqlalchemy import select

from app.core.config import settings
from app.db.session import SyncSession
from app.models import (Attachment, Campaign, CampaignRecipient, EmailEvent,
                        EmailTemplate, GmailAccount, OAuthToken, Recipient,
                        SendAttempt, SendJob, SubjectLine, SuppressionEntry)
from app.services import attachments_service, message_builder
from app.services.gmail_service import build_gmail_client_from_account
from app.services.variable_engine import GeneratedValues, VariableEngine
from app.worker.celery_app import celery_app

log = logging.getLogger("send")

try:
    rdb = redis_lib.from_url(settings.REDIS_URL, socket_timeout=2)
except Exception:
    rdb = None

TEMPORARY_PATTERNS = ["rate limit", "rate-limit", "429", "quota", "backend error", "transient",
                      "temporarily", "503", "502", "500", "timeout", "connection", "try again",
                      "resourceexhausted", "tempfail", "unavailable", "internal error", "deadline"]
PERMANENT_PATTERNS = ["invalid grant", "401", "403", "unauthorized", "forbidden",
                      "invalidrecipient", "failedprecondition", "invalid argument", "malformed",
                      "does not exist", "no such user", "550", "553", "554", "rejected",
                      "policy violation", "disabled", "authentication"]


def classify_error(msg: str) -> str:
    m = (msg or "").lower()
    for p in TEMPORARY_PATTERNS:
        if p in m:
            return "temporary"
    for p in PERMANENT_PATTERNS:
        if p in m:
            return "permanent"
    return "unknown"


def now():
    return datetime.now(timezone.utc)


def _effective_invoice_date(invoice_date_str: str) -> date:
    """Parse the effective invoice date (ISO) safely; fall back to today."""
    try:
        return date.fromisoformat(invoice_date_str)
    except (TypeError, ValueError):
        return date.today()


class SendTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        log.error("Task %s failed: %s", task_id, str(exc)[:200])


@celery_app.task(bind=True, base=SendTask, name="app.tasks.send_tasks.send_one",
                 ignore_result=False)
def send_one(self, campaign_id: int, campaign_recipient_id: int):
    """Process one recipient: build message, send through Gmail, record real outcome."""
    with SyncSession() as db:
        campaign = db.get(Campaign, campaign_id)
        if campaign is None:
            return {"ok": False, "status": "cancelled", "reason": "campaign deleted"}
        if campaign.status in ("paused", "cancelled", "completed"):
            return {"ok": False, "status": "cancelled", "reason": f"campaign {campaign.status}"}

        cr = db.get(CampaignRecipient, campaign_recipient_id)
        if cr is None:
            return {"ok": False, "status": "cancelled", "reason": "campaign_recipient deleted"}
        if cr.status in ("sent", "suppressed", "perm_failed"):
            return {"ok": True, "status": cr.status, "reason": "already finalized"}

        job = (db.execute(select(SendJob)
                          .where(SendJob.campaign_recipient_id == cr.id,
                                 SendJob.campaign_id == campaign.id))).scalar_one_or_none()

        # Re-check suppression at send time (recipient may have unsubscribed since enqueue)
        sup = (db.execute(select(SuppressionEntry)
                          .where(SuppressionEntry.email == cr.email))).scalar_one_or_none()
        if sup:
            cr.status = "suppressed"
            campaign.suppressed_count += 1
            cr.last_error = "Suppressed at send time"
            db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                              event_type="suppressed", detail="Suppressed at send time"))
            db.commit()
            _publish_progress(campaign.id)
            return {"ok": True, "status": "suppressed", "reason": "suppression hit"}

        account = db.get(GmailAccount, cr.assigned_account_id) if cr.assigned_account_id else None
        if account is None or not account.is_active or account.status != "authorized":
            cr.status = "skipped"
            campaign.skipped_count += 1
            cr.last_error = "Assigned account unavailable/unauthorized"
            db.commit()
            _publish_progress(campaign.id)
            return {"ok": False, "status": "skipped", "reason": "account unavailable"}

        # Per-account daily cap (never exceed provider limits)
        today = now().date()
        if account.last_send_date and account.last_send_date.date() == today \
                and account.sends_today >= account.daily_limit:
            campaign.status = "paused"
            db.commit()
            _publish_control(campaign.id, "pause",
                             f"Account {account.email} reached daily limit ({account.daily_limit})")
            return {"ok": False, "status": "paused", "reason": "daily limit reached"}

        if account.cooldown_until and account.cooldown_until > now():
            _reschedule(db, self, campaign.id, cr, job, account.cooldown_until - now(),
                        "account cooling down")
            return {"ok": False, "status": "rescheduled", "reason": "cooldown"}

        # Per-account rate window (0 = disabled; `or` coercion would silently
        # turn an explicit 0 into 1.0)
        interval = (float(campaign.send_interval_seconds)
                    if campaign.send_interval_seconds is not None else 1.0)
        ok, wait = _rate_window(f"acct:{account.id}", interval)
        if not ok:
            _reschedule(db, self, campaign.id, cr, job, timedelta(seconds=wait),
                        "rate limit window")
            return {"ok": False, "status": "rescheduled", "reason": "rate window"}

        # ------------------------------------------------------------- build message
        recipient = db.get(Recipient, cr.recipient_id)
        subject_row = db.get(SubjectLine, cr.generated_subject_id) if cr.generated_subject_id else None
        template = db.get(EmailTemplate, cr.generated_template_id) if cr.generated_template_id else None
        subject = subject_row.subject if subject_row else campaign.name
        if template is None:
            cr.status = "perm_failed"
            cr.last_error = "No template assigned"
            campaign.failed_count += 1
            db.commit()
            return {"ok": False, "status": "perm_failed", "reason": "no template"}

        var_cfg = dict(campaign.variable_config or {})
        engine = VariableEngine(var_cfg)
        _today = date.today()
        g = GeneratedValues(
            invoice_number=cr.generated_invoice_number,
            amount=cr.generated_amount,
            shipping=cr.generated_shipping,
            random_name=engine.generate_random_name(),
            invoice_date=var_cfg.get("invoice_date") or _today.isoformat(),
            due_date=var_cfg.get("due_date")
                      or (_effective_invoice_date(var_cfg.get("invoice_date") or _today.isoformat())
                          + timedelta(days=int(var_cfg.get("due_days", 14)))).isoformat(),
            phone=cr.generated_phone or var_cfg.get("company_phone", ""),
        )
        unsub_url = _unsubscribe_url(recipient, campaign)
        rec = _recipient_dict(recipient)
        cf = (recipient.custom_fields if recipient else None) or {}
        # Render variables in the subject line too (%RandomTotal%, %FirstName%,
        # %InvoiceNumber% …) — same engine and context as the body, so the
        # subject a recipient sees matches the rendered email body.
        subject = engine.substitute(subject, rec, g, custom_fields=cf,
                                    extra={"unsubscribe_link": unsub_url}).text
        result_html = engine.substitute(template.body_html, rec, g, custom_fields=cf,
                                        extra={"unsubscribe_link": unsub_url}).text
        result_text = engine.substitute(template.body_text or _plain_fallback(result_html),
                                        rec, g, custom_fields=cf,
                                        extra={"unsubscribe_link": unsub_url}).text

        attachments, att_names = [], []
        # ---- auto-conversion modes: convert the rendered email BODY itself ----
        # body_pdf / body_image / body_word take the fully substituted HTML body
        # and attach it as a real PDF / PNG / Word document per campaign config.
        if campaign.attach_mode in ("body_pdf", "body_image", "body_word"):
            _conv = {
                "body_pdf": ("pdf", message_builder.html_to_pdf, "application/pdf"),
                "body_image": ("png", message_builder.html_to_image, "image/png"),
                "body_word": ("docx", message_builder.html_to_docx,
                              "application/vnd.openxmlformats-officedocument"
                              ".wordprocessingml.document"),
            }[campaign.attach_mode]
            _ext, _fn, _mime = _conv
            try:
                conv_bytes = _fn(result_html)
            except message_builder.HtmlValidationError as e:
                cr.status = "perm_failed"
                cr.last_error = f"Body conversion failed: {e}"
                campaign.failed_count += 1
                db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                                  event_type="perm_failed", detail=f"conversion failed: {e}"))
                db.commit()
                _publish_progress(campaign.id)
                return {"ok": False, "status": "perm_failed", "reason": "body conversion failed"}
            fname = message_builder.generate_attachment_filename(
                var_cfg.get("attachment_kind", "invoice"), cr.generated_invoice_number, ext=_ext)
            attachments.append({"filename": fname, "content": conv_bytes, "mime_type": _mime})
            att_names.append(fname)
        if campaign.attach_mode == "pdf" and campaign.attach_pdf:
            att_html = ""
            if campaign.attach_pdf_template_id:
                att_tpl = db.get(EmailTemplate, campaign.attach_pdf_template_id)
                if att_tpl:
                    att_html = engine.substitute(att_tpl.body_html, rec, g, custom_fields=cf,
                                                 extra={"unsubscribe_link": unsub_url}).text
            elif campaign.attachment_id:
                att = db.get(Attachment, campaign.attachment_id)
                if att and att.kind == "html":
                    att_html = engine.substitute(att.html_attachment_body or "", rec, g,
                                                 custom_fields=cf,
                                                 extra={"unsubscribe_link": unsub_url}).text
            if att_html:
                pdf_bytes = message_builder.html_to_pdf(att_html)
                fname = message_builder.generate_attachment_filename(
                    var_cfg.get("attachment_kind", "invoice"), cr.generated_invoice_number)
                attachments.append({"filename": fname, "content": pdf_bytes,
                                    "mime_type": "application/pdf"})
                att_names.append(fname)
        if campaign.attach_mode in ("file", "image") and campaign.attachment_id:
            att = db.get(Attachment, campaign.attachment_id)
            if att:
                data = attachments_service.read_attachment_bytes(att)
                if data is not None:
                    attachments.append({"filename": att.filename, "content": data,
                                        "mime_type": att.mime_type})
                    att_names.append(att.filename)

        html_problems = message_builder.validate_html(result_html)
        if html_problems:
            log.warning("HTML validation warnings for %s: %s", cr.email, html_problems)

        mime = message_builder.build_mime_message(
            from_email=account.email,   # real authenticated account; never spoofed
            to_email=cr.email,
            subject=subject,
            html_body=result_html if campaign.message_type != "plain" else "",
            text_body=result_text,
            attachments=attachments,
            unsubscribe_url=unsub_url if campaign.unsubscribe_enabled else "",
            unsubscribe_mailto=f"{account.email}?subject=unsubscribe",
        )

        # ------------------------------------------------------------- send for real
        # Plain SMTP app-password accounts ("Bulk Add via SMTP") authenticate
        # with the encrypted password on the account row itself - no OAuth
        # token exists (or is needed) for them.
        is_smtp_password_account = getattr(account, "auth_provider", "google") == "smtp_password"
        token = None
        if not is_smtp_password_account:
            token = (db.execute(select(OAuthToken)
                                .where(OAuthToken.account_id == account.id))).scalar_one_or_none()
            if token is None:
                cr.status = "perm_failed"
                cr.last_error = "No OAuth token for account"
                campaign.failed_count += 1
                db.commit()
                return {"ok": False, "status": "perm_failed", "reason": "missing token"}

        cr.status = "processing"
        cr.attempt_count += 1
        if job:
            job.status = "processing"
            job.attempts = cr.attempt_count
        attempt = SendAttempt(
            send_job_id=job.id if job else 0,
            campaign_id=campaign.id,
            gmail_account_id=account.id,
            recipient_email=cr.email,
            subject=subject,
            template_name=template.name,
            attachment_names=att_names,
            method=account.sending_method,
            outcome="submitted",
            retry_no=cr.attempt_count - 1,
        )
        db.add(attempt)
        db.commit()

        try:
            client = build_gmail_client_from_account(account, token)
        except Exception as e:
            return _handle_send_error(db, self, campaign, cr, job, account, attempt, str(e))

        try:
            if client.method == "smtp":
                out = client.send_mime_smtp(mime)
            else:
                out = client.send_mime_api(mime)
        except googleapiclient.errors.HttpError as e:
            content = e.content.decode(errors="replace") if isinstance(e.content, bytes) else str(e.content)
            return _handle_send_error(db, self, campaign, cr, job, account, attempt,
                                      f"Gmail API error {e.resp.status}: {content[:300]}")
        except Exception as e:
            return _handle_send_error(db, self, campaign, cr, job, account, attempt, str(e))

        if not out.get("ok"):
            return _handle_send_error(db, self, campaign, cr, job, account, attempt,
                                      out.get("error", "unknown error"))

        # ------------------------------------------------------------- accepted
        msg_id = out.get("message_id", "")
        attempt.outcome = "accepted"
        attempt.message_id = msg_id
        attempt.thread_id = out.get("thread_id", "")
        attempt.provider_response = _sanitize_response(out.get("raw_response"))
        cr.status = "sent"
        cr.msg_id = msg_id
        cr.thread_id = out.get("thread_id", "")
        cr.sent_at = now()
        cr.last_error = ""
        campaign.sent_count += 1
        account.sends_today = (account.sends_today + 1) if (
            account.last_send_date and account.last_send_date.date() == today) else 1
        account.last_send_date = now()
        account.last_success_at = now()
        account.error_count = max(0, account.error_count - 1)
        if job:
            job.status = "sent"
        db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                          gmail_account_id=account.id, event_type="sent",
                          message_id=msg_id, detail=f"accepted via {client.method}"))
        db.commit()
        _publish_progress(campaign.id)
        return {"ok": True, "status": "sent", "message_id": msg_id,
                "thread_id": out.get("thread_id")}


# ---------------------------------------------------------------------------
def _recipient_dict(r) -> dict:
    return {"Email": r.email, "FirstName": r.first_name or "", "LastName": r.last_name or "",
            "CompanyName": r.company_name or "", "CompanyPhone": r.company_phone or "",
            "CustomerEmail": r.customer_email or r.email}


def _plain_fallback(html: str) -> str:
    import html as _html
    text = _re.sub(r"<style.*?</style>", "", html, flags=_re.S | _re.I)
    text = _re.sub(r"<[^>]+>", " ", text)
    return _html.unescape(_re.sub(r"\s+", " ", text)).strip()[:20000]


def _unsubscribe_url(recipient, campaign) -> str:
    base = settings.UNSUBSCRIBE_BASE or settings.BASE_URL
    if recipient and recipient.unsub_token:
        return f"{base}/unsubscribe?token={recipient.unsub_token}"
    return f"{base}/unsubscribe"


def _sanitize_response(resp) -> str:
    s = str(resp)
    for w in ("refresh_token", "access_token", "client_secret"):
        s = s.replace(w, "[REDACTED]")
    return s[:1000]


def _rate_window(key: str, interval: float):
    """Per-key fixed-interval limiter in Redis (SET NX + compare)."""
    if interval <= 0 or rdb is None:
        return True, 0
    try:
        now_ts = time.time()
        k = f"rlw:{key}"
        pipe = rdb.pipeline()
        pipe.set(k, now_ts, nx=True, ex=max(60, int(interval * 4)))
        res = pipe.execute()
        if res[0]:  # newly set -> we own this window
            return True, 0
        cur = float(rdb.get(k) or 0)
        elapsed = now_ts - cur
        if elapsed >= interval:
            rdb.set(k, now_ts, ex=max(60, int(interval * 4)))
            return True, 0
        return False, max(0.5, interval - elapsed)
    except Exception:
        return True, 0


def _reschedule(db, task, campaign_id, cr, job, delay: timedelta, reason: str):
    cr.status = "queued"
    cr.last_error = reason
    if job:
        job.status = "retrying"
        job.retry_after = now() + delay
    db.commit()
    task.apply_async(args=[campaign_id, cr.id],
                     countdown=max(1, int(delay.total_seconds()) + 1))


def _handle_send_error(db, task, campaign, cr, job, account, attempt, error: str):
    err_class = classify_error(error)
    max_retries = campaign.max_retries if campaign.max_retries is not None else settings.DEFAULT_RETRIES
    delay = int(campaign.retry_delay_seconds or settings.DEFAULT_RETRY_DELAY_SECONDS)

    cr.last_error = error[:1000]
    account.error_count += 1
    low = (error or "").lower()

    if err_class == "permanent":
        attempt.outcome = "perm_failed"
        attempt.failure_reason = error[:1000]
        attempt.error_class = err_class
        cr.status = "perm_failed"
        campaign.failed_count += 1
        if job:
            job.status = "failed"
        db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                          gmail_account_id=account.id, event_type="failed", detail=error[:500]))
        if any(p in low for p in ("invalid grant", "401", "unauthorized", "forbidden")):
            account.status = "expired"
            campaign.status = "paused"
        db.commit()
        _publish_progress(campaign.id)
        return {"ok": False, "status": "perm_failed", "reason": error[:200]}

    # quota/rate-limit (temporary class): put the account on a 30-minute cooldown
    if any(p in low for p in ("quota", "rate limit", "429", "resourceexhausted",
                              "rate-limit")):
        account.cooldown_until = now() + timedelta(minutes=30)

    if cr.attempt_count >= max_retries + 1:
        # Dead-letter: never retry indefinitely
        attempt.outcome = "temp_failed"
        attempt.failure_reason = error[:1000]
        attempt.error_class = err_class
        cr.status = "temp_failed"
        if job:
            job.status = "dead"
        db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id,
                          gmail_account_id=account.id, event_type="failed",
                          detail=f"dead-letter after {cr.attempt_count} attempts: {error[:300]}"))
        campaign.failed_count += 1
        db.commit()
        _push_dlq(campaign.id, cr.id, error)
        _publish_progress(campaign.id)
        return {"ok": False, "status": "dead_letter", "reason": error[:200]}

    # temporary -> exponential backoff with jitter
    attempt.outcome = "temp_failed"
    attempt.failure_reason = error[:1000]
    attempt.error_class = err_class
    backoff = delay * (2 ** max(0, cr.attempt_count - 1)) + random.uniform(0, 5)
    db.commit()
    _reschedule(db, task, campaign.id, cr, job, timedelta(seconds=backoff),
                f"temporary error: {error[:120]}")
    return {"ok": False, "status": "retrying", "reason": error[:200]}


def _push_dlq(campaign_id, cr_id, error):
    if rdb is None:
        return
    try:
        rdb.rpush("dlq:send", f"{campaign_id}:{cr_id}:{error[:200]}")
    except Exception:
        pass


def _publish_control(campaign_id, action, reason=""):
    if rdb is None:
        return
    try:
        rdb.publish(f"campaign:{campaign_id}:control", f"{action}|{reason}")
    except Exception:
        pass


def _publish_progress(campaign_id):
    if rdb is None:
        return
    try:
        rdb.publish(f"campaign:{campaign_id}:progress", "tick")
    except Exception:
        pass