"""Render and send campaign emails with SMTP rotation + hourly rate limits.

Per campaign, for its publisher, the pools are loaded once:
  * SMTPs   -> all smtp_manager (publisher_id int, type 3, mail_type 1)
  * subject -> template_subjects (template_type "cfp")   -> random per email
  * body    -> templates (cfp, template_for "1", publisher_id) -> matched by the
               recipient's country_level, random within that level
  * footer  -> templates (library_type 1)                -> random per email

Each email picks a random SMTP that still has hourly capacity
(`sent_mails_hourly_count` < `hourly_count`). An SMTP whose hour has elapsed
(`last_reset_at` older than 1h) is reset to 0 first, so it is released again.
This spreads sends across every domain and honours the per-SMTP hourly cap. A
campaign that can't finish within the caps is left resumable.
"""

import asyncio
import base64
import random
import re
import smtplib
import ssl
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage
from email.utils import formataddr
from typing import Optional

from bson import ObjectId

from .db import get_db

HTTP = "https://"
HOUR = timedelta(hours=1)
# Daily used-count (DUL) resets at IST midnight, tracked by a `used_reset_on`
# date string on each smtp_manager row.
IST = timezone(timedelta(hours=5, minutes=30))
# A claim (send_started_on) older than this with no sent_on is treated as a dead
# run and may be re-claimed, so an interrupted send never stays stuck "Sending".
STALE_CLAIM = timedelta(hours=2)


def _replace(text: str, mapping: dict) -> str:
    for tag, value in mapping.items():
        text = text.replace(tag, "" if value is None else str(value))
    return text


def _bare_domain(value: str) -> str:
    value = (value or "").strip()
    for prefix in ("https://", "http://", "https//", "http//"):
        if value.lower().startswith(prefix):
            return value[len(prefix):]
    return value


def _remaining_tags(html: str) -> list[str]:
    return sorted(set(re.findall(r"\[[A-Za-z0-9_ /]{1,40}\]", html or "")))


def _num(value) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def _aware(dt):
    if dt is None:
        return None
    return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt


async def _load_context(campaign_id: ObjectId) -> dict:
    """Load the publisher pools once: SMTPs (with hourly state), subjects,
    footers, and body templates grouped by country_level."""
    db = get_db()
    warnings: list[str] = []

    camp = await db["cd_campaigns"].find_one({"_id": campaign_id})
    if not camp:
        raise ValueError("campaign not found")
    pub_user = await db["users"].find_one({"_id": camp.get("publisher_id")}) or {}
    pub_int = pub_user.get("publisher_id")

    # Both kinds are used, picked at random per email: mail_type 1 (SMTP,
    # e.g. sendgrid) and mail_type 2 (Mail, self-hosted Postal with subuser auth).
    smtps: list[dict] = []
    async for s in db["smtp_manager"].find(
            {"publisher_id": pub_int, "type": {"$in": [3, "3"]},
             "mail_type": {"$in": [1, "1", 2, "2"]}}):
        subs = []
        async for su in db["smtp_subusers"].find(
                {"smtp_id": {"$in": [s["_id"], str(s["_id"])]}}, {"subuser": 1, "password": 1}):
            if su.get("subuser"):
                subs.append({"subuser": su["subuser"], "password": su.get("password") or ""})
        smtps.append({
            "doc": s,
            "limit": _num(s.get("hourly_count")),
            "count": _num(s.get("sent_mails_hourly_count")),
            "last_reset": s.get("last_reset_at"),
            "total_base": _num(s.get("total_smtp_used_count")),
            "used_base": _num(s.get("used_count")),
            "subusers": subs, "run_sent": 0, "dirty": False,
        })
    if not smtps:
        warnings.append("no type-3 SMTP for this publisher")

    # Daily DUL reset: zero used_count once per IST day, before any send bumps it.
    today_ist = datetime.now(IST).strftime("%Y-%m-%d")
    for st in smtps:
        if str(st["doc"].get("used_reset_on") or "") != today_ist:
            st["used_base"] = 0
            await db["smtp_manager"].update_one(
                {"_id": st["doc"]["_id"]},
                {"$set": {"used_count": 0, "used_reset_on": today_ist}})

    subjects = [d.get("subject_line") for d in await db["template_subjects"].find(
        {"template_type": "cfp", "status": {"$in": [1, "1"]}},
        {"subject_line": 1}).to_list(length=None) if d.get("subject_line")]
    footers = await db["templates"].find({"library_type": {"$in": [1, "1"]}}).to_list(length=None)
    bodies = await db["templates"].find(
        {"template_type": "cfp", "template_for": "1",
         "publisher_id": {"$in": [pub_int, str(pub_int)]}}).to_list(length=None)
    if not bodies:
        warnings.append("no CFP body template for this publisher")
    by_level: dict = {}
    for b in bodies:
        by_level.setdefault(str(b.get("country_level")), []).append(b)

    return {"campaign": camp, "pub_user": pub_user, "pub_int": pub_int,
            "smtps": smtps, "subjects": subjects, "footers": footers,
            "bodies": bodies, "by_level": by_level, "warnings": warnings}


