"""
ARQ Background Worker for INDELIS.

Replaces Celery+RabbitMQ with ARQ (async Redis queue).
ARQ is fully async and uses Redis — no separate message broker needed.

Run worker:
    arq src.worker.arq_app.WorkerSettings
"""
import logging
from typing import Optional
from uuid import UUID

import arq
from arq.connections import RedisSettings

from src.core.config import settings
from src.database.session import AsyncSessionLocal
from src.database.load_models import load_all_models

logger = logging.getLogger(__name__)


# ── Task Definitions ──────────────────────────────────────────────────────────

async def send_invitation_email(
    ctx: dict,
    user_id: str,
    tenant_id: str,
    invited_by_id: Optional[str] = None,
    resend: bool = False,
):
    """Send a staff-invitation email via the templated dispatch engine (INDL-54).

    `resend=True` selects the `staff_invitation_resend` template instead of
    `staff_invitation`.
    """
    from uuid import UUID
    from sqlalchemy import select
    from src.apps.auth.models.user import User
    from src.apps.tenants.models.account import Account
    from src.core.config import settings
    from src.core.constants import EmailTriggerKey
    from src.core.email_dispatch import cemetery_context, email_dispatch_service
    from src.core.security import create_invitation_token

    async with AsyncSessionLocal() as db:
        user = (await db.execute(select(User).where(User.id == UUID(user_id)))).scalar_one_or_none()
        tenant = (await db.execute(select(Account).where(Account.id == UUID(tenant_id)))).scalar_one_or_none()

        invited_by = None
        if invited_by_id:
            invited_by = (await db.execute(select(User).where(User.id == UUID(invited_by_id)))).scalar_one_or_none()

        if not user or not tenant:
            logger.warning("[email] send_invitation_email: user or tenant not found — %s / %s", user_id, tenant_id)
            return {"status": "skipped", "reason": "not_found"}

        invitee_name = f"{user.first_name} {user.last_name}".strip() or user.email
        invited_by_name = (
            f"{invited_by.first_name} {invited_by.last_name}".strip()
            if invited_by
            else tenant.organization_name or "Your administrator"
        )
        role_display = {
            "administrator": "Administrator",
            "manager": "Manager",
            "staff": "Sales Agent",
            "view_only": "Viewer",
        }.get(user.role, user.role.replace("_", " ").title())

        token = create_invitation_token(user_id=user_id, email=user.email, tenant_id=tenant_id)
        admin_portal_url = getattr(settings, "ADMIN_PORTAL_URL", "http://localhost:3000").rstrip("/")
        invite_url = f"{admin_portal_url}/accept-invitation?token={token}"

        trigger = (
            EmailTriggerKey.STAFF_INVITATION_RESEND if resend else EmailTriggerKey.STAFF_INVITATION
        )
        context = {
            "invitee_name": invitee_name,
            "invited_by_name": invited_by_name,
            "organization_name": tenant.organization_name or tenant.subdomain,
            "role": role_display,
            "invite_url": invite_url,
            **cemetery_context(tenant),
        }
        result = await email_dispatch_service.send(
            db, trigger_key=trigger, tenant_id=tenant.id, to=user.email, context=context
        )
        await db.commit()

    logger.info("[email] Invitation dispatch for %s → sent=%s", _mask_recipient(user.email), result.sent)
    return {"status": "sent" if result.sent else "skipped", "email": user.email}


