from __future__ import annotations import mimetypes import re import smtplib from pathlib import Path from datetime import datetime, timezone, timedelta from email.message import EmailMessage from typing import Any from sqlalchemy import select from sqlalchemy.orm import Session from app.modules.email_integration.models import EmailLog, EmailSetting, EmailTemplate from app.modules.email_integration.attachment_service import EmailAttachment from app.modules.core.iam.models import User from app.core.settings import get_settings DEFAULT_TEMPLATES: dict[str, dict[str, str]] = { "AUTH_LOGIN_OTP": { "name": "Login OTP", "subject": "Your OTP for {{ firm_name }} ERP login", "body": "Dear {{ user_name }},\n\nYour OTP for logging into {{ firm_name }} ERP is:\n\n{{ otp_code }}\n\nThis OTP is valid for {{ expiry_minutes }} minutes. Do not share it with anyone.\n\nIf you did not request this login, please contact your firm administrator immediately.\n\nRegards,\n{{ firm_name }}", }, "AUTH_PASSWORD_RESET": { "name": "Password Reset OTP", "subject": "Password reset OTP for {{ firm_name }} ERP", "body": "Dear {{ user_name }},\n\nWe received a request to reset your password for {{ firm_name }} ERP.\n\nYour password reset OTP is:\n\n{{ otp_code }}\n\nThis OTP is valid for {{ expiry_minutes }} minutes. If you did not request this reset, please contact your firm administrator immediately.\n\nRegards,\n{{ firm_name }}", }, "AUTH_PASSWORD_CHANGE": { "name": "Password Change OTP", "subject": "Confirm password change for {{ firm_name }} ERP", "body": "Dear {{ user_name }},\n\nYour OTP to confirm password change for {{ firm_name }} ERP is:\n\n{{ otp_code }}\n\nThis OTP is valid for {{ expiry_minutes }} minutes. If you did not request this change, please contact your firm administrator immediately.\n\nRegards,\n{{ firm_name }}", }, "AUTH_PASSWORD_RESET_OTP": { "name": "Password Reset OTP Alias", "subject": "Password reset OTP for {{ firm_name }} ERP", "body": "Dear {{ user_name }},\n\nYour password reset OTP for {{ firm_name }} ERP is:\n\n{{ otp_code }}\n\nPlease use this OTP to continue the password reset process.\n\nRegards,\n{{ firm_name }}", }, "AUTH_PASSWORD_CHANGE_OTP": { "name": "Password Change OTP Alias", "subject": "Confirm password change for {{ firm_name }} ERP", "body": "Dear {{ user_name }},\n\nYour OTP to confirm password change for {{ firm_name }} ERP is:\n\n{{ otp_code }}\n\nRegards,\n{{ firm_name }}", }, "AUTH_PASSWORD_RESET_LINK": { "name": "Password Reset Link", "subject": "Password reset request for {{ firm_name }} ERP", "body": "Dear {{ user_name }},\n\nWe received a request to reset your password for {{ firm_name }} ERP.\n\nClick the link below to reset your password:\n\n{{ reset_link }}\n\nThis link will expire in {{ expiry_hours }} hours. If you did not request this reset, please ignore this email or contact your firm administrator.\n\nRegards,\n{{ firm_name }}", }, "AUTH_USER_INVITE": { "name": "User Invite", "subject": "You are invited to {{ firm_name }} ERP", "body": "Dear {{ user_name }},\n\nYou have been invited to access {{ firm_name }} ERP.\n\nPlease click the link below to set your password and activate your account:\n\n{{ invite_link }}\n\nThis invite link will expire in {{ expiry_hours }} hours. If you were not expecting this invite, please contact the firm administrator.\n\nRegards,\n{{ firm_name }}", }, "AUTH_PASSWORD_CHANGED": { "name": "Password Changed", "subject": "Your {{ firm_name }} ERP password was changed", "body": "Dear {{ user_name }},\n\nYour password for {{ firm_name }} ERP was changed successfully on {{ changed_at }}.\n\nIf this change was not done by you, please contact your firm administrator immediately.\n\nRegards,\n{{ firm_name }}", }, "TASK_ASSIGNED": { "name": "Task Assigned", "subject": "Task assigned: {{ task_title }}", "body": "Dear {{ user_name }},\n\nA task has been assigned to you.\n\nClient: {{ client_name }}\nService: {{ service_name }}\nEngagement: {{ engagement_code }}\nTask: {{ task_title }}\nDue Date: {{ due_date }}\n\nPlease login to the ERP and update the task status.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "TASK_DUE_TODAY": { "name": "Task Due Today", "subject": "Task due today: {{ task_title }}", "body": "Dear {{ user_name }},\n\nThe following task is due today.\n\nClient: {{ client_name }}\nService: {{ service_name }}\nTask: {{ task_title }}\nDue Date: {{ due_date }}\n\nPlease complete or update the status in the ERP.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "TASK_OVERDUE": { "name": "Task Overdue", "subject": "Overdue task: {{ task_title }}", "body": "Dear {{ user_name }},\n\nThe following task is overdue.\n\nClient: {{ client_name }}\nService: {{ service_name }}\nEngagement: {{ engagement_code }}\nTask: {{ task_title }}\nDue Date: {{ due_date }}\n\nPlease update the status immediately.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "TASK_STATUS_UPDATED": { "name": "Task Status Updated", "subject": "Task status updated: {{ task_title }}", "body": "Dear {{ recipient_name }},\n\nThe task status has been updated.\n\nClient: {{ client_name }}\nTask: {{ task_title }}\nOld Status: {{ old_status }}\nNew Status: {{ new_status }}\nUpdated By: {{ updated_by }}\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "CLIENT_DOCUMENT_REQUEST": { "name": "Client Document Request", "subject": "Documents required - {{ service_name }} - {{ firm_name }}", "body": "Dear {{ client_name }},\n\nWe request you to provide the following documents for {{ service_name }}.\n\n{{ document_list }}\n\nReference: {{ reference_code }}\nDue Date: {{ due_date }}\n\nYou may upload the documents through the client portal.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "CLIENT_CLARIFICATION_REQUEST": { "name": "Client Clarification Request", "subject": "Clarification required - {{ service_name }} - {{ firm_name }}", "body": "Dear {{ client_name }},\n\nWe require your clarification for the following matter.\n\nService: {{ service_name }}\nReference: {{ reference_code }}\nClarification Required: {{ clarification_text }}\n\nPlease reply through the client portal or contact your auditor.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "CLIENT_DOCUMENT_RECEIVED": { "name": "Client Document Received", "subject": "Document received - {{ client_name }}", "body": "Dear {{ recipient_name }},\n\nA document has been received from the client.\n\nClient: {{ client_name }}\nDocument: {{ document_name }}\nService: {{ service_name }}\nUploaded By: {{ uploaded_by }}\n\nPlease review it in the ERP.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "PARTNER_REVIEW_REQUIRED": { "name": "Partner Review Required", "subject": "Review required: {{ work_title }}", "body": "Dear {{ partner_name }},\n\nThe following work is pending for your review.\n\nClient: {{ client_name }}\nService: {{ service_name }}\nEngagement: {{ engagement_code }}\nWork: {{ work_title }}\nDue Date: {{ due_date }}\n\nPlease review and approve or send for rework.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "PARTNER_REWORK_ASSIGNED": { "name": "Partner Rework Assigned", "subject": "Rework assigned: {{ work_title }}", "body": "Dear {{ user_name }},\n\nThe partner has requested rework on the following item.\n\nClient: {{ client_name }}\nService: {{ service_name }}\nWork: {{ work_title }}\nReview Note: {{ review_note }}\n\nPlease update the work and resubmit for review.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "CONSULTANT_ASSIGNMENT": { "name": "Consultant Assignment", "subject": "Assignment from {{ firm_name }}: {{ assignment_title }}", "body": "Dear {{ consultant_name }},\n\nYou have been assigned the following work.\n\nClient: {{ client_name }}\nService: {{ service_name }}\nAssignment: {{ assignment_title }}\nDue Date: {{ due_date }}\n\nPlease login to the consultant portal to view details and submit updates.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "CONSULTANT_CLARIFICATION_REQUEST": { "name": "Consultant Clarification Request", "subject": "Clarification required: {{ assignment_title }}", "body": "Dear {{ consultant_name }},\n\nWe require clarification on the following consultant assignment.\n\nClient: {{ client_name }}\nAssignment: {{ assignment_title }}\nClarification Required: {{ clarification_text }}\n\nPlease reply through the consultant portal.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "CONSULTANT_SUBMISSION_RECEIVED": { "name": "Consultant Submission Received", "subject": "Consultant submission received - {{ client_name }}", "body": "Dear {{ recipient_name }},\n\nA consultant submission has been received.\n\nConsultant: {{ consultant_name }}\nClient: {{ client_name }}\nAssignment: {{ assignment_title }}\nSubmitted On: {{ submitted_at }}\n\nPlease review it in the ERP.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "INVOICE_GENERATED": { "name": "Invoice Generated", "subject": "Invoice {{ invoice_number }} from {{ firm_name }}", "body": "Dear {{ client_name }},\n\nInvoice {{ invoice_number }} has been generated by {{ firm_name }}.\n\nInvoice Amount: {{ invoice_amount }}\nDue Date: {{ due_date }}\n\nYou may view or pay the invoice using the link below.\n\n{{ payment_link }}\n\nRegards,\n{{ firm_name }}", }, "PAYMENT_REMINDER": { "name": "Payment Reminder", "subject": "Payment reminder for invoice {{ invoice_number }}", "body": "Dear {{ client_name }},\n\nThis is a gentle reminder that payment is pending against the following invoice.\n\nInvoice Number: {{ invoice_number }}\nInvoice Amount: {{ invoice_amount }}\nOutstanding Amount: {{ outstanding_amount }}\nDue Date: {{ due_date }}\n\nPayment Link: {{ payment_link }}\n\nIf payment has already been made, please share the payment details with us.\n\nRegards,\n{{ firm_name }}", }, "PAYMENT_RECEIVED_RECEIPT": { "name": "Payment Received Receipt", "subject": "Payment received for invoice {{ invoice_number }}", "body": "Dear {{ client_name }},\n\nWe acknowledge receipt of your payment.\n\nInvoice Number: {{ invoice_number }}\nReceipt Number: {{ receipt_number }}\nAmount Received: {{ payment_amount }}\nPayment Date: {{ payment_date }}\nMode: {{ payment_mode }}\n\nThank you.\n\nRegards,\n{{ firm_name }}", }, "ONLINE_PAYMENT_SUCCESS": { "name": "Online Payment Success", "subject": "Online payment successful - {{ invoice_number }}", "body": "Dear {{ client_name }},\n\nYour online payment has been successfully received.\n\nInvoice Number: {{ invoice_number }}\nAmount Paid: {{ payment_amount }}\nGateway: {{ gateway_name }}\nTransaction Reference: {{ transaction_reference }}\n\nReceipt Number: {{ receipt_number }}\n\nRegards,\n{{ firm_name }}", }, "ONLINE_PAYMENT_FAILED": { "name": "Online Payment Failed", "subject": "Online payment failed - {{ invoice_number }}", "body": "Dear {{ client_name }},\n\nYour online payment attempt could not be completed.\n\nInvoice Number: {{ invoice_number }}\nAmount: {{ invoice_amount }}\nGateway: {{ gateway_name }}\nReason: {{ failure_reason }}\n\nPlease try again using the payment link below or contact us for assistance.\n\n{{ payment_link }}\n\nRegards,\n{{ firm_name }}", }, "LEAVE_REQUEST_SUBMITTED": { "name": "Leave Request Submitted", "subject": "Leave request submitted by {{ employee_name }}", "body": "Dear {{ manager_name }},\n\nA leave request has been submitted.\n\nEmployee: {{ employee_name }}\nLeave Type: {{ leave_type }}\nFrom: {{ from_date }}\nTo: {{ to_date }}\nReason: {{ reason }}\n\nPlease review it in the ERP.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "LEAVE_APPROVED": { "name": "Leave Approved", "subject": "Leave approved - {{ firm_name }}", "body": "Dear {{ employee_name }},\n\nYour leave request has been approved.\n\nLeave Type: {{ leave_type }}\nFrom: {{ from_date }}\nTo: {{ to_date }}\nApproved By: {{ approved_by }}\n\nRegards,\n{{ firm_name }}", }, "LEAVE_REJECTED": { "name": "Leave Rejected", "subject": "Leave request update - {{ firm_name }}", "body": "Dear {{ employee_name }},\n\nYour leave request has been reviewed and rejected.\n\nLeave Type: {{ leave_type }}\nFrom: {{ from_date }}\nTo: {{ to_date }}\nReason/Remarks: {{ remarks }}\n\nRegards,\n{{ firm_name }}", }, "ATTENDANCE_PUNCH_MISSING": { "name": "Attendance Punch Missing", "subject": "Attendance punch missing - {{ attendance_date }}", "body": "Dear {{ employee_name }},\n\nYour attendance record appears incomplete.\n\nDate: {{ attendance_date }}\nMissing Punch: {{ missing_punch }}\n\nPlease regularise or contact your manager.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, "CLIENT_PORTAL_WELCOME": { "name": "Client Portal Welcome", "subject": "Welcome to {{ firm_name }} Client Portal", "body": "Dear {{ client_name }},\n\nWelcome to the {{ firm_name }} client portal.\n\nYou can use the portal to view compliance status, upload documents, reply to clarifications, view invoices and download receipts.\n\nLogin URL: {{ login_url }}\n\nRegards,\n{{ firm_name }}", }, "CONSULTANT_LEAD_FORWARDED": { "name": "Consultant Lead Forwarded", "subject": "New lead referred by {{ consultant_name }}", "body": "Dear {{ recipient_name }},\n\nA consultant has forwarded a new lead to the firm.\n\nConsultant: {{ consultant_name }}\nClient/Lead: {{ client_name }}\nService Required: {{ service_name }}\nContact: {{ client_contact }}\n\nPlease review and convert the lead if suitable.\n\n{{ action_url }}\n\nRegards,\n{{ firm_name }}", }, } _TOKEN_RE = re.compile(r"{{\s*([a-zA-Z0-9_\.]+)\s*}}") def _clean(value: Any) -> str: return "" if value is None else str(value) def render_template_text(template_text: str, context: dict[str, Any]) -> str: def repl(match: re.Match[str]) -> str: key = match.group(1) return _clean(context.get(key, "")) return _TOKEN_RE.sub(repl, template_text or "") def seed_default_email_templates(db: Session, tenant_id: int | None = None, branch_id: int | None = None) -> None: """Seed default templates for one tenant/branch scope without duplicate inserts. This function may be called more than once in the same request, for example when Email Settings creates the default settings row and the page also refreshes the template list. SQLAlchemy pending objects are not always visible to the later SELECT in a way that prevents duplicate INSERTs before commit, so we explicitly check both database rows and pending session rows. """ q = select(EmailTemplate.template_code).where(EmailTemplate.tenant_id == tenant_id) q = q.where(EmailTemplate.branch_id.is_(None)) if branch_id is None else q.where(EmailTemplate.branch_id == branch_id) existing_codes = set(db.execute(q).scalars().all()) for obj in list(db.new): if not isinstance(obj, EmailTemplate): continue if obj.tenant_id == tenant_id and obj.branch_id == branch_id and obj.template_code: existing_codes.add(obj.template_code) for code, payload in DEFAULT_TEMPLATES.items(): if code in existing_codes: continue db.add( EmailTemplate( tenant_id=tenant_id, branch_id=branch_id, template_code=code, template_name=payload["name"], subject_template=payload["subject"], body_template=payload["body"], is_html=False, is_active=True, ) ) existing_codes.add(code) def get_email_setting(db: Session, tenant_id: int | None, branch_id: int | None) -> EmailSetting | None: if tenant_id is None: return None if branch_id is not None: row = db.execute( select(EmailSetting).where(EmailSetting.tenant_id == tenant_id, EmailSetting.branch_id == branch_id) ).scalar_one_or_none() if row: return row return db.execute( select(EmailSetting).where(EmailSetting.tenant_id == tenant_id, EmailSetting.branch_id.is_(None)) ).scalar_one_or_none() def get_or_create_email_setting(db: Session, tenant_id: int, branch_id: int | None, actor_user_id: int | None = None) -> EmailSetting: row = db.execute( select(EmailSetting).where(EmailSetting.tenant_id == tenant_id, EmailSetting.branch_id == branch_id) ).scalar_one_or_none() if row: return row settings = get_settings() row = EmailSetting( tenant_id=tenant_id, branch_id=branch_id, # SMTP defaults from Coolify / environment smtp_host=getattr(settings, "SMTP_HOST", None) or "mail.vavalam.com", smtp_port=int(getattr(settings, "SMTP_PORT", 465) or 465), smtp_security="SSL", smtp_username=getattr(settings, "SMTP_USERNAME", None) or "no-reply@vavalam.com", smtp_password=getattr(settings, "SMTP_PASSWORD", None), # Sender defaults from_email=getattr(settings, "SMTP_FROM_EMAIL", None) or "no-reply@vavalam.com", from_name=getattr(settings, "SMTP_FROM_NAME", None) or "ARRR ERP", # IMAP defaults from Coolify / environment imap_host=getattr(settings, "IMAP_HOST", None) or "mail.vavalam.com", imap_port=int(getattr(settings, "IMAP_PORT", 993) or 993), imap_security="SSL", created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id, ) db.add(row) db.flush() seed_default_email_templates(db, tenant_id=tenant_id, branch_id=branch_id) return row def _get_template(db: Session, tenant_id: int | None, branch_id: int | None, template_code: str) -> EmailTemplate | None: scopes = [] if tenant_id is not None and branch_id is not None: scopes.append((tenant_id, branch_id)) if tenant_id is not None: scopes.append((tenant_id, None)) scopes.append((None, None)) for t_id, b_id in scopes: q = select(EmailTemplate).where( EmailTemplate.tenant_id == t_id, EmailTemplate.template_code == template_code, EmailTemplate.is_active.is_(True), ) q = q.where(EmailTemplate.branch_id == b_id) if b_id is not None else q.where(EmailTemplate.branch_id.is_(None)) row = db.execute(q).scalar_one_or_none() if row: return row defaults = DEFAULT_TEMPLATES.get(template_code) if not defaults: return None return EmailTemplate( tenant_id=tenant_id, branch_id=branch_id, template_code=template_code, template_name=defaults["name"], subject_template=defaults["subject"], body_template=defaults["body"], is_html=False, is_active=True, ) def _firm_name(db: Session, tenant_id: int | None) -> str: try: from app.modules.core.tenancy.models import Tenant tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one_or_none() if tenant_id else None return getattr(tenant, "display_name", None) or getattr(tenant, "name", None) or "Audit Firm" except Exception: return "Audit Firm" def _normalise_attachment(raw: EmailAttachment | str | Path) -> EmailAttachment: if isinstance(raw, EmailAttachment): return raw path = Path(raw) if not path.exists() or not path.is_file(): raise FileNotFoundError(f"Email attachment not found: {path}") content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream" return EmailAttachment(filename=path.name, content=path.read_bytes(), content_type=content_type) def _send_smtp( setting: EmailSetting, recipient_email: str, subject: str, body: str, is_html: bool = False, attachments: list[EmailAttachment | str | Path] | None = None, ) -> None: host = (setting.smtp_host or "").strip() port = int(setting.smtp_port or 0) username = (setting.smtp_username or "").strip() password = setting.smtp_password or "" from_email = (setting.from_email or username or "").strip() from_name = (setting.from_name or "Audit Firm ERP").strip() reply_to = (setting.reply_to_email or from_email).strip() security = (setting.smtp_security or "SSL").upper() if not host or not port or not from_email: raise ValueError("SMTP host, port and from email are required.") msg = EmailMessage() msg["Subject"] = subject msg["From"] = f"{from_name} <{from_email}>" msg["To"] = recipient_email if reply_to: msg["Reply-To"] = reply_to if is_html: msg.set_content("This email requires an HTML compatible email client.") msg.add_alternative(body, subtype="html") else: msg.set_content(body) total_attachment_bytes = 0 for raw_attachment in attachments or []: attachment = _normalise_attachment(raw_attachment) content = attachment.content or b"" total_attachment_bytes += len(content) if total_attachment_bytes > 10 * 1024 * 1024: raise ValueError("Total email attachment size exceeds 10 MB safe limit.") maintype, subtype = (attachment.content_type or "application/octet-stream").split("/", 1) msg.add_attachment( content, maintype=maintype, subtype=subtype, filename=attachment.filename or "attachment", ) timeout = int(setting.smtp_timeout_seconds or 20) if security == "SSL": with smtplib.SMTP_SSL(host, port, timeout=timeout) as smtp: if username: smtp.login(username, password) smtp.send_message(msg) else: with smtplib.SMTP(host, port, timeout=timeout) as smtp: if security == "STARTTLS": smtp.starttls() if username: smtp.login(username, password) smtp.send_message(msg) def is_template_allowed_by_preferences(setting: EmailSetting, template_code: str) -> tuple[bool, str | None]: """Return whether a template may be sent under firm/branch preferences. Phase 7S.1C keeps the switches firm-level so the same SMTP account can be used while selectively enabling/disabling modules. This helper is defensive with getattr() so older databases/files do not break during staged upgrades. """ code = (template_code or "").upper().strip() if code.startswith("AUTH_") and not getattr(setting, "send_auth_emails", False): return False, "Authentication emails disabled in email preferences." if code.startswith("TASK_") and not getattr(setting, "send_task_emails", False): return False, "Task/work emails disabled in email preferences." if code in {"INVOICE_GENERATED", "PAYMENT_REMINDER"}: if not getattr(setting, "send_billing_emails", False): return False, "Billing emails disabled in email preferences." if not getattr(setting, "send_invoice_emails", True): return False, "Invoice/reminder emails disabled in email preferences." if code in {"PAYMENT_RECEIVED_RECEIPT", "ONLINE_PAYMENT_SUCCESS", "ONLINE_PAYMENT_FAILED"}: if not getattr(setting, "send_billing_emails", False): return False, "Billing emails disabled in email preferences." if not getattr(setting, "send_payment_emails", True): return False, "Payment/receipt emails disabled in email preferences." if code.startswith("ONLINE_PAYMENT") and not getattr(setting, "send_online_payment_emails", True): return False, "Online payment emails disabled in email preferences." if code.startswith("CLIENT_") and not getattr(setting, "send_client_emails", True): return False, "Client emails disabled in email preferences." if "DOCUMENT" in code and not getattr(setting, "send_document_emails", True): return False, "Document-request/receipt emails disabled in email preferences." if code.startswith("CONSULTANT_") and not getattr(setting, "send_consultant_emails", True): return False, "Consultant emails disabled in email preferences." if code.startswith("PARTNER_") and not getattr(setting, "send_partner_review_emails", True): return False, "Partner review emails disabled in email preferences." if code.startswith("LEAVE_") or code.startswith("ATTENDANCE_"): if not getattr(setting, "send_leave_attendance_emails", True): return False, "Leave/attendance emails disabled in email preferences." return True, None def _retry_delay(attempt_count: int) -> timedelta: """Small exponential backoff for SMTP failures. Attempt 1 -> 5 minutes, 2 -> 15 minutes, 3+ -> 60 minutes. This keeps local development friendly while preventing repeated immediate SMTP retries when credentials/server are wrong. """ if attempt_count <= 1: return timedelta(minutes=5) if attempt_count == 2: return timedelta(minutes=15) return timedelta(minutes=60) def _attachments_for_log(db: Session, log: EmailLog) -> list[EmailAttachment | str | Path]: """Regenerate known billing attachments during retry. We intentionally do not persist raw attachment bytes in the database. For invoice/receipt emails, attachments are safely regenerated from the related billing records. For all other templates, retries are sent without attachments. """ try: if not log.related_id: return [] if log.template_code == "INVOICE_GENERATED" and log.related_module == "billing_invoice": from app.modules.billing.models import BillingInvoice from app.modules.email_integration.attachment_service import invoice_attachment invoice = db.execute(select(BillingInvoice).where(BillingInvoice.id == int(log.related_id))).scalar_one_or_none() return [invoice_attachment(invoice, firm_name=_firm_name(db, log.tenant_id))] if invoice else [] if log.template_code == "PAYMENT_RECEIVED_RECEIPT" and log.related_module == "billing_payment": from app.modules.billing.models import BillingPayment from app.modules.email_integration.attachment_service import receipt_attachment payment = db.execute(select(BillingPayment).where(BillingPayment.id == int(log.related_id))).scalar_one_or_none() return [receipt_attachment(payment, firm_name=_firm_name(db, log.tenant_id))] if payment else [] except Exception: return [] return [] def send_email_log_now( db: Session, log: EmailLog, *, attachments: list[EmailAttachment | str | Path] | None = None, send_immediately: bool = True, max_attempts: int = 3, queue_priority: int = 100, ) -> EmailLog: """Attempt to send one queued/pending email log and update retry metadata.""" setting = get_email_setting(db, log.tenant_id, log.branch_id) now = datetime.now(timezone.utc) log.processing_started_at = now log.last_attempt_at = now log.attempt_count = int(getattr(log, "attempt_count", 0) or 0) + 1 if not setting or not setting.is_active: log.status = "SKIPPED" log.error_message = "Email settings not configured or inactive." log.is_retryable = False log.processing_started_at = None db.flush() return log try: send_attachments = attachments if attachments is not None else _attachments_for_log(db, log) _send_smtp( setting, log.recipient_email, log.subject, log.body or "", is_html=False, attachments=send_attachments, ) log.status = "SENT" log.sent_at = datetime.now(timezone.utc) log.error_message = None log.next_retry_at = None log.processing_started_at = None log.is_retryable = False except Exception as exc: log.status = "FAILED" log.error_message = str(exc) log.processing_started_at = None max_attempts = int(getattr(log, "max_attempts", 3) or 3) if log.attempt_count < max_attempts and bool(getattr(log, "is_retryable", True)): log.next_retry_at = datetime.now(timezone.utc) + _retry_delay(log.attempt_count) log.is_retryable = True else: log.next_retry_at = None log.is_retryable = False db.flush() return log def process_pending_email_queue( db: Session, *, tenant_id: int | None = None, branch_id: int | None = None, limit: int = 25, ) -> dict[str, int]: """Send retryable pending/failed emails due for retry. This can be called manually from /email/queue/process and later from a background scheduler/worker. It is intentionally conservative and does not affect SKIPPED or permanently failed rows. """ now = datetime.now(timezone.utc) q = select(EmailLog).where( EmailLog.status.in_(["PENDING", "FAILED"]), EmailLog.is_retryable.is_(True), EmailLog.attempt_count < EmailLog.max_attempts, ).where( (EmailLog.next_retry_at.is_(None)) | (EmailLog.next_retry_at <= now) ) if tenant_id is not None: q = q.where(EmailLog.tenant_id == tenant_id) if branch_id is not None: q = q.where(EmailLog.branch_id == branch_id) rows = db.execute( q.order_by(EmailLog.queue_priority.asc(), EmailLog.created_at_utc.asc()).limit(max(1, min(int(limit or 25), 100))) ).scalars().all() result = {"processed": 0, "sent": 0, "failed": 0, "skipped": 0} for row in rows: send_email_log_now(db, row) result["processed"] += 1 if row.status == "SENT": result["sent"] += 1 elif row.status == "SKIPPED": result["skipped"] += 1 else: result["failed"] += 1 db.flush() return result def send_template_email( db: Session, *, tenant_id: int | None, branch_id: int | None, recipient_email: str, template_code: str, context: dict[str, Any] | None = None, related_module: str | None = None, related_id: int | None = None, force_send: bool = False, attachments: list[EmailAttachment | str | Path] | None = None, send_immediately: bool = True, max_attempts: int = 3, queue_priority: int = 100, ) -> EmailLog: context = dict(context or {}) context.setdefault("firm_name", _firm_name(db, tenant_id)) context.setdefault("support_email", "") template = _get_template(db, tenant_id, branch_id, template_code) if not template: subject = template_code body = "" is_html = False else: subject = render_template_text(template.subject_template, context) body = render_template_text(template.body_template, context) is_html = bool(template.is_html) log = EmailLog( tenant_id=tenant_id, branch_id=branch_id, recipient_email=recipient_email, subject=subject[:500], body=body, status="PENDING", related_module=related_module, related_id=related_id, template_code=template_code, queued_at=datetime.now(timezone.utc), max_attempts=max(1, int(max_attempts or 3)), queue_priority=int(queue_priority or 100), is_retryable=True, ) db.add(log) db.flush() setting = get_email_setting(db, tenant_id, branch_id) if not setting or not setting.is_active: log.status = "SKIPPED" log.error_message = "Email settings not configured or inactive." db.flush() return log if not force_send: allowed, reason = is_template_allowed_by_preferences(setting, template_code) if not allowed: log.status = "SKIPPED" log.error_message = reason or "Email disabled in email preferences." db.flush() return log if not send_immediately: db.flush() return log # Immediate first attempt keeps OTP/test email behaviour familiar, while # Phase 7S.1E retry metadata ensures SMTP failures can be retried later # from the email queue page without blocking the business transaction. send_email_log_now(db, log, attachments=attachments) db.flush() return log def _public_base_url() -> str: base = (get_settings().ERP_PUBLIC_BASE_URL or "").strip().rstrip("/") return base or "http://localhost:8000" def _support_email_from_setting(db: Session, tenant_id: int | None, branch_id: int | None) -> str: setting = get_email_setting(db, tenant_id, branch_id) return (getattr(setting, "reply_to_email", None) or getattr(setting, "from_email", None) or "").strip() if setting else "" def _user_display_name(user: User) -> str: return getattr(user, "full_name", None) or str(getattr(user, "email", "User")) def send_auth_otp_email(db: Session, *, user: User, otp_code: str, purpose: str) -> EmailLog | None: code_map = { "login": "AUTH_LOGIN_OTP", "password_reset": "AUTH_PASSWORD_RESET", "password_change": "AUTH_PASSWORD_CHANGE", } template_code = code_map.get(purpose, "AUTH_LOGIN_OTP") if not getattr(user, "email", None): return None tenant_id = getattr(user, "tenant_id", None) branch_id = getattr(user, "branch_id", None) return send_template_email( db, tenant_id=tenant_id, branch_id=branch_id, recipient_email=str(user.email), template_code=template_code, context={ "user_name": _user_display_name(user), "user_email": str(user.email), "otp_code": otp_code, "expiry_minutes": "10", "support_email": _support_email_from_setting(db, tenant_id, branch_id), }, related_module="auth", related_id=int(user.id), force_send=True, queue_priority=10, ) def send_password_reset_link_email(db: Session, *, user: User, reset_token: str) -> EmailLog | None: if not getattr(user, "email", None): return None tenant_id = getattr(user, "tenant_id", None) branch_id = getattr(user, "branch_id", None) reset_link = f"{_public_base_url()}/password-reset/accept?token={reset_token}" return send_template_email( db, tenant_id=tenant_id, branch_id=branch_id, recipient_email=str(user.email), template_code="AUTH_PASSWORD_RESET_LINK", context={ "user_name": _user_display_name(user), "user_email": str(user.email), "reset_link": reset_link, "expiry_hours": str(get_settings().PASSWORD_RESET_HOURS), "support_email": _support_email_from_setting(db, tenant_id, branch_id), }, related_module="auth_password_reset", related_id=int(user.id), force_send=True, queue_priority=10, ) def send_user_invite_email(db: Session, *, user: User, invite_token: str) -> EmailLog | None: if not getattr(user, "email", None): return None tenant_id = getattr(user, "tenant_id", None) branch_id = getattr(user, "branch_id", None) invite_link = f"{_public_base_url()}/invite/accept?token={invite_token}" return send_template_email( db, tenant_id=tenant_id, branch_id=branch_id, recipient_email=str(user.email), template_code="AUTH_USER_INVITE", context={ "user_name": _user_display_name(user), "user_email": str(user.email), "invite_link": invite_link, "expiry_hours": str(get_settings().INVITE_TOKEN_HOURS), "support_email": _support_email_from_setting(db, tenant_id, branch_id), }, related_module="auth_invite", related_id=int(user.id), force_send=True, queue_priority=10, ) def send_password_changed_email(db: Session, *, user: User) -> EmailLog | None: if not getattr(user, "email", None): return None tenant_id = getattr(user, "tenant_id", None) branch_id = getattr(user, "branch_id", None) return send_template_email( db, tenant_id=tenant_id, branch_id=branch_id, recipient_email=str(user.email), template_code="AUTH_PASSWORD_CHANGED", context={ "user_name": _user_display_name(user), "user_email": str(user.email), "changed_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"), "support_email": _support_email_from_setting(db, tenant_id, branch_id), }, related_module="auth_password_changed", related_id=int(user.id), force_send=True, queue_priority=20, )