def _pick_smtp(smtps: list, now: datetime, respect_limit: bool = True) -> Optional[dict]:
    """Reset any SMTP whose hour has elapsed, then return a random SMTP that
    still has hourly capacity (or any, when respect_limit is False)."""
    if not smtps:
        return None
    for st in smtps:
        last = _aware(st["last_reset"])
        if last is None or (now - last) >= HOUR:
            st["count"] = 0
            st["last_reset"] = now
            st["dirty"] = True
    if respect_limit:
        avail = [st for st in smtps if st["limit"] <= 0 or st["count"] < st["limit"]]
    else:
        avail = list(smtps)
    return random.choice(avail) if avail else None


def _select(context: dict, st: Optional[dict], record: dict) -> dict:
    """Build the per-email selection: SMTP domain/from, login credentials, body
    (by country_level), subject and footer."""
    doc = st["doc"] if st else {}
    web_domain = _bare_domain(doc.get("web_domain") or doc.get("domain") or "")
    domain = (doc.get("domain") or doc.get("web_domain") or "").strip()
    mail_type = str(doc.get("mail_type"))
    chosen = random.choice(st["subusers"]) if (st and st["subusers"]) else None
    if chosen:
        sub = (chosen.get("subuser") or "").strip()
        from_email = sub if "@" in sub else f"{sub}@{domain}"
    else:
        user = (doc.get("user") or "").strip()
        from_email = user if "@" in user else (f"noreply@{domain}" if domain else "")

    # Mail (mail_type 2): log in as the subuser (subuser@domain / its password),
    # skipping cert verification. SMTP (mail_type 1): the smtp_manager credentials.
    if mail_type == "2" and chosen:
        login_user, login_password, cert_skip = from_email, (chosen.get("password") or ""), True
    else:
        login_user, login_password, cert_skip = (doc.get("user") or "").strip(), \
            (doc.get("password") or ""), False

    pool = context["by_level"].get(str(record.get("country_level"))) or context["bodies"]
    body = random.choice(pool) if pool else {}
    subject = random.choice(context["subjects"]) if context["subjects"] else ""
    footer = random.choice(context["footers"]) if context["footers"] else {}
    return {"st": st, "smtp": doc, "web_domain": web_domain, "from_email": from_email,
            "login_user": login_user, "login_password": login_password, "cert_skip": cert_skip,
            "subject_line": subject, "body": body.get("template") or "",
            "temp_id": body.get("template_id") or "", "body_id": body.get("_id"),
            "footer": footer.get("template") or ""}