async def send_invoice_reminder(ctx: dict, invoice_id: str, tenant_id: str, days_overdue: int):
    """Send an overdue-invoice reminder via the dispatch engine (INDL-54e).

    Idempotency at the 7/14/30-day marks is enforced upstream in
    `InvoiceService.transition_overdue` (reminder_count / last_reminder_at),
    which is what enqueues this job — this job just renders and sends.
    """
    from uuid import UUID
    from sqlalchemy import select
    from src.apps.billing.models.invoice import Invoice
    from src.apps.tenants.models.account import Account
    from src.core.constants import EmailTriggerKey
    from src.core.email_dispatch import cemetery_context, email_dispatch_service

    async with AsyncSessionLocal() as db:
        invoice = (
            await db.execute(
                select(Invoice).where(
                    Invoice.id == UUID(invoice_id),
                    Invoice.tenant_id == UUID(tenant_id),
                )
            )
        ).scalar_one_or_none()
        if not invoice:
            logger.warning("[email] send_invoice_reminder: invoice %s not found", invoice_id)
            return {"status": "skipped", "reason": "not_found"}
        if not invoice.purchaser_email:
            logger.info("[email] send_invoice_reminder: no purchaser email for %s", invoice_id)
            return {"status": "skipped", "reason": "no_recipient"}

        account = (
            await db.execute(select(Account).where(Account.id == UUID(tenant_id)))
        ).scalar_one_or_none()

        context = {
            "purchaser_name": invoice.purchaser_name or "Valued Customer",
            "invoice_number": invoice.invoice_number or "",
            "invoice_total": f"{invoice.total_amount:,.2f}" if invoice.total_amount is not None else "",
            "amount_due": f"{invoice.balance_due:,.2f}" if invoice.balance_due is not None else "",
            "due_date": invoice.due_date.strftime("%B %d, %Y") if invoice.due_date else "",
            "days_overdue": str(days_overdue),
            **cemetery_context(account),
        }
        result = await email_dispatch_service.send(
            db,
            trigger_key=EmailTriggerKey.INVOICE_OVERDUE_REMINDER,
            tenant_id=UUID(tenant_id),
            to=invoice.purchaser_email,
            context=context,
        )
        await db.commit()

    logger.info("[email] Invoice %s %dd reminder → sent=%s", invoice_id, days_overdue, result.sent)
    return {"status": "sent" if result.sent else "skipped", "invoice_id": invoice_id}


async def generate_qr_code(ctx: dict, qr_code_id: str):
    """Fetch qr_codes row → SVG → S3 → PDF → S3 → update row."""
    from io import BytesIO
    import asyncio
    import qrcode
    import qrcode.image.svg
    import boto3
    from reportlab.pdfgen import canvas
    from reportlab.lib.pagesizes import A4
    from reportlab.lib.utils import ImageReader
    from sqlalchemy import select
    from src.database.session import AsyncSessionLocal
    from src.apps.settings.models.qr_code import QRCode
    from src.apps.tenants.models.account import Account
    from src.apps.settings.services.qr_code_service import QRCodeService
    from src.core.exceptions import NotFoundError
    from src.core.config import settings

    async with AsyncSessionLocal() as db:
        result = await db.execute(select(QRCode).where(QRCode.id == UUID(qr_code_id)))
        qr = result.scalar_one_or_none()
        if not qr:
            logger.error(f"[qr] QRCode row {qr_code_id} not found")
            return {"status": "not_found", "qr_code_id": qr_code_id}

        tenant_id = str(qr.tenant_id)
        qr_type = qr.qr_type
        reference_id = qr.reference_id

        # SEC-08/API10: never trust qr.content_url as stored — re-derive it here
        # from the current DB state (tenant subdomain, and for 'headstone' the
        # live Plot -> Record -> Memorial slug) so a stale/tampered stored URL
        # never gets encoded into the generated QR image. Applies to all
        # qr_types since the job already has full DB access at execution time.
        acct_result = await db.execute(select(Account).where(Account.id == qr.tenant_id))
        account = acct_result.scalar_one_or_none()
        subdomain = account.subdomain if account else tenant_id

        memorial_slug = None
        if qr_type == "headstone":
            from src.apps.plots.models.plot import Plot

            plot_result = await db.execute(
                select(Plot.id).where(
                    Plot.tenant_id == qr.tenant_id, Plot.plot_ref == reference_id
                )
            )
            plot_id = plot_result.scalar_one_or_none()
            if plot_id is not None:
                memorial_slug = await QRCodeService._resolve_memorial_slug(
                    db, qr.tenant_id, plot_id
                )

        try:
            content_url = QRCodeService.build_content_url(
                subdomain, qr_type, reference_id, memorial_slug=memorial_slug
            )
        except NotFoundError as exc:
            logger.error(
                f"[qr] Could not re-derive content_url for {qr_code_id} "
                f"(qr_type={qr_type}): {exc.message}"
            )
            return {"status": "error", "qr_code_id": qr_code_id, "reason": exc.message}

        qr.content_url = content_url
        await db.commit()

    # Generate SVG
    svg_factory = qrcode.image.svg.SvgImage
    qr_obj = qrcode.QRCode(version=1, box_size=10, border=4)
    qr_obj.add_data(content_url)
    qr_obj.make(fit=True)
    svg_img = qr_obj.make_image(image_factory=svg_factory)
    svg_buf = BytesIO()
    svg_img.save(svg_buf)
    svg_bytes = svg_buf.getvalue()

    # Generate PNG for PDF embed
    png_buf = BytesIO()
    qr_png = qrcode.QRCode(version=1, box_size=10, border=4)
    qr_png.add_data(content_url)
    qr_png.make(fit=True)
    png_img = qr_png.make_image(fill_color="black", back_color="white")
    png_img.save(png_buf, format="PNG")
    png_bytes = png_buf.getvalue()

    # Generate PDF label
    pdf_buf = BytesIO()
    c = canvas.Canvas(pdf_buf, pagesize=A4)
    width, height = A4
    # Center QR
    qr_size = 200
    x = (width - qr_size) / 2
    y = (height - qr_size) / 2
    png_reader = ImageReader(BytesIO(png_bytes))
    c.drawImage(png_reader, x, y, width=qr_size, height=qr_size)
    # Add URL text below
    c.setFont("Helvetica", 9)
    c.drawCentredString(width / 2, y - 20, content_url[:80])
    c.save()
    pdf_bytes = pdf_buf.getvalue()

    # Upload to S3
    svg_key = f"tenant/{tenant_id}/qr/{qr_type}/{reference_id}.svg"
    pdf_key = f"tenant/{tenant_id}/qr/{qr_type}/{reference_id}.pdf"

    s3 = boto3.client(
        "s3",
        region_name=settings.AWS_REGION,
        aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
        aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
    )
    await asyncio.to_thread(
        s3.put_object,
        Bucket=settings.S3_BUCKET,
        Key=svg_key,
        Body=svg_bytes,
        ContentType="image/svg+xml",
    )
    await asyncio.to_thread(
        s3.put_object,
        Bucket=settings.S3_BUCKET,
        Key=pdf_key,
        Body=pdf_bytes,
        ContentType="application/pdf",
    )

    # Update DB row
    from datetime import datetime, timezone
    async with AsyncSessionLocal() as db:
        result = await db.execute(select(QRCode).where(QRCode.id == UUID(qr_code_id)))
        qr = result.scalar_one_or_none()
        if qr:
            qr.svg_s3_key = svg_key
            qr.pdf_s3_key = pdf_key
            qr.generated_at = datetime.now(timezone.utc)
            await db.commit()

    logger.info(f"[qr] QR code generated: {qr_code_id}")
    return {"status": "generated", "qr_code_id": qr_code_id}


