""" Celery task: poll and dispatch due invoice reminders. Task name: invoice.send_reminder_poll Schedule: every 900 seconds (15 min) via Celery Beat For each due reminder: - Acquire a Redis lock keyed on reminder.id (TTL = 14 min) for idempotency. - Re-fetch the invoice status — skip if PAID or CANCELLED. - Dispatch via email / SMS based on reminder.channel. - Update reminder_status = 'sent', sent_at = NOW(). - Write InvoiceActivity(INVOICE_REMINDER_SENT) or (INVOICE_REMINDER_FAILED). """ import logging from datetime import datetime, timezone from celery.utils.log import get_task_logger from src.worker.celery_app import celery_app from src.core.database import SessionCelery from src.core.utils.enums import InvoiceStatusTypes, InvoiceActivityTypes, ReminderStatus logger = get_task_logger(__name__) # Statuses that should suppress reminder sending _TERMINAL_STATUSES = {InvoiceStatusTypes.PAID, InvoiceStatusTypes.CANCELLED} # Redis lock TTL: 14 minutes (just under the 15-min poll interval) _LOCK_TTL_SECONDS = 840 @celery_app.task(name="invoice.send_reminder_poll") def send_invoice_reminder_poll() -> dict: """ Poll every 15 min. Sends all due reminders. Idempotent via Redis lock. Returns: Dict with counts: sent, skipped, failed. """ logger.info("send_invoice_reminder_poll starting") sent_count = 0 skipped_count = 0 failed_count = 0 with SessionCelery() as db: from src.apps.invoices import crud as invoice_crud from src.apps.invoices.models.invoice import Invoice from sqlalchemy import select due_reminders = invoice_crud.get_due_reminders(db) logger.info("send_invoice_reminder_poll: %d reminders due", len(due_reminders)) redis_client = _get_redis_client() for reminder in due_reminders: lock_key = f"invoice_reminder_lock:{reminder.id}" # Try to acquire Redis lock (idempotency guard) if redis_client is not None: acquired = redis_client.set(lock_key, "1", nx=True, ex=_LOCK_TTL_SECONDS) if not acquired: logger.debug("Reminder id=%s already locked, skipping", reminder.id) skipped_count += 1 continue try: # Re-fetch invoice status with merchant_id cross-check to prevent # a reminder belonging to one merchant from triggering actions on # an invoice owned by a different merchant. invoice_stmt = select(Invoice).where( Invoice.id == reminder.invoice_id, Invoice.merchant_id == reminder.merchant_id, Invoice.deleted_at == None, ) invoice = db.execute(invoice_stmt).scalar_one_or_none() if invoice is None: logger.warning("Reminder id=%s: invoice_id=%s not found", reminder.id, reminder.invoice_id) _mark_reminder_failed(db, reminder, invoice_crud, "invoice not found") failed_count += 1 continue # Suppress for terminal statuses if invoice.status in _TERMINAL_STATUSES: logger.info( "Reminder id=%s: invoice status=%s is terminal, marking cancelled", reminder.id, invoice.status, ) reminder.reminder_status = ReminderStatus.CANCELLED reminder.updated_at = datetime.now(timezone.utc) db.flush() skipped_count += 1 continue # Determine recipient recipient_email = _get_email(invoice) recipient_phone = _get_phone(invoice) channel = (reminder.channel or "email").lower() success = True try: if channel in ("email", "both") and recipient_email: _dispatch_email_reminder(invoice, reminder, recipient_email) if channel in ("sms", "both") and recipient_phone: _dispatch_sms_reminder(invoice, reminder, recipient_phone) except Exception as exc: logger.error("Reminder id=%s dispatch failed: %s", reminder.id, exc) _mark_reminder_failed(db, reminder, invoice_crud, str(exc)) failed_count += 1 success = False if success: now = datetime.now(timezone.utc) reminder.reminder_status = ReminderStatus.SENT reminder.sent_at = now reminder.updated_at = now db.flush() invoice_crud.write_activity( db=db, invoice_id=invoice.id, activity_type=InvoiceActivityTypes.INVOICE_REMINDER_SENT, description=f"Reminder sent via {channel}", actor_type="system", actor_id=None, metadata={ "reminder_id": reminder.reminder_id, "channel": channel, "recipient_email": recipient_email, }, ) sent_count += 1 except Exception as exc: logger.error("Unexpected error processing reminder id=%s: %s", reminder.id, exc) failed_count += 1 logger.info( "send_invoice_reminder_poll completed: sent=%d skipped=%d failed=%d", sent_count, skipped_count, failed_count, ) return {"sent": sent_count, "skipped": skipped_count, "failed": failed_count} def _mark_reminder_failed(db, reminder, invoice_crud, reason: str) -> None: now = datetime.now(timezone.utc) reminder.reminder_status = ReminderStatus.FAILED reminder.failure_reason = reason reminder.updated_at = now db.flush() if reminder.invoice_id: invoice_crud.write_activity( db=db, invoice_id=reminder.invoice_id, activity_type=InvoiceActivityTypes.INVOICE_REMINDER_FAILED, description=f"Reminder failed: {reason}", actor_type="system", actor_id=None, metadata={"reminder_id": reminder.reminder_id, "reason": reason}, ) def _get_email(invoice) -> str | None: if invoice.payer and hasattr(invoice.payer, "email"): return invoice.payer.email if invoice.customer and hasattr(invoice.customer, "email"): return invoice.customer.email return None def _get_phone(invoice) -> str | None: if invoice.payer and hasattr(invoice.payer, "phone"): return invoice.payer.phone if invoice.customer and hasattr(invoice.customer, "phone"): return invoice.customer.phone return None def _dispatch_email_reminder(invoice, reminder, recipient_email: str) -> None: try: from src.apps.notifications.services import send_email # type: ignore literal = invoice.invoice_literal or str(invoice.id) subject = f"Reminder: Invoice {literal} is due" body = ( f"This is a reminder that invoice {literal} " f"for ${invoice.amount:.2f} is due" + (f" on {invoice.due_date.strftime('%B %d, %Y')}" if invoice.due_date else "") + "." ) send_email(to_email=recipient_email, subject=subject, body=body) except ImportError: logger.warning("_dispatch_email_reminder: notifications service unavailable") def _dispatch_sms_reminder(invoice, reminder, recipient_phone: str) -> None: try: from src.apps.notifications.services import send_sms # type: ignore literal = invoice.invoice_literal or str(invoice.id) message = ( f"Reminder: Invoice {literal} for ${invoice.amount:.2f} is due" + (f" on {invoice.due_date.strftime('%b %d')}" if invoice.due_date else "") + "." ) send_sms(to_phone=recipient_phone, message=message) except ImportError: logger.warning("_dispatch_sms_reminder: notifications service unavailable") def _get_redis_client(): """Return a Redis client if Redis is configured; else return None (non-blocking).""" try: import redis from src.core.config import settings backend_url = getattr(settings, "REDIS_URL", None) or getattr(settings, "CELERY_RESULT_BACKEND", None) if backend_url and "redis" in backend_url: return redis.from_url(backend_url, decode_responses=True) except Exception: pass return None