diff --git a/backend/open_webui/events.py b/backend/open_webui/events.py new file mode 100644 index 0000000000..5b6bafbf21 --- /dev/null +++ b/backend/open_webui/events.py @@ -0,0 +1,360 @@ +from __future__ import annotations + +import logging +import time +import uuid +from dataclasses import asdict, dataclass, field +from enum import StrEnum +from typing import Any + +from open_webui.env import VERSION +from open_webui.models.config import Config +from open_webui.utils.webhook import post_webhook + +log = logging.getLogger(__name__) + +EVENT_VERSION = 'event.v1' +MAX_STRING_LENGTH = 1000 + + +class EVENTS(StrEnum): + SYSTEM_STARTUP_STARTED = 'system.startup.started' + SYSTEM_STARTUP_COMPLETED = 'system.startup.completed' + SYSTEM_SHUTDOWN_STARTED = 'system.shutdown.started' + SYSTEM_SHUTDOWN_COMPLETED = 'system.shutdown.completed' + CONFIG_IMPORTED = 'config.imported' + CONFIG_UPDATED = 'config.updated' + CONFIG_WEBHOOK_UPDATED = 'config.webhook.updated' + CONFIG_CONNECTIONS_UPDATED = 'config.connections.updated' + CONFIG_TOOL_SERVERS_UPDATED = 'config.tool_servers.updated' + CONFIG_TERMINAL_SERVERS_UPDATED = 'config.terminal_servers.updated' + CONFIG_CODE_EXECUTION_UPDATED = 'config.code_execution.updated' + CONFIG_MODELS_UPDATED = 'config.models.updated' + CONFIG_BANNERS_UPDATED = 'config.banners.updated' + CONFIG_SUGGESTIONS_UPDATED = 'config.suggestions.updated' + AUTH_SIGNUP = 'auth.signup' + AUTH_LOGIN = 'auth.login' + AUTH_LOGOUT = 'auth.logout' + AUTH_PASSWORD_CHANGED = 'auth.password_changed' + AUTH_API_KEY_CREATED = 'auth.api_key.created' + AUTH_API_KEY_DELETED = 'auth.api_key.deleted' + AUTH_OAUTH_SESSION_DELETED = 'auth.oauth_session.deleted' + USER_CREATED = 'user.created' + USER_UPDATED = 'user.updated' + USER_DELETED = 'user.deleted' + USER_ROLE_UPDATED = 'user.role_updated' + USER_STATUS_UPDATED = 'user.status_updated' + USER_SETTINGS_UPDATED = 'user.settings_updated' + USER_PROFILE_UPDATED = 'user.profile_updated' + USER_PERMISSIONS_UPDATED = 'user.permissions_updated' + GROUP_CREATED = 'group.created' + GROUP_UPDATED = 'group.updated' + GROUP_DELETED = 'group.deleted' + GROUP_MEMBER_ADDED = 'group.member_added' + GROUP_MEMBER_REMOVED = 'group.member_removed' + CHAT_CREATED = 'chat.created' + CHAT_IMPORTED = 'chat.imported' + CHAT_UPDATED = 'chat.updated' + CHAT_DELETED = 'chat.deleted' + CHAT_DELETED_ALL = 'chat.deleted_all' + CHAT_COMPACTED = 'chat.compacted' + CHAT_PINNED = 'chat.pinned' + CHAT_UNPINNED = 'chat.unpinned' + CHAT_CLONED = 'chat.cloned' + CHAT_ARCHIVED = 'chat.archived' + CHAT_UNARCHIVED = 'chat.unarchived' + CHAT_SHARED = 'chat.shared' + CHAT_UNSHARED = 'chat.unshared' + CHAT_FOLDER_UPDATED = 'chat.folder_updated' + CHAT_TAG_ADDED = 'chat.tag_added' + CHAT_TAG_REMOVED = 'chat.tag_removed' + MESSAGE_CREATED = 'message.created' + MESSAGE_UPDATED = 'message.updated' + MESSAGE_DELETED = 'message.deleted' + MESSAGE_EVENT_RECEIVED = 'message.event_received' + MESSAGE_REACTION_ADDED = 'message.reaction_added' + MESSAGE_REACTION_REMOVED = 'message.reaction_removed' + MESSAGE_PINNED = 'message.pinned' + MESSAGE_UNPINNED = 'message.unpinned' + CHANNEL_CREATED = 'channel.created' + CHANNEL_UPDATED = 'channel.updated' + CHANNEL_DELETED = 'channel.deleted' + CHANNEL_MEMBER_ADDED = 'channel.member_added' + CHANNEL_MEMBER_REMOVED = 'channel.member_removed' + CHANNEL_MEMBER_ACTIVE_UPDATED = 'channel.member_active_updated' + CHANNEL_WEBHOOK_CREATED = 'channel.webhook.created' + CHANNEL_WEBHOOK_UPDATED = 'channel.webhook.updated' + CHANNEL_WEBHOOK_DELETED = 'channel.webhook.deleted' + FILE_UPLOADED = 'file.uploaded' + FILE_CONTENT_UPDATED = 'file.content_updated' + FILE_RENAMED = 'file.renamed' + FILE_DELETED = 'file.deleted' + FILE_DELETED_ALL = 'file.deleted_all' + FOLDER_CREATED = 'folder.created' + FOLDER_UPDATED = 'folder.updated' + FOLDER_PARENT_UPDATED = 'folder.parent_updated' + FOLDER_ACCESS_UPDATED = 'folder.access_updated' + FOLDER_DELETED = 'folder.deleted' + NOTE_CREATED = 'note.created' + NOTE_UPDATED = 'note.updated' + NOTE_ACCESS_UPDATED = 'note.access_updated' + NOTE_PINNED = 'note.pinned' + NOTE_UNPINNED = 'note.unpinned' + NOTE_DELETED = 'note.deleted' + MEMORY_CREATED = 'memory.created' + MEMORY_UPDATED = 'memory.updated' + MEMORY_DELETED = 'memory.deleted' + MEMORY_RESET = 'memory.reset' + KNOWLEDGE_CREATED = 'knowledge.created' + KNOWLEDGE_UPDATED = 'knowledge.updated' + KNOWLEDGE_DELETED = 'knowledge.deleted' + KNOWLEDGE_RESET = 'knowledge.reset' + KNOWLEDGE_REINDEXED = 'knowledge.reindexed' + KNOWLEDGE_ACCESS_UPDATED = 'knowledge.access_updated' + KNOWLEDGE_FILE_ADDED = 'knowledge.file.added' + KNOWLEDGE_FILE_UPDATED = 'knowledge.file.updated' + KNOWLEDGE_FILE_REMOVED = 'knowledge.file.removed' + KNOWLEDGE_FILE_MOVED = 'knowledge.file.moved' + KNOWLEDGE_DIRECTORY_CREATED = 'knowledge.directory.created' + KNOWLEDGE_DIRECTORY_UPDATED = 'knowledge.directory.updated' + KNOWLEDGE_DIRECTORY_DELETED = 'knowledge.directory.deleted' + KNOWLEDGE_EXTERNAL_CONNECTION_CREATED = 'knowledge.external_connection.created' + KNOWLEDGE_EXTERNAL_CONNECTION_UPDATED = 'knowledge.external_connection.updated' + KNOWLEDGE_EXTERNAL_CONNECTION_DELETED = 'knowledge.external_connection.deleted' + RETRIEVAL_CONTENT_PROCESSED = 'retrieval.content.processed' + RETRIEVAL_COLLECTION_DELETED = 'retrieval.collection.deleted' + RETRIEVAL_VECTOR_DB_RESET = 'retrieval.vector_db.reset' + RETRIEVAL_UPLOADS_RESET = 'retrieval.uploads.reset' + MODEL_CREATED = 'model.created' + MODEL_IMPORTED = 'model.imported' + MODEL_SYNCED = 'model.synced' + MODEL_UPDATED = 'model.updated' + MODEL_DELETED = 'model.deleted' + MODEL_ENABLED = 'model.enabled' + MODEL_DISABLED = 'model.disabled' + MODEL_ACCESS_UPDATED = 'model.access_updated' + MODEL_PROVIDER_CONFIG_UPDATED = 'model.provider_config.updated' + MODEL_PROVIDER_MODEL_CREATED = 'model.provider_model.created' + MODEL_PROVIDER_MODEL_DELETED = 'model.provider_model.deleted' + FUNCTION_CREATED = 'function.created' + FUNCTION_UPDATED = 'function.updated' + FUNCTION_DELETED = 'function.deleted' + FUNCTION_ENABLED = 'function.enabled' + FUNCTION_DISABLED = 'function.disabled' + FUNCTION_VALVES_UPDATED = 'function.valves_updated' + TOOL_CREATED = 'tool.created' + TOOL_UPDATED = 'tool.updated' + TOOL_DELETED = 'tool.deleted' + TOOL_ACCESS_UPDATED = 'tool.access_updated' + TOOL_VALVES_UPDATED = 'tool.valves_updated' + SKILL_CREATED = 'skill.created' + SKILL_UPDATED = 'skill.updated' + SKILL_DELETED = 'skill.deleted' + SKILL_ENABLED = 'skill.enabled' + SKILL_DISABLED = 'skill.disabled' + PROMPT_CREATED = 'prompt.created' + PROMPT_UPDATED = 'prompt.updated' + PROMPT_DELETED = 'prompt.deleted' + PROMPT_ENABLED = 'prompt.enabled' + PROMPT_DISABLED = 'prompt.disabled' + PROMPT_VERSION_UPDATED = 'prompt.version_updated' + PROMPT_ACCESS_UPDATED = 'prompt.access_updated' + PIPELINE_UPLOADED = 'pipeline.uploaded' + PIPELINE_ADDED = 'pipeline.added' + PIPELINE_DELETED = 'pipeline.deleted' + PIPELINE_VALVES_UPDATED = 'pipeline.valves_updated' + CALENDAR_CREATED = 'calendar.created' + CALENDAR_UPDATED = 'calendar.updated' + CALENDAR_DELETED = 'calendar.deleted' + CALENDAR_DEFAULT_UPDATED = 'calendar.default_updated' + CALENDAR_EVENT_CREATED = 'calendar.event.created' + CALENDAR_EVENT_UPDATED = 'calendar.event.updated' + CALENDAR_EVENT_DELETED = 'calendar.event.deleted' + CALENDAR_EVENT_RSVP_UPDATED = 'calendar.event.rsvp_updated' + AUTOMATION_CREATED = 'automation.created' + AUTOMATION_UPDATED = 'automation.updated' + AUTOMATION_ENABLED = 'automation.enabled' + AUTOMATION_DISABLED = 'automation.disabled' + AUTOMATION_DELETED = 'automation.deleted' + AUTOMATION_RUN_STARTED = 'automation.run_started' + AUTOMATION_RUN_COMPLETED = 'automation.run_completed' + AUTOMATION_RUN_FAILED = 'automation.run_failed' + FEEDBACK_CREATED = 'feedback.created' + FEEDBACK_UPDATED = 'feedback.updated' + FEEDBACK_DELETED = 'feedback.deleted' + FEEDBACK_DELETED_ALL = 'feedback.deleted_all' + IMAGE_GENERATED = 'image.generated' + IMAGE_EDITED = 'image.edited' + AUDIO_SPEECH_REQUESTED = 'audio.speech_requested' + AUDIO_TRANSCRIPTION_REQUESTED = 'audio.transcription_requested' + TERMINAL_SESSION_OPENED = 'terminal.session.opened' + TERMINAL_SESSION_CLOSED = 'terminal.session.closed' + + +EVENT_CATALOG = tuple(event.value for event in EVENTS) + +SENSITIVE_KEYS = { + 'password', + 'hashed_password', + 'token', + 'access_token', + 'refresh_token', + 'id_token', + 'api_key', + 'secret', + 'key', + 'authorization', + 'cookie', + 'webhook_token', +} + +SAFE_ACTOR_FIELDS = ('id', 'name', 'email', 'role', 'created_at', 'updated_at') + + +@dataclass +class Event: + schema: str + id: str + event: str + resource: str + operation: str + created_at: int + instance_id: str | None + version: str + source: str + actor: dict[str, Any] | None = None + subject: dict[str, Any] | None = None + data: dict[str, Any] = field(default_factory=dict) + message: str | None = None + + def model_dump(self) -> dict[str, Any]: + return asdict(self) + + +def _sensitive(key: Any) -> bool: + normalized = str(key).lower().replace('-', '_') + return ( + normalized in SENSITIVE_KEYS + or normalized.endswith('_token') + or normalized.endswith('_secret') + or normalized.endswith('_api_key') + or normalized.endswith('_key') + ) + + +def _sanitize(value: Any) -> Any: + if hasattr(value, 'model_dump'): + value = value.model_dump() + + if isinstance(value, dict): + return {key: _sanitize(item) for key, item in value.items() if not _sensitive(key)} + + if isinstance(value, (list, tuple, set)): + return [_sanitize(item) for item in value] + + if isinstance(value, str) and len(value) > MAX_STRING_LENGTH: + return f'{value[:MAX_STRING_LENGTH]}...' + + return value + + +def _actor(actor: Any | None) -> dict[str, Any] | None: + actor = _sanitize(actor) + if not actor: + return None + + get = actor.get if isinstance(actor, dict) else lambda key: getattr(actor, key, None) + data = {field: get(field) for field in SAFE_ACTOR_FIELDS if get(field) is not None} + if not data: + return None + + data['type'] = get('type') or 'user' + return data + + +def build_event( + request_or_app: Any, + event: EVENTS, + *, + actor: Any | None = None, + subject_id: Any | None = None, + subject_type: str | None = None, + source: str = 'api', + data: dict | None = None, + message: str | None = None, +) -> Event: + event_name = event.value + app = getattr(request_or_app, 'app', request_or_app) + parts = event_name.split('.') + resource = '.'.join(parts[:-1]) + instance_id = getattr(getattr(app, 'state', None), 'instance_id', None) + subject = ( + {'type': subject_type or resource, 'id': subject_id} + if subject_id is not None or subject_type is not None + else None + ) + + return Event( + schema=EVENT_VERSION, + id=str(uuid.uuid4()), + event=event_name, + resource=resource, + operation=parts[-1], + created_at=int(time.time()), + instance_id=instance_id, + version=VERSION, + source=source, + actor=_actor(actor), + subject=_sanitize(subject) if subject else None, + data=_sanitize(data or {}), + message=message, + ) + + +class WebhookEventSink: + async def handle_event(self, app: Any, event: Event) -> None: + url = await Config.get('webhook_url') + if not url: + return + + name = getattr(getattr(app, 'state', None), 'WEBUI_NAME', 'Open WebUI') + subject = event.subject or {} + subject_id = subject.get('id') + message = event.message or f'{event.event}: {subject.get("type") or event.resource}' + if subject_id: + message = f'{message} ({subject_id})' + + await post_webhook(name, url, message, event.model_dump()) + + +EVENT_SINKS = [WebhookEventSink()] + + +async def publish_event( + request_or_app: Any, + event: EVENTS, + *, + actor: Any | None = None, + subject_id: Any | None = None, + subject_type: str | None = None, + source: str = 'api', + data: dict | None = None, + message: str | None = None, +) -> None: + app = getattr(request_or_app, 'app', request_or_app) + event_payload = build_event( + request_or_app, + event, + actor=actor, + subject_id=subject_id, + subject_type=subject_type, + source=source, + data=data, + message=message, + ) + + for sink in EVENT_SINKS: + try: + await sink.handle_event(app, event_payload) + except Exception: + log.exception('Event sink failed for %s', event.value) diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py index 5cdfb791fe..91267faaab 100644 --- a/backend/open_webui/main.py +++ b/backend/open_webui/main.py @@ -121,6 +121,7 @@ from open_webui.models.access_grants import AccessGrants from open_webui.models.channels import Channels from open_webui.models.chats import ChatForm, Chats from open_webui.models.config import Config +from open_webui.events import EVENT_CATALOG, EVENT_VERSION, EVENTS, publish_event from open_webui.models.functions import Functions from open_webui.models.messages import Messages from open_webui.models.models import Models @@ -185,6 +186,7 @@ from open_webui.tasks import ( stop_task, ) # Import from tasks.py from open_webui.utils import logger +from open_webui.utils.access_control import has_permission from open_webui.utils.actions import chat_action as chat_action_handler from open_webui.utils.asgi_middleware import ( AuthTokenMiddleware, @@ -296,6 +298,7 @@ async def lifespan(app: FastAPI): await import_legacy_config_json() await seed_registered_defaults() await initialize_runtime_config(app) + await publish_event(app, EVENTS.SYSTEM_STARTUP_STARTED, source='system') if LICENSE_KEY: get_license_data(app, LICENSE_KEY) @@ -387,9 +390,12 @@ async def lifespan(app: FastAPI): # Mark application as ready to accept traffic from a startup perspective. app.state.startup_complete = True + await publish_event(app, EVENTS.SYSTEM_STARTUP_COMPLETED, source='system') yield + await publish_event(app, EVENTS.SYSTEM_SHUTDOWN_STARTED, source='system') + # Shutdown: clean up shared resources from open_webui.utils.session_pool import close_session @@ -398,6 +404,8 @@ async def lifespan(app: FastAPI): if hasattr(app.state, 'redis_task_command_listener'): app.state.redis_task_command_listener.cancel() + await publish_event(app, EVENTS.SYSTEM_SHUTDOWN_COMPLETED, source='system') + app = FastAPI( title='Open WebUI', @@ -1249,6 +1257,39 @@ async def chat_completion( folder_id=metadata.get('folder_id'), ), ) + await publish_event( + request, + EVENTS.CHAT_CREATED, + actor=user, + subject_id=chat_id, + data={'title': 'New Chat'}, + ) + if user_message_id: + await publish_event( + request, + EVENTS.MESSAGE_CREATED, + actor=user, + subject_id=user_message_id, + data={ + 'chat_id': chat_id, + 'role': 'user', + 'content_preview': user_message.get('content', '')[:300], + }, + ) + for entry in message_ids: + assistant_message_id = entry.get('message_id') + if assistant_message_id: + await publish_event( + request, + EVENTS.MESSAGE_CREATED, + actor=user, + subject_id=assistant_message_id, + data={ + 'chat_id': chat_id, + 'role': 'assistant', + 'model': entry.get('model_id'), + }, + ) # Insert chat files from user message if any user_message_files = user_message.get('files', []) @@ -1293,6 +1334,17 @@ async def chat_completion( user_message['id'], user_message, ) + await publish_event( + request, + EVENTS.MESSAGE_CREATED, + actor=user, + subject_id=user_message['id'], + data={ + 'chat_id': chat_id, + 'role': user_message.get('role', 'user'), + 'content_preview': user_message.get('content', '')[:300], + }, + ) # Link grandparent → user message (childrenIds) grandparent_id = user_message.get('parentId') @@ -1361,6 +1413,17 @@ async def chat_completion( 'timestamp': int(time.time()), }, ) + await publish_event( + request, + EVENTS.MESSAGE_CREATED, + actor=user, + subject_id=assistant_message_id, + data={ + 'chat_id': chat_id, + 'role': 'assistant', + 'model': target_model_id, + }, + ) request.state.metadata = metadata form_data['metadata'] = metadata @@ -1953,10 +2016,25 @@ async def get_webhook_url(user=Depends(get_admin_user)): } +@app.get('/api/events') +async def get_event_catalog(user=Depends(get_admin_user)): + return { + 'schema': EVENT_VERSION, + 'events': list(EVENT_CATALOG), + } + + @app.post('/api/webhook') async def update_webhook_url(form_data: UrlForm, user=Depends(get_admin_user)): await Config.upsert({'webhook_url': form_data.url}) app.state.WEBHOOK_URL = form_data.url + await publish_event( + app, + EVENTS.CONFIG_WEBHOOK_UPDATED, + actor=user, + subject_id='webhook_url', subject_type='config', + data={'enabled': bool(form_data.url)}, + ) return {'url': form_data.url} diff --git a/backend/open_webui/routers/audio.py b/backend/open_webui/routers/audio.py index 83177483b1..9a439d8c91 100644 --- a/backend/open_webui/routers/audio.py +++ b/backend/open_webui/routers/audio.py @@ -52,6 +52,7 @@ from open_webui.env import ( ENABLE_FORWARD_USER_INFO_HEADERS, ENV, ) +from open_webui.events import EVENTS, publish_event from open_webui.models.config import Config from open_webui.utils.access_control import has_permission from open_webui.utils.auth import get_admin_user, get_verified_user @@ -294,7 +295,18 @@ async def update_audio_config(request: Request, form_data: AudioConfigUpdateForm else: request.app.state.faster_whisper_model = None - return await get_audio_config(request, user) + config = await get_audio_config(request, user) + await publish_event( + request, + EVENTS.CONFIG_UPDATED, + actor=user, + subject_id='audio', + data={ + 'tts_engine': config.get('tts', {}).get('ENGINE'), + 'stt_engine': config.get('stt', {}).get('ENGINE'), + }, + ) + return config def load_speech_pipeline(request): @@ -563,6 +575,13 @@ async def speech(request: Request, user=Depends(get_verified_user)): # Return cached result if available if file_path.is_file(): + await publish_event( + request, + EVENTS.AUDIO_SPEECH_REQUESTED, + actor=user, + subject_id=name, + data={'engine': engine, 'cached': True}, + ) return FileResponse(file_path) try: @@ -575,7 +594,20 @@ async def speech(request: Request, user=Depends(get_verified_user)): if handler is None: raise HTTPException(status_code=400, detail=f'Unsupported TTS engine: {engine}') - return await handler(request, payload, file_path, file_body_path, user) + response = await handler(request, payload, file_path, file_body_path, user) + await publish_event( + request, + EVENTS.AUDIO_SPEECH_REQUESTED, + actor=user, + subject_id=name, + data={ + 'engine': engine, + 'model': payload.get('model'), + 'input_preview': str(payload.get('input', ''))[:300], + 'cached': False, + }, + ) + return response async def _transcribe_whisper(request, file_path, languages, file_dir, id): @@ -1166,6 +1198,17 @@ async def transcription( result = await transcribe(request, file_path, metadata, user) + await publish_event( + request, + EVENTS.AUDIO_TRANSCRIPTION_REQUESTED, + actor=user, + subject_id=str(id), + data={ + 'filename': safe_name, + 'content_type': file.content_type, + 'language': language, + }, + ) return { **result, 'filename': os.path.basename(file_path), diff --git a/backend/open_webui/routers/auths.py b/backend/open_webui/routers/auths.py index d0e06fb3f5..c52d1d279c 100644 --- a/backend/open_webui/routers/auths.py +++ b/backend/open_webui/routers/auths.py @@ -18,7 +18,8 @@ from open_webui.config import ( ENABLE_PASSWORD_AUTH, OAUTH_PROVIDERS, ) -from open_webui.constants import ERROR_MESSAGES, WEBHOOK_MESSAGES +from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import ( AIOHTTP_CLIENT_SESSION_SSL, ENABLE_INITIAL_ADMIN_SIGNUP, @@ -72,7 +73,6 @@ from open_webui.utils.groups import apply_default_group_assignment from open_webui.utils.misc import parse_duration, validate_email_format from open_webui.utils.rate_limit import RateLimiter from open_webui.utils.redis import get_redis_client -from open_webui.utils.webhook import post_webhook from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession @@ -140,7 +140,12 @@ def config_updates(data: dict, key_map: dict[str, str]) -> dict: async def create_session_response( - request: Request, user, db, response: Response = None, set_cookie: bool = False + request: Request, + user, + db, + response: Response = None, + set_cookie: bool = False, + source: str = 'api', ) -> dict: """ Create JWT token and build session response for a user. @@ -177,6 +182,14 @@ async def create_session_response( ) user_permissions = await get_permissions(user.id, await Config.get('user.permissions'), db=db) + await publish_event( + request, + EVENTS.AUTH_LOGIN, + actor=user, + subject_id=user.id, subject_type='user', + source=source, + data={'auth_method': source}, + ) return { 'token': token, @@ -279,6 +292,7 @@ async def get_session_user( @router.post('/update/profile', response_model=UserProfileImageResponse) async def update_profile( + request: Request, form_data: UpdateProfileForm, session_user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), @@ -290,6 +304,13 @@ async def update_profile( db=db, ) if user: + await publish_event( + request, + EVENTS.USER_PROFILE_UPDATED, + actor=session_user, + subject_id=session_user.id, + data={'updated_fields': list(form_data.model_dump().keys())}, + ) return user else: raise HTTPException(400, detail=ERROR_MESSAGES.DEFAULT()) @@ -308,6 +329,7 @@ class UpdateTimezoneForm(BaseModel): @router.post('/update/timezone') async def update_timezone( + request: Request, form_data: UpdateTimezoneForm, session_user=Depends(get_current_user), db: AsyncSession = Depends(get_async_session), @@ -318,6 +340,13 @@ async def update_timezone( {'timezone': form_data.timezone}, db=db, ) + await publish_event( + request, + EVENTS.USER_UPDATED, + actor=session_user, + subject_id=session_user.id, + data={'updated_fields': ['timezone']}, + ) return {'status': True} else: raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED) @@ -330,6 +359,7 @@ async def update_timezone( @router.post('/update/password', response_model=bool) async def update_password( + request: Request, form_data: UpdatePasswordForm, session_user=Depends(get_current_user), db: AsyncSession = Depends(get_async_session), @@ -350,7 +380,15 @@ async def update_password( except Exception as e: raise HTTPException(400, detail=str(e)) hashed = get_password_hash(form_data.new_password) - return await Auths.update_user_password_by_id(user.id, hashed, db=db) + success = await Auths.update_user_password_by_id(user.id, hashed, db=db) + if success: + await publish_event( + request, + EVENTS.AUTH_PASSWORD_CHANGED, + actor=user, + subject_id=user.id, subject_type='user', + ) + return success else: raise HTTPException(400, detail=ERROR_MESSAGES.INCORRECT_PASSWORD) else: @@ -567,17 +605,14 @@ async def ldap_auth( db=db, ) - if await Config.get('webhook_url'): - await post_webhook( - request.app.state.WEBUI_NAME, - await Config.get('webhook_url'), - WEBHOOK_MESSAGES.USER_SIGNUP(user.name), - { - 'action': 'signup', - 'message': WEBHOOK_MESSAGES.USER_SIGNUP(user.name), - 'user': user.model_dump_json(exclude_none=True), - }, - ) + await publish_event( + request, + EVENTS.USER_CREATED, + actor=user, + subject_id=user.id, + source='ldap', + data={'role': user.role}, + ) except HTTPException: raise @@ -597,7 +632,7 @@ async def ldap_auth( except Exception as e: log.error(f'Failed to sync groups for user {user.id}: {e}') - return await create_session_response(request, user, db, response, set_cookie=True) + return await create_session_response(request, user, db, response, set_cookie=True, source='ldap') else: raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED) else: @@ -625,7 +660,10 @@ async def signin( detail=ERROR_MESSAGES.ACTION_PROHIBITED, ) + auth_source = 'password' + if WEBUI_AUTH_TRUSTED_EMAIL_HEADER: + auth_source = 'trusted_header' if WEBUI_AUTH_TRUSTED_EMAIL_HEADER not in request.headers: raise HTTPException(status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.INVALID_TRUSTED_HEADER) @@ -646,6 +684,7 @@ async def signin( str(uuid.uuid4()), name, db=db, + source='trusted_header', ) user = await Auths.authenticate_user_by_email(email, db=db) @@ -666,6 +705,7 @@ async def signin( log.warning(f'Ignoring invalid trusted role header value: {trusted_role}') elif WEBUI_AUTH == False: + auth_source = 'system' admin_email = 'admin@localhost' admin_password = 'admin' @@ -685,6 +725,7 @@ async def signin( admin_password, 'User', db=db, + source='system', ) user = await Auths.authenticate_user( @@ -715,7 +756,7 @@ async def signin( ) if user: - return await create_session_response(request, user, db, response, set_cookie=True) + return await create_session_response(request, user, db, response, set_cookie=True, source=auth_source) else: raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED) @@ -733,6 +774,7 @@ async def signup_handler( profile_image_url: str = '/user.png', *, db: AsyncSession, + source: str = 'api', ) -> UserModel: """ Core user-creation logic shared by the signup endpoint and @@ -764,24 +806,21 @@ async def signup_handler( user = await Users.get_user_by_id(user.id, db=db) await Config.upsert({'ui.enable_signup': False}) - if await Config.get('webhook_url'): - await post_webhook( - request.app.state.WEBUI_NAME, - await Config.get('webhook_url'), - WEBHOOK_MESSAGES.USER_SIGNUP(user.name), - { - 'action': 'signup', - 'message': WEBHOOK_MESSAGES.USER_SIGNUP(user.name), - 'user': user.model_dump_json(exclude_none=True), - }, - ) - await apply_default_group_assignment( await Config.get('ui.default_group_id'), user.id, db=db, ) + await publish_event( + request, + EVENTS.USER_CREATED, + actor=user, + subject_id=user.id, + source=source, + data={'role': user.role}, + ) + return user @@ -825,6 +864,14 @@ async def signup( form_data.profile_image_url, db=db, ) + await publish_event( + request, + EVENTS.AUTH_SIGNUP, + actor=user, + subject_id=user.id, + subject_type='user', + data={'email': user.email}, + ) return await create_session_response(request, user, db, response, set_cookie=True) except HTTPException: raise @@ -846,7 +893,18 @@ async def signout(request: Request, response: Response, db: AsyncSession = Depen token = request.cookies.get('token') if token: + actor = None + data = decode_token(token) + if data and data.get('id'): + actor = await Users.get_user_by_id(data['id'], db=db) await invalidate_token(request, token) + await publish_event( + request, + EVENTS.AUTH_LOGOUT, + actor=actor, + subject_id=actor.id if actor else None, + subject_type='user' if actor else None, + ) response.delete_cookie('token') response.delete_cookie('oui-session') @@ -930,6 +988,7 @@ async def signout(request: Request, response: Response, db: AsyncSession = Depen @router.delete('/oauth/sessions/{provider:path}', response_model=bool) async def delete_oauth_session_by_provider( + request: Request, provider: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), @@ -945,6 +1004,14 @@ async def delete_oauth_session_by_provider( status_code=status.HTTP_404_NOT_FOUND, detail='No OAuth session found for this provider', ) + await publish_event( + request, + EVENTS.AUTH_OAUTH_SESSION_DELETED, + actor=user, + subject_id=user.id, + subject_type='user', + data={'provider': provider}, + ) return True @@ -960,6 +1027,7 @@ async def add_user( user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session), ): + admin_user = user if not validate_email_format(form_data.email.lower()): raise HTTPException(status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.INVALID_EMAIL_FORMAT) @@ -988,6 +1056,14 @@ async def add_user( user.id, db=db, ) + await publish_event( + request, + EVENTS.USER_CREATED, + actor=admin_user, + subject_id=user.id, + source='admin', + data={'role': user.role}, + ) expires_delta = parse_duration(await Config.get('auth.jwt_expiry')) token = create_token(data={'id': user.id}, expires_delta=expires_delta) @@ -1326,6 +1402,12 @@ async def generate_api_key( success = await Users.update_user_api_key_by_id(user.id, api_key, db=db) if success: + await publish_event( + request, + EVENTS.AUTH_API_KEY_CREATED, + actor=user, + subject_id=user.id, subject_type='user', + ) return { 'api_key': api_key, } @@ -1339,7 +1421,15 @@ async def delete_api_key( request: Request, user=Depends(get_current_user), db: AsyncSession = Depends(get_async_session) ): await _check_api_key_permission(request, user, db) - return await Users.delete_user_api_key_by_id(user.id, db=db) + success = await Users.delete_user_api_key_by_id(user.id, db=db) + if success: + await publish_event( + request, + EVENTS.AUTH_API_KEY_DELETED, + actor=user, + subject_id=user.id, subject_type='user', + ) + return success # get api key @@ -1467,4 +1557,4 @@ async def token_exchange( detail='User not found. Please sign in via the web interface first.', ) - return await create_session_response(request, user, db) + return await create_session_response(request, user, db, source='oauth') diff --git a/backend/open_webui/routers/automations.py b/backend/open_webui/routers/automations.py index b7f6b05e1d..0630e83fb2 100644 --- a/backend/open_webui/routers/automations.py +++ b/backend/open_webui/routers/automations.py @@ -4,6 +4,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.automations import ( AutomationForm, @@ -175,7 +176,15 @@ async def create_new_automation( tz = user.timezone automation = await Automations.insert(user.id, form_data, next_run_ns(form_data.data.rrule, tz=tz), db=db) - return await enrich_automation(automation, db, tz=tz) + response = await enrich_automation(automation, db, tz=tz) + await publish_event( + request, + EVENTS.AUTOMATION_CREATED, + actor=user, + subject_id=automation.id, + data={'name': automation.name, 'is_active': automation.is_active}, + ) + return response ############################ @@ -225,7 +234,15 @@ async def update_automation_by_id( tz = user.timezone updated = await Automations.update_by_id(id, form_data, next_run_ns(form_data.data.rrule, tz=tz), db=db) - return await enrich_automation(updated, db, tz=tz) + response = await enrich_automation(updated, db, tz=tz) + await publish_event( + request, + EVENTS.AUTOMATION_UPDATED, + actor=user, + subject_id=updated.id, + data={'name': updated.name, 'is_active': updated.is_active}, + ) + return response ############################ @@ -244,7 +261,15 @@ async def toggle_automation_by_id( automation = await Automations.get_by_id(id, db=db) check_automation_access(automation, user) toggled = await Automations.toggle(id, next_run_ns(automation.data['rrule'], tz=user.timezone), db=db) - return await enrich_automation(toggled, db, tz=user.timezone) + response = await enrich_automation(toggled, db, tz=user.timezone) + await publish_event( + request, + EVENTS.AUTOMATION_ENABLED if toggled.is_active else EVENTS.AUTOMATION_DISABLED, + actor=user, + subject_id=toggled.id, subject_type='automation', + data={'name': toggled.name}, + ) + return response ############################ @@ -263,6 +288,13 @@ async def run_automation_by_id( automation = await Automations.get_by_id(id, db=db) check_automation_access(automation, user) asyncio.create_task(execute_automation(request.app, automation)) + await publish_event( + request, + EVENTS.AUTOMATION_RUN_STARTED, + actor=user, + subject_id=automation.id, + data={'name': automation.name}, + ) return await enrich_automation(automation, db, tz=user.timezone) @@ -282,7 +314,16 @@ async def delete_automation_by_id( automation = await Automations.get_by_id(id, db=db) check_automation_access(automation, user) await AutomationRuns.delete_by_automation(id, db=db) - return await Automations.delete(id, db=db) + result = await Automations.delete(id, db=db) + if result: + await publish_event( + request, + EVENTS.AUTOMATION_DELETED, + actor=user, + subject_id=id, + data={'name': automation.name}, + ) + return result ############################ diff --git a/backend/open_webui/routers/calendar.py b/backend/open_webui/routers/calendar.py index 48327eb7ac..d640b5f34e 100644 --- a/backend/open_webui/routers/calendar.py +++ b/backend/open_webui/routers/calendar.py @@ -4,6 +4,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.models.access_grants import AccessGrants from open_webui.models.calendar import ( CalendarEventAttendees, @@ -125,7 +126,15 @@ async def create_calendar(request: Request, form_data: CalendarForm, user: UserM form_data.access_grants, 'sharing.public_calendars', ) - return await Calendars.insert_new_calendar(user.id, form_data) + calendar = await Calendars.insert_new_calendar(user.id, form_data) + await publish_event( + request, + EVENTS.CALENDAR_CREATED, + actor=user, + subject_id=calendar.id, + data={'name': calendar.name}, + ) + return calendar #################### @@ -266,7 +275,15 @@ async def get_events( async def create_event(request: Request, form_data: CalendarEventForm, user: UserModel = Depends(get_verified_user)): await check_calendar_permission(request, user) await _check_calendar_access(form_data.calendar_id, user, 'write') - return await CalendarEvents.insert_new_event(user.id, form_data) + event = await CalendarEvents.insert_new_event(user.id, form_data) + await publish_event( + request, + EVENTS.CALENDAR_EVENT_CREATED, + actor=user, + subject_id=event.id, + data={'calendar_id': event.calendar_id, 'title': event.title}, + ) + return event @router.get('/events/search', response_model=CalendarEventListResponse) @@ -313,6 +330,13 @@ async def update_event( updated = await CalendarEvents.update_event_by_id(event_id, form_data) if not updated: raise HTTPException(status_code=500, detail='Failed to update') + await publish_event( + request, + EVENTS.CALENDAR_EVENT_UPDATED, + actor=user, + subject_id=updated.id, + data={'calendar_id': updated.calendar_id, 'title': updated.title}, + ) return updated @@ -328,6 +352,13 @@ async def delete_event(request: Request, event_id: str, user: UserModel = Depend result = await CalendarEvents.delete_event_by_id(event_id) if not result: raise HTTPException(status_code=500, detail='Failed to delete') + await publish_event( + request, + EVENTS.CALENDAR_EVENT_DELETED, + actor=user, + subject_id=event_id, + data={'calendar_id': event.calendar_id, 'title': event.title}, + ) return {'status': True} @@ -343,6 +374,13 @@ async def rsvp_event( result = await CalendarEventAttendees.update_rsvp(event_id, user.id, form_data.status) if not result: raise HTTPException(status_code=404, detail='Not an attendee of this event') + await publish_event( + request, + EVENTS.CALENDAR_EVENT_RSVP_UPDATED, + actor=user, + subject_id=event_id, + data={'status': result.status}, + ) return {'status': True, 'rsvp': result.status} @@ -386,6 +424,13 @@ async def update_calendar( updated = await Calendars.update_calendar_by_id(calendar_id, form_data) if not updated: raise HTTPException(status_code=500, detail='Failed to update') + await publish_event( + request, + EVENTS.CALENDAR_UPDATED, + actor=user, + subject_id=updated.id, + data={'name': updated.name}, + ) return updated @@ -410,6 +455,13 @@ async def delete_calendar(request: Request, calendar_id: str, user: UserModel = result = await Calendars.delete_calendar_by_id(calendar_id) if not result: raise HTTPException(status_code=500, detail='Failed to delete') + await publish_event( + request, + EVENTS.CALENDAR_DELETED, + actor=user, + subject_id=calendar_id, + data={'name': cal.name}, + ) return {'status': True} @@ -419,4 +471,11 @@ async def set_default_calendar(request: Request, calendar_id: str, user: UserMod cal = await Calendars.set_default_calendar(user.id, calendar_id) if not cal: raise HTTPException(status_code=404, detail='Calendar not found') + await publish_event( + request, + EVENTS.CALENDAR_DEFAULT_UPDATED, + actor=user, + subject_id=cal.id, + data={'name': cal.name}, + ) return cal diff --git a/backend/open_webui/routers/channels.py b/backend/open_webui/routers/channels.py index 08e44ac763..c3f6a93dbc 100644 --- a/backend/open_webui/routers/channels.py +++ b/backend/open_webui/routers/channels.py @@ -8,6 +8,7 @@ from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, from fastapi.responses import FileResponse, Response, StreamingResponse from open_webui.config import ENABLE_ADMIN_CHAT_ACCESS, ENABLE_ADMIN_EXPORT from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import STATIC_DIR from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_public_write_access_grant @@ -315,6 +316,13 @@ async def create_new_channel( await enter_room_for_users(f'channel:{existing_channel.id}', participant_ids) await Channels.update_member_active_status(existing_channel.id, user.id, True, db=db) + await publish_event( + request, + EVENTS.CHANNEL_MEMBER_ACTIVE_UPDATED, + actor=user, + subject_id=existing_channel.id, + data={'is_active': True}, + ) return ChannelModel(**existing_channel.model_dump()) channel = await Channels.insert_new_channel(form_data, user.id, db=db) @@ -329,6 +337,13 @@ async def create_new_channel( ) await enter_room_for_users(f'channel:{channel.id}', participant_ids) + await publish_event( + request, + EVENTS.CHANNEL_CREATED, + actor=user, + subject_id=channel.id, + data={'type': channel.type, 'name': channel.name}, + ) return ChannelModel(**channel.model_dump()) else: raise Exception('Error creating channel') @@ -566,6 +581,13 @@ async def update_is_active_member_by_id_and_user_id( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) await Channels.update_member_active_status(channel.id, user.id, form_data.is_active, db=db) + await publish_event( + request, + EVENTS.CHANNEL_MEMBER_ACTIVE_UPDATED, + actor=user, + subject_id=channel.id, + data={'is_active': form_data.is_active}, + ) return True @@ -600,6 +622,13 @@ async def add_members_by_id( channel.id, user.id, form_data.user_ids, form_data.group_ids, db=db ) + await publish_event( + request, + EVENTS.CHANNEL_MEMBER_ADDED, + actor=user, + subject_id=channel.id, + data={'user_ids': form_data.user_ids, 'group_ids': form_data.group_ids}, + ) return memberships except Exception as e: log.exception(e) @@ -635,6 +664,13 @@ async def remove_members_by_id( try: deleted = await Channels.remove_members_from_channel(channel.id, form_data.user_ids, db=db) + await publish_event( + request, + EVENTS.CHANNEL_MEMBER_REMOVED, + actor=user, + subject_id=channel.id, + data={'user_ids': form_data.user_ids}, + ) return deleted except Exception as e: log.exception(e) @@ -673,6 +709,13 @@ async def update_channel_by_id( try: channel = await Channels.update_channel_by_id(id, form_data, db=db) + await publish_event( + request, + EVENTS.CHANNEL_UPDATED, + actor=user, + subject_id=id, + data={'name': channel.name, 'type': channel.type}, + ) return ChannelModel(**channel.model_dump()) except Exception as e: log.exception(e) @@ -702,6 +745,13 @@ async def delete_channel_by_id( try: await Channels.delete_channel_by_id(id, db=db) + await publish_event( + request, + EVENTS.CHANNEL_DELETED, + actor=user, + subject_id=id, + data={'name': channel.name, 'type': channel.type}, + ) return True except Exception as e: log.exception(e) @@ -1160,6 +1210,16 @@ async def post_new_message( background_tasks.add_task(background_handler) + await publish_event( + request, + EVENTS.MESSAGE_CREATED, + actor=user, + subject_id=message.id, + data={ + 'channel_id': channel.id, + 'content_preview': message.content[:300], + }, + ) return message except HTTPException as e: @@ -1287,6 +1347,13 @@ async def pin_channel_message( await Messages.update_is_pinned_by_id(message_id, form_data.is_pinned, user.id, db=db) message = await Messages.get_message_by_id(message_id, db=db) message_user = await Users.get_user_by_id(message.user_id, db=db) + await publish_event( + request, + EVENTS.MESSAGE_PINNED if form_data.is_pinned else EVENTS.MESSAGE_UNPINNED, + actor=user, + subject_id=message_id, subject_type='message', + data={'channel_id': id}, + ) return MessageUserResponse( **{ **message.model_dump(), @@ -1416,6 +1483,13 @@ async def update_message_by_id( to=f'channel:{channel.id}', ) + await publish_event( + request, + EVENTS.MESSAGE_UPDATED, + actor=user, + subject_id=message_id, + data={'channel_id': id, 'content_preview': form_data.content[:300]}, + ) return MessageModel(**message.model_dump()) except Exception as e: log.exception(e) @@ -1487,6 +1561,13 @@ async def add_reaction_to_message( to=f'channel:{channel.id}', ) + await publish_event( + request, + EVENTS.MESSAGE_REACTION_ADDED, + actor=user, + subject_id=message_id, + data={'channel_id': id, 'reaction': form_data.name}, + ) return True except Exception as e: log.exception(e) @@ -1555,6 +1636,13 @@ async def remove_reaction_by_id_and_user_id_and_name( to=f'channel:{channel.id}', ) + await publish_event( + request, + EVENTS.MESSAGE_REACTION_REMOVED, + actor=user, + subject_id=message_id, + data={'channel_id': id, 'reaction': form_data.name}, + ) return True except Exception as e: log.exception(e) @@ -1646,6 +1734,13 @@ async def delete_message_by_id( to=f'channel:{channel.id}', ) + await publish_event( + request, + EVENTS.MESSAGE_DELETED, + actor=user, + subject_id=message_id, + data={'channel_id': id}, + ) return True except Exception as e: log.exception(e) @@ -1731,6 +1826,13 @@ async def create_channel_webhook( if not webhook: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT()) + await publish_event( + request, + EVENTS.CHANNEL_WEBHOOK_CREATED, + actor=user, + subject_id=webhook.id, + data={'channel_id': id, 'name': webhook.name}, + ) return webhook @@ -1760,6 +1862,13 @@ async def update_channel_webhook( if not updated: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT()) + await publish_event( + request, + EVENTS.CHANNEL_WEBHOOK_UPDATED, + actor=user, + subject_id=webhook_id, + data={'channel_id': id, 'name': updated.name}, + ) return updated @@ -1784,7 +1893,16 @@ async def delete_channel_webhook( if not webhook or webhook.channel_id != id: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) - return await Channels.delete_webhook_by_id(webhook_id, db=db) + deleted = await Channels.delete_webhook_by_id(webhook_id, db=db) + if deleted: + await publish_event( + request, + EVENTS.CHANNEL_WEBHOOK_DELETED, + actor=user, + subject_id=webhook_id, + data={'channel_id': id}, + ) + return deleted ############################ @@ -1867,4 +1985,12 @@ async def post_webhook_message( to=f'channel:{channel.id}', ) + await publish_event( + request, + EVENTS.MESSAGE_CREATED, + actor={'id': webhook.id, 'name': webhook.name, 'role': 'webhook', 'type': 'webhook'}, + subject_id=message.id, + source='channel_webhook', + data={'channel_id': channel.id, 'content_preview': form_data.content[:300]}, + ) return {'success': True, 'message_id': message.id} diff --git a/backend/open_webui/routers/chats.py b/backend/open_webui/routers/chats.py index 6d1e8f1b41..537207f849 100644 --- a/backend/open_webui/routers/chats.py +++ b/backend/open_webui/routers/chats.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.responses import StreamingResponse from open_webui.config import ENABLE_ADMIN_CHAT_ACCESS, ENABLE_ADMIN_EXPORT from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants from open_webui.models.config import Config @@ -544,6 +545,13 @@ async def delete_all_user_chats( ) result = await Chats.delete_chats_by_user_id(user.id, db=db) + if result: + await publish_event( + request, + EVENTS.CHAT_DELETED_ALL, + actor=user, + subject_id=user.id, subject_type='user', + ) return result @@ -590,6 +598,7 @@ async def get_user_chat_list_by_user_id( @router.post('/new', response_model=ChatResponse | None) async def create_new_chat( + request: Request, form_data: ChatForm, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), @@ -611,6 +620,13 @@ async def create_new_chat( try: chat = await Chats.insert_new_chat(str(uuid4()), user.id, form_data, db=db) + await publish_event( + request, + EVENTS.CHAT_CREATED, + actor=user, + subject_id=chat.id, + data={'title': chat.title, 'folder_id': chat.folder_id}, + ) return ChatResponse(**chat.model_dump()) except Exception as e: log.exception(e) @@ -633,6 +649,13 @@ async def import_chats( try: chats = await Chats.import_chats(user.id, form_data.chats, db=db) + await publish_event( + request, + EVENTS.CHAT_IMPORTED, + actor=user, + subject_type='chat.import', + data={'count': len(chats), 'chat_ids': [chat.id for chat in chats]}, + ) return chats except Exception as e: log.exception(e) @@ -889,8 +912,13 @@ async def get_archived_session_user_chat_count( @router.post('/archive/all', response_model=bool) -async def archive_all_chats(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): - return await Chats.archive_all_chats_by_user_id(user.id, db=db) +async def archive_all_chats( + request: Request, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): + result = await Chats.archive_all_chats_by_user_id(user.id, db=db) + if result: + await publish_event(request, EVENTS.CHAT_ARCHIVED, actor=user, subject_id=user.id, subject_type='user') + return result ############################ @@ -899,8 +927,13 @@ async def archive_all_chats(user=Depends(get_verified_user), db: AsyncSession = @router.post('/unarchive/all', response_model=bool) -async def unarchive_all_chats(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): - return await Chats.unarchive_all_chats_by_user_id(user.id, db=db) +async def unarchive_all_chats( + request: Request, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): + result = await Chats.unarchive_all_chats_by_user_id(user.id, db=db) + if result: + await publish_event(request, EVENTS.CHAT_UNARCHIVED, actor=user, subject_id=user.id, subject_type='user') + return result ############################ @@ -909,7 +942,9 @@ async def unarchive_all_chats(user=Depends(get_verified_user), db: AsyncSession @router.delete('/share/all', response_model=bool) -async def unshare_all_chats(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): +async def unshare_all_chats( + request: Request, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): # Collect chat_ids that have shares so we can clear share_id and access grants shared_list = await SharedChats.get_by_user_id(user.id, db=db) chat_ids = [s.chat_id for s in shared_list] @@ -922,6 +957,14 @@ async def unshare_all_chats(user=Depends(get_verified_user), db: AsyncSession = await Chats.update_chat_share_id_by_id(chat_id, None, db=db) await AccessGrants.set_access_grants('shared_chat', chat_id, [], db=db) + if result: + await publish_event( + request, + EVENTS.CHAT_UNSHARED, + actor=user, + subject_id=user.id, subject_type='user', + data={'count': len(chat_ids), 'chat_ids': chat_ids}, + ) return result @@ -1067,7 +1110,16 @@ async def compact_chat_by_id( if not model_id: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='No model found for context compaction.') - return await compact_chat_branch(request, user, chat, model_id, request.app.state.MODELS) + result = await compact_chat_branch(request, user, chat, model_id, request.app.state.MODELS) + if result.get('compacted'): + await publish_event( + request, + EVENTS.CHAT_COMPACTED, + actor=user, + subject_id=id, + data={'dropped_messages': result.get('dropped_messages')}, + ) + return result ############################ @@ -1115,6 +1167,7 @@ async def get_chat_by_id(id: str, user=Depends(get_verified_user), db: AsyncSess @router.post('/{id}', response_model=ChatResponse | None) async def update_chat_by_id( + request: Request, id: str, form_data: ChatForm, user=Depends(get_verified_user), @@ -1143,6 +1196,13 @@ async def update_chat_by_id( if messages: await Chats.reconcile_messages_by_chat_id(id, user.id, messages) + await publish_event( + request, + EVENTS.CHAT_UPDATED, + actor=user, + subject_id=id, + data={'title': chat.title}, + ) return ChatResponse(**chat.model_dump()) else: raise HTTPException( @@ -1160,6 +1220,7 @@ class MessageForm(BaseModel): @router.post('/{id}/messages/{message_id}', response_model=ChatResponse | None) async def update_chat_message_by_id( + request: Request, id: str, message_id: str, form_data: MessageForm, @@ -1209,6 +1270,13 @@ async def update_chat_message_by_id( } ) + await publish_event( + request, + EVENTS.MESSAGE_UPDATED, + actor=user, + subject_id=message_id, + data={'chat_id': id, 'content_preview': form_data.content[:300]}, + ) return ChatResponse(**chat.model_dump()) @@ -1222,6 +1290,7 @@ class EventForm(BaseModel): @router.post('/{id}/messages/{message_id}/event', response_model=bool | None) async def send_chat_message_event_by_id( + request: Request, id: str, message_id: str, form_data: EventForm, @@ -1255,6 +1324,13 @@ async def send_chat_message_event_by_id( await event_emitter(form_data.model_dump()) else: return False + await publish_event( + request, + EVENTS.MESSAGE_EVENT_RECEIVED, + actor=user, + subject_id=message_id, + data={'chat_id': id, 'event_type': form_data.type}, + ) return True except Exception: return False @@ -1287,6 +1363,14 @@ async def delete_chat_by_id( result = await Chats.delete_chat_by_id(id, db=db) + if result: + await publish_event( + request, + EVENTS.CHAT_DELETED, + actor=user, + subject_id=id, + data={'owner_id': chat.user_id}, + ) return result else: if not await has_permission(user.id, 'chat.delete', await Config.get('user.permissions')): @@ -1304,6 +1388,14 @@ async def delete_chat_by_id( await Chats.delete_orphan_tags_for_user(chat.meta.get('tags', []), user.id, threshold=1, db=db) result = await Chats.delete_chat_by_id_and_user_id(id, user.id, db=db) + if result: + await publish_event( + request, + EVENTS.CHAT_DELETED, + actor=user, + subject_id=id, + data={'owner_id': user.id}, + ) return result @@ -1329,10 +1421,18 @@ async def get_pinned_status_by_id( @router.post('/{id}/pin', response_model=ChatResponse | None) -async def pin_chat_by_id(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): +async def pin_chat_by_id( + request: Request, id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db) if chat: chat = await Chats.toggle_chat_pinned_by_id(id, db=db) + await publish_event( + request, + EVENTS.CHAT_PINNED if chat.pinned else EVENTS.CHAT_UNPINNED, + actor=user, + subject_id=id, subject_type='chat', + ) return chat else: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT()) @@ -1383,6 +1483,13 @@ async def clone_chat_by_id( if chats: chat = chats[0] + await publish_event( + request, + EVENTS.CHAT_CLONED, + actor=user, + subject_id=chat.id, + data={'original_chat_id': id}, + ) return ChatResponse(**chat.model_dump()) else: raise HTTPException( @@ -1493,6 +1600,12 @@ async def archive_chat_by_id( # Unarchived — ensure tag rows exist await Tags.ensure_tags_exist(tag_ids, user.id, db=db) + await publish_event( + request, + EVENTS.CHAT_ARCHIVED if chat.archived else EVENTS.CHAT_UNARCHIVED, + actor=user, + subject_id=id, subject_type='chat', + ) return ChatResponse(**chat.model_dump()) else: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT()) @@ -1522,6 +1635,13 @@ async def share_chat_by_id( shared = await SharedChats.update(chat.share_id, db=db) if shared: chat = await Chats.get_chat_by_id(id, db=db) + await publish_event( + request, + EVENTS.CHAT_SHARED, + actor=user, + subject_id=id, + data={'share_id': chat.share_id, 'updated': True}, + ) return ChatResponse(**chat.model_dump()) # Create a new share @@ -1533,6 +1653,13 @@ async def share_chat_by_id( if not chat: raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=ERROR_MESSAGES.DEFAULT()) + await publish_event( + request, + EVENTS.CHAT_SHARED, + actor=user, + subject_id=id, + data={'share_id': shared.id}, + ) return ChatResponse(**chat.model_dump()) @@ -1541,7 +1668,7 @@ async def share_chat_by_id( @router.delete('/{id}/share', response_model=bool | None) async def delete_shared_chat_by_id( - id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) + request: Request, id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) ): chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db) if not chat: @@ -1554,6 +1681,13 @@ async def delete_shared_chat_by_id( await Chats.update_chat_share_id_by_id(id, None, db=db) await AccessGrants.set_access_grants('shared_chat', id, [], db=db) + await publish_event( + request, + EVENTS.CHAT_UNSHARED, + actor=user, + subject_id=id, + data={'share_id': chat.share_id}, + ) return True @@ -1641,6 +1775,7 @@ class ChatFolderIdForm(BaseModel): @router.post('/{id}/folder', response_model=ChatResponse | None) async def update_chat_folder_id_by_id( + request: Request, id: str, form_data: ChatFolderIdForm, user=Depends(get_verified_user), @@ -1661,6 +1796,13 @@ async def update_chat_folder_id_by_id( ) chat = await Chats.update_chat_folder_id_by_id_and_user_id(id, user.id, form_data.folder_id, db=db) + await publish_event( + request, + EVENTS.CHAT_FOLDER_UPDATED, + actor=user, + subject_id=id, + data={'folder_id': form_data.folder_id}, + ) return ChatResponse(**chat.model_dump()) else: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=ERROR_MESSAGES.DEFAULT()) @@ -1688,6 +1830,7 @@ async def get_chat_tags_by_id(id: str, user=Depends(get_verified_user), db: Asyn @router.post('/{id}/tags', response_model=list[TagModel]) async def add_tag_by_id_and_tag_name( + request: Request, id: str, form_data: TagForm, user=Depends(get_verified_user), @@ -1706,6 +1849,13 @@ async def add_tag_by_id_and_tag_name( if tag_id not in tags: await Chats.add_chat_tag_by_id_and_user_id_and_tag_name(id, user.id, form_data.name, db=db) + await publish_event( + request, + EVENTS.CHAT_TAG_ADDED, + actor=user, + subject_id=id, + data={'tag': form_data.name}, + ) chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db) tags = chat.meta.get('tags', []) @@ -1721,6 +1871,7 @@ async def add_tag_by_id_and_tag_name( @router.delete('/{id}/tags', response_model=list[TagModel]) async def delete_tag_by_id_and_tag_name( + request: Request, id: str, form_data: TagForm, user=Depends(get_verified_user), @@ -1729,6 +1880,13 @@ async def delete_tag_by_id_and_tag_name( chat = await Chats.get_chat_by_id_and_user_id(id, user.id, db=db) if chat: await Chats.delete_tag_by_id_and_user_id_and_tag_name(id, user.id, form_data.name, db=db) + await publish_event( + request, + EVENTS.CHAT_TAG_REMOVED, + actor=user, + subject_id=id, + data={'tag': form_data.name}, + ) if await Chats.count_chats_by_tag_name_and_user_id(form_data.name, user.id, db=db) == 0: await Tags.delete_tag_by_name_and_user_id(form_data.name, user.id, db=db) diff --git a/backend/open_webui/routers/configs.py b/backend/open_webui/routers/configs.py index 32f2ac0379..a210b72894 100644 --- a/backend/open_webui/routers/configs.py +++ b/backend/open_webui/routers/configs.py @@ -8,8 +8,9 @@ import aiohttp from fastapi import APIRouter, Depends, HTTPException, Request from mcp.shared.auth import OAuthMetadata from open_webui.config import BannerModel -from open_webui.models.config import Config from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT +from open_webui.events import EVENTS, publish_event +from open_webui.models.config import Config from open_webui.models.oauth_sessions import OAuthSessions from open_webui.utils.auth import get_admin_user, get_verified_user from open_webui.utils.headers import get_custom_headers @@ -89,6 +90,13 @@ class ImportConfigForm(BaseModel): @router.post('/import', response_model=dict) async def import_config(request: Request, form_data: ImportConfigForm, user=Depends(get_admin_user)): await Config.upsert(form_data.config) + await publish_event( + request, + EVENTS.CONFIG_IMPORTED, + actor=user, + subject_id='import', + data={'keys': list(form_data.config.keys())}, + ) return await Config.get_all() @@ -129,7 +137,15 @@ async def set_connections_config( user=Depends(get_admin_user), ): await Config.upsert(config_updates(form_data.model_dump(), CONNECTIONS_CONFIG_KEYS)) - return await get_config_values(CONNECTIONS_CONFIG_KEYS) + values = await get_config_values(CONNECTIONS_CONFIG_KEYS) + await publish_event( + request, + EVENTS.CONFIG_CONNECTIONS_UPDATED, + actor=user, + subject_id='connections', subject_type='config', + data=values, + ) + return values class OAuthClientRegistrationForm(BaseModel): @@ -253,6 +269,13 @@ async def set_tool_servers_config( log.debug(f'Failed to add OAuth client for MCP tool server: {e}') continue + await publish_event( + request, + EVENTS.CONFIG_TOOL_SERVERS_UPDATED, + actor=user, + subject_id='tool_server.connections', subject_type='config', + data={'count': len(connections), 'types': [connection.get('type', 'openapi') for connection in connections]}, + ) return {'TOOL_SERVER_CONNECTIONS': connections} @@ -298,6 +321,13 @@ async def set_terminal_servers_config( await set_terminal_servers(request) + await publish_event( + request, + EVENTS.CONFIG_TERMINAL_SERVERS_UPDATED, + actor=user, + subject_id='terminal_server.connections', subject_type='config', + data={'count': len(connections)}, + ) return {'TERMINAL_SERVER_CONNECTIONS': connections} @@ -659,7 +689,20 @@ async def set_code_execution_config( request: Request, form_data: CodeInterpreterConfigForm, user=Depends(get_admin_user) ): await Config.upsert(config_updates(form_data.model_dump(), CODE_EXECUTION_CONFIG_KEYS)) - return await get_config_values(CODE_EXECUTION_CONFIG_KEYS) + values = await get_config_values(CODE_EXECUTION_CONFIG_KEYS) + await publish_event( + request, + EVENTS.CONFIG_CODE_EXECUTION_UPDATED, + actor=user, + subject_id='code_execution', subject_type='config', + data={ + 'code_execution_enabled': values.get('ENABLE_CODE_EXECUTION'), + 'code_execution_engine': values.get('CODE_EXECUTION_ENGINE'), + 'code_interpreter_enabled': values.get('ENABLE_CODE_INTERPRETER'), + 'code_interpreter_engine': values.get('CODE_INTERPRETER_ENGINE'), + }, + ) + return values ############################ @@ -688,7 +731,19 @@ async def get_models_config(request: Request, user=Depends(get_admin_user)): @router.post('/models', response_model=ModelsConfigForm) async def set_models_config(request: Request, form_data: ModelsConfigForm, user=Depends(get_admin_user)): await Config.upsert(config_updates(form_data.model_dump(), MODELS_CONFIG_KEYS)) - return await get_config_values(MODELS_CONFIG_KEYS) + values = await get_config_values(MODELS_CONFIG_KEYS) + await publish_event( + request, + EVENTS.CONFIG_MODELS_UPDATED, + actor=user, + subject_id='models', subject_type='config', + data={ + 'default_models': values.get('DEFAULT_MODELS'), + 'default_pinned_models': values.get('DEFAULT_PINNED_MODELS'), + 'model_order_count': len(values.get('MODEL_ORDER_LIST') or []), + }, + ) + return values class PromptSuggestion(BaseModel): @@ -708,7 +763,15 @@ async def set_default_suggestions( ): data = form_data.model_dump() await Config.upsert({'ui.prompt_suggestions': data['suggestions']}) - return await Config.get('ui.prompt_suggestions') + suggestions = await Config.get('ui.prompt_suggestions') + await publish_event( + request, + EVENTS.CONFIG_SUGGESTIONS_UPDATED, + actor=user, + subject_id='ui.prompt_suggestions', subject_type='config', + data={'count': len(suggestions or [])}, + ) + return suggestions ############################ @@ -728,7 +791,15 @@ async def set_banners( ): data = form_data.model_dump() await Config.upsert({'ui.banners': data['banners']}) - return await Config.get('ui.banners') + banners = await Config.get('ui.banners') + await publish_event( + request, + EVENTS.CONFIG_BANNERS_UPDATED, + actor=user, + subject_id='ui.banners', subject_type='config', + data={'count': len(banners or [])}, + ) + return banners @router.get('/banners', response_model=list[BannerModel]) diff --git a/backend/open_webui/routers/evaluations.py b/backend/open_webui/routers/evaluations.py index 5f0c48af60..f1247d5a00 100644 --- a/backend/open_webui/routers/evaluations.py +++ b/backend/open_webui/routers/evaluations.py @@ -4,6 +4,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.concurrency import run_in_threadpool from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.config import Config from open_webui.models.feedbacks import ( @@ -291,7 +292,19 @@ async def update_config( if form_data.EVALUATION_ARENA_MODELS is not None: updates['evaluation.arena.models'] = form_data.EVALUATION_ARENA_MODELS await Config.upsert(updates) - return await get_config_values(EVALUATION_CONFIG_KEYS) + values = await get_config_values(EVALUATION_CONFIG_KEYS) + await publish_event( + request, + EVENTS.CONFIG_UPDATED, + actor=user, + subject_id='evaluation', + data={ + 'keys': list(updates.keys()), + 'arena_enabled': values.get('ENABLE_EVALUATION_ARENA_MODELS'), + 'arena_model_count': len(values.get('EVALUATION_ARENA_MODELS') or []), + }, + ) + return values @router.get('/feedbacks/models', response_model=list[str]) @@ -305,8 +318,19 @@ async def get_all_feedback_ids(user=Depends(get_admin_user), db: AsyncSession = @router.delete('/feedbacks/all') -async def delete_all_feedbacks(user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def delete_all_feedbacks( + request: Request, + user=Depends(get_admin_user), + db: AsyncSession = Depends(get_async_session), +): success = await Feedbacks.delete_all_feedbacks(db=db) + if success: + await publish_event( + request, + EVENTS.FEEDBACK_DELETED_ALL, + actor=user, + subject_id='all', + ) return success @@ -338,8 +362,19 @@ async def get_user_feedbacks( @router.delete('/feedbacks', response_model=bool) -async def delete_feedbacks(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): +async def delete_feedbacks( + request: Request, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): success = await Feedbacks.delete_feedbacks_by_user_id(user.id, db=db) + if success: + await publish_event( + request, + EVENTS.FEEDBACK_DELETED_ALL, + actor=user, + subject_id=user.id, subject_type='user', + ) return success @@ -383,6 +418,13 @@ async def create_feedback( detail=ERROR_MESSAGES.DEFAULT(), ) + await publish_event( + request, + EVENTS.FEEDBACK_CREATED, + actor=user, + subject_id=feedback.id, + data={'rating': getattr(feedback, 'rating', None)}, + ) return feedback @@ -401,6 +443,7 @@ async def get_feedback_by_id(id: str, user=Depends(get_verified_user), db: Async @router.post('/feedback/{id}', response_model=FeedbackModel) async def update_feedback_by_id( + request: Request, id: str, form_data: FeedbackForm, user=Depends(get_verified_user), @@ -414,12 +457,22 @@ async def update_feedback_by_id( if not feedback: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) + await publish_event( + request, + EVENTS.FEEDBACK_UPDATED, + actor=user, + subject_id=feedback.id, + data={'rating': getattr(feedback, 'rating', None)}, + ) return feedback @router.delete('/feedback/{id}') async def delete_feedback_by_id( - id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) + request: Request, + id: str, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), ): if user.role == 'admin': success = await Feedbacks.delete_feedback_by_id(id=id, db=db) @@ -429,4 +482,10 @@ async def delete_feedback_by_id( if not success: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) + await publish_event( + request, + EVENTS.FEEDBACK_DELETED, + actor=user, + subject_id=id, + ) return success diff --git a/backend/open_webui/routers/files.py b/backend/open_webui/routers/files.py index cf3be4a08a..26114a8276 100644 --- a/backend/open_webui/routers/files.py +++ b/backend/open_webui/routers/files.py @@ -23,6 +23,7 @@ from fastapi import ( from fastapi.responses import FileResponse, StreamingResponse from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL, STORAGE_LOCAL_CACHE, STORAGE_PROVIDER, UPLOAD_DIR from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_db_context, get_async_session from open_webui.models.access_grants import AccessGrants from open_webui.models.channels import Channels @@ -247,7 +248,7 @@ async def upload_file( user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), ): - return await upload_file_handler( + result = await upload_file_handler( request, file=file, metadata=metadata, @@ -258,6 +259,29 @@ async def upload_file( db=db, ) + if isinstance(result, dict): + result_id = result.get('id') + result_filename = result.get('filename') + result_meta = result.get('meta') or {} + else: + result_id = result.id + result_filename = result.filename + result_meta = result.meta or {} + + result_content_type = ( + result_meta.get('content_type') + if isinstance(result_meta, dict) + else getattr(result_meta, 'content_type', None) + ) + await publish_event( + request, + EVENTS.FILE_UPLOADED, + actor=user, + subject_id=result_id, + data={'filename': result_filename, 'content_type': result_content_type}, + ) + return result + async def upload_file_handler( request: Request, @@ -483,7 +507,7 @@ async def count_files( @router.delete('/all') -async def delete_all_files(user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def delete_all_files(request: Request, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): result = await Files.delete_all_files(db=db) if result: try: @@ -496,6 +520,7 @@ async def delete_all_files(user=Depends(get_admin_user), db: AsyncSession = Depe status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT('Error deleting files'), ) + await publish_event(request, EVENTS.FILE_DELETED_ALL, actor=user, subject_type='file') return {'message': 'All files deleted successfully'} else: raise HTTPException( @@ -669,6 +694,13 @@ async def update_file_data_content_by_id( except Exception as e: log.warning(f'Failed to update knowledge {knowledge.id} after content change for file {id}: {e}') + await publish_event( + request, + EVENTS.FILE_CONTENT_UPDATED, + actor=user, + subject_id=id, + data={'content_preview': form_data.content[:300]}, + ) return {'content': file.data.get('content', '')} else: raise HTTPException( @@ -858,6 +890,7 @@ class FileRenameForm(BaseModel): @router.post('/{id}/rename') async def rename_file_by_id( + request: Request, id: str, form_data: FileRenameForm, user=Depends(get_verified_user), @@ -874,6 +907,13 @@ async def rename_file_by_id( if file.user_id == user.id or user.role == 'admin' or await has_access_to_file(id, 'write', user, db=db): result = await Files.update_file_name_by_id(id, form_data.filename, db=db) if result: + await publish_event( + request, + EVENTS.FILE_RENAMED, + actor=user, + subject_id=id, + data={'filename': form_data.filename}, + ) return result else: raise HTTPException( @@ -893,7 +933,9 @@ async def rename_file_by_id( @router.delete('/{id}') -async def delete_file_by_id(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): +async def delete_file_by_id( + request: Request, id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): file = await Files.get_file_by_id(id, db=db) if not file: @@ -928,6 +970,13 @@ async def delete_file_by_id(id: str, user=Depends(get_verified_user), db: AsyncS status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT('Error deleting files'), ) + await publish_event( + request, + EVENTS.FILE_DELETED, + actor=user, + subject_id=id, + data={'filename': file.filename}, + ) return {'message': 'File deleted successfully'} else: raise HTTPException( diff --git a/backend/open_webui/routers/folders.py b/backend/open_webui/routers/folders.py index 43d3a11498..71ee9b9cf0 100644 --- a/backend/open_webui/routers/folders.py +++ b/backend/open_webui/routers/folders.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile from fastapi.responses import FileResponse, StreamingResponse from open_webui.config import UPLOAD_DIR from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.config import Config from open_webui.models.chats import Chats @@ -131,6 +132,13 @@ async def create_folder( # Create as the folder owner's subfolder (keep tree consistent) try: folder = await Folders.insert_new_folder(parent.user_id, form_data, form_data.parent_id, db=db) + await publish_event( + request, + EVENTS.FOLDER_CREATED, + actor=user, + subject_id=folder.id, + data={'name': folder.name, 'parent_id': folder.parent_id, 'owner_id': folder.user_id}, + ) return folder except Exception as e: log.exception(e) @@ -141,6 +149,13 @@ async def create_folder( try: folder = await Folders.insert_new_folder(user.id, form_data, form_data.parent_id, db=db) + await publish_event( + request, + EVENTS.FOLDER_CREATED, + actor=user, + subject_id=folder.id, + data={'name': folder.name, 'parent_id': folder.parent_id, 'owner_id': folder.user_id}, + ) return folder except Exception as e: log.exception(e) @@ -283,6 +298,13 @@ async def update_folder_name_by_id( try: folder = await Folders.update_folder_by_id_and_user_id(id, folder.user_id, form_data, db=db) + await publish_event( + request, + EVENTS.FOLDER_UPDATED, + actor=user, + subject_id=id, + data={'name': folder.name}, + ) return folder except Exception as e: log.exception(e) @@ -325,6 +347,13 @@ async def update_folder_parent_id_by_id( try: folder = await Folders.update_folder_parent_id_by_id_and_user_id(id, user.id, form_data.parent_id, db=db) + await publish_event( + request, + EVENTS.FOLDER_PARENT_UPDATED, + actor=user, + subject_id=id, + data={'parent_id': form_data.parent_id}, + ) return folder except Exception as e: log.exception(e) @@ -423,6 +452,13 @@ async def update_folder_access_by_id( await AccessGrants.set_access_grants('folder', id, form_data.access_grants, db=db) grants = await AccessGrants.get_grants_by_resource('folder', id, db=db) + await publish_event( + request, + EVENTS.FOLDER_ACCESS_UPDATED, + actor=user, + subject_id=id, + data={'grant_count': len(grants)}, + ) return { **folder.model_dump(), 'access_grants': [g.model_dump() for g in grants], @@ -549,6 +585,13 @@ async def delete_folder_by_id( # Clean up access grants for this folder await AccessGrants.revoke_all_access('folder', folder_id, db=db) + await publish_event( + request, + EVENTS.FOLDER_DELETED, + actor=user, + subject_id=id, + data={'folder_ids': folder_ids, 'delete_contents': delete_contents}, + ) return True except Exception as e: log.exception(e) diff --git a/backend/open_webui/routers/functions.py b/backend/open_webui/routers/functions.py index d6eab3e0bd..ea6b66e0d3 100644 --- a/backend/open_webui/routers/functions.py +++ b/backend/open_webui/routers/functions.py @@ -11,6 +11,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status from open_webui.config import CACHE_DIR from open_webui.constants import ERROR_MESSAGES from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.functions import ( FunctionForm, @@ -218,6 +219,13 @@ async def create_new_function( await Functions.update_function_metadata_by_id(form_data.id, {'toggle': True}, db=db) if function: + await publish_event( + request, + EVENTS.FUNCTION_CREATED, + actor=user, + subject_id=function.id, + data={'type': function.type, 'name': function.name}, + ) return function else: raise HTTPException( @@ -261,12 +269,24 @@ async def get_function_by_id(id: str, user=Depends(get_admin_user), db: AsyncSes @router.post('/id/{id}/toggle', response_model=FunctionModel | None) -async def toggle_function_by_id(id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def toggle_function_by_id( + request: Request, + id: str, + user=Depends(get_admin_user), + db: AsyncSession = Depends(get_async_session), +): function = await Functions.get_function_by_id(id, db=db) if function: function = await Functions.update_function_by_id(id, {'is_active': not function.is_active}, db=db) if function: + await publish_event( + request, + EVENTS.FUNCTION_ENABLED if function.is_active else EVENTS.FUNCTION_DISABLED, + actor=user, + subject_id=function.id, subject_type='function', + data={'type': function.type, 'name': function.name}, + ) return function else: raise HTTPException( @@ -286,12 +306,24 @@ async def toggle_function_by_id(id: str, user=Depends(get_admin_user), db: Async @router.post('/id/{id}/toggle/global', response_model=FunctionModel | None) -async def toggle_global_by_id(id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def toggle_global_by_id( + request: Request, + id: str, + user=Depends(get_admin_user), + db: AsyncSession = Depends(get_async_session), +): function = await Functions.get_function_by_id(id, db=db) if function: function = await Functions.update_function_by_id(id, {'is_global': not function.is_global}, db=db) if function: + await publish_event( + request, + EVENTS.FUNCTION_UPDATED, + actor=user, + subject_id=function.id, + data={'type': function.type, 'name': function.name, 'is_global': function.is_global}, + ) return function else: raise HTTPException( @@ -335,6 +367,13 @@ async def update_function_by_id( await Functions.update_function_metadata_by_id(id, {'toggle': True}, db=db) if function: + await publish_event( + request, + EVENTS.FUNCTION_UPDATED, + actor=user, + subject_id=function.id, + data={'type': function.type, 'name': function.name}, + ) return function else: raise HTTPException( @@ -366,6 +405,12 @@ async def delete_function_by_id( if result: FUNCTIONS = get_functions_cache(request) FUNCTIONS.pop(id, None) + await publish_event( + request, + EVENTS.FUNCTION_DELETED, + actor=user, + subject_id=id, + ) return result @@ -452,6 +497,12 @@ async def update_function_valves_by_id( valves_dict = valves.model_dump(exclude_unset=True) await Functions.update_function_valves_by_id(id, valves_dict, db=db) + await publish_event( + request, + EVENTS.FUNCTION_VALVES_UPDATED, + actor=user, + subject_id=id, + ) return valves_dict except Exception as e: log.exception(f'Error updating function values by id {id}: {e}') @@ -544,6 +595,13 @@ async def update_function_user_valves_by_id( user_valves = UserValves(**form_data) user_valves_dict = user_valves.model_dump(exclude_unset=True) await Functions.update_user_valves_by_id_and_user_id(id, user.id, user_valves_dict, db=db) + await publish_event( + request, + EVENTS.FUNCTION_VALVES_UPDATED, + actor=user, + subject_id=id, + data={'scope': 'user'}, + ) return user_valves_dict except Exception as e: log.exception(f'Error updating function user valves by id {id}: {e}') diff --git a/backend/open_webui/routers/groups.py b/backend/open_webui/routers/groups.py index 6efcd3946e..b4afa897bf 100755 --- a/backend/open_webui/routers/groups.py +++ b/backend/open_webui/routers/groups.py @@ -6,6 +6,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from open_webui.config import CACHE_DIR from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants from open_webui.models.groups import ( @@ -58,6 +59,7 @@ async def get_groups( @router.post('/create', response_model=Optional[GroupResponse]) async def create_new_group( + request: Request, form_data: GroupForm, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session), @@ -65,6 +67,13 @@ async def create_new_group( try: group = await Groups.insert_new_group(user.id, form_data, db=db) if group: + await publish_event( + request, + EVENTS.GROUP_CREATED, + actor=user, + subject_id=group.id, + data={'name': group.name}, + ) return GroupResponse( **group.model_dump(), member_count=await Groups.get_group_member_count_by_id(group.id, db=db), @@ -168,6 +177,7 @@ async def get_users_in_group(id: str, user=Depends(get_admin_user), db: AsyncSes @router.post('/id/{id}/update', response_model=Optional[GroupResponse]) async def update_group_by_id( + request: Request, id: str, form_data: GroupUpdateForm, user=Depends(get_admin_user), @@ -176,6 +186,13 @@ async def update_group_by_id( try: group = await Groups.update_group_by_id(id, form_data, db=db) if group: + await publish_event( + request, + EVENTS.GROUP_UPDATED, + actor=user, + subject_id=id, + data={'name': group.name}, + ) return GroupResponse( **group.model_dump(), member_count=await Groups.get_group_member_count_by_id(group.id, db=db), @@ -200,6 +217,7 @@ async def update_group_by_id( @router.post('/id/{id}/users/add', response_model=Optional[GroupResponse]) async def add_user_to_group( + request: Request, id: str, form_data: UserIdsForm, user=Depends(get_admin_user), @@ -211,6 +229,13 @@ async def add_user_to_group( group = await Groups.add_users_to_group(id, form_data.user_ids, db=db) if group: + await publish_event( + request, + EVENTS.GROUP_MEMBER_ADDED, + actor=user, + subject_id=id, + data={'user_ids': form_data.user_ids}, + ) return GroupResponse( **group.model_dump(), member_count=await Groups.get_group_member_count_by_id(group.id, db=db), @@ -230,6 +255,7 @@ async def add_user_to_group( @router.post('/id/{id}/users/remove', response_model=Optional[GroupResponse]) async def remove_users_from_group( + request: Request, id: str, form_data: UserIdsForm, user=Depends(get_admin_user), @@ -238,6 +264,13 @@ async def remove_users_from_group( try: group = await Groups.remove_users_from_group(id, form_data.user_ids, db=db) if group: + await publish_event( + request, + EVENTS.GROUP_MEMBER_REMOVED, + actor=user, + subject_id=id, + data={'user_ids': form_data.user_ids}, + ) return GroupResponse( **group.model_dump(), member_count=await Groups.get_group_member_count_by_id(group.id, db=db), @@ -261,10 +294,18 @@ async def remove_users_from_group( @router.delete('/id/{id}/delete', response_model=bool) -async def delete_group_by_id(id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def delete_group_by_id( + request: Request, id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session) +): try: result = await Groups.delete_group_by_id(id, db=db) if result: + await publish_event( + request, + EVENTS.GROUP_DELETED, + actor=user, + subject_id=id, + ) return result else: raise HTTPException( diff --git a/backend/open_webui/routers/images.py b/backend/open_webui/routers/images.py index 17f1acb7e5..28008b25db 100644 --- a/backend/open_webui/routers/images.py +++ b/backend/open_webui/routers/images.py @@ -23,6 +23,7 @@ from open_webui.config import ( ) from open_webui.constants import ERROR_MESSAGES from open_webui.env import AIOHTTP_CLIENT_ALLOW_REDIRECTS, AIOHTTP_CLIENT_SESSION_SSL, ENABLE_FORWARD_USER_INFO_HEADERS +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.chats import Chats from open_webui.models.config import Config @@ -246,7 +247,20 @@ async def update_config(request: Request, form_data: ImagesConfig, user=Depends( updates['images.edit.comfyui.base_url'] = form_data.IMAGES_EDIT_COMFYUI_BASE_URL.strip('/') await Config.upsert(updates) await set_image_model(request, form_data.IMAGE_GENERATION_MODEL) - return await get_config_values(IMAGE_CONFIG_KEYS) + values = await get_config_values(IMAGE_CONFIG_KEYS) + await publish_event( + request, + EVENTS.CONFIG_UPDATED, + actor=user, + subject_id='images', + data={ + 'image_generation_enabled': values.get('ENABLE_IMAGE_GENERATION'), + 'image_edit_enabled': values.get('ENABLE_IMAGE_EDIT'), + 'image_generation_engine': values.get('IMAGE_GENERATION_ENGINE'), + 'image_edit_engine': values.get('IMAGE_EDIT_ENGINE'), + }, + ) + return values def get_automatic1111_api_auth(image_config): @@ -501,7 +515,20 @@ async def generate_images(request: Request, form_data: CreateImageForm, user=Dep detail=ERROR_MESSAGES.ACCESS_PROHIBITED, ) - return await image_generations(request, form_data, user=user) + result = await image_generations(request, form_data, user=user) + await publish_event( + request, + EVENTS.IMAGE_GENERATED, + actor=user, + subject_id=None, subject_type='image', + data={ + 'model': form_data.model, + 'size': form_data.size, + 'n': form_data.n, + 'prompt_preview': form_data.prompt[:300], + }, + ) + return result async def image_generations( @@ -788,7 +815,20 @@ async def edit_images(request: Request, form_data: EditImageForm, user=Depends(g detail=ERROR_MESSAGES.ACCESS_PROHIBITED, ) - return await image_edits(request, form_data, user=user) + result = await image_edits(request, form_data, user=user) + await publish_event( + request, + EVENTS.IMAGE_EDITED, + actor=user, + subject_id=None, subject_type='image', + data={ + 'model': form_data.model, + 'size': form_data.size, + 'n': form_data.n, + 'prompt_preview': form_data.prompt[:300], + }, + ) + return result async def image_edits( diff --git a/backend/open_webui/routers/knowledge.py b/backend/open_webui/routers/knowledge.py index 54c6177245..4b91d5879b 100644 --- a/backend/open_webui/routers/knowledge.py +++ b/backend/open_webui/routers/knowledge.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import io +import json import logging import time import uuid @@ -13,6 +14,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from fastapi.responses import StreamingResponse from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants from open_webui.models.config import Config @@ -300,6 +302,13 @@ async def create_new_knowledge( knowledge.name, knowledge.description, ) + await publish_event( + request, + EVENTS.KNOWLEDGE_CREATED, + actor=user, + subject_id=knowledge.id, + data={'name': knowledge.name}, + ) return knowledge else: raise HTTPException( @@ -358,12 +367,19 @@ async def reindex_knowledge_files( # Don't raise, just continue continue - if failed_files: - log.warning(f'Failed to process {len(failed_files)} files in knowledge base {knowledge_base.id}') - for failed in failed_files: - log.warning(f'File ID: {failed["file_id"]}, Error: {failed["error"]}') + if failed_files: + log.warning(f'Failed to process {len(failed_files)} files in knowledge base {knowledge_base.id}') + for failed in failed_files: + log.warning(f'File ID: {failed["file_id"]}, Error: {failed["error"]}') log.info(f'Reindexing completed.') + await publish_event( + request, + EVENTS.KNOWLEDGE_REINDEXED, + actor=user, + subject_id='all', + data={'count': len(knowledge_bases)}, + ) return True @@ -597,6 +613,7 @@ async def get_external_knowledge_connections( @router.post('/external/connections', response_model=dict) async def create_external_knowledge_connection( + request: Request, form_data: ExternalKnowledgeConnectionForm, user=Depends(get_admin_user), ): @@ -604,7 +621,15 @@ async def create_external_knowledge_connection( connection = _external_connection_dict(form_data, user.id) connections.append(connection) await _set_external_connections(connections) - return _sanitize_external_connection(connection) + sanitized = _sanitize_external_connection(connection) + await publish_event( + request, + EVENTS.KNOWLEDGE_EXTERNAL_CONNECTION_CREATED, + actor=user, + subject_id=connection.get('id'), + data={'name': sanitized.get('name'), 'provider': sanitized.get('provider')}, + ) + return sanitized @router.get('/external/connections/{id}', response_model=dict) @@ -621,6 +646,7 @@ async def get_external_knowledge_connection( @router.patch('/external/connections/{id}', response_model=dict) async def update_external_knowledge_connection( + request: Request, id: str, form_data: ExternalKnowledgeConnectionForm, user=Depends(get_admin_user), @@ -633,11 +659,20 @@ async def update_external_knowledge_connection( connection = _external_connection_update_dict(form_data, connections[idx]) connections[idx] = connection await _set_external_connections(connections) - return _sanitize_external_connection(connection) + sanitized = _sanitize_external_connection(connection) + await publish_event( + request, + EVENTS.KNOWLEDGE_EXTERNAL_CONNECTION_UPDATED, + actor=user, + subject_id=id, + data={'name': sanitized.get('name'), 'provider': sanitized.get('provider')}, + ) + return sanitized @router.delete('/external/connections/{id}', response_model=bool) async def delete_external_knowledge_connection( + request: Request, id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session), @@ -654,6 +689,13 @@ async def delete_external_knowledge_connection( connections = [connection for connection in await _get_external_connections() if connection.get('id') != id] await _set_external_connections(connections) + await publish_event( + request, + EVENTS.KNOWLEDGE_EXTERNAL_CONNECTION_DELETED, + actor=user, + subject_id=id, + data={'name': connection.get('name'), 'provider': connection.get('provider')}, + ) return True @@ -1071,10 +1113,18 @@ async def update_knowledge_by_id( knowledge.name, knowledge.description, ) - return KnowledgeFilesResponse( + response = KnowledgeFilesResponse( **knowledge.model_dump(), files=await Knowledges.get_file_metadatas_by_id(knowledge.id), ) + await publish_event( + request, + EVENTS.KNOWLEDGE_UPDATED, + actor=user, + subject_id=knowledge.id, + data={'name': knowledge.name}, + ) + return response else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -1132,10 +1182,18 @@ async def update_knowledge_access_by_id( knowledge.access_grants = await AccessGrants.set_access_grants('knowledge', id, form_data.access_grants, db=db) - return KnowledgeFilesResponse( + response = KnowledgeFilesResponse( **knowledge.model_dump(), files=await Knowledges.get_file_metadatas_by_id(id, db=db), ) + await publish_event( + request, + EVENTS.KNOWLEDGE_ACCESS_UPDATED, + actor=user, + subject_id=knowledge.id, + data={'name': knowledge.name}, + ) + return response ############################ @@ -1360,10 +1418,18 @@ async def add_file_to_knowledge_by_id( ) if knowledge: - return KnowledgeFilesResponse( + response = KnowledgeFilesResponse( **knowledge.model_dump(), files=await Knowledges.get_file_metadatas_by_id(knowledge.id, db=db), ) + await publish_event( + request, + EVENTS.KNOWLEDGE_FILE_ADDED, + actor=user, + subject_id=form_data.file_id, + data={'knowledge_id': knowledge.id, 'directory_id': form_data.directory_id}, + ) + return response else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -1436,10 +1502,18 @@ async def update_file_from_knowledge_by_id( ) if knowledge: - return KnowledgeFilesResponse( + response = KnowledgeFilesResponse( **knowledge.model_dump(), files=await Knowledges.get_file_metadatas_by_id(knowledge.id, db=db), ) + await publish_event( + request, + EVENTS.KNOWLEDGE_FILE_UPDATED, + actor=user, + subject_id=form_data.file_id, + data={'knowledge_id': knowledge.id}, + ) + return response else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -1454,6 +1528,7 @@ async def update_file_from_knowledge_by_id( @router.post('/{id}/file/remove', response_model=KnowledgeFilesResponse | None) async def remove_file_from_knowledge_by_id( + request: Request, id: str, form_data: KnowledgeFileIdForm, delete_file: bool = Query(True), @@ -1531,10 +1606,18 @@ async def remove_file_from_knowledge_by_id( await Files.delete_file_by_id(form_data.file_id, db=db) if knowledge: - return KnowledgeFilesResponse( + response = KnowledgeFilesResponse( **knowledge.model_dump(), files=await Knowledges.get_file_metadatas_by_id(knowledge.id, db=db), ) + await publish_event( + request, + EVENTS.KNOWLEDGE_FILE_REMOVED, + actor=user, + subject_id=form_data.file_id, + data={'knowledge_id': knowledge.id, 'delete_file': delete_file}, + ) + return response else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -1549,7 +1632,10 @@ async def remove_file_from_knowledge_by_id( @router.delete('/{id}/delete', response_model=bool) async def delete_knowledge_by_id( - id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) + request: Request, + id: str, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), ): knowledge = await Knowledges.get_knowledge_by_id(id=id, db=db) if not knowledge: @@ -1615,6 +1701,14 @@ async def delete_knowledge_by_id( await remove_knowledge_base_metadata_embedding(id) result = await Knowledges.delete_knowledge_by_id(id=id, db=db) + if result: + await publish_event( + request, + EVENTS.KNOWLEDGE_DELETED, + actor=user, + subject_id=id, + data={'name': knowledge.name}, + ) return result @@ -1625,6 +1719,7 @@ async def delete_knowledge_by_id( @router.post('/{id}/reset', response_model=KnowledgeResponse | None) async def reset_knowledge_by_id( + request: Request, id: str, include_directories: bool = Query(True), user=Depends(get_verified_user), @@ -1662,6 +1757,14 @@ async def reset_knowledge_by_id( pass knowledge = await Knowledges.reset_knowledge_by_id(id=id, include_directories=include_directories, db=db) + if knowledge: + await publish_event( + request, + EVENTS.KNOWLEDGE_RESET, + actor=user, + subject_id=id, + data={'include_directories': include_directories}, + ) return knowledge @@ -2070,6 +2173,7 @@ async def _verify_knowledge_write_access(id: str, user, db: AsyncSession): @router.post('/{id}/dirs/create', response_model=KnowledgeDirectoryModel) async def create_knowledge_directory( + request: Request, id: str, form_data: KnowledgeDirectoryCreateForm, user=Depends(get_verified_user), @@ -2089,11 +2193,19 @@ async def create_knowledge_directory( status_code=status.HTTP_400_BAD_REQUEST, detail='Failed to create directory. A directory with this name may already exist at this level.', ) + await publish_event( + request, + EVENTS.KNOWLEDGE_DIRECTORY_CREATED, + actor=user, + subject_id=directory.id, + data={'knowledge_id': id, 'name': directory.name, 'parent_id': directory.parent_id}, + ) return directory @router.post('/{id}/dirs/{dir_id}/update', response_model=KnowledgeDirectoryModel) async def update_knowledge_directory( + request: Request, id: str, dir_id: str, form_data: KnowledgeDirectoryUpdateForm, @@ -2121,11 +2233,19 @@ async def update_knowledge_directory( status_code=status.HTTP_400_BAD_REQUEST, detail='Failed to update directory. This may be caused by a naming conflict or circular move.', ) + await publish_event( + request, + EVENTS.KNOWLEDGE_DIRECTORY_UPDATED, + actor=user, + subject_id=result.id, + data={'knowledge_id': id, 'name': result.name, 'parent_id': result.parent_id}, + ) return result @router.delete('/{id}/dirs/{dir_id}/delete') async def delete_knowledge_directory( + request: Request, id: str, dir_id: str, move_files: bool = Query(True, description='If true, move contained files to parent. If false, delete them.'), @@ -2152,11 +2272,19 @@ async def delete_knowledge_directory( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail='Failed to delete directory.', ) + await publish_event( + request, + EVENTS.KNOWLEDGE_DIRECTORY_DELETED, + actor=user, + subject_id=dir_id, + data={'knowledge_id': id, 'move_files': move_files}, + ) return {'status': True} @router.post('/{id}/file/move') async def move_file_in_knowledge( + request: Request, id: str, form_data: KnowledgeFileMoveForm, user=Depends(get_verified_user), @@ -2191,4 +2319,11 @@ async def move_file_in_knowledge( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail='Failed to move file.', ) + await publish_event( + request, + EVENTS.KNOWLEDGE_FILE_MOVED, + actor=user, + subject_id=form_data.file_id, + data={'knowledge_id': id, 'directory_id': form_data.directory_id}, + ) return {'status': True} diff --git a/backend/open_webui/routers/memories.py b/backend/open_webui/routers/memories.py index 1e0e4a9ed4..a7dc3875cf 100644 --- a/backend/open_webui/routers/memories.py +++ b/backend/open_webui/routers/memories.py @@ -6,6 +6,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.config import Config from open_webui.models.memories import Memories, MemoryModel @@ -99,6 +100,13 @@ async def add_memory( ], ) + await publish_event( + request, + EVENTS.MEMORY_CREATED, + actor=user, + subject_id=memory.id, + data={'content_preview': memory.content[:300]}, + ) return memory @@ -214,6 +222,13 @@ async def reset_memory_from_vector_db( ], ) + await publish_event( + request, + EVENTS.MEMORY_RESET, + actor=user, + subject_id=user.id, subject_type='user', + data={'count': len(memories)}, + ) return True @@ -237,6 +252,12 @@ async def delete_memory_by_user_id( await ASYNC_VECTOR_DB_CLIENT.delete_collection(f'user-memory-{user.id}') except Exception as e: log.error(e) + await publish_event( + request, + EVENTS.MEMORY_DELETED, + actor=user, + subject_id=user.id, subject_type='user', + ) return True return False @@ -282,6 +303,13 @@ async def update_memory_by_id( ], ) + await publish_event( + request, + EVENTS.MEMORY_UPDATED, + actor=user, + subject_id=memory.id, + data={'content_preview': memory.content[:300]}, + ) return memory @@ -303,6 +331,12 @@ async def delete_memory_by_id( if result: await ASYNC_VECTOR_DB_CLIENT.delete(collection_name=f'user-memory-{user.id}', ids=[memory_id]) + await publish_event( + request, + EVENTS.MEMORY_DELETED, + actor=user, + subject_id=memory_id, + ) return True return False diff --git a/backend/open_webui/routers/models.py b/backend/open_webui/routers/models.py index ed4d221bc6..af16958e8e 100644 --- a/backend/open_webui/routers/models.py +++ b/backend/open_webui/routers/models.py @@ -20,6 +20,7 @@ from fastapi import ( from fastapi.responses import RedirectResponse, StreamingResponse from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import ENABLE_PROFILE_IMAGE_URL_FORWARDING, PROFILE_IMAGE_ALLOWED_MIME_TYPES from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants @@ -268,6 +269,13 @@ async def create_new_model( model = await Models.insert_new_model(form_data, user.id, db=db) if model: + await publish_event( + request, + EVENTS.MODEL_CREATED, + actor=user, + subject_id=model.id, + data={'name': model.name}, + ) return model else: raise HTTPException( @@ -360,10 +368,12 @@ async def import_models( else: writable_model_ids = set(existing_model_ids) + imported_ids = [] for model_data in data: model_id = model_data.get('id') if model_id and is_valid_model_id(model_id): + imported_ids.append(model_id) # Defense-in-depth: skip models referencing inaccessible files try: await _verify_knowledge_file_access( @@ -424,6 +434,13 @@ async def import_models( 'sharing.public_models', ) await Models.insert_new_model(user_id=user.id, form_data=new_model, db=db) + await publish_event( + request, + EVENTS.MODEL_IMPORTED, + actor=user, + subject_type='model', + data={'count': len(imported_ids), 'model_ids': imported_ids}, + ) return True else: raise HTTPException(status_code=400, detail='Invalid JSON format') @@ -448,7 +465,15 @@ async def sync_models( user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session), ): - return await Models.sync_models(user.id, form_data.models, db=db) + models = await Models.sync_models(user.id, form_data.models, db=db) + await publish_event( + request, + EVENTS.MODEL_SYNCED, + actor=user, + subject_type='model', + data={'count': len(models), 'model_ids': [model.id for model in models]}, + ) + return models ########################### @@ -596,7 +621,9 @@ async def get_model_profile_image( @router.post('/model/toggle', response_model=ModelResponse | None) -async def toggle_model_by_id(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): +async def toggle_model_by_id( + request: Request, id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) +): model = await Models.get_model_by_id(id, db=db) if model: if ( @@ -613,6 +640,13 @@ async def toggle_model_by_id(id: str, user=Depends(get_verified_user), db: Async model = await Models.toggle_model_by_id(id, db=db) if model: + await publish_event( + request, + EVENTS.MODEL_ENABLED if model.is_active else EVENTS.MODEL_DISABLED, + actor=user, + subject_id=model.id, subject_type='model', + data={'name': model.name}, + ) return model else: raise HTTPException( @@ -682,6 +716,14 @@ async def update_model_by_id( ) model = await Models.update_model_by_id(form_data.id, ModelForm(**form_data.model_dump()), db=db) + if model: + await publish_event( + request, + EVENTS.MODEL_UPDATED, + actor=user, + subject_id=model.id, + data={'name': model.name}, + ) return model @@ -757,7 +799,14 @@ async def update_model_access_by_id( await Models.update_model_updated_at_by_id(form_data.id, db=db) - return await Models.get_model_by_id(form_data.id, db=db) + model = await Models.get_model_by_id(form_data.id, db=db) + await publish_event( + request, + EVENTS.MODEL_ACCESS_UPDATED, + actor=user, + subject_id=form_data.id, + ) + return model ############################ @@ -767,6 +816,7 @@ async def update_model_access_by_id( @router.post('/model/delete', response_model=bool) async def delete_model_by_id( + request: Request, form_data: ModelIdForm, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session), @@ -795,10 +845,20 @@ async def delete_model_by_id( ) result = await Models.delete_model_by_id(form_data.id, db=db) + if result: + await publish_event( + request, + EVENTS.MODEL_DELETED, + actor=user, + subject_id=form_data.id, + data={'name': model.name}, + ) return result @router.delete('/delete/all', response_model=bool) -async def delete_all_models(user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def delete_all_models(request: Request, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): result = await Models.delete_all_models(db=db) + if result: + await publish_event(request, EVENTS.MODEL_DELETED, actor=user, subject_type='model') return result diff --git a/backend/open_webui/routers/notes.py b/backend/open_webui/routers/notes.py index 8438a517c5..c477e03eca 100644 --- a/backend/open_webui/routers/notes.py +++ b/backend/open_webui/routers/notes.py @@ -9,6 +9,7 @@ from open_webui.config import ( ENABLE_ADMIN_EXPORT, ) from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants from open_webui.models.config import Config @@ -227,6 +228,13 @@ async def create_new_note( try: note = await Notes.insert_new_note(user.id, form_data, db=db) + await publish_event( + request, + EVENTS.NOTE_CREATED, + actor=user, + subject_id=note.id, + data={'title': note.title}, + ) return note except Exception as e: log.exception(e) @@ -352,6 +360,13 @@ async def update_note_by_id( to=f'note:{note.id}', ) + await publish_event( + request, + EVENTS.NOTE_UPDATED, + actor=user, + subject_id=note.id, + data={'title': note.title}, + ) return note except Exception as e: log.exception(e) @@ -412,6 +427,12 @@ async def update_note_access_by_id( note = await Notes.get_note_by_id(id, db=db) pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db) note.is_pinned = note.id in pinned_note_ids + await publish_event( + request, + EVENTS.NOTE_ACCESS_UPDATED, + actor=user, + subject_id=note.id, + ) return note @@ -454,6 +475,12 @@ async def pin_note_by_id( note = await Notes.toggle_note_pinned_by_id(id, user.id, db=db) pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db) note.is_pinned = note.id in pinned_note_ids + await publish_event( + request, + EVENTS.NOTE_PINNED if note.is_pinned else EVENTS.NOTE_UNPINNED, + actor=user, + subject_id=note.id, subject_type='note', + ) return note @@ -495,6 +522,12 @@ async def delete_note_by_id( try: note = await Notes.delete_note_by_id(id, db=db) + await publish_event( + request, + EVENTS.NOTE_DELETED, + actor=user, + subject_id=id, + ) return True except Exception as e: log.exception(e) diff --git a/backend/open_webui/routers/ollama.py b/backend/open_webui/routers/ollama.py index 955ef1500b..60fb82b87c 100644 --- a/backend/open_webui/routers/ollama.py +++ b/backend/open_webui/routers/ollama.py @@ -20,6 +20,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from open_webui.config import UPLOAD_DIR from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import ( AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT, @@ -291,6 +292,18 @@ async def update_config( 'ollama.api_configs': api_configs, } ) + await publish_event( + request, + EVENTS.MODEL_PROVIDER_CONFIG_UPDATED, + actor=user, + subject_id='ollama', + subject_type='model.provider_config', + data={ + 'provider': 'ollama', + 'enabled': form_data.ENABLE_OLLAMA_API, + 'base_url_count': len(form_data.OLLAMA_BASE_URLS), + }, + ) return { 'ENABLE_OLLAMA_API': form_data.ENABLE_OLLAMA_API, 'OLLAMA_BASE_URLS': form_data.OLLAMA_BASE_URLS, @@ -699,6 +712,13 @@ async def copy_model( key=key, user=user, ) + await publish_event( + request, + EVENTS.MODEL_PROVIDER_MODEL_CREATED, + actor=user, + subject_id=form_data.destination, + data={'provider': 'ollama', 'source': form_data.source, 'url_idx': url_idx}, + ) return True @@ -735,6 +755,13 @@ async def delete_model( key=key, user=user, ) + await publish_event( + request, + EVENTS.MODEL_PROVIDER_MODEL_DELETED, + actor=user, + subject_id=model, + data={'provider': 'ollama', 'url_idx': url_idx}, + ) return True diff --git a/backend/open_webui/routers/openai.py b/backend/open_webui/routers/openai.py index 4280e4e9fd..2631011a5b 100644 --- a/backend/open_webui/routers/openai.py +++ b/backend/open_webui/routers/openai.py @@ -22,6 +22,7 @@ from open_webui.config import ( CACHE_DIR, ) from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import ( AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT, @@ -310,6 +311,18 @@ async def update_config(request: Request, form_data: OpenAIConfigForm, user=Depe 'openai.api_configs': api_configs, } ) + await publish_event( + request, + EVENTS.MODEL_PROVIDER_CONFIG_UPDATED, + actor=user, + subject_id='openai', + subject_type='model.provider_config', + data={ + 'provider': 'openai', + 'enabled': form_data.ENABLE_OPENAI_API, + 'base_url_count': len(form_data.OPENAI_API_BASE_URLS), + }, + ) return { 'ENABLE_OPENAI_API': form_data.ENABLE_OPENAI_API, diff --git a/backend/open_webui/routers/pipelines.py b/backend/open_webui/routers/pipelines.py index 3f50d23f86..fc3022d117 100644 --- a/backend/open_webui/routers/pipelines.py +++ b/backend/open_webui/routers/pipelines.py @@ -411,13 +411,6 @@ async def get_pipelines(request: Request, urlIdx: Optional[int] = None, user=Dep response.raise_for_status() data = await response.json() - await publish_event( - request, - EVENTS.PIPELINE_VALVES_UPDATED, - actor=user, - subject_id=pipeline_id, - data={'url_idx': urlIdx}, - ) return {**data} except Exception as e: # Handle connection error here @@ -458,6 +451,13 @@ async def get_pipeline_valves( response.raise_for_status() data = await response.json() + await publish_event( + request, + EVENTS.PIPELINE_VALVES_UPDATED, + actor=user, + subject_id=pipeline_id, + data={'url_idx': urlIdx}, + ) return {**data} except Exception as e: # Handle connection error here diff --git a/backend/open_webui/routers/prompts.py b/backend/open_webui/routers/prompts.py index 2698bb2af6..373323a59c 100644 --- a/backend/open_webui/routers/prompts.py +++ b/backend/open_webui/routers/prompts.py @@ -5,6 +5,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants from open_webui.models.config import Config @@ -178,6 +179,13 @@ async def create_new_prompt( prompt = await Prompts.insert_new_prompt(user.id, form_data, db=db) if prompt: + await publish_event( + request, + EVENTS.PROMPT_CREATED, + actor=user, + subject_id=prompt.id, + data={'name': prompt.name, 'command': prompt.command}, + ) return prompt raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -292,6 +300,13 @@ async def update_prompt_by_id( # Use the ID from the found prompt updated_prompt = await Prompts.update_prompt_by_id(prompt.id, form_data, user.id, db=db) if updated_prompt: + await publish_event( + request, + EVENTS.PROMPT_UPDATED, + actor=user, + subject_id=updated_prompt.id, + data={'name': updated_prompt.name, 'command': updated_prompt.command}, + ) return updated_prompt else: raise HTTPException( @@ -307,6 +322,7 @@ async def update_prompt_by_id( @router.post('/id/{prompt_id}/update/meta', response_model=PromptModel | None) async def update_prompt_metadata( + request: Request, prompt_id: str, form_data: PromptMetadataForm, user=Depends(get_verified_user), @@ -350,6 +366,13 @@ async def update_prompt_metadata( prompt.id, form_data.name, form_data.command, form_data.tags, db=db ) if updated_prompt: + await publish_event( + request, + EVENTS.PROMPT_UPDATED, + actor=user, + subject_id=updated_prompt.id, + data={'name': updated_prompt.name, 'command': updated_prompt.command}, + ) return updated_prompt else: raise HTTPException( @@ -360,6 +383,7 @@ async def update_prompt_metadata( @router.post('/id/{prompt_id}/update/version', response_model=PromptModel | None) async def set_prompt_version( + request: Request, prompt_id: str, form_data: PromptVersionUpdateForm, user=Depends(get_verified_user), @@ -390,6 +414,13 @@ async def set_prompt_version( updated_prompt = await Prompts.update_prompt_version(prompt.id, form_data.version_id, db=db) if updated_prompt: + await publish_event( + request, + EVENTS.PROMPT_VERSION_UPDATED, + actor=user, + subject_id=updated_prompt.id, + data={'version_id': updated_prompt.version_id}, + ) return updated_prompt else: raise HTTPException( @@ -448,7 +479,15 @@ async def update_prompt_access_by_id( await AccessGrants.set_access_grants('prompt', prompt_id, form_data.access_grants, db=db) - return await Prompts.get_prompt_by_id(prompt_id, db=db) + updated_prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) + await publish_event( + request, + EVENTS.PROMPT_ACCESS_UPDATED, + actor=user, + subject_id=prompt_id, + data={'name': updated_prompt.name if updated_prompt else None}, + ) + return updated_prompt ############################ @@ -458,7 +497,10 @@ async def update_prompt_access_by_id( @router.post('/id/{prompt_id}/toggle', response_model=PromptModel | None) async def toggle_prompt_active( - prompt_id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) + request: Request, + prompt_id: str, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), ): prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) @@ -486,6 +528,13 @@ async def toggle_prompt_active( result = await Prompts.toggle_prompt_active(prompt.id, db=db) if result: + await publish_event( + request, + EVENTS.PROMPT_ENABLED if result.is_active else EVENTS.PROMPT_DISABLED, + actor=user, + subject_id=result.id, subject_type='prompt', + data={'name': result.name, 'command': result.command}, + ) return result raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -500,7 +549,10 @@ async def toggle_prompt_active( @router.delete('/id/{prompt_id}/delete', response_model=bool) async def delete_prompt_by_id( - prompt_id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) + request: Request, + prompt_id: str, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), ): prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) @@ -527,6 +579,14 @@ async def delete_prompt_by_id( ) result = await Prompts.delete_prompt_by_id(prompt.id, db=db) + if result: + await publish_event( + request, + EVENTS.PROMPT_DELETED, + actor=user, + subject_id=prompt.id, + data={'name': prompt.name, 'command': prompt.command}, + ) return result diff --git a/backend/open_webui/routers/retrieval.py b/backend/open_webui/routers/retrieval.py index 0017a13b02..ebba3ff575 100644 --- a/backend/open_webui/routers/retrieval.py +++ b/backend/open_webui/routers/retrieval.py @@ -56,6 +56,7 @@ from open_webui.env import ( SENTENCE_TRANSFORMERS_CROSS_ENCODER_SIGMOID_ACTIVATION_FUNCTION, SENTENCE_TRANSFORMERS_MODEL_KWARGS, ) +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_db, get_async_session from open_webui.models.files import FileModel, Files, FileUpdateForm from open_webui.models.knowledge import Knowledges @@ -1898,6 +1899,13 @@ async def process_file( if config.BYPASS_EMBEDDING_AND_RETRIEVAL: await Files.update_file_data_by_id(file.id, {'status': 'completed'}, db=db) await Files.update_file_hash_by_id(file.id, hash, db=db) + await publish_event( + request, + EVENTS.RETRIEVAL_CONTENT_PROCESSED, + actor=user, + subject_id=file.id, subject_type='file', + data={'collection_name': None, 'filename': file.filename}, + ) return { 'status': True, 'collection_name': None, @@ -1950,6 +1958,13 @@ async def process_file( ) await Files.update_file_hash_by_id(file.id, hash, db=session) + await publish_event( + request, + EVENTS.RETRIEVAL_CONTENT_PROCESSED, + actor=user, + subject_id=file.id, subject_type='file', + data={'collection_name': collection_name, 'filename': file.filename}, + ) return { 'status': True, 'collection_name': collection_name, @@ -2018,6 +2033,13 @@ async def process_text( config = await get_retrieval_config() result = await run_in_threadpool(save_docs_to_vector_db, request, docs, collection_name, config, user=user) if result: + await publish_event( + request, + EVENTS.RETRIEVAL_CONTENT_PROCESSED, + actor=user, + subject_id=collection_name, subject_type='retrieval.collection', + data={'name': form_data.name, 'content_preview': text_content[:300]}, + ) return { 'status': True, 'collection_name': collection_name, @@ -2730,6 +2752,7 @@ class DeleteForm(BaseModel): @router.post('/delete') async def delete_entries_from_collection( + request: Request, form_data: DeleteForm, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session), @@ -2768,6 +2791,13 @@ async def delete_entries_from_collection( collection_name=form_data.collection_name, filter={'hash': hash}, ) + await publish_event( + request, + EVENTS.RETRIEVAL_COLLECTION_DELETED, + actor=user, + subject_id=form_data.collection_name, + data={'file_id': form_data.file_id}, + ) return {'status': True} else: return {'status': False} @@ -2781,13 +2811,23 @@ async def delete_entries_from_collection( @router.post('/reset/db') -async def reset_vector_db(user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def reset_vector_db( + request: Request, + user=Depends(get_admin_user), + db: AsyncSession = Depends(get_async_session), +): await ASYNC_VECTOR_DB_CLIENT.reset() await Knowledges.delete_all_knowledge(db=db) + await publish_event( + request, + EVENTS.RETRIEVAL_VECTOR_DB_RESET, + actor=user, + subject_id='default', + ) @router.post('/reset/uploads') -async def reset_upload_dir(user=Depends(get_admin_user)) -> bool: +async def reset_upload_dir(request: Request, user=Depends(get_admin_user)) -> bool: folder = f'{UPLOAD_DIR}' try: # Check if the directory exists @@ -2806,6 +2846,12 @@ async def reset_upload_dir(user=Depends(get_admin_user)) -> bool: log.warning(f'The directory {folder} does not exist') except Exception as e: log.exception(f'Failed to process the directory {folder}. Reason: {e}') + await publish_event( + request, + EVENTS.RETRIEVAL_UPLOADS_RESET, + actor=user, + subject_id='all', subject_type='file.uploads', + ) return True @@ -2936,4 +2982,15 @@ async def process_files_batch( file_result.status = 'failed' file_errors.append(BatchProcessFilesResult(file_id=file_result.file_id, status='failed', error=str(e))) - return BatchProcessFilesResponse(results=file_results, errors=file_errors) + response = BatchProcessFilesResponse(results=file_results, errors=file_errors) + await publish_event( + request, + EVENTS.RETRIEVAL_CONTENT_PROCESSED, + actor=user, + subject_id=collection_name, subject_type='retrieval.collection', + data={ + 'count': len([item for item in file_results if item.status == 'completed']), + 'errors': len(file_errors), + }, + ) + return response diff --git a/backend/open_webui/routers/scim.py b/backend/open_webui/routers/scim.py index 6e2ebb892b..42d24a4a53 100644 --- a/backend/open_webui/routers/scim.py +++ b/backend/open_webui/routers/scim.py @@ -16,6 +16,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, s from fastapi.responses import JSONResponse from open_webui.config import OAUTH_PROVIDERS from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import SCIM_AUTH_PROVIDER from open_webui.internal.db import get_async_session from open_webui.models.groups import GroupModel, Groups @@ -629,6 +630,18 @@ async def create_user( await Users.update_user_scim_by_id(user_id, provider, user_data.externalId, db=db) new_user = await Users.get_user_by_id(user_id, db=db) + await publish_event( + request, + EVENTS.USER_CREATED, + subject_id=new_user.id, + source='scim', + data={ + 'email': new_user.email, + 'role': new_user.role, + 'external_id': user_data.externalId, + }, + ) + return await user_to_scim(new_user, request, db=db) @@ -687,6 +700,16 @@ async def update_user( await Users.update_user_scim_by_id(user_id, provider, user_data.externalId, db=db) updated_user = await Users.get_user_by_id(user_id, db=db) + await publish_event( + request, + EVENTS.USER_UPDATED, + subject_id=user_id, + source='scim', + data={ + 'updated_fields': list(update_data.keys()) + (['externalId'] if user_data.externalId else []), + }, + ) + return await user_to_scim(updated_user, request, db=db) @@ -741,6 +764,14 @@ async def patch_user( else: updated_user = user + await publish_event( + request, + EVENTS.USER_UPDATED, + subject_id=user_id, + source='scim', + data={'updated_fields': list(update_data.keys())}, + ) + return await user_to_scim(updated_user, request, db=db) @@ -766,6 +797,14 @@ async def delete_user( detail='Failed to delete user', ) + await publish_event( + request, + EVENTS.USER_DELETED, + subject_id=user_id, + source='scim', + data={'email': user.email}, + ) + return None @@ -883,6 +922,22 @@ async def create_group( new_group = await Groups.get_group_by_id(new_group.id, db=db) + await publish_event( + request, + EVENTS.GROUP_CREATED, + subject_id=new_group.id, + source='scim', + data={'name': new_group.name, 'member_ids': member_ids, 'member_count': len(member_ids)}, + ) + if member_ids: + await publish_event( + request, + EVENTS.GROUP_MEMBER_ADDED, + subject_id=new_group.id, + source='scim', + data={'member_ids': member_ids, 'count': len(member_ids)}, + ) + return await group_to_scim(new_group, request, db=db) @@ -911,9 +966,15 @@ async def update_group( ) # Handle members if provided + added_member_ids = [] + removed_member_ids = [] if group_data.members is not None: + old_member_ids = set(await Groups.get_group_user_ids_by_id(group_id, db) or []) member_ids = [member.value for member in group_data.members] await Groups.set_group_user_ids_by_id(group_id, member_ids, db=db) + new_member_ids = set(member_ids) + added_member_ids = sorted(new_member_ids - old_member_ids) + removed_member_ids = sorted(old_member_ids - new_member_ids) # Update group updated_group = await Groups.update_group_by_id(group_id, update_form, db=db) @@ -923,6 +984,30 @@ async def update_group( detail='Failed to update group', ) + await publish_event( + request, + EVENTS.GROUP_UPDATED, + subject_id=group_id, + source='scim', + data={'updated_fields': ['name', 'members'] if group_data.members is not None else ['name']}, + ) + if added_member_ids: + await publish_event( + request, + EVENTS.GROUP_MEMBER_ADDED, + subject_id=group_id, + source='scim', + data={'member_ids': added_member_ids, 'count': len(added_member_ids)}, + ) + if removed_member_ids: + await publish_event( + request, + EVENTS.GROUP_MEMBER_REMOVED, + subject_id=group_id, + source='scim', + data={'member_ids': removed_member_ids, 'count': len(removed_member_ids)}, + ) + return await group_to_scim(updated_group, request, db=db) @@ -948,6 +1033,8 @@ async def patch_group( name=group.name, description=group.description, ) + added_member_ids = [] + removed_member_ids = [] for operation in patch_data.Operations: op = operation.op.lower() @@ -959,7 +1046,12 @@ async def patch_group( update_form.name = value elif path == 'members': # Replace all members - await Groups.set_group_user_ids_by_id(group_id, [member['value'] for member in value], db=db) + old_member_ids = set(await Groups.get_group_user_ids_by_id(group_id, db) or []) + new_member_ids = [member['value'] for member in value] + await Groups.set_group_user_ids_by_id(group_id, new_member_ids, db=db) + new_member_ids_set = set(new_member_ids) + added_member_ids.extend(sorted(new_member_ids_set - old_member_ids)) + removed_member_ids.extend(sorted(old_member_ids - new_member_ids_set)) elif op == 'add': if path == 'members': @@ -968,11 +1060,13 @@ async def patch_group( for member in value: if isinstance(member, dict) and 'value' in member: await Groups.add_users_to_group(group_id, [member['value']], db=db) + added_member_ids.append(member['value']) elif op == 'remove': if path and path.startswith('members[value eq'): # Remove specific member member_id = path.split('"')[1] await Groups.remove_users_from_group(group_id, [member_id], db=db) + removed_member_ids.append(member_id) # Update group updated_group = await Groups.update_group_by_id(group_id, update_form, db=db) @@ -982,6 +1076,30 @@ async def patch_group( detail='Failed to update group', ) + await publish_event( + request, + EVENTS.GROUP_UPDATED, + subject_id=group_id, + source='scim', + data={'operation_count': len(patch_data.Operations)}, + ) + if added_member_ids: + await publish_event( + request, + EVENTS.GROUP_MEMBER_ADDED, + subject_id=group_id, + source='scim', + data={'member_ids': sorted(set(added_member_ids)), 'count': len(set(added_member_ids))}, + ) + if removed_member_ids: + await publish_event( + request, + EVENTS.GROUP_MEMBER_REMOVED, + subject_id=group_id, + source='scim', + data={'member_ids': sorted(set(removed_member_ids)), 'count': len(set(removed_member_ids))}, + ) + return await group_to_scim(updated_group, request, db=db) @@ -1007,4 +1125,12 @@ async def delete_group( detail='Failed to delete group', ) + await publish_event( + request, + EVENTS.GROUP_DELETED, + subject_id=group_id, + source='scim', + data={'name': group.name}, + ) + return None diff --git a/backend/open_webui/routers/terminals.py b/backend/open_webui/routers/terminals.py index 269e0cd764..e1236b08b0 100644 --- a/backend/open_webui/routers/terminals.py +++ b/backend/open_webui/routers/terminals.py @@ -13,6 +13,7 @@ import aiohttp from fastapi import APIRouter, Depends, Request, Response, WebSocket from fastapi.responses import JSONResponse, StreamingResponse from open_webui.config import TERMINAL_PROXY_HEADERS +from open_webui.events import EVENTS, publish_event from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL from open_webui.models.config import Config from open_webui.models.groups import Groups @@ -297,6 +298,8 @@ async def ws_terminal( if upstream_params: upstream_url += f'?{urllib.parse.urlencode(upstream_params)}' + app = ws.scope.get('app') + opened = False session = aiohttp.ClientSession() try: async with session.ws_connect(upstream_url, ssl=AIOHTTP_CLIENT_SESSION_SSL) as upstream: @@ -309,6 +312,16 @@ async def ws_terminal( key = connection.get('key', '') await upstream.send_str(_json.dumps({'type': 'auth', 'token': key})) + await publish_event( + app, + EVENTS.TERMINAL_SESSION_OPENED, + actor=user, + subject_id=session_id, + subject_type='terminal.session', + data={'server_id': server_id}, + ) + opened = True + async def _client_to_upstream(): """Forward client → upstream.""" try: @@ -357,6 +370,15 @@ async def ws_terminal( log.exception('Terminal WebSocket proxy error: %s', e) finally: await session.close() + if opened: + await publish_event( + app, + EVENTS.TERMINAL_SESSION_CLOSED, + actor=user, + subject_id=session_id, + subject_type='terminal.session', + data={'server_id': server_id}, + ) try: await ws.close() except Exception: diff --git a/backend/open_webui/routers/tools.py b/backend/open_webui/routers/tools.py index 0c3ee91c60..a085d2be01 100644 --- a/backend/open_webui/routers/tools.py +++ b/backend/open_webui/routers/tools.py @@ -11,6 +11,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL, CACHE_DIR from open_webui.constants import ERROR_MESSAGES from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL, AIOHTTP_CLIENT_TIMEOUT +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_session from open_webui.models.access_grants import AccessGrants from open_webui.models.config import Config @@ -380,6 +381,13 @@ async def create_new_tools( tool_cache_dir.mkdir(parents=True, exist_ok=True) if tools: + await publish_event( + request, + EVENTS.TOOL_CREATED, + actor=user, + subject_id=tools.id, + data={'name': tools.name}, + ) return tools else: raise HTTPException( @@ -522,6 +530,13 @@ async def update_tools_by_id( tools = await Tools.update_tool_by_id(id, updated, db=db) if tools: + await publish_event( + request, + EVENTS.TOOL_UPDATED, + actor=user, + subject_id=tools.id, + data={'name': tools.name}, + ) return tools else: raise HTTPException( @@ -586,7 +601,15 @@ async def update_tool_access_by_id( await AccessGrants.set_access_grants('tool', id, form_data.access_grants, db=db) - return await Tools.get_tool_by_id(id, db=db) + tools = await Tools.get_tool_by_id(id, db=db) + await publish_event( + request, + EVENTS.TOOL_ACCESS_UPDATED, + actor=user, + subject_id=id, + data={'name': tools.name if tools else None}, + ) + return tools ############################ @@ -628,6 +651,13 @@ async def delete_tools_by_id( if result: TOOLS = get_tools_cache(request) TOOLS.pop(id, None) + await publish_event( + request, + EVENTS.TOOL_DELETED, + actor=user, + subject_id=id, + data={'name': tools.name}, + ) return result @@ -770,6 +800,12 @@ async def update_tools_valves_by_id( valves = Valves(**form_data) valves_dict = valves.model_dump(exclude_unset=True) await Tools.update_tool_valves_by_id(id, valves_dict, db=db) + await publish_event( + request, + EVENTS.TOOL_VALVES_UPDATED, + actor=user, + subject_id=id, + ) return valves_dict except Exception as e: log.exception(f'Failed to update tool valves by id {id}: {e}') @@ -903,6 +939,13 @@ async def update_tools_user_valves_by_id( user_valves = UserValves(**form_data) user_valves_dict = user_valves.model_dump(exclude_unset=True) await Tools.update_user_valves_by_id_and_user_id(id, user.id, user_valves_dict, db=db) + await publish_event( + request, + EVENTS.TOOL_VALVES_UPDATED, + actor=user, + subject_id=id, + data={'scope': 'user'}, + ) return user_valves_dict except Exception as e: log.exception(f'Failed to update user valves by id {id}: {e}') diff --git a/backend/open_webui/routers/users.py b/backend/open_webui/routers/users.py index 53d500477c..2d5de22cb5 100644 --- a/backend/open_webui/routers/users.py +++ b/backend/open_webui/routers/users.py @@ -9,6 +9,7 @@ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.responses import FileResponse, Response, StreamingResponse from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import ENABLE_PROFILE_IMAGE_URL_FORWARDING, PROFILE_IMAGE_ALLOWED_MIME_TYPES, STATIC_DIR from open_webui.internal.db import get_async_session from open_webui.models.auths import Auths @@ -273,6 +274,12 @@ async def get_default_user_permissions(request: Request, user=Depends(get_admin_ async def update_default_user_permissions(request: Request, form_data: UserPermissions, user=Depends(get_admin_user)): user_permissions = form_data.model_dump(by_alias=True) await Config.upsert({'user.permissions': user_permissions}) + await publish_event( + request, + EVENTS.USER_PERMISSIONS_UPDATED, + actor=user, + subject_id='user.permissions', subject_type='config', + ) return user_permissions @@ -332,6 +339,12 @@ async def update_user_settings_by_session_user( user = await Users.update_user_settings_by_id(user.id, updated_user_settings, db=db) if user: + await publish_event( + request, + EVENTS.USER_SETTINGS_UPDATED, + actor=user, + subject_id=user.id, + ) return user.settings else: raise HTTPException( @@ -380,6 +393,12 @@ async def update_user_status_by_session_user( # user already fetched by get_verified_user — no need to refetch updated = await Users.update_user_status_by_id(user.id, form_data, db=db) if updated: + await publish_event( + request, + EVENTS.USER_STATUS_UPDATED, + actor=user, + subject_id=user.id, + ) return updated raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, @@ -562,6 +581,7 @@ async def get_user_active_status_by_id( @router.post('/{user_id}/update', response_model=UserModel | None) async def update_user_by_id( + request: Request, user_id: str, form_data: UserUpdateForm, session_user: UserModel = Depends(get_admin_user), @@ -641,6 +661,29 @@ async def update_user_by_id( # privileges cached in SESSION_POOL are invalidated. if updated_user.role != user.role: await disconnect_user_sessions(user_id) + await publish_event( + request, + EVENTS.USER_ROLE_UPDATED, + actor=session_user, + subject_id=user_id, + data={'role': updated_user.role}, + ) + else: + await publish_event( + request, + EVENTS.USER_UPDATED, + actor=session_user, + subject_id=user_id, + data={'updated_fields': list(update_data.keys())}, + ) + if form_data.password: + await publish_event( + request, + EVENTS.AUTH_PASSWORD_CHANGED, + actor=session_user, + subject_id=user_id, subject_type='user', + source='admin', + ) return updated_user raise HTTPException( @@ -660,7 +703,9 @@ async def update_user_by_id( @router.delete('/{user_id}', response_model=bool) -async def delete_user_by_id(user_id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session)): +async def delete_user_by_id( + request: Request, user_id: str, user=Depends(get_admin_user), db: AsyncSession = Depends(get_async_session) +): # Prevent deletion of the primary admin user try: first_user = await Users.get_first_user(db=db) @@ -683,6 +728,12 @@ async def delete_user_by_id(user_id: str, user=Depends(get_admin_user), db: Asyn if result: await disconnect_user_sessions(user_id) + await publish_event( + request, + EVENTS.USER_DELETED, + actor=user, + subject_id=user_id, + ) return True raise HTTPException( diff --git a/backend/open_webui/test/test_events.py b/backend/open_webui/test/test_events.py new file mode 100644 index 0000000000..b8ca8ff38e --- /dev/null +++ b/backend/open_webui/test/test_events.py @@ -0,0 +1,119 @@ +from types import SimpleNamespace + +import pytest + +from open_webui import events + + +class DummyRequest: + def __init__(self): + self.app = SimpleNamespace(state=SimpleNamespace(instance_id='instance-1', WEBUI_NAME='Test WebUI')) + + +@pytest.mark.asyncio +async def test_build_event_derives_resource_operation_and_sanitizes(): + request = DummyRequest() + + event = events.build_event( + request, + events.EVENTS.KNOWLEDGE_FILE_ADDED, + actor={ + 'id': 'user-1', + 'name': 'Ada', + 'email': 'ada@example.com', + 'role': 'admin', + 'api_key': 'secret', + }, + subject_id='file-1', + data={ + 'safe': 'value', + 'token': 'hidden', + 'nested': {'refresh_token': 'hidden', 'name': 'visible'}, + 'content': 'x' * (events.MAX_STRING_LENGTH + 10), + }, + ) + + payload = event.model_dump() + + assert payload['schema'] == events.EVENT_VERSION + assert payload['event'] == 'knowledge.file.added' + assert payload['resource'] == 'knowledge.file' + assert payload['operation'] == 'added' + assert payload['instance_id'] == 'instance-1' + assert payload['actor'] == { + 'id': 'user-1', + 'name': 'Ada', + 'email': 'ada@example.com', + 'role': 'admin', + 'type': 'user', + } + assert 'token' not in payload['data'] + assert 'refresh_token' not in payload['data']['nested'] + assert payload['data']['nested']['name'] == 'visible' + assert payload['data']['content'].endswith('...') + + +def test_build_event_accepts_events_enum(): + request = DummyRequest() + + event = events.build_event( + request, + events.EVENTS.MESSAGE_CREATED, + actor={'id': 'user-1'}, + subject_id='message-1', + ) + + payload = event.model_dump() + assert payload['event'] == 'message.created' + assert payload['resource'] == 'message' + assert payload['operation'] == 'created' + assert events.EVENTS.MESSAGE_CREATED.value in events.EVENT_CATALOG + + +@pytest.mark.asyncio +async def test_webhook_sink_sends_canonical_json(monkeypatch): + request = DummyRequest() + sent = {} + + async def fake_config_get(key, default=None): + assert key == 'webhook_url' + return 'https://example.com/events' + + async def fake_post_webhook(name, url, message, event_data): + sent.update({'name': name, 'url': url, 'message': message, 'event_data': event_data}) + return True + + monkeypatch.setattr(events.Config, 'get', fake_config_get) + monkeypatch.setattr(events, 'post_webhook', fake_post_webhook) + + event = events.build_event( + request, + events.EVENTS.USER_CREATED, + actor={'id': 'admin-1', 'name': 'Admin', 'role': 'admin'}, + subject_id='user-1', + ) + + await events.WebhookEventSink().handle_event(request.app, event) + + assert sent['name'] == 'Test WebUI' + assert sent['url'] == 'https://example.com/events' + assert sent['event_data'] == event.model_dump() + assert sent['event_data']['event'] == 'user.created' + + +@pytest.mark.asyncio +async def test_publish_event_swallows_sink_failure(monkeypatch): + request = DummyRequest() + + class FailingSink: + async def handle_event(self, app, event): + raise RuntimeError('boom') + + monkeypatch.setattr(events, 'EVENT_SINKS', [FailingSink()]) + + await events.publish_event( + request, + events.EVENTS.USER_CREATED, + actor={'id': 'admin-1'}, + subject_id='user-1', + ) diff --git a/backend/open_webui/utils/automations.py b/backend/open_webui/utils/automations.py index e651c426d8..60c359db1e 100644 --- a/backend/open_webui/utils/automations.py +++ b/backend/open_webui/utils/automations.py @@ -26,6 +26,7 @@ from zoneinfo import ZoneInfo from dateutil.rrule import rrulestr from fastapi import Request from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.internal.db import get_async_db from open_webui.models.automations import AutomationModel, AutomationRuns, Automations from open_webui.models.chats import ChatForm, Chats @@ -357,6 +358,12 @@ async def execute_automation(app, automation: AutomationModel) -> None: user = await Users.get_user_by_id(automation.user_id) if not user: await _record_run(automation.id, 'error', error='User not found') + await publish_event( + app, + EVENTS.AUTOMATION_RUN_FAILED, + subject_id=automation.id, + data={'name': automation.name, 'error': 'User not found'}, + ) return # Re-gate the rehydrated owner: a demoted/deactivated or de-permissioned owner must not run. @@ -366,7 +373,15 @@ async def execute_automation(app, automation: AutomationModel) -> None: user.role != 'admin' and not await has_permission(user.id, 'features.automations', await Config.get('user.permissions')) ): - await _record_run(automation.id, 'error', error='Owner no longer permitted to run automations') + error = 'Owner no longer permitted to run automations' + await _record_run(automation.id, 'error', error=error) + await publish_event( + app, + EVENTS.AUTOMATION_RUN_FAILED, + actor=user, + subject_id=automation.id, + data={'name': automation.name, 'error': error}, + ) return prompt = await prompt_template(automation.data['prompt'], user) @@ -418,7 +433,15 @@ async def execute_automation(app, automation: AutomationModel) -> None: ) if not chat: - await _record_run(automation.id, 'error', error='Failed to create chat') + error = 'Failed to create chat' + await _record_run(automation.id, 'error', error=error) + await publish_event( + app, + EVENTS.AUTOMATION_RUN_FAILED, + actor=user, + subject_id=automation.id, + data={'name': automation.name, 'error': error}, + ) return # Notify frontend to refresh chat list @@ -488,10 +511,24 @@ async def execute_automation(app, automation: AutomationModel) -> None: ) await _record_run(automation.id, 'success', chat_id=chat.id) + await publish_event( + app, + EVENTS.AUTOMATION_RUN_COMPLETED, + actor=user, + subject_id=automation.id, + data={'name': automation.name, 'chat_id': chat.id}, + ) except Exception as e: log.exception(f'Automation {automation.id} failed') - await _record_run(automation.id, 'error', error=str(e)[:4000]) + error = str(e)[:4000] + await _record_run(automation.id, 'error', error=error) + await publish_event( + app, + EVENTS.AUTOMATION_RUN_FAILED, + subject_id=automation.id, + data={'name': automation.name, 'error': error}, + ) #################### diff --git a/backend/open_webui/utils/oauth.py b/backend/open_webui/utils/oauth.py index f92f33a93f..fedb137887 100644 --- a/backend/open_webui/utils/oauth.py +++ b/backend/open_webui/utils/oauth.py @@ -62,7 +62,8 @@ from open_webui.config import ( OAUTH_USERNAME_CLAIM, WEBHOOK_URL, ) -from open_webui.constants import ERROR_MESSAGES, WEBHOOK_MESSAGES +from open_webui.constants import ERROR_MESSAGES +from open_webui.events import EVENTS, publish_event from open_webui.env import ( AIOHTTP_CLIENT_ALLOW_REDIRECTS, AIOHTTP_CLIENT_SESSION_SSL, @@ -73,7 +74,6 @@ from open_webui.env import ( REDIS_KEY_PREFIX, WEBUI_AUTH_COOKIE_SAME_SITE, WEBUI_AUTH_COOKIE_SECURE, - WEBUI_NAME, ) from open_webui.models.auths import Auths from open_webui.models.config import Config @@ -84,7 +84,6 @@ from open_webui.retrieval.web.utils import validate_url from open_webui.utils.auth import create_token, get_password_hash from open_webui.utils.groups import apply_default_group_assignment from open_webui.utils.misc import parse_duration -from open_webui.utils.webhook import post_webhook from starlette.responses import RedirectResponse @@ -1248,6 +1247,7 @@ class OAuthManager: """ provider = session.provider token_data = session.token + auth_config = await get_oauth_runtime_config() if not token_data.get('refresh_token'): log.warning(f'No refresh token available for session {session.id}') @@ -1837,20 +1837,16 @@ class OAuthManager: await Users.update_user_role_by_id(user.id, 'admin', db=db) user = await Users.get_user_by_id(user.id, db=db) - if auth_config.WEBHOOK_URL: - await post_webhook( - WEBUI_NAME, - auth_config.WEBHOOK_URL, - WEBHOOK_MESSAGES.USER_SIGNUP(user.name), - { - 'action': 'signup', - 'message': WEBHOOK_MESSAGES.USER_SIGNUP(user.name), - 'user': user.model_dump_json(exclude_none=True), - }, - ) - default_group_id = await Config.get('ui.default_group_id') await apply_default_group_assignment(default_group_id, user.id, db=db) + await publish_event( + request, + EVENTS.USER_CREATED, + actor=user, + subject_id=user.id, + source='oauth', + data={'role': user.role, 'provider': provider}, + ) else: raise HTTPException( diff --git a/backend/open_webui/utils/webhook.py b/backend/open_webui/utils/webhook.py index 8a65f348a8..5c4163c031 100644 --- a/backend/open_webui/utils/webhook.py +++ b/backend/open_webui/utils/webhook.py @@ -34,12 +34,14 @@ async def post_webhook(name: str, url: str, message: str, event_data: dict) -> b # Microsoft Teams Webhooks elif 'webhook.office.com' in url: action = event_data.get('action', 'undefined') - user_data = event_data.get('user', '{}') + user_data = event_data.get('user') or event_data.get('actor') or {} if isinstance(user_data, dict): user_dict = user_data else: user_dict = json.loads(user_data) - facts = [{'name': name, 'value': value} for name, value in user_dict.items()] + facts = [{'name': key, 'value': value} for key, value in user_dict.items()] + if event_data.get('event'): + facts.insert(0, {'name': 'event', 'value': event_data.get('event')}) payload = { '@type': 'MessageCard', '@context': 'http://schema.org/extensions', @@ -57,7 +59,7 @@ async def post_webhook(name: str, url: str, message: str, event_data: dict) -> b } # Default Payload else: - payload = {**event_data} + payload = event_data log.debug(f'payload: {payload}') async with aiohttp.ClientSession(