async def regenerate_all_qr(ctx: dict, tenant_id: str, qr_type: Optional[str] = None):
    """Fan-out: enqueue generate_qr_code per active qr_codes row."""
    from sqlalchemy import select
    from src.database.session import AsyncSessionLocal
    from src.apps.settings.models.qr_code import QRCode

    tenant_uuid = UUID(tenant_id)
    async with AsyncSessionLocal() as db:
        filters = [QRCode.tenant_id == tenant_uuid, QRCode.is_active.is_(True)]
        if qr_type:
            filters.append(QRCode.qr_type == qr_type)
        result = await db.execute(select(QRCode.id).where(*filters))
        ids = [str(row[0]) for row in result.all()]

    redis = ctx["redis"]
    for qr_id in ids:
        await redis.enqueue_job("generate_qr_code", qr_id)

    logger.info(f"[qr] Enqueued {len(ids)} generate_qr_code jobs for tenant {tenant_id}")
    return {"enqueued": len(ids)}


async def generate_contract_pdf(ctx: dict, tenant_id: str, contract_id: str):
    """Generate signed contract PDF, upload to S3, and email the purchaser."""
    import asyncio
    from uuid import UUID

    import boto3
    from botocore.exceptions import NoCredentialsError, ClientError
    from sqlalchemy import select
    from sqlalchemy.orm import selectinload

    from src.apps.sales.models.contract import Contract
    from src.apps.sales.models.contract_line_item import ContractLineItem  # noqa: F401
    from src.apps.sales.services.contract_pdf_service import build_contract_pdf_bytes

    logger.info("[pdf] Generating contract PDF for contract %s", contract_id)

    # ── 1. Load contract with line items ─────────────────────────────────────
    tenant_uuid = UUID(tenant_id)
    contract_uuid = UUID(contract_id)

    async with AsyncSessionLocal() as db:
        result = await db.execute(
            select(Contract)
            .options(selectinload(Contract.line_items))
            .where(
                Contract.id == contract_uuid,
                Contract.tenant_id == tenant_uuid,
                Contract.deleted_at.is_(None),
            )
        )
        contract = result.scalar_one_or_none()

    if not contract:
        logger.error("[pdf] Contract %s not found for tenant %s", contract_id, tenant_id)
        return {"status": "not_found", "contract_id": contract_id}

    # ── 2. Build PDF with ReportLab ──────────────────────────────────────────
    pdf_bytes = await asyncio.to_thread(build_contract_pdf_bytes, contract)

    # ── 3. Upload to S3 ──────────────────────────────────────────────────────
    s3_key = f"contracts/{tenant_id}/{contract.contract_number or contract_id}.pdf"
    pdf_s3_key_saved = None

    if not settings.AWS_ACCESS_KEY_ID or not settings.AWS_SECRET_ACCESS_KEY:
        logger.warning(
            "[pdf] S3 credentials not configured — skipping upload for contract %s",
            contract_id,
        )
    else:
        try:
            s3 = boto3.client(
                "s3",
                region_name=settings.AWS_REGION,
                aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
                aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            )
            await asyncio.to_thread(
                s3.put_object,
                Bucket=settings.S3_BUCKET,
                Key=s3_key,
                Body=pdf_bytes,
                ContentType="application/pdf",
            )
            pdf_s3_key_saved = s3_key
            logger.info("[pdf] Uploaded contract PDF to s3://%s/%s", settings.S3_BUCKET, s3_key)
        except (NoCredentialsError, ClientError) as exc:
            logger.warning("[pdf] S3 upload failed for contract %s: %s", contract_id, exc)

    # ── 4. Update contract.pdf_s3_key ────────────────────────────────────────
    if pdf_s3_key_saved:
        async with AsyncSessionLocal() as db:
            result = await db.execute(
                select(Contract).where(
                    Contract.id == contract_uuid,
                    Contract.tenant_id == tenant_uuid,
                    Contract.deleted_at.is_(None),
                )
            )
            c = result.scalar_one_or_none()
            if c:
                c.pdf_s3_key = pdf_s3_key_saved
                await db.commit()

    # ── 5. Send confirmation via the templated dispatch engine (SMTP) ──────────
    # (INDL-54f: previously SendGrid — now consolidated onto SMTP via _dispatch)
    if contract.purchaser_email:
        from src.apps.tenants.models.account import Account
        from src.core.constants import EmailTriggerKey
        from src.core.email_dispatch import cemetery_context, email_dispatch_service

        async with AsyncSessionLocal() as db:
            account = (
                await db.execute(select(Account).where(Account.id == tenant_uuid))
            ).scalar_one_or_none()
            context = {
                "purchaser_name": contract.purchaser_name or "Valued Customer",
                "contract_number": contract.contract_number or "",
                **cemetery_context(account),
            }
            result = await email_dispatch_service.send(
                db,
                trigger_key=EmailTriggerKey.CONTRACT_PDF_CONFIRMATION,
                tenant_id=tenant_uuid,
                to=contract.purchaser_email,
                context=context,
            )
            await db.commit()
        logger.info("[pdf] Contract confirmation dispatch for contract %s → sent=%s",
                    contract_id, result.sent)

    return {
        "status": "generated",
        "contract_id": contract_id,
        "pdf_uploaded": pdf_s3_key_saved is not None,
    }