def _render(context: dict, sel: dict, record: dict, collector: dict,
            to_override: Optional[str] = None) -> dict:
    web_domain = sel["web_domain"]
    pub_user = context["pub_user"]
    pub_int = context["pub_int"]
    email = (record.get("email") or "").strip()
    to_email = (to_override or email).strip()
    b64email = base64.b64encode(email.encode()).decode() if email else ""
    rec_id = record.get("de_id") or ""
    date21 = (datetime.now(timezone.utc) + timedelta(days=21)).strftime("%d %B %Y")

    footer = _replace(sel["footer"], {
        "[EMP_NAME]": collector.get("pen_name"),
        "[EMP_ID]": collector.get("user_id"),
        "[ADDRESS]": pub_user.get("address"),
        "[WHATSAPP]": pub_user.get("whatsapp_number"),
        "[ONLINECHAT]": pub_user.get("jivo_chat"),
        "[TEMP_ID]": sel["temp_id"],
    })
    links = {
        "[SUBMISSION_LINK]": f"{HTTP}{web_domain}/submission",
        "[SUBSCRIBELINK]": f"{HTTP}{web_domain}/subscribe.php?p={pub_int}&e={rec_id}&ae={b64email}",
        "[SUBSCRIBE_D_LINK]": f"{web_domain}/subscribe.php?p={pub_int}&e={rec_id}&ae={b64email}",
        "[HOME]": f"{HTTP}{web_domain}/index.php?p={pub_int}&e={rec_id}&ae={b64email}",
        "[HOME_D_LINK]": f"{web_domain}/index.php?p={pub_int}&e={rec_id}&ae={b64email}",
        "[CONTACT]": f"{HTTP}{web_domain}/contact",
    }
    fields = {
        "[TITLE]": record.get("title"),
        "[F1]": record.get("name"),
        "[DATE]": date21,
        "[JOURNAL]": pub_user.get("fname"),
        "[Journal Name]": pub_user.get("fname"),
        "[Journal / Publisher Name]": pub_user.get("fname"),
        "[FOOTER]": footer,
    }
    html = _replace(sel["body"], {**fields, **links})
    subject = _replace(sel["subject_line"], {**fields, **links})
    return {"subject": subject, "html": html, "recipient": to_email,
            "from_email": sel["from_email"], "from_name": (collector.get("pen_name") or "").strip()}


async def render_campaign_email(campaign_id: ObjectId, record: Optional[dict] = None) -> dict:
    """Preview one recipient's email (no send, no rate limit)."""
    db = get_db()
    context = await _load_context(campaign_id)
    if record is None:
        record = await db["cd_campaign_data"].find_one({"campaign_id": campaign_id}) or {}
        if not record:
            context["warnings"].append("campaign has no recipient rows")
    collector = await db["users"].find_one({"_id": record.get("user_id")}) or {}
    st = _pick_smtp(context["smtps"], datetime.now(timezone.utc), respect_limit=False)
    sel = _select(context, st, record)
    r = _render(context, sel, record, collector)
    warnings = list(context["warnings"])
    left = _remaining_tags(r["html"])
    if left:
        warnings.append(f"unreplaced tags: {', '.join(left)}")
    return {
        "campaign_id": str(campaign_id),
        "publisher": context["pub_user"].get("fname") or "",
        "smtp_host": (st["doc"].get("host") if st else "") or "",
        "web_domain": sel["web_domain"],
        "recipient": r["recipient"], "subject": r["subject"], "html": r["html"],
        "warnings": warnings,
    }


def _send_html(smtp: dict, from_email: str, from_name: str, to_email: str,
               subject: str, html: str, login_user: Optional[str] = None,
               login_password: Optional[str] = None, cert_skip: bool = False) -> None:
    host = (smtp.get("host") or "").strip()
    port = int(smtp.get("port") or 587)
    user = login_user if login_user is not None else (smtp.get("user") or "").strip()
    password = login_password if login_password is not None else (smtp.get("password") or "")
    helo = (from_email.split("@")[-1].strip() or smtp.get("domain") or "").strip() or None

    msg = EmailMessage()
    msg["Subject"] = subject or "(no subject)"
    msg["From"] = formataddr((from_name or "", from_email))
    msg["To"] = to_email
    msg["X-SMTPAPI"] = '{"filters":{"clicktrack":{"settings":{"enable":0}}}}'
    msg.set_content("This email is best viewed in an HTML-capable client.")
    msg.add_alternative(html or "", subtype="html")

    context = ssl.create_default_context()
    if cert_skip:  # mail_type 2 (Postal) presents a mismatched cert
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
    if port == 465:
        with smtplib.SMTP_SSL(host, port, timeout=25, context=context,
                              local_hostname=helo) as server:
            if user and password:
                server.login(user, password)
            server.send_message(msg)
    else:
        with smtplib.SMTP(host, port, timeout=25, local_hostname=helo) as server:
            server.ehlo()
            if server.has_extn("starttls"):
                server.starttls(context=context)
                server.ehlo()
            if user and password:
                server.login(user, password)
            server.send_message(msg)


