"""ARQ job queue integration.""" from __future__ import annotations import logging from typing import Any from arq import create_pool from arq.connections import ArqRedis, RedisSettings from app.config import get_settings logger = logging.getLogger(__name__) # ── Global ARQ pool singleton ──────────────────────────────────────────────── _job_pool: ArqRedis | None = None async def init_job_pool() -> ArqRedis: """Create and store the global ARQ job pool. Called once during app lifespan startup so every subsequent enqueue reuses the same connection instead of opening a new one per call. """ global _job_pool if _job_pool is not None: logger.warning("init_job_pool() called but pool already initialized") return _job_pool settings = get_settings() redis_settings = RedisSettings.from_dsn(settings.redis_url) _job_pool = await create_pool(redis_settings) logger.info("Global ARQ job pool initialized") return _job_pool async def close_job_pool() -> None: """Close the global ARQ job pool. Called during app lifespan shutdown.""" global _job_pool if _job_pool is not None: await _job_pool.close() _job_pool = None logger.info("Global ARQ job pool closed") async def get_job_pool() -> ArqRedis: """Return the global ARQ job pool singleton. If init_job_pool() has not been called yet (e.g. during testing), a new pool is created lazily so callers always get a working connection. """ global _job_pool if _job_pool is None: settings = get_settings() redis_settings = RedisSettings.from_dsn(settings.redis_url) _job_pool = await create_pool(redis_settings) logger.debug("ARQ job pool created lazily (init_job_pool not called)") return _job_pool async def enqueue_job(job_name: str, *args: Any, **kwargs: Any) -> str | None: """Enqueue a background job. Returns job_id or None.""" pool = await get_job_pool() job = await pool.enqueue_job(job_name, *args, **kwargs) if job: return job.job_id return None async def get_job_status(job_id: str) -> dict[str, Any] | None: """Get the status of a background job.""" pool = await get_job_pool() job_info = await pool.job_info(job_id) if job_info is None: return None return { "job_id": job_id, "status": str(job_info.status), "result": job_info.result, "enqueue_time": job_info.enqueue_time.isoformat() if job_info.enqueue_time else None, "start_time": job_info.start_time.isoformat() if job_info.start_time else None, "finish_time": job_info.finish_time.isoformat() if job_info.finish_time else None, } # ── Password Reset Email Job ───────────────────────────────────────────────── async def send_password_reset_email( ctx: dict[str, Any], *, user_id: str, email: str, raw_token: str, expires_at: str, ) -> None: """Send a password reset email via SMTP. This is an ARQ worker function. It is registered with the job registry so the worker can execute it when the auth service enqueues it. """ from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText import aiosmtplib settings = get_settings() # Build the reset URL reset_url = f"{settings.frontend_url.rstrip('/')}/reset-password?token={raw_token}" # Build the email msg = MIMEMultipart("alternative") msg["From"] = settings.smtp_from_email msg["To"] = email msg["Subject"] = "LeoCRM — Passwort zurücksetzen" text_body = ( f"Sie haben angefordert, Ihr Passwort zurückzusetzen.\n\n" f"Klicken Sie auf den folgenden Link, um ein neues Passwort zu setzen:\n" f"{reset_url}\n\n" f"Dieser Link ist gültig bis {expires_at}.\n\n" f"Falls Sie diese Anfrage nicht gestellt haben, können Sie diese\n" f"E-Mail ignorieren. Ihr Passwort bleibt unverändert.\n" ) html_body = ( f"" f"

Passwort zurücksetzen

" f"

Sie haben angefordert, Ihr Passwort zurückzusetzen.

" f"

Passwort jetzt zurücksetzen

" f"

Dieser Link ist gültig bis {expires_at}.

" f"

Falls Sie diese Anfrage nicht gestellt haben, können Sie diese " f"E-Mail ignorieren. Ihr Passwort bleibt unverändert.

