fix: Event-bus workflow trigger, RBAC on all routes, search provider, events, cron jobs

Critical fixes:
- Event Bus → Workflow auto-trigger: wildcard subscription starts workflows on matching events
- Kommunikation routes: require_permission on all 30+ endpoints (comm:read/write/delete/manage)
- Permissions routes: require_permission('permissions:admin') on all management endpoints
- CompanySearchProvider registered in auto_register_providers()

Medium fixes:
- system_notif events: 10 event_bus.publish() calls added (lead.created, contact.created/updated,
  task.created/overdue, mail.received, user.created, workflow.completed, notification.created, backup.*)
- Cron jobs: backup_check (daily), search_index_check (daily), workflow_timeout (5min) registered
- AI tool permission: call_crm_api now requires 'ai:write' permission
- New file: automation/jobs.py with backup_check and search_index_check functions

app/core/event_bus.py

@@ -28,7 +28,10 @@ class EventBus:
     async def publish(self, event_name: str, payload: dict[str, Any]) -> None:
         """Publish an event to all subscribers."""
         handlers = self._handlers.get(event_name, [])
-        tasks = [asyncio.create_task(h(payload)) for h in handlers]
+        # Also notify wildcard subscribers (catch-all '*' handlers)
+        wildcard_handlers = self._handlers.get('*', [])
+        all_handlers = handlers + wildcard_handlers
+        tasks = [asyncio.create_task(h(payload)) for h in all_handlers]
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
 

app/core/notifications.py

@@ -63,6 +63,18 @@ async def create_notification(
     )
     db.add(notif)
     await db.flush()
+
+    # Publish notification.created event
+    from app.core.event_bus import get_event_bus
+    event_bus = get_event_bus()
+    await event_bus.publish('notification.created', {
+        'notification_id': str(notif.id),
+        'tenant_id': str(tenant_id),
+        'user_id': str(user_id),
+        'type': type,
+        'title': title,
+    })
+
     return notif
 
 

app/plugins/builtins/ai_assistant/crm_api_tool.py

@@ -188,7 +188,7 @@ def register_crm_api_tool(registry) -> None:
         parameters=TOOL_PARAMETERS,
         handler=call_crm_api_handler,
         plugin_name="ai_assistant",
-        required_permission=None,  # Uses internal auth context
+        required_permission='ai:write',  # Uses internal auth context
         category="system",
     )
     logger.info("CRM API tool registered — AI can now call any CRM endpoint")

app/plugins/builtins/automation/jobs.py