async def send_campaign_test(campaign_id: ObjectId, to_email: str) -> dict:
    db = get_db()
    context = await _load_context(campaign_id)
    st = _pick_smtp(context["smtps"], datetime.now(timezone.utc), respect_limit=False)
    if st is None:
        return {"ok": False, "detail": "no type-3 SMTP for this publisher"}
    record = await db["cd_campaign_data"].find_one({"campaign_id": campaign_id}) or {}
    collector = await db["users"].find_one({"_id": record.get("user_id")}) or {}
    sel = _select(context, st, record)
    r = _render(context, sel, record, collector, to_override=to_email)
    from_disp = formataddr((r["from_name"] or "", r["from_email"]))
    try:
        await asyncio.to_thread(_send_html, st["doc"], r["from_email"], r["from_name"],
                                to_email, r["subject"], r["html"],
                                sel["login_user"], sel["login_password"], sel["cert_skip"])
        return {"ok": True, "detail": f"Test email sent to {to_email} (from {from_disp})."}
    except Exception as exc:
        return {"ok": False, "detail": f"{type(exc).__name__}: {exc}"}


async def _mirror_to_collected(db, de_id) -> None:
    """After a successful send, copy the source collected_data_de record into
    collected_data (fresh mailing state). Best-effort: never blocks the send."""
    if de_id is None:
        return
    try:
        de = await db["collected_data_de"].find_one({"_id": de_id})
        if not de:
            return
        await db["collected_data"].insert_one({
            "user_id": de.get("user_id"),
            "section_id": de.get("section_id"),
            "data_type": de.get("data_type"),
            "title": de.get("title"),
            "topic": de.get("topic"),
            "name": de.get("name"),
            "email": de.get("email"),
            "country_id": de.get("country_id"),
            "country_level": de.get("country_level"),
            "country_name": de.get("country_name"),
            "mail_send_status": 0,
            "subscribe_send_status": 0,
            "posted_on": de.get("posted_on"),
            "primary_topic_id": de.get("primary_topic_id"),
            "primary_topic": de.get("primary_topic"),
            "created_on": datetime.now(timezone.utc),
        })
    except Exception:
        pass