" f"" ) msg.attach(MIMEText(text_body, "plain", "utf-8")) msg.attach(MIMEText(html_body, "html", "utf-8")) # Send via SMTP (port 465 = implicit TLS, port 587 = STARTTLS) use_tls = settings.smtp_port == 465 start_tls = settings.smtp_use_tls and not use_tls await aiosmtplib.send( msg, hostname=settings.smtp_host, port=settings.smtp_port, username=settings.smtp_username, password=settings.smtp_password, use_tls=use_tls, start_tls=start_tls, ) logger.info("Password reset email sent to %s for user %s", email, user_id) # Register the job so the worker can find it from app.core.job_registry import register_job # noqa: E402 register_job("send_password_reset_email", send_password_reset_email) # ── DSAR Processing Job (G1 DSGVO: Art. 15 Auskunft / Art. 17 Löschung) ───── async def _dsar_collect_user_data(db: Any, tenant_id: str, user_id: str) -> dict[str, Any]: """Collect every data category the dsgvo-export route promises. Shared by type=access (full export) so both paths stay consistent. """ from datetime import UTC, datetime from uuid import UUID as PyUUID from sqlalchemy import or_ as sa_or_ from sqlalchemy import select as sa_select from app.models.audit import AuditLog from app.models.contact import Contact from app.models.notification import Notification from app.models.user import User from app.plugins.builtins.contracts import get_contract uid = PyUUID(user_id) tid = PyUUID(tenant_id) export_data: dict[str, Any] = { "user_id": user_id, "exported_at": datetime.now(UTC).isoformat(), "legal_basis": "GDPR Art. 15 (access) / Art. 20 (portability)", "data": {}, } # Profile user = ( await db.execute(sa_select(User).where(User.id == uid)) ).scalar_one_or_none() if user: export_data["data"]["profile"] = { "email": user.email, "name": user.name, "is_active": user.is_active, "created_at": user.created_at.isoformat() if user.created_at else None, } # Contacts owned by the user contacts = ( await db.execute( sa_select(Contact).where( Contact.tenant_id == tid, Contact.owner_id == uid, Contact.deleted_at.is_(None), ) ) ).scalars().all() export_data["data"]["contacts"] = [ { "id": str(c.id), "type": c.type, "displayname": c.displayname, "email_1": c.email_1, "email_2": c.email_2, } for c in contacts ] # Audit trail entries by/about the user (bounded to keep payloads sane) audit_entries = ( await db.execute( sa_select(AuditLog).where( AuditLog.tenant_id == tid, AuditLog.user_id == uid, ).limit(1000) ) ).scalars().all() export_data["data"]["audit_log"] = [ { "action": a.action, "entity_type": a.entity_type, "timestamp": a.timestamp.isoformat() if a.timestamp else None, } for a in audit_entries ] # Notifications addressed to the user notifications = ( await db.execute( sa_select(Notification).where( Notification.tenant_id == tid, Notification.owner_id == uid, ).limit(1000) ) ).scalars().all() export_data["data"]["notifications"] = [ { "id": str(n.id), "type": getattr(n, "type", None), "title": getattr(n, "title", None), "created_at": n.created_at.isoformat() if n.created_at else None, } for n in notifications ] # ── Categories promised by the dsgvo-export route docstring ── # (G1-b: mail accounts, tasks, calendar entries, comm messages) try: mail_contract = get_contract("mail") if mail_contract is None: raise ImportError("mail plugin not active") mail_account_m = mail_contract.MailAccount mail_accounts = ( await db.execute( sa_select(mail_account_m).where( mail_account_m.tenant_id == tid, mail_account_m.user_id == uid, ).limit(500) ) ).scalars().all() export_data["data"]["mail_accounts"] = [ { "id": str(a.id), "email_address": a.email_address, "display_name": a.display_name, "is_shared": a.is_shared, "is_active": a.is_active, } for a in mail_accounts ] except ImportError: pass try: tasks_contract = get_contract("tasks") if tasks_contract is None: raise ImportError("tasks plugin not active") task_m = tasks_contract.Task tasks = ( await db.execute( sa_select(task_m).where( sa_or_(task_m.owner_id == uid, task_m.assigned_to == uid), task_m.tenant_id == tid, task_m.deleted_at.is_(None), ).limit(1000) ) ).scalars().all() export_data["data"]["tasks"] = [ { "id": str(t.id), "title": t.title, "status": t.status, "priority": t.priority, "due_date": t.due_date.isoformat() if t.due_date else None, } for t in tasks ] except ImportError: pass try: cal_contract = get_contract("calendar") if cal_contract is None: raise ImportError("calendar plugin not active") cal_entry_m = cal_contract.CalendarEntry cal_entries = ( await db.execute( sa_select(cal_entry_m).where( cal_entry_m.tenant_id == tid, cal_entry_m.owner_id == uid, cal_entry_m.deleted_at.is_(None), ).limit(1000) ) ).scalars().all() export_data["data"]["calendar_entries"] = [ { "id": str(e.id), "title": e.title, "entry_type": e.entry_type, "start_at": e.start_at.isoformat() if e.start_at else None, "end_at": e.end_at.isoformat() if e.end_at else None, } for e in cal_entries ] except ImportError: pass try: komm_contract = get_contract("kommunikation") if komm_contract is None: raise ImportError("kommunikation plugin not active") comm_msg_m = komm_contract.CommMessage comm_messages = ( await db.execute( sa_select(comm_msg_m).where( comm_msg_m.sender_id == uid, comm_msg_m.tenant_id == tid, ).limit(1000) ) ).scalars().all() export_data["data"]["comm_messages"] = [ { "id": str(m.id), "sender_type": m.sender_type, "content": m.content[:500], "created_at": m.created_at.isoformat() if m.created_at else None, } for m in comm_messages ] except ImportError: pass return export_data async def _dsar_execute_deletion(db: Any, tenant_id: str, user_id: str) -> dict[str, int]: """Execute GDPR Art. 17 erasure for a user within one tenant. Strategy (respects retention duties): - Contacts owned by the user → soft-delete via deleted_at (audit history must remain intact — it is not personal data of the subject but business record; retention policy governs its cleanup) - Notifications owned by the user → hard delete - User account → deactivate (is_active=False), clear personal fields, scramble password hash and email (keeps FK integrity for audit rows) Returns counters for the audit entry. """ from datetime import UTC, datetime from uuid import UUID as PyUUID from sqlalchemy import select as sa_select from sqlalchemy import update as sa_update from app.core.audit import log_audit from app.models.contact import Contact from app.models.notification import Notification from app.models.user import User uid = PyUUID(user_id) tid = PyUUID(tenant_id) counts: dict[str, int] = {} # 1. Soft-delete contacts owned by the user contact_result = await db.execute( sa_select(Contact).where( Contact.tenant_id == tid, Contact.owner_id == uid, Contact.deleted_at.is_(None), ) ) contacts = contact_result.scalars().all() for c in contacts: c.deleted_at = datetime.now(UTC) counts["contacts_soft_deleted"] = len(contacts) # 2. Hard-delete notifications owned by the user notif_result = await db.execute( sa_select(Notification).where( Notification.tenant_id == tid, Notification.owner_id == uid, ) ) notifications = notif_result.scalars().all() for n in notifications: await db.delete(n) counts["notifications_deleted"] = len(notifications) # 3. Anonymize + deactivate the account (FK integrity for audit rows kept) await db.execute( sa_update(User) .where(User.id == uid) .values( email=f"erased.{uid.hex[:16]}@anonymized.invalid", name="[gelöscht gemäß DSGVO Art. 17]", first_name=None, last_name=None, avatar_url=None, password_hash="!dsar-erased", is_active=False, preferences={}, ) ) counts["user_anonymized"] = 1 # 4. Audit the erasure itself (who/what/when — required by Art. 17 recital) await log_audit( db, tid, user_id, "dsar_erasure", "user", uid, {"target_user": user_id, **counts}, ) return counts async def process_dsar( ctx: dict[str, Any], *, user_id: str, tenant_id: str, request_type: str, ) -> dict[str, Any]: """Process a GDPR Data Subject Access Request (DSAR). ARQ worker function registered as "process_dsar". request_type: - "access": collect all data categories (Art. 15/20) and post a system message that the export is ready (served via the existing dsgvo-export endpoint). - "deletion": execute Art. 17 erasure (soft-delete contacts, hard-delete notifications, anonymize+deactivate account) and audit it. - "rectification": post a system message asking admins to handle the correction manually. Returns a summary dict for the job result. """ import logging import uuid as uuid_module from app.core.db import get_worker_session_factory from app.core.notifications import post_system_message logger = logging.getLogger(__name__) tid = uuid_module.UUID(tenant_id) uid = uuid_module.UUID(user_id) factory = get_worker_session_factory() async with factory() as db: try: if request_type == "access": data = await _dsar_collect_user_data(db, tenant_id, user_id) await db.commit() categories = list(data.get("data", {}).keys()) await post_system_message( db, tid, uid, "dsar_access_ready", "DSGVO-Auskunft bereit", f"Datenkategorien: {', '.join(categories)}", severity="info", ) await db.commit() logger.info("DSAR access processed for user %s", user_id) return {"type": request_type, "status": "completed", "categories": categories} if request_type == "deletion": counts = await _dsar_execute_deletion(db, tenant_id, user_id) await db.commit() await post_system_message( db, tid, uid, "dsar_deletion_done", "DSGVO-Löschung ausgeführt", f"Kontakten soft-gelöscht: {counts.get('contacts_soft_deleted', 0)}; Konto anonymisiert.", severity="info", ) await db.commit() logger.info("DSAR deletion executed for user %s: %s", user_id, counts) return {"type": request_type, "status": "completed", **counts} if request_type == "rectification": await post_system_message( db, tid, uid, "dsar_rectification_requested", "DSGVO-Berichtigung angefordert", f"Manuelle Bearbeitung für User {user_id} erforderlich.", severity="warning", ) await db.commit() logger.info("DSAR rectification requested for user %s", user_id) return {"type": request_type, "status": "queued_for_manual_handling"} logger.warning("Unknown DSAR request_type '%s' for user %s", request_type, user_id) return {"type": request_type, "status": "unknown_type"} except Exception: await db.rollback() raise register_job("process_dsar", process_dsar)