async def export_week_pdf(ctx: dict, tenant_id: str, week_start: str):
    """Generate weekly run-sheet PDF → S3 → return presigned URL."""
    logger.info(f"[pdf] Generating week PDF for tenant {tenant_id}, week {week_start}")
    # TODO: ReportLab PDF generation + S3 upload
    return {"status": "generated", "week_start": week_start, "presigned_url": None}


def _mask_recipient(email: str) -> str:
    """Mask a recipient address for audit-log storage (SEC-AC-06): j***@example.com."""
    local, _, domain = email.partition("@")
    if not domain:
        return "***"
    return f"{local[0]}***@{domain}" if local else f"***@{domain}"


def _is_valid_recipient(email: Optional[str]) -> bool:
    """RFC 5321 sanity check before any SES/SMTP call (SEC-AC-05)."""
    if not email:
        return False
    try:
        from email_validator import validate_email

        validate_email(email, check_deliverability=False)
        return True
    except Exception:
        return False


def _scheduling_email_fields(event) -> dict:
    """Shared date/time formatting for the confirmation + reminder emails."""
    date_formatted = event.scheduled_date.strftime("%B %d, %Y") if event.scheduled_date else "To be confirmed"
    start_time = event.scheduled_time.strftime("%I:%M %p").lstrip("0") if event.scheduled_time else None
    end_time = None
    if event.scheduled_time and event.duration_minutes:
        from datetime import datetime as _dt, timedelta as _td

        start_dt = _dt.combine(event.scheduled_date, event.scheduled_time)
        end_time = (start_dt + _td(minutes=event.duration_minutes)).strftime("%I:%M %p").lstrip("0")
    return {"date_formatted": date_formatted, "start_time": start_time, "end_time": end_time}