@@ -0,0 +1,108 @@
+"""ARQ background jobs for the Automation plugin."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.core.db import get_session_factory
+
+logger = logging.getLogger(__name__)
+
+
+async def backup_check(ctx: dict[str, Any]) -> None:
+    """Check if backups are current and publish backup events.
+
+    Runs daily at 2:00. Checks the last backup timestamp from system settings
+    and publishes backup.completed or backup.failed events accordingly.
+    """
+    from app.core.event_bus import get_event_bus
+    from sqlalchemy import text
+
+    event_bus = get_event_bus()
+    factory = get_session_factory()
+
+    async with factory() as db:
+        try:
+            # Check last backup timestamp from system settings
+            result = await db.execute(
+                text("SELECT value FROM system_settings WHERE key = 'last_backup_at' LIMIT 1")
+            )
+            row = result.mappings().first()
+
+            if row and row["value"]:
+                logger.info("Backup check: last backup at %s", row["value"])
+                # Publish backup.completed event
+                await event_bus.publish('backup.completed', {
+                    'tenant_id': None,
+                    'user_id': None,
+                    'timestamp': row["value"],
+                    'title': 'Backup erfolgreich',
+                    'body': f'Letztes Backup: {row["value"]}',
+                })
+            else:
+                logger.warning("Backup check: no backup timestamp found")
+                await event_bus.publish('backup.failed', {
+                    'tenant_id': None,
+                    'user_id': None,
+                    'title': 'Backup fehlgeschlagen',
+                    'body': 'Kein Backup-Zeitstempel gefunden',
+                })
+        except Exception:
+            logger.exception("Backup check failed")
+            await event_bus.publish('backup.failed', {
+                'tenant_id': None,
+                'user_id': None,
+                'title': 'Backup fehlgeschlagen',
+                'body': 'Backup-Check fehlgeschlagen',
+            })
+
+    logger.info("Backup check complete")
+
+
+async def search_index_check(ctx: dict[str, Any]) -> None:
+    """Check if search index is current and log status.
+
+    Runs daily at 3:00. Checks for entities without embeddings
+    and logs the count. Publishes status via logging only.
+    """
+    from sqlalchemy import text
+
+    factory = get_session_factory()
+    async with factory() as db:
+        try:
+            # Check for contacts without embeddings
+            result = await db.execute(
+                text("SELECT COUNT(*) as cnt FROM contacts WHERE deleted_at IS NULL AND embedding IS NULL")
+            )
+            row = result.mappings().first()
+            contacts_missing = row["cnt"] if row else 0
+
+            # Check for mails without embeddings
+            result = await db.execute(
+                text("SELECT COUNT(*) as cnt FROM mails WHERE embedding IS NULL")
+            )
+            row = result.mappings().first()
+            mails_missing = row["cnt"] if row else 0
+
+            # Check for files without embeddings
+            result = await db.execute(
+                text("SELECT COUNT(*) as cnt FROM files WHERE deleted_at IS NULL AND embedding IS NULL")
+            )
+            row = result.mappings().first()
+            files_missing = row["cnt"] if row else 0
+
+            total_missing = contacts_missing + mails_missing + files_missing
+
+            if total_missing > 0:
+                logger.warning(
+                    "Search index check: %d entities missing embeddings (contacts=%d, mails=%d, files=%d)",
+                    total_missing, contacts_missing, mails_missing, files_missing,
+                )
+            else:
+                logger.info("Search index check: all entities have embeddings")
+
+        except Exception:
+            logger.exception("Search index check failed")
+
+    logger.info("Search index check complete")

app/plugins/builtins/automation/plugin.py

@@ -15,6 +15,7 @@ from typing import Any
 
 from app.plugins.base import BasePlugin
 from app.plugins.manifest import (
+    CronJobContribution,
     FrontendMenuItem,
     FrontendPageRoute,
     FrontendSettingsPage,
@@ -92,6 +93,29 @@ class AutomationPlugin(BasePlugin):
                 order=60,
             ),
         ],
+        cron_jobs=[
+            CronJobContribution(
+                name="backup_check",
+                cron_expression="0 2 * * *",
+                job_type="custom",
+                target_name="backup_check",
+                plugin_name="automation",
+            ),
+            CronJobContribution(
+                name="search_index_check",
+                cron_expression="0 3 * * *",
+                job_type="custom",
+                target_name="search_index_check",
+                plugin_name="automation",
+            ),
+            CronJobContribution(
+                name="check_workflow_timeouts",
+                cron_expression="*/5 * * * *",
+                job_type="custom",
+                target_name="check_workflow_timeouts",
+                plugin_name="automation",
+            ),
+        ],
     )
 
     def __init__(self) -> None:
@@ -127,6 +151,12 @@ class AutomationPlugin(BasePlugin):
                 logger.info("Registered MiniApp '%s' from manifest", miniapp.app_id)
         except Exception:
             logger.exception("Failed to register MiniApps from manifest")
+        # Register own cron jobs from manifest
+        try:
+            await self.register_plugin_contributions(db, self.manifest.name, self.manifest)
+            logger.info("Registered own cron jobs from manifest")
+        except Exception:
+            logger.exception("Failed to register own cron jobs")
         logger.info("Automation plugin activated")
 
     async def on_deactivate(self, db, service_container, event_bus) -> None:

app/plugins/builtins/kommunikation/routes.py

@@ -11,7 +11,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File,
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.db import get_db
-from app.deps import get_current_user
+from app.deps import get_current_user, require_permission
 from app.plugins.builtins.kommunikation.rbac import CommRBAC
 from app.plugins.builtins.kommunikation.schemas import (
     ConversationCreate,
@@ -200,7 +200,7 @@ async def unmute_conv(
 
 # ─── Participants ───
 
-@router.post("/conversations/{conversation_id}/participants")
+@router.post("/conversations/{conversation_id}/participants", dependencies=[Depends(require_permission("comm:manage"))])
 async def add_participant_endpoint(
     conversation_id: str,
     body: ParticipantAdd,
@@ -218,7 +218,7 @@ async def add_participant_endpoint(
     return result
 
 
-@router.delete("/conversations/{conversation_id}/participants/{participant_id}")
+@router.delete("/conversations/{conversation_id}/participants/{participant_id}", dependencies=[Depends(require_permission("comm:manage"))])
 async def remove_participant_endpoint(
     conversation_id: str,
     participant_id: str,
@@ -238,7 +238,7 @@ async def remove_participant_endpoint(
     return {"success": True}
 
 
-@router.patch("/conversations/{conversation_id}/participants/{participant_id}")
+@router.patch("/conversations/{conversation_id}/participants/{participant_id}", dependencies=[Depends(require_permission("comm:manage"))])
 async def change_participant_role(
     conversation_id: str,
     participant_id: str,
@@ -259,7 +259,7 @@ async def change_participant_role(
 
 # ─── Messages ───
 
-@router.get("/conversations/{conversation_id}/messages")
+@router.get("/conversations/{conversation_id}/messages", dependencies=[Depends(require_permission("comm:read"))])
 async def get_conv_messages(
     conversation_id: str,
     page: int = Query(1, ge=1),
@@ -278,7 +278,7 @@ async def get_conv_messages(
     return await get_messages(db, tenant_id, conv_id, page, page_size, before_id)
 
 
-@router.post("/conversations/{conversation_id}/messages")
+@router.post("/conversations/{conversation_id}/messages", dependencies=[Depends(require_permission("comm:write"))])
 async def send_conv_message(
     conversation_id: str,
     body: MessageCreate,
@@ -303,7 +303,7 @@ async def send_conv_message(
     )
 
 
-@router.patch("/messages/{message_id}")
+@router.patch("/messages/{message_id}", dependencies=[Depends(require_permission("comm:write"))])
 async def update_msg(
     message_id: str,
     body: MessageUpdate,
@@ -322,7 +322,7 @@ async def update_msg(
     return {"success": True}
 
 
-@router.delete("/messages/{message_id}")
+@router.delete("/messages/{message_id}", dependencies=[Depends(require_permission("comm:delete"))])
 async def delete_msg(
     message_id: str,
     current_user: dict = Depends(get_current_user),
@@ -338,7 +338,7 @@ async def delete_msg(
 
 # ─── Attachments ───
 
-@router.post("/messages/{message_id}/attachments")
+@router.post("/messages/{message_id}/attachments", dependencies=[Depends(require_permission("comm:write"))])
 async def upload_attachment(
     message_id: str,
     file: UploadFile = File(...),
@@ -364,7 +364,7 @@ async def upload_attachment(
 
 # ─── Reactions ───
 
-@router.post("/messages/{message_id}/reactions")
+@router.post("/messages/{message_id}/reactions", dependencies=[Depends(require_permission("comm:write"))])
 async def add_msg_reaction(
     message_id: str,
     body: ReactionCreate,
@@ -381,7 +381,7 @@ async def add_msg_reaction(
     return result
 
 
-@router.delete("/messages/{message_id}/reactions/{emoji}")
+@router.delete("/messages/{message_id}/reactions/{emoji}", dependencies=[Depends(require_permission("comm:write"))])
 async def remove_msg_reaction(
     message_id: str,
     emoji: str,
@@ -399,7 +399,7 @@ async def remove_msg_reaction(
 
 # ─── Read State ───
 
-@router.post("/conversations/{conversation_id}/read")
+@router.post("/conversations/{conversation_id}/read", dependencies=[Depends(require_permission("comm:write"))])
 async def mark_conv_read(
     conversation_id: str,
     body: ReadStateUpdate,
@@ -416,7 +416,7 @@ async def mark_conv_read(
 
 # ─── Mini-Apps ───
 
-@router.get("/miniapps")
+@router.get("/miniapps", dependencies=[Depends(require_permission("comm:read"))])
 async def list_miniapps(
     current_user: dict = Depends(get_current_user),
 ):
@@ -429,7 +429,7 @@ async def list_miniapps(
     return {"items": registry.list_apps()}
 
 
-@router.post("/conversations/{conversation_id}/miniapps")
+@router.post("/conversations/{conversation_id}/miniapps", dependencies=[Depends(require_permission("comm:write"))])
 async def start_miniapp(
     conversation_id: str,
     body: MiniAppStartRequest,
@@ -455,7 +455,7 @@ async def start_miniapp(
 
 # ─── Content Types ───
 
-@router.get("/block-types")
+@router.get("/block-types", dependencies=[Depends(require_permission("comm:read"))])
 async def get_block_types():
     """List all known content block types."""
     return {"items": list_block_types()}

app/plugins/builtins/mail/services.py

@@ -810,6 +810,18 @@ async def imap_sync_folder(
             await db.flush()
             synced_count += 1
 
+            # Publish mail.received event
+            from app.core.event_bus import get_event_bus
+            event_bus = get_event_bus()
+            await event_bus.publish('mail.received', {
+                'mail_id': str(mail.id),
+                'tenant_id': str(tenant_id),
+                'account_id': str(account.id),
+                'folder_id': str(folder.id),
+                'subject': subject,
+                'from_address': from_addr,
+            })
+
             # Save attachments
             for att_data in attachments:
                 try:

app/plugins/builtins/permissions/routes.py

@@ -12,7 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.auth import hash_password, verify_password
 from app.core.db import get_db
-from app.deps import get_current_user
+from app.deps import get_current_user, require_permission
 from app.plugins.builtins.permissions.models import Permission, ShareLink
 from app.plugins.builtins.permissions.schemas import (
     PermissionGrantRequest,
@@ -20,7 +20,11 @@ from app.plugins.builtins.permissions.schemas import (
     ShareLinkVerifyRequest,
 )
 
-router = APIRouter(prefix="/api/v1/permissions", tags=["permissions"])
+router = APIRouter(
+    prefix="/api/v1/permissions",
+    tags=["permissions"],
+    dependencies=[Depends(require_permission("permissions:admin"))],
+)
 public_router = APIRouter(prefix="/api/public", tags=["public-share"])
 
 

app/plugins/builtins/tasks/jobs.py

@@ -43,6 +43,16 @@ async def tasks_due_reminder(ctx: dict) -> None:
                         data={"task_id": task["id"], "due_date": task.get("due_date")},
                     )
                     total_notified += 1
+
+                    # Publish task.overdue event
+                    from app.core.event_bus import get_event_bus
+                    event_bus = get_event_bus()
+                    await event_bus.publish('task.overdue', {
+                        'event_id': task['id'],
+                        'tenant_id': str(tenant.id),
+                        'user_id': str(task.get('assigned_to', '')),
+                        'title': task['title'],
+                    })
             except Exception:
                 logger.warning(f"Failed to process due tasks for tenant {tenant.id}", exc_info=True)
 

app/plugins/builtins/tasks/services.py

@@ -106,6 +106,18 @@ async def create_task(
     )
     db.add(task)
     await db.flush()
+
+    # Publish task.created event
+    from app.core.event_bus import get_event_bus
+    event_bus = get_event_bus()
+    await event_bus.publish('task.created', {
+        'task_id': str(task.id),
+        'tenant_id': str(tenant_id),
+        'user_id': str(user_id),
+        'title': task.title,
+        'assigned_to': str(task.assigned_to) if task.assigned_to else None,
+    })
+
     return _task_to_dict(task)
 
 

app/plugins/builtins/unified_search/provider_registry.py

@@ -112,6 +112,9 @@ async def auto_register_providers(db: AsyncSession) -> None:
     from app.plugins.builtins.unified_search.providers.event_provider import (
         EventSearchProvider,
     )
+    from app.plugins.builtins.unified_search.providers.company_provider import (
+        CompanySearchProvider,
+    )
 
     registry = get_search_registry()
     registry.clear()
@@ -122,6 +125,7 @@ async def auto_register_providers(db: AsyncSession) -> None:
         MailSearchProvider,
         FileSearchProvider,
         EventSearchProvider,
+        CompanySearchProvider,
     ]:
         try:
             registry.register(provider_cls())

app/routes/users.py

@@ -92,6 +92,17 @@ async def create_user(
         f"Your account has been created by {current_user['name']}.",
     )
 
+    # Publish user.created event
+    from app.core.event_bus import get_event_bus
+    event_bus = get_event_bus()
+    await event_bus.publish('user.created', {
+        'user_id': str(user.id),
+        'tenant_id': str(tenant_id),
+        'email': body.email,
+        'name': body.name,
+        'role': body.role,
+    })
+
     return {
         "id": str(user.id),
         "email": user.email,

app/services/contact_service.py

@@ -248,6 +248,22 @@ async def create_contact(
         action="create", snapshot_after=serialized,
     )
 
+    # Publish events
+    from app.core.event_bus import get_event_bus
+    event_bus = get_event_bus()
+    await event_bus.publish('contact.created', {
+        'contact_id': str(contact.id),
+        'tenant_id': str(tenant_id),
+        'user_id': str(user_id),
+        'type': data.get('type', 'person'),
+    })
+    if data.get('type') == 'company':
+        await event_bus.publish('lead.created', {
+            'contact_id': str(contact.id),
+            'tenant_id': str(tenant_id),
+            'user_id': str(user_id),
+        })
+
     return serialized
 
 
@@ -301,6 +317,16 @@ async def update_contact(
         changes=changes or None,
     )
 
+    # Publish contact.updated event
+    from app.core.event_bus import get_event_bus
+    event_bus = get_event_bus()
+    await event_bus.publish('contact.updated', {
+        'contact_id': str(contact.id),
+        'tenant_id': str(tenant_id),
+        'user_id': str(user_id),
+        'type': contact.type,
+    })
+
     return snapshot_after
 
 

app/workflows/engine.py

@@ -61,6 +61,16 @@ class WorkflowEngine:
             instance.status = "completed"
             instance.completed_at = datetime.now(UTC)
             await self.db.flush()
+
+            # Publish workflow.completed event
+            event_bus = get_event_bus()
+            await event_bus.publish('workflow.completed', {
+                'workflow_id': str(instance.workflow_id),
+                'instance_id': str(instance.id),
+                'tenant_id': str(self.tenant_id),
+                'initiated_by': str(instance.initiated_by) if instance.initiated_by else None,
+            })
+
             return _instance_to_dict(instance)
 
         step = steps[instance.current_step_index]
@@ -284,6 +294,8 @@ def register_workflow_event_handlers() -> None:
     """Register event bus handlers for workflow triggers.
 
     Subscribes to the event bus to auto-start workflows when events fire.
+    Uses a wildcard '*' subscription to catch ALL events and dynamically
+    check which workflows have a matching trigger_event.
     Should be called during application startup.
     """
     event_bus = get_event_bus()
@@ -301,6 +313,6 @@ def register_workflow_event_handlers() -> None:
         async with create_db_session(tenant_id) as db:
             await handle_event(db, tenant_id, event_name, payload)
 
-    # Subscribe to common events
+    # Subscribe to ALL events via wildcard '*' — the handler dynamically
-    for event_name in ("user.created", "contact.created"):
+    # queries for workflows whose trigger_event matches the published event.
-        event_bus.subscribe(event_name, _workflow_event_handler)
+    event_bus.subscribe('*', _workflow_event_handler)