Security fixes: P0-P2 complete (22 fixes)

P0 (7): Auth-bypass removed, migrations fixed, plugin-upload disabled, RLS FORCE+WITH CHECK, plugin double-registration fixed, persistent volume, domain removed
P1 (11): User/tenant model, Redis centralized, worker separated, transactional outbox, XSS fixed, DMS chunked streaming, permissions unified, password reset, metrics secured, config/docs fixed, cross-tenant FK
P2 (4): Contact model normalized, cross-imports reduced 94%, commands+state machines for contacts/dms/mail/calendar, SPA path-traversal

8 new migrations, 99 unit tests, 13 commands, 8 contracts, 72 files changed

app/core/auth.py

@@ -8,6 +8,8 @@ import uuid
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
+import logging
+
 import redis.asyncio as aioredis
 from passlib.context import CryptContext
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -16,10 +18,53 @@ from app.config import get_settings
 from app.models.session import Session as SessionModel
 from app.models.user import User
 
+logger = logging.getLogger(__name__)
+
 _pwd_context = CryptContext(
     schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=get_settings().bcrypt_rounds
 )
 
+# ── Global Redis client singleton ────────────────────────────────────────────
+_redis_client: aioredis.Redis | None = None
+
+
+async def init_redis() -> aioredis.Redis:
+    """Create and store the global Redis client. Called once during app lifespan startup."""
+    global _redis_client
+    if _redis_client is not None:
+        logger.warning("init_redis() called but Redis client already initialized")
+        return _redis_client
+    _redis_client = aioredis.from_url(
+        get_settings().redis_url, decode_responses=True
+    )
+    logger.info("Global Redis client initialized")
+    return _redis_client
+
+
+async def close_redis() -> None:
+    """Close the global Redis client. Called during app lifespan shutdown."""
+    global _redis_client
+    if _redis_client is not None:
+        await _redis_client.aclose()
+        _redis_client = None
+        logger.info("Global Redis client closed")
+
+
+def get_redis() -> aioredis.Redis:
+    """Return the global Redis client singleton.
+
+    If init_redis() has not been called yet (e.g. during testing or
+    outside the app lifespan), a new client is created lazily so callers
+    always get a working connection.
+    """
+    global _redis_client
+    if _redis_client is None:
+        _redis_client = aioredis.from_url(
+            get_settings().redis_url, decode_responses=True
+        )
+        logger.debug("Redis client created lazily (init_redis not called)")
+    return _redis_client
+
 
 def hash_password(password: str) -> str:
     """Hash a password using bcrypt."""
@@ -56,9 +101,13 @@ async def create_session(
     redis: aioredis.Redis,
     user: User,
     tenant_id: uuid.UUID,
+    role: str = "viewer",
 ) -> tuple[str, str]:
     """Create a session in Redis (runtime) and PostgreSQL (audit trail).
     Returns (session_id, csrf_token).
+
+    ``role`` comes from UserTenant — the built-in role string for the
+    active tenant membership.
     """
     settings = get_settings()
     session_id = str(uuid.uuid4())