async def send_scheduling_confirmation(ctx: dict, service_id: str, tenant_id: str) -> dict:
    """Family-facing service confirmation email (INDL-38 AC-01/AC-02/AC-08)."""
    from uuid import UUID

    from sqlalchemy import select

    from src.apps.scheduling.models.service_event import ServiceEvent
    from src.apps.scheduling.services.scheduling_service import SchedulingService
    from src.apps.tenants.models.account import Account
    from src.core.constants import EmailTriggerKey
    from src.core.email_dispatch import cemetery_context, email_dispatch_service

    tenant_uuid, service_uuid = UUID(tenant_id), UUID(service_id)

    async with AsyncSessionLocal() as db:
        # Re-validate tenant ownership fresh from the DB — never trust the
        # payload alone (SEC-AC-01 / API1).
        event = (
            await db.execute(
                select(ServiceEvent).where(
                    ServiceEvent.id == service_uuid,
                    ServiceEvent.tenant_id == tenant_uuid,
                )
            )
        ).scalar_one_or_none()
        if event is None or event.deleted_at is not None:
            logger.info("[scheduling-email] service %s not found/deleted — skipping confirmation", service_id)
            return {"status": "skipped", "reason": "not_found_or_deleted"}

        # AC-09: draft services never trigger emails.
        if event.status not in ("awaiting_family_confirm", "confirmed"):
            logger.info("[scheduling-email] service %s is draft — skipping confirmation", service_id)
            return {"status": "skipped", "reason": "draft_status"}

        svc = SchedulingService(db)

        if not _is_valid_recipient(event.family_contact_email):
            await svc.log_event(
                service_id=event.id,
                tenant_id=event.tenant_id,
                actor_id=None,
                event_type="email_error",
                description="Confirmation email not sent: family_contact_email is empty or invalid.",
            )
            await db.commit()
            return {"status": "skipped", "reason": "invalid_email"}

        account = (
            await db.execute(select(Account).where(Account.id == tenant_uuid))
        ).scalar_one_or_none()
        fields = _scheduling_email_fields(event)
        location = ", ".join(p for p in [event.plot_location, event.section] if p) or "To be confirmed"
        time_range = (
            f"{fields['start_time']} – {fields['end_time']}"
            if fields["start_time"] and fields["end_time"]
            else (fields["start_time"] or "To be confirmed")
        )
        context = {
            "family_name": event.family_contact_name or "Family",
            "service_type": event.service_type,
            "decedent_name": event.decedent_name or "",
            "service_date": fields["date_formatted"],
            "service_time": time_range,
            "location": location,
            "officiant_name": event.officiant or "To be confirmed",
            **cemetery_context(account),
        }
        result = await email_dispatch_service.send(
            db,
            trigger_key=EmailTriggerKey.SERVICE_SCHEDULED_FAMILY,
            tenant_id=tenant_uuid,
            to=event.family_contact_email,
            context=context,
        )
        if not result.sent and not result.skipped_reason:
            await svc.log_event(
                service_id=event.id,
                tenant_id=event.tenant_id,
                actor_id=None,
                event_type="email_error",
                description="Confirmation email dispatch failed.",
            )
            await db.commit()
            raise RuntimeError("confirmation email dispatch failed")  # ARQ retry (max 3)

        await svc.log_event(
            service_id=event.id,
            tenant_id=event.tenant_id,
            actor_id=None,
            event_type="confirmation_email_sent",
            description=f"Confirmation email sent to {_mask_recipient(event.family_contact_email)}.",
            metadata={"recipient": _mask_recipient(event.family_contact_email)},
        )
        await db.commit()

    logger.info("[scheduling-email] confirmation sent for service %s", service_id)
    return {"status": "sent", "service_id": service_id}


