diff --git a/app/core/event_bus.py b/app/core/event_bus.py index edd615f..80c0dcc 100644 --- a/app/core/event_bus.py +++ b/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) diff --git a/app/core/notifications.py b/app/core/notifications.py index 8dbfad4..d0a1de6 100644 --- a/app/core/notifications.py +++ b/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 diff --git a/app/plugins/builtins/ai_assistant/crm_api_tool.py b/app/plugins/builtins/ai_assistant/crm_api_tool.py index e1eb73c..8393242 100644 --- a/app/plugins/builtins/ai_assistant/crm_api_tool.py +++ b/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") diff --git a/app/plugins/builtins/automation/jobs.py b/app/plugins/builtins/automation/jobs.py new file mode 100644 index 0000000..5254e5d --- /dev/null +++ b/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") diff --git a/app/plugins/builtins/automation/plugin.py b/app/plugins/builtins/automation/plugin.py index 1c010a7..14c935a 100644 --- a/app/plugins/builtins/automation/plugin.py +++ b/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: diff --git a/app/plugins/builtins/kommunikation/routes.py b/app/plugins/builtins/kommunikation/routes.py index 2a8c59a..3f46192 100644 --- a/app/plugins/builtins/kommunikation/routes.py +++ b/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()} diff --git a/app/plugins/builtins/mail/services.py b/app/plugins/builtins/mail/services.py index 25cb92d..14f15d6 100644 --- a/app/plugins/builtins/mail/services.py +++ b/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: diff --git a/app/plugins/builtins/permissions/routes.py b/app/plugins/builtins/permissions/routes.py index d724cdc..38457b5 100644 --- a/app/plugins/builtins/permissions/routes.py +++ b/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"]) diff --git a/app/plugins/builtins/tasks/jobs.py b/app/plugins/builtins/tasks/jobs.py index f64d8ae..b84d11d 100644 --- a/app/plugins/builtins/tasks/jobs.py +++ b/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) diff --git a/app/plugins/builtins/tasks/services.py b/app/plugins/builtins/tasks/services.py index d69271f..c2f8ba4 100644 --- a/app/plugins/builtins/tasks/services.py +++ b/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) diff --git a/app/plugins/builtins/unified_search/provider_registry.py b/app/plugins/builtins/unified_search/provider_registry.py index 1822564..fbbab28 100644 --- a/app/plugins/builtins/unified_search/provider_registry.py +++ b/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()) diff --git a/app/routes/users.py b/app/routes/users.py index 17eb812..74b2ca5 100644 --- a/app/routes/users.py +++ b/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, diff --git a/app/services/contact_service.py b/app/services/contact_service.py index 8862a5a..d611a37 100644 --- a/app/services/contact_service.py +++ b/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 diff --git a/app/workflows/engine.py b/app/workflows/engine.py index 18542c8..078d93c 100644 --- a/app/workflows/engine.py +++ b/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 - for event_name in ("user.created", "contact.created"): - event_bus.subscribe(event_name, _workflow_event_handler) + # Subscribe to ALL events via wildcard '*' — the handler dynamically + # queries for workflows whose trigger_event matches the published event. + event_bus.subscribe('*', _workflow_event_handler)