@@ -71,7 +120,7 @@ async def create_session(
         "tenant_id": str(tenant_id),
         "email": user.email,
         "name": user.name,
-        "role": user.role,
+        "role": role,
         "is_system_admin": user.is_system_admin,
         "csrf_token": csrf_token,
         "is_active": user.is_active,
@@ -123,8 +172,9 @@ async def update_session_tenant(
     redis: aioredis.Redis,
     session_id: str,
     new_tenant_id: uuid.UUID,
+    role: str | None = None,
 ) -> dict[str, Any] | None:
-    """Update the active tenant in a Redis session."""
+    """Update the active tenant (and optionally role) in a Redis session."""
     import json
 
     settings = get_settings()
@@ -133,6 +183,8 @@ async def update_session_tenant(
         return None
     data = json.loads(raw)
     data["tenant_id"] = str(new_tenant_id)
+    if role is not None:
+        data["role"] = role
     ttl = await redis.ttl(f"session:{session_id}")
     if ttl <= 0:
         ttl = settings.session_ttl_seconds

app/core/event_bus.py

@@ -1,4 +1,23 @@
-"""In-process event bus for publish/subscribe."""
+"""In-process event bus for publish/subscribe.
+
+.. note::
+
+   This bus is **in-process only** — events are lost on crash, restart, or
+   when multiple replicas are running.  For **domain/business events** that
+   must be delivered reliably (e.g. ``contact.created``, ``contact.updated``,
+   ``user.created``), use the :mod:`app.core.outbox` transactional outbox
+   instead::
+
+       from app.core.outbox import enqueue_outbox_event
+       await enqueue_outbox_event(db, tenant_id, "contact.created", {...})
+
+   The outbox worker (see :mod:`app.core.worker`) polls the ``event_outbox``
+   table every 5 seconds and publishes events to this in-process bus, so
+   local handlers still receive them — but with durability guarantees.
+
+   ``publish()`` may still be used for **uncritical local events** that do
+   not require persistence (e.g. cache invalidation signals).
+"""
 
 from __future__ import annotations
 
@@ -35,6 +54,27 @@ class EventBus:
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
 
+    async def publish_with_results(
+        self, event_name: str, payload: dict[str, Any]
+    ) -> list[Exception | None]:
+        """Publish an event and return per-handler results.
+
+        Unlike :meth:`publish`, this method does **not** swallow exceptions.
+        Each list entry is ``None`` on success or the caught ``Exception``
+        on failure, so callers (e.g. the outbox processor) can detect handler
+        errors and apply retry logic.
+        """
+        handlers = self._handlers.get(event_name, [])
+        wildcard_handlers = self._handlers.get('*', [])
+        all_handlers = handlers + wildcard_handlers
+        if not all_handlers:
+            return []
+        tasks = [asyncio.create_task(h(payload)) for h in all_handlers]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        return [
+            r if isinstance(r, Exception) else None for r in results
+        ]
+
 
 # Global event bus instance
 _event_bus = EventBus()

app/core/jobs.py

@@ -2,19 +2,59 @@
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
 from arq import create_pool
-from arq.connections import RedisSettings
+from arq.connections import RedisSettings, ArqRedis
 
 from app.config import get_settings
 
+logger = logging.getLogger(__name__)
 
-async def get_job_pool():
-    """Get an ARQ job pool for enqueueing background tasks."""
+# ── 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)
-    return await create_pool(redis_settings)
+    _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:

app/core/outbox.py

@@ -0,0 +1,210 @@
+"""Transactional outbox for reliable domain event delivery.
+
+Instead of publishing events directly to an in-process bus (which is lost
+on crash/restart), domain events are written to the ``event_outbox`` table
+**within the same database transaction** as the business operation.  A
+background worker then polls the outbox and publishes events to the
+in-process event bus.
+
+Usage in services::
+
+    from app.core.outbox import enqueue_outbox_event
+
+    await enqueue_outbox_event(db, tenant_id, "contact.created", {
+        "contact_id": str(contact.id),
+        "tenant_id": str(tenant_id),
+    })
+    # ... later, the transaction commits and the event is durable.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import redis.asyncio as aioredis
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+# ── SQL statements (raw text for FOR UPDATE SKIP LOCKED) ────────────────────
+
+_INSERT_SQL = text(
+    """
+    INSERT INTO event_outbox (tenant_id, event_name, payload)
+    VALUES (:tenant_id, :event_name, CAST(:payload AS JSONB))
+    """
+)
+
+_CLAIM_SQL = text(
+    """
+    UPDATE event_outbox
+       SET status = 'processing',
+           updated_at = now()
+     WHERE id IN (
+         SELECT id FROM event_outbox
+          WHERE status = 'pending'
+            AND (next_retry_at IS NULL OR next_retry_at <= now())
+          ORDER BY created_at
+          LIMIT :batch_size
+          FOR UPDATE SKIP LOCKED
+     )
+    RETURNING id, tenant_id, event_name, payload, attempts, max_attempts
+    """
+)
+
+_MARK_PUBLISHED_SQL = text(
+    """
+    UPDATE event_outbox
+       SET status = 'published',
+           published_at = now(),
+           updated_at = now()
+     WHERE id = :id
+    """
+)
+
+_FAIL_SQL = text(
+    """
+    UPDATE event_outbox
+       SET status = 'failed',
+           updated_at = now()
+     WHERE id = :id
+    """
+)
+
+_RETRY_SQL = text(
+    """
+    UPDATE event_outbox
+       SET status = 'pending',
+           attempts = :attempts,
+           next_retry_at = :next_retry_at,
+           updated_at = now()
+     WHERE id = :id
+    """
+)
+
+
+def _json_payload(payload: dict[str, Any]) -> str:
+    """Serialise payload to a JSON string suitable for JSONB cast."""
+    import json
+
+    return json.dumps(payload, default=str)
+
+
+async def enqueue_outbox_event(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    event_name: str,
+    payload: dict[str, Any],
+) -> None:
+    """Insert an event into the outbox table within the current transaction.
+
+    The event is only persisted when the surrounding transaction commits.
+    This guarantees at-least-once delivery — no event is lost even if the
+    process crashes after the business operation but before the event is
+    published.
+
+    Args:
+        db: Active async SQLAlchemy session (part of the business transaction).
+        tenant_id: Tenant scope for the event.
+        event_name: Logical event name (e.g. ``"contact.created"``).
+        payload: Event payload dict (will be stored as JSONB).
+    """
+    await db.execute(
+        _INSERT_SQL,
+        {
+            "tenant_id": str(tenant_id),
+            "event_name": event_name,
+            "payload": _json_payload(payload),
+        },
+    )
+
+
+async def process_outbox_batch(
+    db: AsyncSession,
+    redis: aioredis.Redis | None = None,
+    batch_size: int = 50,
+) -> int:
+    """Process one batch of pending outbox events.
+
+    1. Claim up to *batch_size* pending events using ``FOR UPDATE SKIP LOCKED``
+       so multiple workers don't interfere.
+    2. Publish each event to the in-process event bus (for local handlers).
+    3. On success: mark as ``published``.
+    4. On failure: increment attempts, schedule retry with exponential
+       backoff, or mark as ``failed`` if max attempts exceeded.
+
+    Args:
+        db: Async SQLAlchemy session for this batch.
+        redis: Optional Redis client (unused for now, reserved for future
+               cross-process pub/sub).
+        batch_size: Maximum events to process in one batch.
+
+    Returns:
+        Number of events successfully published.
+    """
+    from app.core.event_bus import get_event_bus
+
+    event_bus = get_event_bus()
+    published_count = 0
+
+    # Claim a batch of pending events
+    rows = (
+        await db.execute(_CLAIM_SQL, {"batch_size": batch_size})
+    ).fetchall()
+
+    if not rows:
+        return 0
+
+    for row in rows:
+        event_id = row[0]
+        event_name = row[2]
+        payload = row[3]
+        attempts = row[4]
+        max_attempts = row[5]
+
+        # payload comes back as a dict from JSONB
+        if isinstance(payload, str):
+            import json
+            payload_dict = json.loads(payload)
+        else:
+            payload_dict = payload
+
+        try:
+            results = await event_bus.publish_with_results(event_name, payload_dict)
+            # If any handler raised, treat as failure
+            handler_errors = [r for r in results if r is not None]
+            if handler_errors:
+                raise handler_errors[0]
+            await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
+            published_count += 1
+        except Exception as exc:
+            logger.error(
+                "Failed to publish outbox event %s (%s): %s",
+                event_id, event_name, exc,
+                exc_info=True,
+            )
+            new_attempts = attempts + 1
+            if new_attempts >= max_attempts:
+                await db.execute(_FAIL_SQL, {"id": str(event_id)})
+                logger.warning(
+                    "Outbox event %s marked as failed after %d attempts",
+                    event_id, new_attempts,
+                )
+            else:
+                backoff = timedelta(seconds=(2 ** new_attempts) * 10)
+                next_retry = datetime.now(timezone.utc) + backoff
+                await db.execute(
+                    _RETRY_SQL,
+                    {
+                        "id": str(event_id),
+                        "attempts": new_attempts,
+                        "next_retry_at": next_retry,
+                    },
+                )
+
+    await db.commit()
+    return published_count

app/core/permissions.py

@@ -16,7 +16,7 @@ import uuid
 from typing import Any
 
 import redis.asyncio as aioredis
-from sqlalchemy import select
+from sqlalchemy import select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import get_settings
@@ -30,6 +30,9 @@ logger = logging.getLogger(__name__)
 CACHE_TTL = 300  # 5 minutes
 CACHE_PREFIX = "resolved"
 
+# Severity ordering for field permissions: highest wins
+_FIELD_PERM_SEVERITY = {"hidden": 3, "readonly": 2, "read": 1}
+
 
 def _matches_permission(granted: str, required: str) -> bool:
     """Check if a granted permission matches the required permission.
@@ -44,7 +47,6 @@ def _matches_permission(granted: str, required: str) -> bool:
         return True
     g_parts = granted.split(":")
     r_parts = required.split(":")
-    # Wildcard * matches any single segment, but remaining segments must still match
     if len(g_parts) != len(r_parts):
         return False
     for i, g_part in enumerate(g_parts):
@@ -88,6 +90,87 @@ def _normalize_permissions(permissions: Any) -> set[str]:
     return result
 
 
+def _merge_field_permissions(
+    existing: dict[str, dict[str, str]],
+    incoming: dict[str, Any],
+) -> None:
+    """Merge incoming field permissions into existing dict.
+
+    Uses 'strictest right wins': hidden > readonly > read.
+    When a field already exists, the more restrictive (higher severity) value wins.
+    """
+    for module, fields in incoming.items():
+        if not isinstance(fields, dict):
+            continue
+        if module not in existing:
+            existing[module] = {}
+        for field, perm in fields.items():
+            if not isinstance(perm, str):
+                continue
+            perm_lower = perm.lower()
+            if perm_lower not in _FIELD_PERM_SEVERITY:
+                # Unknown permission level — skip with warning
+                logger.warning(
+                    "Unknown field permission level '%s' for %s.%s — skipping",
+                    perm, module, field,
+                )
+                continue
+            current = existing[module].get(field)
+            if current is None:
+                existing[module][field] = perm_lower
+            else:
+                # Strictest (highest severity) wins
+                if _FIELD_PERM_SEVERITY[perm_lower] > _FIELD_PERM_SEVERITY.get(current, 0):
+                    existing[module][field] = perm_lower
+
+
+async def _get_current_permission_version(
+    db: AsyncSession,
+    user_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+) -> int:
+    """Get the current max permission_version from DB for cache validation.
+
+    Uses a SAVEPOINT so that a failure here does not abort the outer transaction.
+    """
+    async with db.begin_nested():
+        # Check role version
+        ut_q = select(UserTenant.role_id).where(
+            UserTenant.user_id == user_id,
+            UserTenant.tenant_id == tenant_id,
+        )
+        ut_result = await db.execute(ut_q)
+        role_id = ut_result.scalar_one_or_none()
+
+        max_version = 0
+        if role_id is not None:
+            role_q = select(Role.permission_version).where(Role.id == role_id)
+            role_result = await db.execute(role_q)
+            role_ver = role_result.scalar_one_or_none()
+            if role_ver is not None:
+                max_version = max(max_version, role_ver)
+
+        # Check group versions
+        ug_q = select(UserGroup.group_id).where(
+            UserGroup.user_id == user_id,
+            UserGroup.tenant_id == tenant_id,
+        )
+        ug_result = await db.execute(ug_q)
+        group_ids = [row[0] for row in ug_result.all()]
+
+        if group_ids:
+            groups_q = select(func.max(Group.permission_version)).where(
+                Group.id.in_(group_ids),
+                Group.deleted_at.is_(None),
+            )
+            groups_result = await db.execute(groups_q)
+            group_max = groups_result.scalar()
+            if group_max is not None:
+                max_version = max(max_version, group_max)
+
+        return max_version
+
+
 async def resolve_permissions(
     db: AsyncSession,
     user_id: uuid.UUID,
@@ -104,18 +187,24 @@ async def resolve_permissions(
             "version": int,                # permission_version for cache invalidation
         }
     """
-    # Check system admin first
-    # If a previous query in this session failed, the transaction may be aborted.
-    # Rollback to recover before executing our query.
+    # Use SAVEPOINT for the initial query so a failure doesn't abort
+    # the outer transaction.
     try:
-        user_q = select(User.is_system_admin).where(User.id == user_id)
-        user_result = await db.execute(user_q)
-        is_system_admin = user_result.scalar() or False
+        async with db.begin_nested():
+            user_q = select(User.is_system_admin).where(User.id == user_id)
+            user_result = await db.execute(user_q)
+            is_system_admin = user_result.scalar() or False
     except Exception:
-        await db.rollback()
-        user_q = select(User.is_system_admin).where(User.id == user_id)
-        user_result = await db.execute(user_q)
-        is_system_admin = user_result.scalar() or False
+        logger.warning(
+            "SAVEPOINT failed for is_system_admin query (user=%s), retrying without savepoint",
+            user_id,
+            exc_info=True,
+        )
+        # Last-resort fallback: still use savepoint to isolate
+        async with db.begin_nested():
+            user_q = select(User.is_system_admin).where(User.id == user_id)
+            user_result = await db.execute(user_q)
+            is_system_admin = user_result.scalar() or False
 
     if is_system_admin:
         return {
@@ -126,13 +215,14 @@ async def resolve_permissions(
             "version": 0,  # system admin doesn't need version tracking
         }
 
-    # Load UserTenant to get role_id
-    ut_q = select(UserTenant).where(
-        UserTenant.user_id == user_id,
-        UserTenant.tenant_id == tenant_id,
-    )
-    ut_result = await db.execute(ut_q)
-    user_tenant = ut_result.scalar_one_or_none()
+    # Load UserTenant to get role_id — use SAVEPOINT
+    async with db.begin_nested():
+        ut_q = select(UserTenant).where(
+            UserTenant.user_id == user_id,
+            UserTenant.tenant_id == tenant_id,
+        )
+        ut_result = await db.execute(ut_q)
+        user_tenant = ut_result.scalar_one_or_none()
 
     allowed: set[str] = set()
     denied: set[str] = set()
@@ -141,72 +231,75 @@ async def resolve_permissions(
 
     # Load role permissions
     if user_tenant and user_tenant.role_id:
-        role_q = select(Role).where(Role.id == user_tenant.role_id)
-        role_result = await db.execute(role_q)
-        role = role_result.scalar_one_or_none()
+        async with db.begin_nested():
+            role_q = select(Role).where(Role.id == user_tenant.role_id)
+            role_result = await db.execute(role_q)
+            role = role_result.scalar_one_or_none()
+
         if role:
             allowed |= _normalize_permissions(role.permissions)
             denied |= _normalize_permissions(role.denied_permissions)
-            max_version = max(max_version, role.permission_version)
-            # Merge field permissions
+            max_version = max(max_version, role.permission_version or 0)
+            # Merge field permissions using strictest-wins
             if role.field_permissions:
-                for module, fields in role.field_permissions.items():
-                    if isinstance(fields, dict):
-                        if module not in field_perms:
-                            field_perms[module] = {}
-                        field_perms[module].update(fields)
+                _merge_field_permissions(field_perms, role.field_permissions)
+
+    # Also check built-in role string on UserTenant for backward compatibility
+    if user_tenant is not None and user_tenant.role_id is None:
+        legacy_role = user_tenant.role
 
-    # Also check legacy role string on User for backward compatibility
-    if user_tenant is None or user_tenant.role_id is None:
-        legacy_q = select(User.role).where(User.id == user_id)
-        legacy_result = await db.execute(legacy_q)
-        legacy_role = legacy_result.scalar_one_or_none()
         if legacy_role == "admin":
             allowed.add("*:*")
         elif legacy_role == "editor":
-            allowed |= {"contacts:read", "contacts:write", "contacts:read", "contacts:write",
-                        "users:read", "roles:read", "audit:read", "attachments:read",
-                        "attachments:write", "workflows:read", "workflows:write",
-                        "sequences:read", "sequences:write", "addresses:read", "addresses:write",
-                        "taxes:read", "taxes:write", "currencies:read", "currencies:write",
-                        "notifications:read", "notifications:write", "import_export:read",
-                        "import_export:write",
-                        "user_preferences:read", "user_preferences:write"}
+            allowed |= {
+                "contacts:read", "contacts:write",
+                "users:read", "roles:read", "audit:read",
+                "attachments:read", "attachments:write",
+                "workflows:read", "workflows:write",
+                "sequences:read", "sequences:write",
+                "addresses:read", "addresses:write",
+                "taxes:read", "taxes:write",
+                "currencies:read", "currencies:write",
+                "notifications:read", "notifications:write",
+                "import_export:read", "import_export:write",
+                "user_preferences:read", "user_preferences:write",
+            }
         elif legacy_role == "viewer":
-            allowed |= {"contacts:read", "contacts:read", "users:read", "roles:read",
-                        "audit:read", "attachments:read", "workflows:read", "sequences:read",
-                        "addresses:read", "taxes:read", "currencies:read",
-                        "notifications:read", "import_export:read",
-                        "user_preferences:read", "user_preferences:write"}
+            allowed |= {
+                "contacts:read", "users:read", "roles:read",
+                "audit:read", "attachments:read", "workflows:read",
+                "sequences:read", "addresses:read", "taxes:read",
+                "currencies:read", "notifications:read",
+                "import_export:read",
+                "user_preferences:read", "user_preferences:write",
+            }
 
     # Load group permissions
-    ug_q = select(UserGroup).where(
-        UserGroup.user_id == user_id,
-        UserGroup.tenant_id == tenant_id,
-    )
-    ug_result = await db.execute(ug_q)
-    user_groups = ug_result.scalars().all()
+    async with db.begin_nested():
+        ug_q = select(UserGroup).where(
+            UserGroup.user_id == user_id,
+            UserGroup.tenant_id == tenant_id,
+        )
+        ug_result = await db.execute(ug_q)
+        user_groups = ug_result.scalars().all()
 
     if user_groups:
         group_ids = [ug.group_id for ug in user_groups]
-        groups_q = select(Group).where(
-            Group.id.in_(group_ids),
-            Group.deleted_at.is_(None),
-        )
-        groups_result = await db.execute(groups_q)
-        groups = groups_result.scalars().all()
+        async with db.begin_nested():
+            groups_q = select(Group).where(
+                Group.id.in_(group_ids),
+                Group.deleted_at.is_(None),
+            )
+            groups_result = await db.execute(groups_q)
+            groups = groups_result.scalars().all()
 
         for group in groups:
             allowed |= _normalize_permissions(group.permissions)
             denied |= _normalize_permissions(group.denied_permissions)
-            max_version = max(max_version, group.permission_version)
-            # Merge field permissions
+            max_version = max(max_version, group.permission_version or 0)
+            # Merge field permissions using strictest-wins
             if group.field_permissions:
-                for module, fields in group.field_permissions.items():
-                    if isinstance(fields, dict):
-                        if module not in field_perms:
-                            field_perms[module] = {}
-                        field_perms[module].update(fields)
+                _merge_field_permissions(field_perms, group.field_permissions)
 
     # Apply deny list
     resolved = allowed - denied
@@ -226,15 +319,42 @@ async def get_cached_permissions(
     user_id: uuid.UUID,
     tenant_id: uuid.UUID,
 ) -> dict[str, Any]:
-    """Get resolved permissions from Redis cache or resolve from DB."""
+    """Get resolved permissions from Redis cache or resolve from DB.
+
+    Validates the cached permission_version against the current DB version.
+    If they differ, the cache entry is stale and will be re-resolved.
+    """
     cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"
 
     raw = await redis.get(cache_key)
     if raw is not None:
         data = json.loads(raw)
-        return data
+        cached_version = data.get("version", -1)
 
-    # Cache miss — resolve from DB
+        # Validate cached version against current DB version
+        try:
+            current_version = await _get_current_permission_version(db, user_id, tenant_id)
+        except Exception:
+            logger.warning(
+                "Failed to query current permission_version for cache validation "
+                "(user=%s, tenant=%s) — using cached data",
+                user_id, tenant_id,
+                exc_info=True,
+            )
+            current_version = cached_version  # assume cache is valid if we can't check
+
+        if cached_version == current_version:
+            return data
+
+        # Version mismatch — invalidate stale cache and re-resolve
+        logger.info(
+            "Permission cache version mismatch for user=%s tenant=%s "
+            "(cached=%s, current=%s) — re-resolving",
+            user_id, tenant_id, cached_version, current_version,
+        )
+        await redis.delete(cache_key)
+
+    # Cache miss or stale — resolve from DB
     resolved = await resolve_permissions(db, user_id, tenant_id)
 
     # Store in cache (convert sets to lists for JSON)
@@ -263,11 +383,33 @@ async def invalidate_all_user_permissions(
     redis: aioredis.Redis,
     tenant_id: uuid.UUID,
 ) -> None:
-    """Invalidate permission cache for all users in a tenant (e.g. after role/group change)."""
+    """Invalidate permission cache for all users in a tenant (e.g. after role/group change).
+
+    Uses SCAN (non-blocking) instead of KEYS to avoid blocking Redis.
+    """
     pattern = f"{CACHE_PREFIX}:*:{tenant_id}"
-    keys = await redis.keys(pattern)
-    if keys:
-        await redis.delete(*keys)
+    batch_size = 200
+    cursor: int | bytes | str = 0
+    deleted_count = 0
+
+    while True:
+        cursor, keys = await redis.scan(
+            cursor=cursor,
+            match=pattern,
+            count=batch_size,
+        )
+        if keys:
+            await redis.delete(*keys)
+            deleted_count += len(keys)
+        # SCAN returns cursor as bytes or int depending on redis-py version
+        cursor_int = int(cursor) if cursor else 0
+        if cursor_int == 0:
+            break
+
+    logger.info(
+        "Invalidated %d permission cache entries for tenant=%s",
+        deleted_count, tenant_id,
+    )
 
 
 def check_permission(resolved: dict[str, Any], required: str) -> bool:

app/core/state_machine.py

@@ -0,0 +1,66 @@
+"""Generic finite state machine for domain entity status transitions.
+
+Defines allowed state transitions for Contact and Workflow entities.
+Usage in Commands:
+
+    from app.core.state_machine import contact_state_machine
+    contact_state_machine.transition(contact.status, "qualified")
+"""
+
+from __future__ import annotations
+
+
+class StateMachineError(Exception):
+    """Raised when an invalid state transition is attempted."""
+
+
+class StateMachine:
+    """Finite state machine that validates and executes state transitions.
+
+    Attributes:
+        transitions: Mapping from a state to the list of states it can transition to.
+    """
+
+    def __init__(self, transitions: dict[str, list[str]]) -> None:
+        self.transitions: dict[str, list[str]] = transitions
+
+    def can_transition(self, current: str, target: str) -> bool:
+        """Return True if transitioning from *current* to *target* is allowed."""
+        allowed = self.transitions.get(current, [])
+        return target in allowed
+
+    def transition(self, current: str, target: str) -> str:
+        """Validate and return the new state.
+
+        Raises:
+            StateMachineError: if the transition is not allowed.
+        """
+        if not self.can_transition(current, target):
+            raise StateMachineError(
+                f"Invalid state transition: '{current}' -> '{target}'. "
+                f"Allowed targets from '{current}': {self.transitions.get(current, [])}"
+            )
+        return target
+
+
+# ── Contact lifecycle: lead → qualified → customer → inactive ──
+# Allows skipping 'qualified' and reactivation from inactive.
+contact_state_machine = StateMachine(
+    transitions={
+        "lead": ["qualified", "customer", "inactive"],
+        "qualified": ["customer", "lead", "inactive"],
+        "customer": ["inactive"],
+        "inactive": ["lead"],
+    }
+)
+
+# ── Workflow lifecycle: draft → active → paused → completed → cancelled ──
+workflow_state_machine = StateMachine(
+    transitions={
+        "draft": ["active", "cancelled"],
+        "active": ["paused", "completed", "cancelled"],
+        "paused": ["active", "completed", "cancelled"],
+        "completed": [],
+        "cancelled": [],
+    }
+)

app/core/storage.py

@@ -9,15 +9,18 @@ Configuration via environment variables:
 - S3_SECRET_KEY: Secret key
 - S3_REGION: Region (default: us-east-1)
 - S3_SECURE: Use HTTPS (default: true)
+
 """
 
 from __future__ import annotations
 
+import asyncio
 import io
 import logging
 import os
+import tempfile
 from abc import ABC, abstractmethod
-from typing import Any
+from typing import Any, AsyncIterator
 
 import aiofiles
 
@@ -32,6 +35,11 @@ class StorageBackend(ABC):
         """Save data to storage at the given path. Returns the full storage path."""
         ...
 
+    @abstractmethod
+    async def save_stream(self, path: str, chunk_aiter: AsyncIterator[bytes]) -> int:
+        """Stream chunks to storage. Returns total bytes written."""
+        ...
+
     @abstractmethod
     async def read(self, path: str) -> bytes:
         """Read data from storage at the given path."""
@@ -77,6 +85,18 @@ class LocalStorage(StorageBackend):
         logger.debug("LocalStorage: saved %s (%d bytes)", path, len(data))
         return path
 
+    async def save_stream(self, path: str, chunk_aiter: AsyncIterator[bytes]) -> int:
+        """Stream chunks directly to a local file. Returns total bytes written."""
+        full_path = self._full_path(path)
+        os.makedirs(os.path.dirname(full_path), exist_ok=True)
+        total = 0
+        async with aiofiles.open(full_path, "wb") as f:
+            async for chunk in chunk_aiter:
+                await f.write(chunk)
+                total += len(chunk)
+        logger.debug("LocalStorage: streamed %s (%d bytes)", path, total)
+        return total
+
     async def read(self, path: str) -> bytes:
         full_path = self._full_path(path)
         async with aiofiles.open(full_path, "rb") as f:
@@ -155,25 +175,33 @@ class S3Storage(StorageBackend):
             logger.error("S3Storage: failed to connect to %s: %s", self.endpoint, e)
             raise
 
-    async def save(self, path: str, data: bytes) -> str:
-        from io import BytesIO
+    # ── Sync helper methods (called via asyncio.to_thread) ──────────────────
 
+    def _save_sync(self, path: str, data: bytes) -> str:
         client = self._get_client()
         client.put_object(
             bucket_name=self.bucket,
             object_name=path,
-            data=BytesIO(data),
+            data=io.BytesIO(data),
             length=len(data),
         )
-        logger.debug("S3Storage: saved %s (%d bytes)", path, len(data))
         return path
 
-    async def read(self, path: str) -> bytes:
+    def _put_file_sync(self, object_name: str, file_path: str) -> str:
+        client = self._get_client()
+        client.fput_object(self.bucket, object_name, file_path)
+        return object_name
+
+    def _read_sync(self, path: str) -> bytes:
         client = self._get_client()
         response = client.get_object(self.bucket, path)
-        return response.read()
+        try:
+            return response.read()
+        finally:
+            response.close()
+            response.release_conn()
 
-    async def delete(self, path: str) -> bool:
+    def _delete_sync(self, path: str) -> bool:
         client = self._get_client()
         try:
             client.remove_object(self.bucket, path)
@@ -181,7 +209,7 @@ class S3Storage(StorageBackend):
         except Exception:
             return False
 
-    async def exists(self, path: str) -> bool:
+    def _exists_sync(self, path: str) -> bool:
         client = self._get_client()
         try:
             client.stat_object(self.bucket, path)
@@ -189,17 +217,63 @@ class S3Storage(StorageBackend):
         except Exception:
             return False
 
-    async def get_url(self, path: str, expires: int = 3600) -> str:
+    def _get_url_sync(self, path: str, expires: int) -> str:
         from datetime import timedelta
 
         client = self._get_client()
         return client.presigned_get_object(self.bucket, path, expires=timedelta(seconds=expires))
 
-    async def list_files(self, prefix: str) -> list[str]:
+    def _list_files_sync(self, prefix: str) -> list[str]:
         client = self._get_client()
         objects = client.list_objects(self.bucket, prefix=prefix, recursive=True)
         return [obj.object_name for obj in objects]
 
+    # ── Async public API (wraps sync calls in asyncio.to_thread) ─────────────
+
+    async def save(self, path: str, data: bytes) -> str:
+        result = await asyncio.to_thread(self._save_sync, path, data)
+        logger.debug("S3Storage: saved %s (%d bytes)", path, len(data))
+        return result
+
+    async def save_stream(self, path: str, chunk_aiter: AsyncIterator[bytes]) -> int:
+        """Stream chunks to a temp file, then upload to S3 via fput_object.
+
+        This avoids loading the entire file into RAM. The temp file is
+        cleaned up after upload.
+        """
+        tmp_fd, tmp_path = tempfile.mkstemp(prefix="s3_upload_")
+        os.close(tmp_fd)
+        total = 0
+        try:
+            async with aiofiles.open(tmp_path, "wb") as f:
+                async for chunk in chunk_aiter:
+                    await f.write(chunk)
+                    total += len(chunk)
+            await asyncio.to_thread(self._put_file_sync, path, tmp_path)
+            logger.debug("S3Storage: streamed %s (%d bytes)", path, total)
+            return total
+        finally:
+            if os.path.exists(tmp_path):
+                try:
+                    os.remove(tmp_path)
+                except OSError:
+                    logger.warning("S3Storage: failed to clean up temp file %s", tmp_path)
+
+    async def read(self, path: str) -> bytes:
+        return await asyncio.to_thread(self._read_sync, path)
+
+    async def delete(self, path: str) -> bool:
+        return await asyncio.to_thread(self._delete_sync, path)
+
+    async def exists(self, path: str) -> bool:
+        return await asyncio.to_thread(self._exists_sync, path)
+
+    async def get_url(self, path: str, expires: int = 3600) -> str:
+        return await asyncio.to_thread(self._get_url_sync, path, expires)
+
+    async def list_files(self, prefix: str) -> list[str]:
+        return await asyncio.to_thread(self._list_files_sync, prefix)
+
 
 # ─── Factory ───
 

app/core/worker.py

@@ -14,6 +14,73 @@ from app.core.job_registry import get_all_jobs, get_job, register_job
 logger = logging.getLogger(__name__)
 
 
+# ── Distributed lock helpers ─────────────────────────────────────────────────
+# When multiple worker replicas run concurrently, cron jobs must not fire
+# on every replica.  We use a short-lived Redis SET NX lock per cron call
+# so only one replica actually executes the job.
+
+import redis.asyncio as aioredis  # noqa: E402
+import uuid  # noqa: E402
+
+
+async def _acquire_cron_lock(job_name: str, ttl_seconds: int = 120) -> str | None:
+    """Try to acquire a distributed lock for a cron job.
+
+    Returns a lock token (random UUID) if acquired, or None if another
+    replica already holds the lock.  The lock auto-expires after
+    *ttl_seconds* to avoid deadlocks if a worker crashes mid-job.
+    """
+    settings = get_settings()
+    client = aioredis.from_url(settings.redis_url)
+    token = str(uuid.uuid4())
+    lock_key = f"leocrm:cron_lock:{job_name}"
+    try:
+        acquired = await client.set(lock_key, token, nx=True, ex=ttl_seconds)
+        return token if acquired else None
+    finally:
+        await client.aclose()
+
+
+async def _release_cron_lock(job_name: str, token: str) -> None:
+    """Release a previously acquired cron lock using a safe compare-and-delete."""
+    settings = get_settings()
+    client = aioredis.from_url(settings.redis_url)
+    lock_key = f"leocrm:cron_lock:{job_name}"
+    try:
+        # Lua script ensures we only delete if the token matches (avoid
+        # releasing a lock that was already expired and re-acquired).
+        script = (
+            b"if redis.call('get', KEYS[1]) == ARGV[1] "
+            b"then return redis.call('del', KEYS[1]) "
+            b"else return 0 end"
+        )
+        await client.eval(script, 1, lock_key, token.encode())
+    finally:
+        await client.aclose()
+
+
+def _wrap_cron_with_lock(job_name: str, func: Any, ttl_seconds: int = 120) -> Any:
+    """Wrap a cron callable so it acquires a distributed lock first.
+
+    If the lock cannot be acquired (another replica is handling it), the
+    wrapped function is silently skipped.
+    """
+    import functools
+
+    @functools.wraps(func)
+    async def _locked_wrapper(ctx: dict[str, Any], *args: Any, **kwargs: Any) -> Any:
+        token = await _acquire_cron_lock(job_name, ttl_seconds=ttl_seconds)
+        if token is None:
+            logger.debug("Cron job '%s' skipped — lock held by another replica", job_name)
+            return None
+        try:
+            return await func(ctx, *args, **kwargs)
+        finally:
+            await _release_cron_lock(job_name, token)
+
+    return _locked_wrapper
+
+
 def _get_redis_settings() -> RedisSettings:
     """Get Redis settings from app config."""
     settings = get_settings()
@@ -70,6 +137,32 @@ def _lazy_register_plugin_jobs() -> None:
 _lazy_register_plugin_jobs()
 
 
+# ── Outbox processor job ────────────────────────────────────────────────────
+
+async def process_outbox_job(ctx: dict[str, Any]) -> None:
+    """Poll the transactional outbox and publish pending events.
+
+    Uses a distributed Redis lock so only one worker replica processes the
+    outbox at a time.  Runs every 5 seconds.
+    """
+    from app.core.db import get_session_factory
+    from app.core.outbox import process_outbox_batch
+
+    factory = get_session_factory()
+    async with factory() as db:
+        try:
+            count = await process_outbox_batch(db, batch_size=50)
+            if count:
+                logger.info("Outbox: published %d events", count)
+        except Exception:
+            logger.error("Outbox processing failed", exc_info=True)
+            await db.rollback()
+
+
+# Register the outbox job so it appears in get_all_jobs()
+register_job("process_outbox", process_outbox_job)
+
+
 class WorkerSettings:
     """ARQ worker settings."""
     functions = get_all_jobs()
@@ -80,6 +173,17 @@ class WorkerSettings:
     job_timeout = 300
     queue_name = "arq:queue"
     cron_jobs = [
-        cron(get_job("scheduler_tick"), minute={0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55}),
-        cron(get_job("tasks_due_reminder"), hour=8, minute=0),
+        cron(
+            _wrap_cron_with_lock("scheduler_tick", get_job("scheduler_tick")),
+            minute={0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55},
+        ),
+        cron(
+            _wrap_cron_with_lock("tasks_due_reminder", get_job("tasks_due_reminder")),
+            hour=8, minute=0,
+        ),
+        # Outbox processor — every 5 seconds, guarded by distributed lock
+        cron(
+            _wrap_cron_with_lock("process_outbox", process_outbox_job, ttl_seconds=30),
+            second="*/5",
+        ),
     ]