async def send_scheduling_reminder(ctx: dict, service_id: str, tenant_id: str) -> dict:
    """24h-before-service reminder email to the family (INDL-38 AC-03/AC-04/AC-08)."""
    from uuid import UUID

    from sqlalchemy import select

    from src.apps.scheduling.models.service_event import ServiceEvent
    from src.apps.scheduling.services.scheduling_service import SchedulingService
    from src.apps.tenants.models.account import Account
    from src.core.constants import EmailTriggerKey
    from src.core.email_dispatch import cemetery_context, email_dispatch_service

    tenant_uuid, service_uuid = UUID(tenant_id), UUID(service_id)

    async with AsyncSessionLocal() as db:
        event = (
            await db.execute(
                select(ServiceEvent).where(
                    ServiceEvent.id == service_uuid,
                    ServiceEvent.tenant_id == tenant_uuid,
                )
            )
        ).scalar_one_or_none()
        if event is None or event.deleted_at is not None:
            logger.info("[scheduling-email] service %s not found/deleted — skipping reminder", service_id)
            return {"status": "skipped", "reason": "not_found_or_deleted"}

        # The flag may have been turned off after the job was scheduled (AC-05
        # safety net) — re-check it at fire time, not just at enqueue time.
        if not event.notify_family_reminder_24h:
            logger.info("[scheduling-email] reminder disabled for service %s — skipping", service_id)
            return {"status": "skipped", "reason": "reminder_disabled"}

        if event.status not in ("awaiting_family_confirm", "confirmed"):
            logger.info("[scheduling-email] service %s is draft — skipping reminder", service_id)
            return {"status": "skipped", "reason": "draft_status"}

        svc = SchedulingService(db)

        if not _is_valid_recipient(event.family_contact_email):
            await svc.log_event(
                service_id=event.id,
                tenant_id=event.tenant_id,
                actor_id=None,
                event_type="email_error",
                description="Reminder email not sent: family_contact_email is empty or invalid.",
            )
            await db.commit()
            return {"status": "skipped", "reason": "invalid_email"}

        account = (
            await db.execute(select(Account).where(Account.id == tenant_uuid))
        ).scalar_one_or_none()
        fields = _scheduling_email_fields(event)
        location = ", ".join(p for p in [event.plot_location, event.section] if p) or "To be confirmed"
        time_range = (
            f"{fields['start_time']} – {fields['end_time']}"
            if fields["start_time"] and fields["end_time"]
            else (fields["start_time"] or "To be confirmed")
        )
        context = {
            "family_name": event.family_contact_name or "Family",
            "service_type": event.service_type,
            "decedent_name": event.decedent_name or "your loved one",
            "service_date": fields["date_formatted"],
            "service_time": time_range,
            "location": location,
            **cemetery_context(account),
        }
        result = await email_dispatch_service.send(
            db,
            trigger_key=EmailTriggerKey.SERVICE_REMINDER_24H,
            tenant_id=tenant_uuid,
            to=event.family_contact_email,
            context=context,
        )
        if not result.sent and not result.skipped_reason:
            await svc.log_event(
                service_id=event.id,
                tenant_id=event.tenant_id,
                actor_id=None,
                event_type="email_error",
                description="Reminder email dispatch failed.",
            )
            await db.commit()
            raise RuntimeError("reminder email dispatch failed")  # ARQ retry (max 3)

        await svc.log_event(
            service_id=event.id,
            tenant_id=event.tenant_id,
            actor_id=None,
            event_type="reminder_email_sent",
            description=f"Reminder email sent to {_mask_recipient(event.family_contact_email)}.",
            metadata={"recipient": _mask_recipient(event.family_contact_email)},
        )
        await db.commit()

    logger.info("[scheduling-email] reminder sent for service %s", service_id)
    return {"status": "sent", "service_id": service_id}


async def check_overdue_invoices(ctx: dict):
    """Daily job: mark overdue invoices and send 7/14/30-day reminders."""
    logger.info("[scheduler] Running check_overdue_invoices")
    from src.database.session import AsyncSessionLocal
    from src.apps.billing.services.invoice_service import InvoiceService

    reminders: list = []
    async with AsyncSessionLocal() as db:
        service = InvoiceService()
        count = await service.transition_overdue(db, reminders_out=reminders)
        # Commit the overdue transitions + reminder_count bumps FIRST, then
        # enqueue — so a crash after commit can only miss a reminder, never
        # send a duplicate to a family (INDL-54e / SEC API4).
        await db.commit()

    redis = ctx.get("redis")
    enqueued = 0
    if redis is not None:
        for r in reminders:
            try:
                await redis.enqueue_job(
                    "send_invoice_reminder",
                    r["invoice_id"], r["tenant_id"], r["days_overdue"],
                )
                enqueued += 1
            except Exception as exc:  # noqa: BLE001
                logger.error("[scheduler] failed to enqueue reminder for invoice %s: %s",
                             r["invoice_id"], exc)

    logger.info("[scheduler] Processed %d overdue invoices, enqueued %d reminders",
                count, enqueued)
    return {"processed": count, "reminders_enqueued": enqueued}


