"""Background email queue with durable outbox pattern. Architecture: - enqueue_email() creates an outbox record inside the same DB transaction as the registration/status update, ensuring no email is lost. - EmailWorker threads poll pending jobs, lock them, process them (build HTML, generate PDF, send SMTP), and update status. - Configurable concurrency, exponential backoff retry, rate limiting. """ import os import sqlite3 import threading import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta from nexora.config import ( EMAIL_BATCH_SIZE, EMAIL_JOB_TIMEOUT, EMAIL_MAX_RETRIES, EMAIL_POLL_INTERVAL, EMAIL_RATE_LIMIT_PER_MINUTE, EMAIL_WORKER_CONCURRENCY, logger, ) from nexora.database.connection import DB_PATH # Wake event: set whenever a new job is enqueued so the worker poll loop # processes it immediately instead of waiting for the next poll interval. _WAKE = threading.Event() def wake_worker(): """Notify the running worker (if any) to poll for new jobs right away.""" _WAKE.set() # ──────────────────────────────────────────────── # Enqueue — called inside the request transaction # ──────────────────────────────────────────────── def enqueue_email(conn, registration_id, recipient_email, recipient_name, email_type, old_status, new_status, reason="", priority=5, idempotency_key="", bulk_operation_id=0): """Insert a pending job into email_outbox. Must be called INSIDE an existing database transaction (``conn``). Returns the outbox job id or None on duplicate. """ if not recipient_email or not recipient_email.strip(): return None now = datetime.now().isoformat(timespec="seconds") try: cur = conn.execute( """INSERT INTO email_outbox (registration_id, recipient_email, recipient_name, email_type, old_status, new_status, reason, priority, idempotency_key, status, attempt_count, max_attempts, last_error, next_retry_at, process_after, locked_by, locked_at, completed_at, bulk_operation_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?, '', '', ?, '', '', '', ?, ?)""", (registration_id, recipient_email.strip(), recipient_name, email_type, old_status, new_status, reason, priority, idempotency_key, EMAIL_MAX_RETRIES, now, bulk_operation_id, now), ) return cur.lastrowid except sqlite3.IntegrityError: return None def enqueue_bulk_emails(conn, registration_ids, status, reason="", sent_by="system", bulk_operation_id=0): """Enqueue separate email jobs for each registration. Returns (operation_id, jobs_created). """ from nexora.services.registrations import registration_details now = datetime.now().isoformat(timespec="seconds") # Create bulk operation record cur = conn.execute( """INSERT INTO bulk_email_operations (operation_type, status, total_jobs, queued_count, processing_count, sent_count, failed_count, pending_count, created_by, created_at, completed_at) VALUES ('bulk_status', 'processing', 0, 0, 0, 0, 0, 0, ?, ?, '')""", (sent_by, now), ) op_id = cur.lastrowid jobs_created = 0 for reg_id in registration_ids: details = registration_details(reg_id) if not details: continue email = (details.get("email") or "").strip() if not email: continue name = details.get("name") or "Participant" idem_key = f"bulk:{op_id}:reg:{reg_id}:status:{status}" job_id = enqueue_email( conn, reg_id, email, name, "bulk_status_change", "", status, reason, priority=7, idempotency_key=idem_key, bulk_operation_id=op_id, ) if job_id: jobs_created += 1 # Also enqueue for group members entry_type = (details.get("entry_type") or "").lower() if entry_type == "group": from nexora.services.tickets import ticket_member_roster roster = ticket_member_roster(details) for member in roster: if member.get("number", 0) == 1: continue m_email = (member.get("email") or "").strip() if not m_email: continue m_name = member.get("name") or f"Member {member.get('number', '')}" m_idem = f"bulk:{op_id}:reg:{reg_id}:m{member.get('number', '')}:status:{status}" m_job = enqueue_email( conn, reg_id, m_email, m_name, "bulk_status_change", "", status, reason, priority=7, idempotency_key=m_idem, bulk_operation_id=op_id, ) if m_job: jobs_created += 1 # Update counts conn.execute( "UPDATE bulk_email_operations SET total_jobs = ?, queued_count = ?, pending_count = ? WHERE id = ?", (jobs_created, jobs_created, jobs_created, op_id), ) return op_id, jobs_created # ──────────────────────────────────────────────── # Worker # ──────────────────────────────────────────────── class EmailWorker: """Background worker that processes email outbox jobs.""" def __init__(self, concurrency=None): self._concurrency = concurrency or EMAIL_WORKER_CONCURRENCY self._executor = ThreadPoolExecutor( max_workers=self._concurrency, thread_name_prefix="email-worker", ) self._running = False self._stop_event = threading.Event() self._rate_limiter = _RateLimiter(EMAIL_RATE_LIMIT_PER_MINUTE) self._active_count = 0 self._lock = threading.Lock() def start(self): if self._running: return self._reset_stale_processing_jobs() self._running = True self._stop_event.clear() t = threading.Thread(target=self._poll_loop, name="email-queue-poll", daemon=True) t.start() logger.info("[email-queue] Worker started (concurrency=%d)", self._concurrency) def stop(self): self._running = False self._stop_event.set() self._executor.shutdown(wait=False, cancel_futures=True) logger.info("[email-queue] Worker stopped") @property def active_count(self): with self._lock: return self._active_count def _poll_loop(self): while self._running: try: self._claim_and_process_batch() except Exception: logger.exception("[email-queue] Poll loop error") self._sleep_with_wake() def _sleep_with_wake(self): """Sleep up to EMAIL_POLL_INTERVAL but return immediately when a job is enqueued (wake_worker) or when the worker is asked to stop.""" deadline = time.monotonic() + EMAIL_POLL_INTERVAL while self._running and time.monotonic() < deadline: if _WAKE.is_set(): _WAKE.clear() return # 0.2s tick keeps stop() responsive too self._stop_event.wait(timeout=0.2) def _claim_and_process_batch(self): jobs = self._claim_jobs(EMAIL_BATCH_SIZE) if not jobs: return futures = [] for job in jobs: self._rate_limiter.wait() with self._lock: self._active_count += 1 future = self._executor.submit(self._process_job, job) future.add_done_callback(self._on_job_done) futures.append(future) def _reset_stale_processing_jobs(self): """Return timed-out processing jobs to pending after a restart.""" cutoff = ( datetime.now() - timedelta(seconds=EMAIL_JOB_TIMEOUT) ).isoformat(timespec="seconds") try: conn = sqlite3.connect(str(DB_PATH), timeout=10) try: conn.execute( """UPDATE email_outbox SET status = 'pending', locked_by = '', locked_at = '', last_error = 'Recovered after worker restart' WHERE status = 'processing' AND (locked_at = '' OR locked_at < ?)""", (cutoff,), ) conn.commit() finally: conn.close() except Exception: logger.exception("[email-queue] Failed to reset stale processing jobs") def _claim_jobs(self, batch_size): """Atomically claim pending jobs for this worker.""" now = datetime.now().isoformat(timespec="seconds") worker_id = f"worker-{os.getpid()}-{threading.current_thread().name}" try: conn = sqlite3.connect(str(DB_PATH), timeout=10) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=5000") try: rows = conn.execute( """SELECT id FROM email_outbox WHERE status = 'pending' AND (process_after = '' OR process_after <= ?) AND attempt_count < max_attempts ORDER BY priority ASC, created_at ASC LIMIT ?""", (now, batch_size), ).fetchall() if not rows: return [] ids = [r["id"] for r in rows] placeholders = ",".join("?" for _ in ids) conn.execute( f"""UPDATE email_outbox SET status = 'processing', locked_by = ?, locked_at = ?, attempt_count = attempt_count + 1 WHERE id IN ({placeholders})""", [worker_id, now] + ids, ) conn.commit() claimed = conn.execute( f"""SELECT * FROM email_outbox WHERE id IN ({placeholders})""", ids, ).fetchall() return [dict(r) for r in claimed] finally: conn.close() except Exception: logger.exception("[email-queue] Failed to claim jobs") return [] def _process_job(self, job): """Process a single email job: build HTML, generate PDF, send.""" try: from nexora.services.email_service import ( _send_smtp_email, build_status_email_html, status_email_subject, ) from nexora.services.pdf_tickets import build_ticket_pdf from nexora.services.registrations import registration_details from nexora.services.tickets import ticket_member_roster reg_id = job["registration_id"] new_status = job["new_status"] reason = job["reason"] recipient_name = job["recipient_name"] email_type = job["email_type"] details = registration_details(reg_id) if not details: self._complete_job(job["id"], "permanent_failure", "Registration not found.") return settings_row = _query_one( "SELECT * FROM settings WHERE id = 1") settings_dict = dict(settings_row) if settings_row else {} event_title = settings_dict.get( "event_title", "Nexora Esports Championship") s = (new_status or "").lower().replace(" ", "_") html_body = build_status_email_html( details, s, recipient_name=recipient_name, event_title=event_title, reason=reason, ) subject = status_email_subject(s) ticket_pdf_bytes = build_ticket_pdf(details, settings_dict) ticket_number = details.get( "ticket_number", f"NX-{reg_id:04d}") ok, err = _send_smtp_email( job["recipient_email"], subject, html_body, [(f"{ticket_number}.pdf", ticket_pdf_bytes)], ) if ok: self._complete_job(job["id"], "sent", "") self._log_email_delivery( reg_id, job["recipient_email"], subject, s, job["old_status"], email_type, True, "", ticket_pdf_bytes, job["idempotency_key"], "worker", ) else: self._handle_failure(job, err) except Exception as exc: logger.exception("[email-queue] Job %d failed", job["id"]) self._handle_failure(job, str(exc)) def _handle_failure(self, job, error_msg): attempt = job.get("attempt_count", 1) max_attempts = job.get("max_attempts", EMAIL_MAX_RETRIES) if attempt >= max_attempts: self._complete_job( job["id"], "permanent_failure", error_msg) self._log_email_delivery( job["registration_id"], job["recipient_email"], "", job["new_status"], job["old_status"], job["email_type"], False, error_msg, b"", job["idempotency_key"], "worker", ) else: backoff = self._backoff_seconds(attempt) next_retry = ( datetime.now() + timedelta(seconds=backoff) ).isoformat(timespec="seconds") self._update_job_status( job["id"], "pending", error_msg, next_retry) def _complete_job(self, job_id, status, error_msg): now = datetime.now().isoformat(timespec="seconds") try: conn = sqlite3.connect(str(DB_PATH), timeout=10) try: conn.execute( """UPDATE email_outbox SET status = ?, last_error = ?, completed_at = ?, locked_by = '' WHERE id = ?""", (status, error_msg, now, job_id), ) conn.commit() finally: conn.close() except Exception: logger.exception("[email-queue] Failed to complete job %d", job_id) # Update bulk operation counts self._update_bulk_counts(job_id) def _update_job_status(self, job_id, status, error_msg, next_retry): try: conn = sqlite3.connect(str(DB_PATH), timeout=10) try: conn.execute( """UPDATE email_outbox SET status = ?, last_error = ?, next_retry_at = ?, locked_by = '' WHERE id = ?""", (status, error_msg, next_retry, job_id), ) conn.commit() finally: conn.close() except Exception: logger.exception("[email-queue] Failed to update job %d", job_id) def _on_job_done(self, future): with self._lock: self._active_count -= 1 try: future.result() except Exception: logger.exception("[email-queue] Worker thread exception") def _log_email_delivery(self, reg_id, email, subject, new_status, old_status, email_type, ok, error_msg, pdf_bytes, idem_key, sent_by): pdf_attached = "1" if (ok and pdf_bytes) else "0" try: conn = sqlite3.connect(str(DB_PATH), timeout=10) try: conn.execute( """INSERT INTO email_logs (registration_id, recipient_email, subject, status, error_message, sent_by, created_at, email_type, previous_status, new_status, pdf_attached, retry_count, idempotency_key) VALUES (?, ?, ?, ?, ?, ?, datetime('now'), ?, ?, ?, ?, 0, ?)""", (reg_id, email, subject, "sent" if ok else "failed", error_msg, sent_by, email_type, old_status, new_status, pdf_attached, idem_key), ) conn.execute( """UPDATE registrations SET last_email_status = ?, last_email_sent_at = datetime('now') WHERE id = ?""", (new_status, reg_id), ) conn.commit() finally: conn.close() except Exception: logger.exception("[email-queue] Failed to log email for reg %d", reg_id) def _update_bulk_counts(self, job_id): try: conn = sqlite3.connect(str(DB_PATH), timeout=10) try: row = conn.execute( "SELECT bulk_operation_id FROM email_outbox WHERE id = ?", (job_id,), ).fetchone() if not row or not row[0]: return op_id = row[0] counts = conn.execute( """SELECT SUM(CASE WHEN status='pending' THEN 1 ELSE 0 END) AS pending, SUM(CASE WHEN status='processing' THEN 1 ELSE 0 END) AS processing, SUM(CASE WHEN status='sent' THEN 1 ELSE 0 END) AS sent, SUM(CASE WHEN status IN ('failed','permanent_failure') THEN 1 ELSE 0 END) AS failed FROM email_outbox WHERE bulk_operation_id = ?""", (op_id,), ).fetchone() pending = counts[0] or 0 processing = counts[1] or 0 sent = counts[2] or 0 failed = counts[3] or 0 all_done = pending == 0 and processing == 0 now = datetime.now().isoformat(timespec="seconds") conn.execute( """UPDATE bulk_email_operations SET pending_count = ?, processing_count = ?, sent_count = ?, failed_count = ?, status = ?, completed_at = ? WHERE id = ?""", (pending, processing, sent, failed, "completed" if all_done else "processing", now if all_done else "", op_id), ) conn.commit() finally: conn.close() except Exception: pass @staticmethod def _backoff_seconds(attempt): """Exponential backoff: 0, 60, 300, 900, 3600 seconds.""" schedule = [0, 60, 300, 900, 3600] idx = min(attempt - 1, len(schedule) - 1) return max(schedule[idx], 0) class _RateLimiter: """Simple token-bucket rate limiter for email sending.""" def __init__(self, per_minute): self._interval = 60.0 / max(1, per_minute) self._last = 0.0 self._lock = threading.Lock() def wait(self): with self._lock: now = time.monotonic() wait_time = self._last + self._interval - now if wait_time > 0: time.sleep(wait_time) self._last = time.monotonic() # ──────────────────────────────────────────────── # Queue manager (singleton) # ──────────────────────────────────────────────── _worker_instance = None _worker_lock = threading.Lock() def start_workers(): global _worker_instance with _worker_lock: if _worker_instance is None: _worker_instance = EmailWorker() _worker_instance.start() def stop_workers(): global _worker_instance with _worker_lock: if _worker_instance: _worker_instance.stop() _worker_instance = None def process_pending_jobs_once(limit=5): """Process a small batch immediately for shared-hosting reliability. Passenger/cPanel apps can pause background threads between requests. The normal worker still handles the queue, but admin actions and registration submissions call this once after enqueueing so emails are not left waiting for a sleeping worker. """ worker = EmailWorker(concurrency=1) processed = 0 try: for job in worker._claim_jobs(max(1, int(limit or 1))): worker._process_job(job) processed += 1 finally: worker._executor.shutdown(wait=False, cancel_futures=True) return processed def get_queue_status(): """Return current queue status for the admin dashboard.""" try: conn = sqlite3.connect(str(DB_PATH), timeout=5) conn.row_factory = sqlite3.Row try: outbox = conn.execute( """SELECT COUNT(*) AS total, SUM(CASE WHEN status='pending' THEN 1 ELSE 0 END) AS pending, SUM(CASE WHEN status='processing' THEN 1 ELSE 0 END) AS processing, SUM(CASE WHEN status='sent' THEN 1 ELSE 0 END) AS sent, SUM(CASE WHEN status='failed' THEN 1 ELSE 0 END) AS failed_retry, SUM(CASE WHEN status='permanent_failure' THEN 1 ELSE 0 END) AS permanent_failed FROM email_outbox""" ).fetchone() recent_ops = conn.execute( """SELECT * FROM bulk_email_operations ORDER BY id DESC LIMIT 10""" ).fetchall() return { "outbox": dict(outbox) if outbox else {}, "bulk_operations": [dict(r) for r in recent_ops], "worker_active": _worker_instance.active_count if _worker_instance else 0, "worker_running": _worker_instance._running if _worker_instance else False, } finally: conn.close() except Exception: return {"outbox": {}, "bulk_operations": [], "worker_active": 0, "worker_running": False} def get_bulk_operation_status(op_id): """Return status for a specific bulk operation.""" try: conn = sqlite3.connect(str(DB_PATH), timeout=5) conn.row_factory = sqlite3.Row try: op = conn.execute( "SELECT * FROM bulk_email_operations WHERE id = ?", (op_id,), ).fetchone() if not op: return None return dict(op) finally: conn.close() except Exception: return None def retry_failed_jobs(bulk_operation_id=None): """Reset failed jobs to pending for retry.""" try: conn = sqlite3.connect(str(DB_PATH), timeout=10) try: if bulk_operation_id: conn.execute( """UPDATE email_outbox SET status = 'pending', attempt_count = 0, last_error = '', next_retry_at = '', locked_by = '' WHERE bulk_operation_id = ? AND status IN ('failed', 'permanent_failure')""", (bulk_operation_id,), ) else: conn.execute( """UPDATE email_outbox SET status = 'pending', attempt_count = 0, last_error = '', next_retry_at = '', locked_by = '' WHERE status = 'failed'""" ) conn.commit() return True finally: conn.close() except Exception: return False def cleanup_old_outbox(): """Remove completed outbox entries older than retention period.""" from nexora.config import EMAIL_OUTBOX_RETENTION_HOURS try: cutoff = ( datetime.now() - timedelta(hours=EMAIL_OUTBOX_RETENTION_HOURS) ).isoformat(timespec="seconds") conn = sqlite3.connect(str(DB_PATH), timeout=10) try: conn.execute( """DELETE FROM email_outbox WHERE status IN ('sent', 'permanent_failure') AND completed_at != '' AND completed_at < ?""", (cutoff,), ) conn.commit() finally: conn.close() except Exception: pass def _query_one(sql, params=()): """Helper: execute a query and return one row as sqlite3.Row.""" conn = sqlite3.connect(str(DB_PATH), timeout=5) conn.row_factory = sqlite3.Row try: return conn.execute(sql, params).fetchone() finally: conn.close()