async def send_campaign(campaign_id: ObjectId) -> dict:
    """Send to every not-yet-sent recipient, rotating SMTPs within their hourly
    caps. Stops if all SMTPs are capped (leaving the rest resumable)."""
    db = get_db()
    context = await _load_context(campaign_id)
    smtps = context["smtps"]
    if not smtps:
        return {"total": 0, "sent": 0, "failed": 0}

    records = await db["cd_campaign_data"].find(
        {"campaign_id": campaign_id, "mail_send_status": {"$ne": 1}}).to_list(length=None)
    collectors: dict = {}
    template_used: dict = {}
    sent = failed = 0

    for rec in records:
        st = _pick_smtp(smtps, datetime.now(timezone.utc), respect_limit=True)
        if st is None:
            break  # every SMTP is at its hourly cap -> leave the rest resumable
        uid = rec.get("user_id")
        if uid not in collectors:
            collectors[uid] = await db["users"].find_one({"_id": uid}) or {}
        sel = _select(context, st, rec)
        r = _render(context, sel, rec, collectors[uid])
        try:
            await asyncio.to_thread(_send_html, st["doc"], r["from_email"], r["from_name"],
                                    rec.get("email"), r["subject"], r["html"],
                                    sel["login_user"], sel["login_password"], sel["cert_skip"])
            st["count"] += 1
            st["run_sent"] += 1
            st["dirty"] = True
            template_used[sel["body_id"]] = template_used.get(sel["body_id"], 0) + 1
            await db["cd_campaign_data"].update_one({"_id": rec["_id"]}, {"$set": {
                "mail_send_status": 1, "sent_on": datetime.now(timezone.utc),
                "smtp_host": st["doc"].get("host"), "smtp_domain": st["doc"].get("domain"),
                "template_id": str(sel["temp_id"]) if sel["temp_id"] not in (None, "") else None}})
            sent += 1
            await _mirror_to_collected(db, rec.get("de_id"))
        except Exception as exc:
            await db["cd_campaign_data"].update_one(
                {"_id": rec["_id"]}, {"$set": {
                    "mail_send_status": 2,
                    "last_attempt_on": datetime.now(timezone.utc),
                    "smtp_host": st["doc"].get("host"),
                    "send_error": f"{type(exc).__name__}: {exc}"[:300]}})
            failed += 1

    # Persist SMTP hourly counters/resets and the total-used bump.
    for st in smtps:
        if st["dirty"]:
            await db["smtp_manager"].update_one({"_id": st["doc"]["_id"]}, {"$set": {
                "sent_mails_hourly_count": st["count"], "last_reset_at": st["last_reset"],
                "used_count": st["used_base"] + st["run_sent"],
                "total_smtp_used_count": st["total_base"] + st["run_sent"]}})
    # Bump template used_count (string-safe).
    for body_id, cnt in template_used.items():
        if body_id is not None and cnt:
            await db["templates"].update_one({"_id": body_id}, [{"$set": {"used_count": {"$add": [
                {"$convert": {"input": "$used_count", "to": "int", "onError": 0, "onNull": 0}}, cnt]}}}])

    now = datetime.now(timezone.utc)
    total_sent = await db["cd_campaign_data"].count_documents(
        {"campaign_id": campaign_id, "mail_send_status": 1})
    total_failed = await db["cd_campaign_data"].count_documents(
        {"campaign_id": campaign_id, "mail_send_status": 2})
    pending = await db["cd_campaign_data"].count_documents(
        {"campaign_id": campaign_id, "mail_send_status": {"$nin": [1, 2]}})
    if pending == 0:
        await db["cd_campaigns"].update_one({"_id": campaign_id}, {"$set": {
            "status": 1, "sent_count": total_sent, "failed_count": total_failed, "sent_on": now}})
    else:
        # Capped out this hour -> record progress and allow a later resume.
        await db["cd_campaigns"].update_one(
            {"_id": campaign_id},
            {"$set": {"sent_count": total_sent, "failed_count": total_failed,
                      "pending_count": pending, "last_run_on": now},
             "$unset": {"send_started_on": ""}})
    return {"total": len(records), "sent": sent, "failed": failed}


# --- desktop-driven send: API prepares, the desktop exe does the sending -----