async def generate_monthly_invoices(ctx: dict) -> dict:
    """1st-of-month cron: generate subscription invoices for all active tenants.

    Skips tenants that already have an invoice created in the current calendar month
    so the job is safe to re-run or to call manually mid-month without duplicates.
    """
    from datetime import date, timedelta
    from decimal import Decimal
    from sqlalchemy import extract, func, select
    from src.apps.billing.models.invoice import Invoice
    from src.apps.tenants.models.account import Account
    from src.core.constants import AccountStatus, PLAN_PRICES_CAD, SubscriptionPlan, TRIAL_DAYS

    today = date.today()
    created = 0
    skipped = 0

    async with AsyncSessionLocal() as db:
        accounts_result = await db.execute(
            select(Account).where(Account.status == AccountStatus.ACTIVE.value)
        )
        accounts = accounts_result.scalars().all()

        for account in accounts:
            # Skip if an invoice already exists for this calendar month
            month_count = (
                await db.execute(
                    select(func.count(Invoice.id)).where(
                        Invoice.tenant_id == account.id,
                        extract("year", Invoice.created_at) == today.year,
                        extract("month", Invoice.created_at) == today.month,
                    )
                )
            ).scalar() or 0

            if month_count > 0:
                skipped += 1
                continue

            plan_key = account.plan or SubscriptionPlan.PROFESSIONAL.value
            try:
                plan_enum = SubscriptionPlan(plan_key)
            except ValueError:
                plan_enum = SubscriptionPlan.PROFESSIONAL
            amount = Decimal(str(PLAN_PRICES_CAD.get(plan_enum, 349.00)))

            seq = (
                (
                    await db.execute(
                        select(func.count(Invoice.id)).where(Invoice.tenant_id == account.id)
                    )
                ).scalar() or 0
            ) + 1
            inv_number = f"INV-{today.year}-{seq:04d}"

            invoice = Invoice(
                tenant_id=account.id,
                invoice_number=inv_number,
                status="outstanding",
                total_amount=amount,
                paid_amount=Decimal("0"),
                balance_due=amount,
                due_date=today + timedelta(days=TRIAL_DAYS),
                purchaser_name=account.organization_name,
                purchaser_email=account.contact_email,
            )
            db.add(invoice)
            created += 1

        await db.commit()

    logger.info(
        "[cron] generate_monthly_invoices: created=%d skipped=%d month=%d-%02d",
        created, skipped, today.year, today.month,
    )
    return {"created": created, "skipped": skipped}


async def auto_close_stale_enquiries(ctx: dict) -> dict:
    """Daily cron: close qualified enquiries not converted after 30 days."""
    from src.apps.site_admin.services.enquiry_service import EnquiryService

    async with AsyncSessionLocal() as db:
        async with db.begin():
            service = EnquiryService(db)
            closed_ids = await service.auto_close_stale_qualified()
            logger.info(
                "[cron] auto_close_stale_enquiries: closed %d enquiries — ids: %s",
                len(closed_ids),
                [str(i) for i in closed_ids],
            )
    return {"closed": len(closed_ids), "ids": [str(i) for i in closed_ids]}


async def send_contract_email(ctx: dict, tenant_id: str, contract_id: str, payment_link_url: str):
    """Send the signed-contract confirmation to the purchaser (contract_signed).

    Rendered through the templated dispatch engine (INDL-54) — admins can edit
    the `contract_signed` template in Settings. `payment_url` is available as a
    merge field for templates that reference it.
    """
    from uuid import UUID
    from sqlalchemy import select
    from sqlalchemy.orm import selectinload
    from src.apps.sales.models.contract import Contract
    from src.apps.tenants.models.account import Account
    from src.core.constants import EmailTriggerKey
    from src.core.email_dispatch import cemetery_context, email_dispatch_service

    tenant_uuid = UUID(tenant_id)
    contract_uuid = UUID(contract_id)

    async with AsyncSessionLocal() as db:
        result = await db.execute(
            select(Contract)
            .options(selectinload(Contract.line_items))
            .where(
                Contract.id == contract_uuid,
                Contract.tenant_id == tenant_uuid,
                Contract.deleted_at.is_(None),
            )
        )
        contract = result.scalar_one_or_none()

        if not contract:
            logger.error("[contract-email] Contract %s not found", contract_id)
            return {"status": "not_found"}

        if not contract.purchaser_email:
            logger.info("[contract-email] No purchaser email for contract %s — skipping", contract_id)
            return {"status": "skipped", "reason": "no_purchaser_email"}

        account = (
            await db.execute(select(Account).where(Account.id == tenant_uuid))
        ).scalar_one_or_none()

        def _fmt(v):
            return f"{v:,.2f}" if isinstance(v, (int, float)) else (str(v) if v is not None else "")

        context = {
            "purchaser_name": contract.purchaser_name or "Valued Customer",
            "contract_number": contract.contract_number or "",
            "contract_type": getattr(contract, "contract_type", None) or "",
            "plot_id": getattr(contract, "plot_id", None) or "",
            "section": getattr(contract, "section", None) or "",
            "contract_total": _fmt(getattr(contract, "total_amount", None)),
            "balance_due": _fmt(getattr(contract, "balance_due", None)),
            "payment_url": payment_link_url or "",
            **cemetery_context(account),
        }
        result_r = await email_dispatch_service.send(
            db,
            trigger_key=EmailTriggerKey.CONTRACT_SIGNED,
            tenant_id=tenant_uuid,
            to=contract.purchaser_email,
            context=context,
        )
        await db.commit()

    logger.info("[contract-email] contract %s dispatch → sent=%s", contract_id, result_r.sent)
    return {"status": "sent" if result_r.sent else "skipped", "email": contract.purchaser_email}