async def prepare_send_jobs(campaign_id: ObjectId) -> dict:
    """Claim the campaign and return one rendered email + its SMTP per not-yet-
    sent recipient. The API does NOT send -- the desktop exe does. SMTP counters
    move only when the desktop reports back (report_send)."""
    db = get_db()
    # Claim so a campaign is never prepared/sent twice at once. A claim that is
    # older than STALE_CLAIM with no sent_on means a previous run died mid-flight
    # (e.g. the app closed, or 0 SMTP capacity) -- re-claim it so it can recover.
    now = datetime.now(timezone.utc)
    stale = now - STALE_CLAIM
    claim = await db["cd_campaigns"].update_one(
        {"_id": campaign_id, "sent_on": {"$exists": False},
         "$or": [{"send_started_on": {"$exists": False}},
                 {"send_started_on": {"$lt": stale}}]},
        {"$set": {"send_started_on": now}})
    if claim.matched_count == 0:
        camp = await db["cd_campaigns"].find_one({"_id": campaign_id}) or {}
        when = str(camp.get("sent_on") or camp.get("send_started_on") or "")[:16]
        return {"claimed": False, "campaign_id": str(campaign_id), "total": 0, "jobs": [],
                "detail": f"Already sent/sending{(' on ' + when) if when else ''}."}

    context = await _load_context(campaign_id)
    smtps = context["smtps"]
    if not smtps:
        await db["cd_campaigns"].update_one({"_id": campaign_id}, {"$unset": {"send_started_on": ""}})
        return {"claimed": False, "campaign_id": str(campaign_id), "total": 0, "jobs": [],
                "detail": "no type-3 SMTP for this publisher"}

    # Release any SMTP whose hour has elapsed, and persist that reset now so the
    # desktop's report increments start from zero.
    for st in smtps:
        last = _aware(st["last_reset"])
        if last is None or (now - last) >= HOUR:
            st["count"] = 0
            st["last_reset"] = now
            await db["smtp_manager"].update_one(
                {"_id": st["doc"]["_id"]},
                {"$set": {"sent_mails_hourly_count": 0, "last_reset_at": now}})

    records = await db["cd_campaign_data"].find(
        {"campaign_id": campaign_id, "mail_send_status": {"$ne": 1}}).to_list(length=None)
    collectors: dict = {}
    jobs: list = []
    for rec in records:
        st = _pick_smtp(smtps, datetime.now(timezone.utc), respect_limit=True)
        if st is None:
            break  # every SMTP is at its hourly cap -> leave the rest resumable
        uid = rec.get("user_id")
        if uid not in collectors:
            collectors[uid] = await db["users"].find_one({"_id": uid}) or {}
        sel = _select(context, st, rec)
        r = _render(context, sel, rec, collectors[uid])
        st["count"] += 1  # reserve in-memory so hourly limits hold within this run
        doc = st["doc"]
        from_email = r["from_email"] or ""
        jobs.append({
            "rec_id": str(rec["_id"]),
            "de_id": str(rec["de_id"]) if rec.get("de_id") else None,
            "to_email": rec.get("email"),
            "subject": r["subject"], "html": r["html"],
            "from_email": from_email, "from_name": r["from_name"],
            "smtp_id": str(doc.get("_id")),
            "smtp_host": doc.get("host"), "smtp_port": int(doc.get("port") or 587),
            "smtp_user": sel["login_user"], "smtp_password": sel["login_password"],
            "cert_skip": sel["cert_skip"],
            "smtp_domain": doc.get("domain"),
            "helo": (from_email.split("@")[-1] if "@" in from_email else doc.get("domain")),
            "template_id": str(sel["temp_id"]) if sel["temp_id"] not in (None, "") else None,
            "body_id": str(sel["body_id"]) if sel.get("body_id") else None,
        })

    if not jobs:
        # Claimed but nothing to send right now (every SMTP at its hourly cap).
        # Release the claim so the campaign isn't stuck "Sending" and can retry.
        await db["cd_campaigns"].update_one(
            {"_id": campaign_id}, {"$unset": {"send_started_on": ""}})
        return {"claimed": False, "campaign_id": str(campaign_id), "total": 0, "jobs": [],
                "detail": "no SMTP hourly capacity right now — try again later"}

    return {"claimed": True, "campaign_id": str(campaign_id), "total": len(jobs),
            "jobs": jobs, "detail": f"{len(jobs)} email(s) ready to send."}