async def send_news_notification(ctx: dict, tenant_id: str, post_id: str):
    """Notify all active subscribers when a news post is published."""
    from uuid import UUID
    from sqlalchemy import select
    from src.apps.news.models.news_post import NewsPost
    from src.apps.news.models.news_subscription import NewsSubscription
    from src.apps.tenants.models.account import Account
    from src.core.config import settings
    from src.core.constants import EmailTriggerKey
    from src.core.email_dispatch import cemetery_context, email_dispatch_service

    tenant_uuid = UUID(tenant_id)
    post_uuid = UUID(post_id)

    async with AsyncSessionLocal() as db:
        post_result = await db.execute(
            select(NewsPost).where(
                NewsPost.id == post_uuid,
                NewsPost.tenant_id == tenant_uuid,
                NewsPost.status == "published",
            )
        )
        post = post_result.scalar_one_or_none()
        if not post:
            logger.warning(
                "[news] Post %s not found or not published — skipping notification",
                post_id,
            )
            return {"status": "skipped", "reason": "post_not_found"}

        subs_result = await db.execute(
            select(NewsSubscription).where(
                NewsSubscription.tenant_id == tenant_uuid,
                NewsSubscription.unsubscribed_at.is_(None),
            )
        )
        subscribers = subs_result.scalars().all()

        if not subscribers:
            logger.info("[news] No active subscribers for tenant %s", tenant_id)
            return {"status": "ok", "sent": 0}

        account = (
            await db.execute(select(Account).where(Account.id == tenant_uuid))
        ).scalar_one_or_none()
        base_url = settings.APP_DOMAIN or "indelis.com"
        globals_ctx = cemetery_context(account)
        sent = 0
        for sub in subscribers:
            article_url = f"https://{base_url}/news/{post.slug}"
            unsubscribe_url = f"https://{base_url}/news/unsubscribe?token={sub.unsubscribe_token}"
            result = await email_dispatch_service.send(
                db,
                trigger_key=EmailTriggerKey.NEWS_NOTIFICATION,
                tenant_id=tenant_uuid,
                to=sub.email,
                context={
                    "post_title": post.title,
                    "post_excerpt": post.excerpt or "",
                    "article_url": article_url,
                    "unsubscribe_url": unsubscribe_url,
                    **globals_ctx,
                },
            )
            if result.sent:
                sent += 1
        await db.commit()

    logger.info(
        "[news] Sent %d/%d notifications for post %s",
        sent,
        len(subscribers),
        post_id,
    )
    return {"status": "ok", "sent": sent, "total": len(subscribers)}


async def startup(ctx: dict):
    load_all_models()
    logger.info("ARQ worker started")


async def shutdown(ctx: dict):
    logger.info("ARQ worker stopped")


# ── Worker Settings ───────────────────────────────────────────────────────────

class WorkerSettings:
    functions = [
        send_invitation_email,
        send_invoice_reminder,
        generate_qr_code,
        regenerate_all_qr,
        generate_contract_pdf,
        send_contract_email,
        check_overdue_invoices,
        generate_monthly_invoices,
        export_week_pdf,
        send_scheduling_confirmation,
        send_scheduling_reminder,
        auto_close_stale_enquiries,
        send_news_notification,
    ]
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = RedisSettings.from_dsn(settings.REDIS_URL)
    max_jobs = 10
    job_timeout = 300  # 5 min
    keep_result = 3600  # 1 hour

    # Scheduled jobs (cron)
    cron_jobs = [
        arq.cron(check_overdue_invoices, hour=0, minute=0),          # 00:00 UTC daily
        arq.cron(auto_close_stale_enquiries, hour=2, minute=0),      # 02:00 UTC daily
        arq.cron(generate_monthly_invoices, day=1, hour=6, minute=0),              # 06:00 UTC on 1st of each month
    ]