async def report_send(campaign_id: ObjectId, results: list, final: bool = True) -> dict:
    """Record what the desktop actually sent: mark rows sent/failed, mirror the
    successes into collected_data, and move SMTP + template counters. The desktop
    calls this in batches as it sends; only the `final` call finalizes the
    campaign's status (sent_on / resumable), so the claim holds until then."""
    db = get_db()
    smtp_sent: dict = {}   # smtp_id -> successes
    tmpl_used: dict = {}   # body_id -> successes
    sent = failed = 0

    for item in results:
        it = item if isinstance(item, dict) else item.model_dump()
        try:
            roid = ObjectId(it.get("rec_id"))
        except Exception:
            continue
        if it.get("ok"):
            await db["cd_campaign_data"].update_one({"_id": roid}, {"$set": {
                "mail_send_status": 1, "sent_on": datetime.now(timezone.utc),
                "smtp_host": it.get("smtp_host"), "smtp_domain": it.get("smtp_domain"),
                "template_id": it.get("template_id")}})
            sent += 1
            de_id = it.get("de_id")
            if de_id:
                try:
                    await _mirror_to_collected(db, ObjectId(de_id))
                except Exception:
                    pass
            if it.get("smtp_id"):
                smtp_sent[it["smtp_id"]] = smtp_sent.get(it["smtp_id"], 0) + 1
            if it.get("body_id"):
                tmpl_used[it["body_id"]] = tmpl_used.get(it["body_id"], 0) + 1
        else:
            await db["cd_campaign_data"].update_one({"_id": roid}, {"$set": {
                "mail_send_status": 2, "last_attempt_on": datetime.now(timezone.utc),
                "smtp_host": it.get("smtp_host"),
                "send_error": (it.get("error") or "")[:300]}})
            failed += 1

    for sid, cnt in smtp_sent.items():
        try:
            soid = ObjectId(sid)
        except Exception:
            continue
        await db["smtp_manager"].update_one({"_id": soid}, [{"$set": {
            "used_count": {"$add": [
                {"$convert": {"input": "$used_count", "to": "int",
                              "onError": 0, "onNull": 0}}, cnt]},
            "total_smtp_used_count": {"$add": [
                {"$convert": {"input": "$total_smtp_used_count", "to": "int",
                              "onError": 0, "onNull": 0}}, cnt]},
            "sent_mails_hourly_count": {"$add": [
                {"$convert": {"input": "$sent_mails_hourly_count", "to": "int",
                              "onError": 0, "onNull": 0}}, cnt]},
        }}])
    for bid, cnt in tmpl_used.items():
        try:
            boid = ObjectId(bid)
        except Exception:
            continue
        await db["templates"].update_one({"_id": boid}, [{"$set": {"used_count": {"$add": [
            {"$convert": {"input": "$used_count", "to": "int", "onError": 0, "onNull": 0}}, cnt]}}}])

    if not final:
        # Mid-send batch: progress is saved. Refresh the claim (heartbeat) so an
        # actively-sending campaign is never seen as a stale/dead run, while one
        # that stopped reporting goes stale on its own after STALE_CLAIM.
        await db["cd_campaigns"].update_one(
            {"_id": campaign_id, "sent_on": {"$exists": False}},
            {"$set": {"send_started_on": datetime.now(timezone.utc)}})
        return {"sent": sent, "failed": failed, "pending": 0, "completed": False,
                "detail": f"{sent} sent, {failed} failed (batch)."}

    now = datetime.now(timezone.utc)
    total_sent = await db["cd_campaign_data"].count_documents(
        {"campaign_id": campaign_id, "mail_send_status": 1})
    total_failed = await db["cd_campaign_data"].count_documents(
        {"campaign_id": campaign_id, "mail_send_status": 2})
    pending = await db["cd_campaign_data"].count_documents(
        {"campaign_id": campaign_id, "mail_send_status": {"$nin": [1, 2]}})
    if pending == 0:
        await db["cd_campaigns"].update_one({"_id": campaign_id}, {"$set": {
            "status": 1, "sent_count": total_sent, "failed_count": total_failed, "sent_on": now}})
    else:
        await db["cd_campaigns"].update_one(
            {"_id": campaign_id},
            {"$set": {"sent_count": total_sent, "failed_count": total_failed,
                      "pending_count": pending, "last_run_on": now},
             "$unset": {"send_started_on": ""}})
    return {"sent": sent, "failed": failed, "pending": pending, "completed": pending == 0,
            "detail": f"{sent} sent, {failed} failed, {pending} pending."}