From 3814b7b2f152c671fd3789e5a0099313d5316467 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sat, 31 Oct 2026 16:50:01 +0900 Subject: [PATCH] refac --- backend/open_webui/config.py | 2 + backend/open_webui/events.py | 90 ++-- backend/open_webui/main.py | 35 +- backend/open_webui/migrations/env.py | 1 + .../e42c8f710a36_add_notification_inbox.py | 52 ++ backend/open_webui/models/access_grants.py | 240 ++-------- backend/open_webui/models/automations.py | 46 ++ backend/open_webui/models/calendar.py | 9 + backend/open_webui/models/channels.py | 5 + backend/open_webui/models/chats.py | 24 +- backend/open_webui/models/comments.py | 18 +- backend/open_webui/models/knowledge.py | 7 +- backend/open_webui/models/messages.py | 46 ++ backend/open_webui/models/models.py | 13 +- backend/open_webui/models/notes.py | 4 +- backend/open_webui/models/notifications.py | 162 +++++++ backend/open_webui/models/prompts.py | 7 +- backend/open_webui/models/shared_chats.py | 4 - backend/open_webui/models/skills.py | 4 +- backend/open_webui/models/tools.py | 4 +- backend/open_webui/models/users.py | 3 + backend/open_webui/routers/auths.py | 4 + backend/open_webui/routers/automations.py | 19 + backend/open_webui/routers/channels.py | 25 +- backend/open_webui/routers/comments.py | 51 ++ backend/open_webui/routers/functions.py | 3 +- backend/open_webui/routers/notifications.py | 125 ++++- backend/open_webui/routers/search.py | 2 +- backend/open_webui/routers/terminals.py | 6 +- backend/open_webui/socket/comments.py | 49 +- backend/open_webui/utils/automations.py | 128 +++-- backend/open_webui/utils/inbox.py | 451 ++++++++++++++++++ src/app.d.ts | 4 + src/lib/apis/notifications/inbox.ts | 56 +++ .../components/admin/Settings/General.svelte | 7 + .../automations/AutomationEditor.svelte | 41 +- .../calendar/CalendarEventModal.svelte | 15 +- src/lib/components/chat/Chat.svelte | 20 +- .../components/comments/CommentsPanel.svelte | 27 +- src/lib/components/icons/Bell.svelte | 20 + src/lib/components/layout/InboxBadge.svelte | 13 + src/lib/components/layout/InboxButton.svelte | 42 ++ src/lib/components/layout/InboxHost.svelte | 81 ++++ src/lib/components/layout/InboxPanel.svelte | 244 ++++++++++ src/lib/components/layout/Sidebar.svelte | 93 ++-- src/lib/i18n/locales/en-US/translation.json | 15 + src/lib/stores/inbox.ts | 20 + src/lib/stores/index.ts | 1 + .../(app)/automations/[id]/+page.svelte | 28 +- src/routes/(app)/calendar/+page.svelte | 34 +- 50 files changed, 1936 insertions(+), 464 deletions(-) create mode 100644 backend/open_webui/migrations/versions/e42c8f710a36_add_notification_inbox.py create mode 100644 backend/open_webui/models/notifications.py create mode 100644 backend/open_webui/utils/inbox.py create mode 100644 src/lib/apis/notifications/inbox.ts create mode 100644 src/lib/components/icons/Bell.svelte create mode 100644 src/lib/components/layout/InboxBadge.svelte create mode 100644 src/lib/components/layout/InboxButton.svelte create mode 100644 src/lib/components/layout/InboxHost.svelte create mode 100644 src/lib/components/layout/InboxPanel.svelte create mode 100644 src/lib/stores/inbox.ts diff --git a/backend/open_webui/config.py b/backend/open_webui/config.py index e20b20ffe3..8c09b5ce82 100644 --- a/backend/open_webui/config.py +++ b/backend/open_webui/config.py @@ -2127,6 +2127,7 @@ AUTOMATION_MIN_INTERVAL = os.getenv('AUTOMATION_MIN_INTERVAL', '') AUTOMATION_AUTH_TOKEN_EXPIRES_IN = os.getenv('AUTOMATION_AUTH_TOKEN_EXPIRES_IN', '1h') ENABLE_COMMENTS = os.getenv('ENABLE_COMMENTS', 'False').lower() == 'true' +ENABLE_NOTIFICATIONS = os.getenv('ENABLE_NOTIFICATIONS', 'True').lower() == 'true' ENABLE_NOTES = os.getenv('ENABLE_NOTES', 'True').lower() == 'true' @@ -3259,6 +3260,7 @@ DEFAULT_CONFIG = { 'automations.auth_token_expires_in': AUTOMATION_AUTH_TOKEN_EXPIRES_IN, 'notes.enable': ENABLE_NOTES, 'comments.enable': ENABLE_COMMENTS, + 'notifications.enable': ENABLE_NOTIFICATIONS, 'users.enable_status': ENABLE_USER_STATUS, 'evaluation.arena.enable': ENABLE_EVALUATION_ARENA_MODELS, 'evaluation.arena.models': EVALUATION_ARENA_MODELS, diff --git a/backend/open_webui/events.py b/backend/open_webui/events.py index 8236e2e992..071f23dec2 100644 --- a/backend/open_webui/events.py +++ b/backend/open_webui/events.py @@ -5,9 +5,9 @@ import inspect import logging import time import uuid -from types import SimpleNamespace from typing import Any +from fastapi import Request from pydantic import BaseModel, ConfigDict, Field, model_validator from open_webui.env import ENABLE_FUNCTIONS, VERSION @@ -43,6 +43,13 @@ class EventDefinition(BaseModel): class EventDefinitions(BaseModel): model_config = ConfigDict(frozen=True) + COMMENT_CREATED: EventDefinition = EventDefinition( + name='comment.created', description='A resource comment or reply was created.', message='Comment created' + ) + COMMENT_UPDATED: EventDefinition = EventDefinition( + name='comment.updated', description='A resource comment was edited.', message='Comment updated' + ) + SYSTEM_STARTUP_STARTED: EventDefinition = EventDefinition( name='system.startup.started', description='Application startup began.', message='Startup started' ) @@ -710,7 +717,6 @@ NOTIFICATION_EVENTS = ( EVENTS.CHAT_FAILED.name, EVENTS.CHANNEL_MESSAGE.name, EVENTS.CALENDAR_ALERT.name, - EVENTS.RETRIEVAL_CONTENT_PROCESS_FAILED.name, EVENTS.AUTOMATION_RUN_FAILED.name, ) @@ -1033,7 +1039,7 @@ def event_name(event: EventDefinition | str) -> str: def build_event( - request_or_app: Any, + request: Request | None, event: EventDefinition | str, *, actor: Any | None = None, @@ -1042,9 +1048,10 @@ def build_event( source: str = 'api', data: dict | None = None, message: str | None = None, + event_id: str | None = None, ) -> Event: event_name_value = event_name(event) - app = getattr(request_or_app, 'app', request_or_app) + app = request.app if request is not None else None parts = event_name_value.split('.') resource = '.'.join(parts[:-1]) instance_id = getattr(getattr(app, 'state', None), 'instance_id', None) @@ -1056,7 +1063,7 @@ def build_event( return Event( schema=VERSION, - id=str(uuid.uuid4()), + id=event_id or str(uuid.uuid4()), event=event_name_value, resource=resource, operation=parts[-1], @@ -1107,8 +1114,8 @@ def schedule_webhook_dispatch(app: Any, event: Event) -> None: class WebhookEventSink: - async def handle_event(self, app: Any, event: Event, request: Any | None = None) -> None: - schedule_webhook_dispatch(app, event) + async def handle_event(self, request: Request, event: Event) -> None: + schedule_webhook_dispatch(request.app, event) def schedule_notification_dispatch(app: Any, event: Event) -> None: @@ -1121,13 +1128,13 @@ def schedule_notification_dispatch(app: Any, event: Event) -> None: class NotificationEventSink: - async def handle_event(self, app: Any, event: Event, request: Any | None = None) -> None: + async def handle_event(self, request: Request, event: Event) -> None: if event.event in NOTIFICATION_EVENTS: - schedule_notification_dispatch(app, event) + schedule_notification_dispatch(request.app, event) class SocketSessionEventSink: - async def handle_event(self, app: Any, event: Event, request: Any | None = None) -> None: + async def handle_event(self, request: Request, event: Event) -> None: from open_webui.socket.main import refresh_chat_access if event.event in { @@ -1162,7 +1169,7 @@ class SocketSessionEventSink: async def dispatch_event_functions( - app: Any, event: Event, request: Any | None = None, extra_function_ids: list[str] | None = None + request: Request, event: Event, extra_function_ids: list[str] | None = None ) -> None: if not ENABLE_FUNCTIONS: return @@ -1170,7 +1177,6 @@ async def dispatch_event_functions( from open_webui.models.functions import Functions from open_webui.utils.plugin import get_function_module_from_cache - context = request or SimpleNamespace(app=app) event_payload = event.model_dump() try: @@ -1187,7 +1193,7 @@ async def dispatch_event_functions( for function in event_functions: try: - function_module, _, _ = await get_function_module_from_cache(context, function.id, function=function) + function_module, _, _ = await get_function_module_from_cache(request, function.id, function=function) handler = getattr(function_module, 'event', None) if not handler: continue @@ -1204,7 +1210,7 @@ async def dispatch_event_functions( '__event__': event, '__event_id__': event.id, '__event_name__': event.event, - '__app__': app, + '__app__': request.app, '__request__': request, } params = {key: value for key, value in extra_params.items() if accepts_kwargs or key in sig.parameters} @@ -1217,24 +1223,32 @@ async def dispatch_event_functions( log.exception('Event function failed for %s', function.id) -def schedule_event_function_dispatch(app: Any, event: Event, request: Any | None = None) -> None: +def schedule_event_function_dispatch(request: Request, event: Event) -> None: try: - asyncio.create_task(dispatch_event_functions(app, event, request)) + asyncio.create_task(dispatch_event_functions(request, event)) except RuntimeError: log.exception('Event functions could not be scheduled for %s', event.event) class EventFunctionSink: - async def handle_event(self, app: Any, event: Event, request: Any | None = None) -> None: - schedule_event_function_dispatch(app, event, request) + async def handle_event(self, request: Request, event: Event) -> None: + schedule_event_function_dispatch(request, event) EVENT_SINKS = [SocketSessionEventSink(), EventFunctionSink(), WebhookEventSink(), NotificationEventSink()] +async def _dispatch_event(request, event_payload): + for sink in EVENT_SINKS: + try: + await sink.handle_event(request, event_payload) + except Exception: + log.exception('Event sink failed for %s', event_payload.event) + + async def publish_event( - request_or_app: Any, - event: EventDefinition | str, + request: Request, + event: EventDefinition | str | Event, *, actor: Any | None = None, subject_id: Any | None = None, @@ -1243,28 +1257,26 @@ async def publish_event( data: dict | None = None, message: str | None = None, ) -> None: - app = getattr(request_or_app, 'app', request_or_app) - request = request_or_app if hasattr(request_or_app, 'app') else None - event_payload = build_event( - request_or_app, - event, - actor=actor, - subject_id=subject_id, - subject_type=subject_type, - source=source, - data=data, - message=message, + """Deliver an event through the existing sinks. Call after committing source changes.""" + event_payload = ( + event + if isinstance(event, Event) + else build_event( + request, + 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, request=request) - except Exception: - log.exception('Event sink failed for %s', event_payload.event) + await _dispatch_event(request, event_payload) async def publish_model_provider_request_failed( - request_or_app: Any, + request: Request, *, actor: Any | None, provider: str, @@ -1325,7 +1337,7 @@ async def publish_model_provider_request_failed( data['upstream_message'] = error await publish_event( - request_or_app, + request, EVENTS.MODEL_PROVIDER_REQUEST_FAILED, actor=actor, subject_id=requested_model, diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py index e7410a1c03..04c5bfbeae 100644 --- a/backend/open_webui/main.py +++ b/backend/open_webui/main.py @@ -392,7 +392,18 @@ async def lifespan(app: FastAPI): await validate_mfa_configuration() await initialize_runtime_config(app) await migrate_legacy_webhook_config() - await publish_event(app, EVENTS.SYSTEM_STARTUP_STARTED, source='system') + request = Request( + { + 'type': 'http', + 'method': 'GET', + 'path': '/internal', + 'query_string': b'', + 'headers': [], + 'scheme': 'http', + 'app': app, + } + ) + await publish_event(request, EVENTS.SYSTEM_STARTUP_STARTED, source='system') license_task = None if LICENSE_KEY: @@ -503,11 +514,11 @@ async def lifespan(app: FastAPI): log.warning(f'License data retrieval failed during startup: {e}') app.state.startup_complete = True - await publish_event(app, EVENTS.SYSTEM_STARTUP_COMPLETED, source='system') + await publish_event(request, EVENTS.SYSTEM_STARTUP_COMPLETED, source='system') yield - await publish_event(app, EVENTS.SYSTEM_SHUTDOWN_STARTED, source='system') + await publish_event(request, EVENTS.SYSTEM_SHUTDOWN_STARTED, source='system') await app_connections.close() # Shutdown: clean up shared resources @@ -529,7 +540,7 @@ async def lifespan(app: FastAPI): app.state.periodic_session_pool_cleanup.cancel() app.state.scheduler_worker_loop.cancel() - await publish_event(app, EVENTS.SYSTEM_SHUTDOWN_COMPLETED, source='system') + await publish_event(request, EVENTS.SYSTEM_SHUTDOWN_COMPLETED, source='system') # ENABLE_ORJSON: orjson for request-body parsing and JSONResponse bodies; @@ -2358,6 +2369,7 @@ async def get_app_config(request: Request): 'automations.enable', 'notes.enable', 'comments.enable', + 'notifications.enable', 'chat.context_compaction.enable', 'chat.tool_permissions.enable', 'web.search.enable', @@ -2466,6 +2478,7 @@ async def get_app_config(request: Request): 'enable_automations': config.get('automations.enable'), 'enable_notes': config.get('notes.enable'), 'enable_comments': config.get('comments.enable'), + 'enable_notifications': config.get('notifications.enable'), 'enable_context_compaction': config.get('chat.context_compaction.enable'), 'enable_tool_permissions': config.get('chat.tool_permissions.enable'), 'enable_web_search': config.get('web.search.enable'), @@ -2621,7 +2634,7 @@ async def get_event_webhooks_api(user=Depends(get_admin_user)): @app.post('/api/events/webhooks') -async def create_event_webhook(form_data: EventWebhookForm, user=Depends(get_admin_user)): +async def create_event_webhook(request: Request, form_data: EventWebhookForm, user=Depends(get_admin_user)): try: webhook = await upsert_event_webhook( { @@ -2636,7 +2649,7 @@ async def create_event_webhook(form_data: EventWebhookForm, user=Depends(get_adm raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) await publish_event( - app, + request, EVENTS.CONFIG_WEBHOOK_UPDATED, actor=user, subject_id=webhook['id'], @@ -2652,7 +2665,9 @@ async def create_event_webhook(form_data: EventWebhookForm, user=Depends(get_adm @app.put('/api/events/webhooks/{webhook_id}') -async def update_event_webhook(webhook_id: str, form_data: EventWebhookUpdateForm, user=Depends(get_admin_user)): +async def update_event_webhook( + request: Request, webhook_id: str, form_data: EventWebhookUpdateForm, user=Depends(get_admin_user) +): webhooks = await get_event_webhooks() existing = next((webhook for webhook in webhooks if webhook.get('id') == webhook_id), None) if not existing: @@ -2670,7 +2685,7 @@ async def update_event_webhook(webhook_id: str, form_data: EventWebhookUpdateFor raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) await publish_event( - app, + request, EVENTS.CONFIG_WEBHOOK_UPDATED, actor=user, subject_id=webhook_id, @@ -2686,13 +2701,13 @@ async def update_event_webhook(webhook_id: str, form_data: EventWebhookUpdateFor @app.delete('/api/events/webhooks/{webhook_id}') -async def delete_event_webhook_api(webhook_id: str, user=Depends(get_admin_user)): +async def delete_event_webhook_api(request: Request, webhook_id: str, user=Depends(get_admin_user)): deleted = await delete_event_webhook(webhook_id) if not deleted: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Webhook not found') await publish_event( - app, + request, EVENTS.CONFIG_WEBHOOK_UPDATED, actor=user, subject_id=webhook_id, diff --git a/backend/open_webui/migrations/env.py b/backend/open_webui/migrations/env.py index 62a73d4f4a..c6e385d74d 100644 --- a/backend/open_webui/migrations/env.py +++ b/backend/open_webui/migrations/env.py @@ -15,6 +15,7 @@ from open_webui.models.calendar import Calendar, CalendarEvent, CalendarEventAtt from open_webui.models.chat_messages import ChatMessage # noqa: F401 from open_webui.models.chats import Chat # noqa: F401 from open_webui.models.comments import Comment, CommentReaction # noqa: F401 +from open_webui.models.notifications import Notification # noqa: F401 from sqlalchemy import create_engine, engine_from_config, inspect, pool alembic_config = alembic.context.config diff --git a/backend/open_webui/migrations/versions/e42c8f710a36_add_notification_inbox.py b/backend/open_webui/migrations/versions/e42c8f710a36_add_notification_inbox.py new file mode 100644 index 0000000000..f5f455a5fa --- /dev/null +++ b/backend/open_webui/migrations/versions/e42c8f710a36_add_notification_inbox.py @@ -0,0 +1,52 @@ +"""Add persistent general inbox notifications. + +Revision ID: e42c8f710a36 +Revises: 6f8c2a91d4e7 +""" + +import sqlalchemy as sa +from alembic import op + +revision = 'e42c8f710a36' +down_revision = '6f8c2a91d4e7' +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + 'notification', + sa.Column('id', sa.Text(), primary_key=True), + sa.Column('user_id', sa.Text(), nullable=False), + sa.Column('event_type', sa.Text(), nullable=False), + sa.Column('event_key', sa.Text(), nullable=False), + sa.Column('actor_type', sa.Text(), nullable=False), + sa.Column('actor_id', sa.Text(), nullable=True), + sa.Column('resource_type', sa.Text(), nullable=True), + sa.Column('resource_id', sa.Text(), nullable=True), + sa.Column('object_type', sa.Text(), nullable=True), + sa.Column('object_id', sa.Text(), nullable=True), + sa.Column('data', sa.JSON(), nullable=False), + sa.Column('seen_at', sa.BigInteger(), nullable=True), + sa.Column('read_at', sa.BigInteger(), nullable=True), + sa.Column('created_at', sa.BigInteger(), nullable=False), + sa.Column('updated_at', sa.BigInteger(), nullable=False), + sa.UniqueConstraint('user_id', 'event_key', name='uq_notification_occurrence'), + sa.CheckConstraint('(resource_type IS NULL) = (resource_id IS NULL)', name='ck_notification_resource'), + sa.CheckConstraint('(object_type IS NULL) = (object_id IS NULL)', name='ck_notification_object'), + sa.CheckConstraint("actor_type IN ('user', 'system')", name='ck_notification_actor'), + ) + op.create_index('ix_notification_user_created', 'notification', ['user_id', 'created_at', 'id']) + op.create_index( + 'ix_notification_unseen', + 'notification', + ['user_id', 'created_at', 'id'], + sqlite_where=sa.text('seen_at IS NULL'), + postgresql_where=sa.text('seen_at IS NULL'), + ) + op.create_index('ix_notification_resource', 'notification', ['resource_type', 'resource_id']) + op.create_index('ix_notification_object', 'notification', ['object_type', 'object_id']) + + +def downgrade(): + op.drop_table('notification') diff --git a/backend/open_webui/models/access_grants.py b/backend/open_webui/models/access_grants.py index b9c2221f7a..53e9a81a69 100644 --- a/backend/open_webui/models/access_grants.py +++ b/backend/open_webui/models/access_grants.py @@ -582,47 +582,17 @@ class AccessGrantsTable: - There's a grant for any of the user's groups with the requested permission """ async with get_async_db_context(db) as db: - # Build conditions for matching grants - conditions = [ - # Public access - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == '*', - ), - # Direct user access - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == user_id, - ), - ] - - # Group access if user_group_ids is None: from open_webui.models.groups import Groups user_groups = await Groups.get_groups_by_member_id(user_id, db=db, include_inherited=True) user_group_ids = {group.id for group in user_groups} - if user_group_ids: - conditions.append( - and_( - AccessGrant.principal_type == 'group', - AccessGrant.principal_id.in_(user_group_ids), - ) + return bool( + await db.scalar( + select(self.has_grant_expression(user_id, resource_type, resource_id, permission, user_group_ids)) ) - - result = await db.execute( - select(AccessGrant) - .filter( - AccessGrant.resource_type == resource_type, - AccessGrant.resource_id == resource_id, - AccessGrant.permission == permission, - or_(*conditions), - ) - .limit(1) ) - grant = result.scalars().first() - return grant is not None async def get_accessible_resource_ids( self, @@ -766,122 +736,55 @@ class AccessGrantsTable: user_ids = await self.get_user_ids_by_access_grants(access_grants, permission, db=db) return previous_user_ids - user_ids + def has_grant_expression(self, user_id, resource_type, resource_id, permission='read', user_group_ids=()): + """SQL grant check for one resource or a correlated resource column; ownership is separate.""" + return ( + select(AccessGrant.id) + .where( + AccessGrant.resource_type == resource_type, + AccessGrant.resource_id == resource_id, + AccessGrant.permission == permission, + or_( + and_( + AccessGrant.principal_type == 'user', + AccessGrant.principal_id.in_(['*', user_id] if user_id else ['*']), + ), + and_(AccessGrant.principal_type == 'group', AccessGrant.principal_id.in_(user_group_ids or ())), + ), + ) + .correlate_except(AccessGrant) + .exists() + ) + def has_permission_filter( self, - db, query, DocumentModel, filter: dict, resource_type: str, permission: str = 'read', + *, + db, ): - """ - Apply access control filtering to a SQLAlchemy query by JOINing with access_grant. - - This replaces the old JSON-column-based filtering with a proper relational JOIN. - - Note: This method builds SQLAlchemy expressions and does NOT perform I/O itself, - so it remains synchronous. The caller is responsible for executing the query - asynchronously with `await db.execute(...)`. - """ + """Apply owner-or-grant filtering without loading resources or duplicating rows.""" + if permission == 'read_only': + return self._has_read_only_permission_filter(query, DocumentModel, filter, resource_type, db=db) group_ids = filter.get('group_ids', []) user_id = filter.get('user_id') - - if permission == 'read_only': - return self._has_read_only_permission_filter(db, query, DocumentModel, filter, resource_type) - - # Build principal conditions - principal_conditions = [] - - if group_ids or user_id: - # Public access: user:* read - principal_conditions.append( - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == '*', - ) - ) - - if user_id: - # Owner always has access - principal_conditions.append(DocumentModel.user_id == user_id) - - # Direct user grant - principal_conditions.append( - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == user_id, - ) - ) - - if group_ids: - # Group grants - principal_conditions.append( - and_( - AccessGrant.principal_type == 'group', - AccessGrant.principal_id.in_(group_ids), - ) - ) - - if not principal_conditions: + if not group_ids and not user_id: return query - - # LEFT JOIN access_grant and filter - # We use a subquery approach to avoid duplicates from multiple matching grants - from sqlalchemy import exists as sa_exists - - grant_exists = ( - select(AccessGrant.id) - .where( - AccessGrant.resource_type == resource_type, - AccessGrant.resource_id == DocumentModel.id, - AccessGrant.permission == permission, - or_( - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == '*', - ), - *( - [ - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == user_id, - ) - ] - if user_id - else [] - ), - *( - [ - and_( - AccessGrant.principal_type == 'group', - AccessGrant.principal_id.in_(group_ids), - ) - ] - if group_ids - else [] - ), - ), - ) - .correlate(DocumentModel) - .exists() - ) - - # Owner OR has a matching grant - owner_or_grant = [grant_exists] + allowed = self.has_grant_expression(user_id, resource_type, DocumentModel.id, permission, group_ids) if user_id: - owner_or_grant.append(DocumentModel.user_id == user_id) - - query = query.filter(or_(*owner_or_grant)) - return query + allowed = or_(DocumentModel.user_id == user_id, allowed) + return query.filter(allowed) def _has_read_only_permission_filter( self, - db, query, DocumentModel, filter: dict, resource_type: str, + db, ): """ Filter for items where user has read BUT NOT write access. @@ -895,81 +798,8 @@ class AccessGrantsTable: group_ids = filter.get('group_ids', []) user_id = filter.get('user_id') - from sqlalchemy import exists as sa_exists - - read_grant_exists = ( - select(AccessGrant.id) - .where( - AccessGrant.resource_type == resource_type, - AccessGrant.resource_id == DocumentModel.id, - AccessGrant.permission == 'read', - or_( - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == '*', - ), - *( - [ - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == user_id, - ) - ] - if user_id - else [] - ), - *( - [ - and_( - AccessGrant.principal_type == 'group', - AccessGrant.principal_id.in_(group_ids), - ) - ] - if group_ids - else [] - ), - ), - ) - .correlate(DocumentModel) - .exists() - ) - - write_grant_exists = ( - select(AccessGrant.id) - .where( - AccessGrant.resource_type == resource_type, - AccessGrant.resource_id == DocumentModel.id, - AccessGrant.permission == 'write', - or_( - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == '*', - ), - *( - [ - and_( - AccessGrant.principal_type == 'user', - AccessGrant.principal_id == user_id, - ) - ] - if user_id - else [] - ), - *( - [ - and_( - AccessGrant.principal_type == 'group', - AccessGrant.principal_id.in_(group_ids), - ) - ] - if group_ids - else [] - ), - ), - ) - .correlate(DocumentModel) - .exists() - ) + read_grant_exists = self.has_grant_expression(user_id, resource_type, DocumentModel.id, 'read', group_ids) + write_grant_exists = self.has_grant_expression(user_id, resource_type, DocumentModel.id, 'write', group_ids) conditions = [read_grant_exists, ~write_grant_exists] diff --git a/backend/open_webui/models/automations.py b/backend/open_webui/models/automations.py index 0aab02e5f6..bba4b1d028 100644 --- a/backend/open_webui/models/automations.py +++ b/backend/open_webui/models/automations.py @@ -3,6 +3,7 @@ import time from typing import Literal, Optional from uuid import uuid4 +from open_webui.models.notifications import emit_changes from open_webui.internal.db import Base, get_async_db_context from open_webui.utils.misc import json_text_variants from pydantic import BaseModel, ConfigDict @@ -291,8 +292,12 @@ class AutomationTable: row = await db.get(Automation, id) if not row: return False + from open_webui.models.notifications import delete_notifications + + recipients = await delete_notifications(resource_type='automation', resource_ids=[id], db=db) await db.delete(row) await db.commit() + await emit_changes(recipients) return True async def claim_due(self, now_ns: int, limit: int = 10, db: Optional[AsyncSession] = None) -> list[AutomationModel]: @@ -361,6 +366,9 @@ class AutomationRunTable: status: str, chat_id: Optional[str] = None, error: Optional[str] = None, + *, + request=None, + event_actor=None, db: Optional[AsyncSession] = None, ) -> AutomationRunModel: async with get_async_db_context(db) as db: @@ -373,7 +381,37 @@ class AutomationRunTable: created_at=int(time.time_ns()), ) db.add(row) + await db.flush() + event = None + recipients = set() + if status == 'error': + from open_webui.events import EVENTS, build_event, publish_event + from open_webui.utils.inbox import consume_event + from open_webui.models.notifications import emit_changes + + automation = await db.get(Automation, automation_id) + if automation: + event = build_event( + request, + EVENTS.AUTOMATION_RUN_FAILED, + actor=event_actor, + subject_id=automation_id, + event_id=f'automation-run:{row.id}', + data={ + 'run_id': row.id, + 'name': automation.name, + 'error': error, + 'user_id': automation.user_id, + 'message': f'{automation.name}: {error}', + **({'chat_id': chat_id} if chat_id else {}), + }, + ) + recipients = await consume_event(event, db=db) await db.commit() + if event is not None: + await emit_changes(recipients) + if request is not None: + await publish_event(request, event) return AutomationRunModel.model_validate(row) async def get_latest(self, automation_id: str, db: Optional[AsyncSession] = None) -> Optional[AutomationRunModel]: @@ -434,8 +472,16 @@ class AutomationRunTable: async def delete_by_automation(self, automation_id: str, db: Optional[AsyncSession] = None) -> int: async with get_async_db_context(db) as db: + from open_webui.models.notifications import delete_notifications + + recipients = await delete_notifications( + object_type='automation_run', + object_ids=select(AutomationRun.id).where(AutomationRun.automation_id == automation_id), + db=db, + ) result = await db.execute(delete(AutomationRun).filter_by(automation_id=automation_id)) await db.commit() + await emit_changes(recipients) return result.rowcount async def get_runs_by_user_range( diff --git a/backend/open_webui/models/calendar.py b/backend/open_webui/models/calendar.py index a7489e925f..42fbe47cea 100644 --- a/backend/open_webui/models/calendar.py +++ b/backend/open_webui/models/calendar.py @@ -3,6 +3,7 @@ import time from typing import Optional from uuid import uuid4 +from open_webui.models.notifications import emit_changes from open_webui.constants import ERROR_MESSAGES from open_webui.internal.db import Base, get_async_db_context from open_webui.models.access_grants import AccessGrantModel, AccessGrants @@ -408,8 +409,12 @@ class CalendarTable: await db.execute(delete(CalendarEvent).filter(CalendarEvent.calendar_id == id)) # Delete calendar + from open_webui.models.notifications import delete_notifications + + recipients = await delete_notifications(resource_type='calendar', resource_ids=[id], db=db) await db.execute(delete(Calendar).filter(Calendar.id == id)) await db.commit() + await emit_changes(recipients) # Revoke access grants in a separate transaction to avoid # write-lock contention on SQLite when session sharing is off. @@ -772,8 +777,12 @@ class CalendarEventTable: try: async with get_async_db_context(db) as db: await db.execute(delete(CalendarEventAttendee).filter(CalendarEventAttendee.event_id == id)) + from open_webui.models.notifications import delete_notifications + + recipients = await delete_notifications(object_type='calendar_event', object_ids=[id], db=db) await db.execute(delete(CalendarEvent).filter(CalendarEvent.id == id)) await db.commit() + await emit_changes(recipients) return True except Exception: return False diff --git a/backend/open_webui/models/channels.py b/backend/open_webui/models/channels.py index 1876f41c03..140562f1c2 100644 --- a/backend/open_webui/models/channels.py +++ b/backend/open_webui/models/channels.py @@ -3,6 +3,7 @@ import time import uuid from typing import Optional +from open_webui.models.notifications import emit_changes from open_webui.internal.db import Base, JSONField, get_async_db_context from open_webui.models.access_grants import ( AccessGrantModel, @@ -963,8 +964,12 @@ class ChannelTable: async def delete_channel_by_id(self, id: str, db: Optional[AsyncSession] = None) -> bool: async with get_async_db_context(db) as db: await AccessGrants.revoke_all_access('channel', id, db=db) + from open_webui.models.notifications import delete_notifications + + recipients = await delete_notifications(resource_type='channel', resource_ids=[id], db=db) await db.execute(delete(Channel).filter(Channel.id == id)) await db.commit() + await emit_changes(recipients) return True #################### diff --git a/backend/open_webui/models/chats.py b/backend/open_webui/models/chats.py index f5977e8ec2..b34704a7c9 100644 --- a/backend/open_webui/models/chats.py +++ b/backend/open_webui/models/chats.py @@ -11,6 +11,7 @@ import uuid from typing import Any, Literal # local imports +from open_webui.models.notifications import emit_changes from open_webui.env import ENABLE_ADMIN_CHAT_ACCESS from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, JSONField, get_async_db, get_async_db_context @@ -2815,9 +2816,10 @@ class ChatTable: async with get_async_db_context(db) as session: await session.execute(update(AutomationRun).filter_by(chat_id=id).values(chat_id=None)) await session.execute(delete(ChatMessage).filter_by(chat_id=id)) - await delete_resource_comments(session, 'chat', select(Chat.id).filter_by(id=id)) + recipients = await delete_resource_comments('chat', select(Chat.id).filter_by(id=id), db=session) await session.execute(delete(Chat).filter_by(id=id)) await session.commit() + await emit_changes(recipients) return True and await self.delete_shared_chat_by_chat_id(id, db=session) except Exception: @@ -2828,9 +2830,12 @@ class ChatTable: async with get_async_db_context(db) as session: await session.execute(update(AutomationRun).filter_by(chat_id=id).values(chat_id=None)) await session.execute(delete(ChatMessage).filter_by(chat_id=id)) - await delete_resource_comments(session, 'chat', select(Chat.id).filter_by(id=id, user_id=user_id)) + recipients = await delete_resource_comments( + 'chat', select(Chat.id).filter_by(id=id, user_id=user_id), db=session + ) await session.execute(delete(Chat).filter_by(id=id, user_id=user_id)) await session.commit() + await emit_changes(recipients) return True and await self.delete_shared_chat_by_chat_id(id, db=session) except Exception: @@ -2850,9 +2855,12 @@ class ChatTable: await session.execute( delete(ChatMessage).filter(ChatMessage.chat_id.in_(select(Chat.id).filter_by(user_id=user_id))) ) - await delete_resource_comments(session, 'chat', select(Chat.id).filter_by(user_id=user_id)) + recipients = await delete_resource_comments( + 'chat', select(Chat.id).filter_by(user_id=user_id), db=session + ) await session.execute(delete(Chat).filter_by(user_id=user_id)) await session.commit() + await emit_changes(recipients) return True except Exception: @@ -2870,11 +2878,16 @@ class ChatTable: update(AutomationRun).filter(AutomationRun.chat_id.in_(chat_ids_stmt)).values(chat_id=None) ) await session.execute(delete(ChatMessage).filter(ChatMessage.chat_id.in_(chat_ids_stmt))) - await delete_resource_comments(session, 'chat', chat_ids_stmt) + recipients = await delete_resource_comments('chat', chat_ids_stmt, db=session) await session.execute(delete(SharedChatTable).filter(SharedChatTable.chat_id.in_(chat_ids_stmt))) - await delete_resource_comments(session, 'chat', select(Chat.id).filter_by(user_id=user_id, folder_id=folder_id)) + recipients.update( + await delete_resource_comments( + 'chat', select(Chat.id).filter_by(user_id=user_id, folder_id=folder_id), db=session + ) + ) await session.execute(delete(Chat).filter_by(user_id=user_id, folder_id=folder_id)) await session.commit() + await emit_changes(recipients) return True except Exception: @@ -2903,7 +2916,6 @@ class ChatTable: try: async with get_async_db_context(db) as session: # Delete shared_chat rows for this user's chats - await delete_resource_comments(session, 'shared_chat', select(SharedChatTable.id).filter_by(user_id=user_id)) await session.execute(delete(SharedChatTable).filter_by(user_id=user_id)) # Clear share_id on all of this user's chats diff --git a/backend/open_webui/models/comments.py b/backend/open_webui/models/comments.py index f9670d308a..cf866295db 100644 --- a/backend/open_webui/models/comments.py +++ b/backend/open_webui/models/comments.py @@ -137,27 +137,21 @@ async def comment_responses(db: AsyncSession, comments: list[Comment], user, own ] -async def delete_resource_comments(db: AsyncSession, resource_type: str, resource_ids): +async def delete_resource_comments(resource_type: str, resource_ids, db: AsyncSession): """Lock resources before cleanup so concurrent comment creation cannot leave orphans.""" from open_webui.utils.comments import resource_models - # Retain cleanup for legacy snapshot comments when their shared chat is deleted. - if resource_type == 'shared_chat': - from open_webui.models.shared_chats import SharedChat - - model = SharedChat - else: - model = resource_models()[resource_type] + model = resource_models()[resource_type] ids = list((await db.scalars(select(model.id).where(model.id.in_(resource_ids)).with_for_update())).all()) if not ids: - return - if resource_type == 'chat': - from open_webui.models.shared_chats import SharedChat + return set() + from open_webui.models.notifications import delete_notifications - await delete_resource_comments(db, 'shared_chat', select(SharedChat.id).where(SharedChat.chat_id.in_(ids))) + recipients = await delete_notifications(resource_type=resource_type, resource_ids=ids, db=db) comment_ids = select(Comment.id).where(Comment.resource_type == resource_type, Comment.resource_id.in_(ids)) await db.execute(delete(CommentReaction).where(CommentReaction.comment_id.in_(comment_ids))) await db.execute(delete(Comment).where(Comment.resource_type == resource_type, Comment.resource_id.in_(ids))) + return recipients async def cleanup_comments_for_deleted_user(db: AsyncSession, user_id: str): diff --git a/backend/open_webui/models/knowledge.py b/backend/open_webui/models/knowledge.py index b1963f2200..b2c6c0f5ec 100644 --- a/backend/open_webui/models/knowledge.py +++ b/backend/open_webui/models/knowledge.py @@ -3,6 +3,7 @@ import time import uuid from typing import Optional +from open_webui.models.notifications import emit_changes from open_webui.config import RAG_FILE_CONTENT_SEARCH_MAX_CHARS from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, JSONField, get_async_db_context @@ -860,9 +861,10 @@ class KnowledgeTable: try: async with get_async_db_context(db) as db: await AccessGrants.revoke_all_access('knowledge', id, db=db) - await delete_resource_comments(db, 'knowledge', [id]) + recipients = await delete_resource_comments('knowledge', [id], db=db) await db.execute(delete(Knowledge).filter_by(id=id)) await db.commit() + await emit_changes(recipients) return True except Exception: return False @@ -874,9 +876,10 @@ class KnowledgeTable: knowledge_ids = [row[0] for row in result.all()] for knowledge_id in knowledge_ids: await AccessGrants.revoke_all_access('knowledge', knowledge_id, db=db) - await delete_resource_comments(db, 'knowledge', knowledge_ids) + recipients = await delete_resource_comments('knowledge', knowledge_ids, db=db) await db.execute(delete(Knowledge)) await db.commit() + await emit_changes(recipients) return True except Exception: diff --git a/backend/open_webui/models/messages.py b/backend/open_webui/models/messages.py index b6e04bb973..8568420f71 100644 --- a/backend/open_webui/models/messages.py +++ b/backend/open_webui/models/messages.py @@ -2,6 +2,7 @@ import time import uuid from typing import Optional +from open_webui.models.notifications import emit_changes from open_webui.internal.db import Base, JSONField, get_async_db_context from open_webui.models.channels import ChannelMember, Channels from open_webui.models.tags import Tag, TagModel, Tags @@ -156,6 +157,10 @@ class MessageTable: form_data: MessageForm, channel_id: str, user_id: str, + *, + request=None, + event_actor=None, + event_source='api', db: Optional[AsyncSession] = None, ) -> Optional[MessageModel]: async with get_async_db_context(db) as db: @@ -185,8 +190,25 @@ class MessageTable: db.add(result) from open_webui.utils.apps import publish_app_message_event + from open_webui.events import EVENTS, build_event, publish_event + from open_webui.utils.inbox import consume_event + from open_webui.models.notifications import emit_changes + await db.flush() + event = build_event( + request, + EVENTS.MESSAGE_CREATED, + actor=event_actor or {'id': user_id}, + subject_id=result.id, + source=event_source, + data={'channel_id': channel_id, 'content_preview': result.content[:300]}, + event_id=f'message:{result.id}:created', + ) + recipients = await consume_event(event, db=db) await db.commit() + await emit_changes(recipients) + if request is not None: + await publish_event(request, event) await db.refresh(result) await publish_app_message_event(result, 'channel.message.created') return MessageModel.model_validate(result) if result else None @@ -209,7 +231,21 @@ class MessageTable: updated_at=now, ) db.add(message) + from open_webui.events import EVENTS, build_event + from open_webui.utils.inbox import consume_event + from open_webui.models.notifications import emit_changes + + await db.flush() + event = build_event( + None, + EVENTS.MESSAGE_CREATED, + subject_id=message.id, + actor={'id': app_id, 'type': 'app'}, + event_id=f'message:{message.id}:created', + ) + recipients = await consume_event(event, db=db) await db.commit() + await emit_changes(recipients) from open_webui.utils.apps import publish_app_message_event await publish_app_message_event(message, 'channel.message.created') @@ -703,8 +739,14 @@ class MessageTable: from open_webui.utils.apps import publish_app_message_event messages = (await db.scalars(select(Message).filter_by(parent_id=id))).all() + from open_webui.models.notifications import delete_notifications + + recipients = await delete_notifications( + object_type='message', object_ids=select(Message.id).where(Message.parent_id == id), db=db + ) await db.execute(delete(Message).filter_by(parent_id=id)) await db.commit() + await emit_changes(recipients) for message in messages: await publish_app_message_event(message, 'channel.message.deleted') return True @@ -714,12 +756,16 @@ class MessageTable: from open_webui.utils.apps import publish_app_message_event message = await db.get(Message, id) + from open_webui.models.notifications import delete_notifications + + recipients = await delete_notifications(object_type='message', object_ids=[id], db=db) await db.execute(delete(Message).filter_by(id=id)) # Delete all reactions to this message await db.execute(delete(MessageReaction).filter_by(message_id=id)) await db.commit() + await emit_changes(recipients) if message: await publish_app_message_event(message, 'channel.message.deleted') return True diff --git a/backend/open_webui/models/models.py b/backend/open_webui/models/models.py index b383572c42..0384d5981c 100755 --- a/backend/open_webui/models/models.py +++ b/backend/open_webui/models/models.py @@ -7,6 +7,7 @@ from copy import deepcopy from typing import Annotated, Any, Literal from fastapi import HTTPException +from open_webui.models.notifications import emit_changes from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, JSONField, get_async_db_context from open_webui.models.access_grants import AccessGrant, AccessGrantModel, AccessGrants @@ -715,18 +716,20 @@ class ModelsTable: return None async def _delete_models(self, session, ids): - await delete_resource_comments(session, 'model', ids) + recipients = await delete_resource_comments('model', ids, db=session) await session.execute( delete(AccessGrant).where(AccessGrant.resource_type == 'model', AccessGrant.resource_id.in_(ids)) ) await session.execute(delete(ModelHistory).where(ModelHistory.model_id.in_(ids))) await session.execute(delete(Model).where(Model.id.in_(ids))) + return recipients async def delete_model_by_id(self, id: str, db: AsyncSession | None = None) -> bool: async with get_async_db_context(db) as session: try: - await self._delete_models(session, [id]) + recipients = await self._delete_models(session, [id]) await session.commit() + await emit_changes(recipients) return True except Exception: await session.rollback() @@ -736,8 +739,9 @@ class ModelsTable: async with get_async_db_context(db) as session: try: ids = (await session.execute(select(Model.id))).scalars().all() - await self._delete_models(session, ids) + recipients = await self._delete_models(session, ids) await session.commit() + await emit_changes(recipients) return True except Exception: await session.rollback() @@ -754,8 +758,9 @@ class ModelsTable: # Imported version IDs are never local history identities. form = ModelForm(**model.model_dump()) written.append(await self._write_model(session, form, user_id, existing.get(model.id))) - await self._delete_models(session, existing.keys() - {model.id for model in models}) + recipients = await self._delete_models(session, existing.keys() - {model.id for model in models}) await session.commit() + await emit_changes(recipients) return [await self._written_model(session, model) for model in written] except Exception: await session.rollback() diff --git a/backend/open_webui/models/notes.py b/backend/open_webui/models/notes.py index 801aaf5352..e7c1fcae78 100644 --- a/backend/open_webui/models/notes.py +++ b/backend/open_webui/models/notes.py @@ -8,6 +8,7 @@ from typing import Literal, Optional import markdown from bs4 import BeautifulSoup +from open_webui.models.notifications import emit_changes from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, get_async_db_context from open_webui.models.access_grants import AccessGrantModel, AccessGrants @@ -843,10 +844,11 @@ class NoteTable: try: async with get_async_db_context(db) as db: await AccessGrants.revoke_all_access('note', id, db=db) - await delete_resource_comments(db, 'note', [id]) + recipients = await delete_resource_comments('note', [id], db=db) await db.execute(delete(PinnedNote).filter(PinnedNote.note_id == id)) await db.execute(delete(Note).filter(Note.id == id)) await db.commit() + await emit_changes(recipients) return True except Exception: return False diff --git a/backend/open_webui/models/notifications.py b/backend/open_webui/models/notifications.py new file mode 100644 index 0000000000..80f5f83066 --- /dev/null +++ b/backend/open_webui/models/notifications.py @@ -0,0 +1,162 @@ +"""Persistent, per-recipient inbox activity. Writers never commit the caller's transaction.""" + +import logging +import time +from uuid import uuid4 + +from open_webui.internal.db import Base +from pydantic import BaseModel, ConfigDict, Field +from sqlalchemy import JSON, BigInteger, CheckConstraint, Column, Index, Text, UniqueConstraint, delete, update + +log = logging.getLogger(__name__) + + +class Notification(Base): + __tablename__ = 'notification' + id = Column(Text, primary_key=True) + user_id = Column(Text, nullable=False) + event_type = Column(Text, nullable=False) + event_key = Column(Text, nullable=False) + actor_type = Column(Text, nullable=False) + actor_id = Column(Text, nullable=True) + resource_type = Column(Text, nullable=True) + resource_id = Column(Text, nullable=True) + object_type = Column(Text, nullable=True) + object_id = Column(Text, nullable=True) + data = Column(JSON, nullable=False) + seen_at = Column(BigInteger, nullable=True) + read_at = Column(BigInteger, nullable=True) + created_at = Column(BigInteger, nullable=False) + updated_at = Column(BigInteger, nullable=False) + __table_args__ = ( + UniqueConstraint('user_id', 'event_key', name='uq_notification_occurrence'), + CheckConstraint('(resource_type IS NULL) = (resource_id IS NULL)', name='ck_notification_resource'), + CheckConstraint('(object_type IS NULL) = (object_id IS NULL)', name='ck_notification_object'), + CheckConstraint("actor_type IN ('user', 'system')", name='ck_notification_actor'), + Index('ix_notification_user_created', 'user_id', 'created_at', 'id'), + Index( + 'ix_notification_unseen', + 'user_id', + 'created_at', + 'id', + sqlite_where=seen_at.is_(None), + postgresql_where=seen_at.is_(None), + ), + Index('ix_notification_resource', 'resource_type', 'resource_id'), + Index('ix_notification_object', 'object_type', 'object_id'), + ) + + +class NotificationData(BaseModel): + model_config = ConfigDict(extra='forbid') + + +class ReminderData(NotificationData): + start_at: int + + +class SystemData(NotificationData): + message: str = Field(min_length=1, max_length=2000) + + +DATA_TYPES = { + 'comment.created': NotificationData, + 'comment.replied': NotificationData, + 'comment.mentioned': NotificationData, + 'channel.mentioned': NotificationData, + 'channel.replied': NotificationData, + 'calendar.alert': ReminderData, + 'automation.run_failed': NotificationData, + 'system.announcement': SystemData, +} + + +async def emit_changes(user_ids): + from open_webui.socket.main import sio + + for user_id in user_ids: + try: + await sio.emit('notifications:changed', {}, room=f'user:{user_id}') + except Exception: + log.exception('Could not broadcast inbox change for %s', user_id) + + +async def insert_notifications( + *, + recipients, + event_key, + actor_id=None, + resource_type=None, + resource_id=None, + object_type=None, + object_id=None, + data=None, + db, +): + """recipients maps user IDs to event types; the producer chooses the strongest reason.""" + from open_webui.models.config import Config + from open_webui.utils.inbox import validate_source, eligible_recipients + + if not recipients or not await Config.get('notifications.enable'): + return set() + if not event_key or len(event_key) > 512: + raise ValueError('Invalid notification occurrence') + payloads = {kind: DATA_TYPES[kind].model_validate(data or {}).model_dump() for kind in set(recipients.values())} + for event_type in payloads: + await validate_source(event_type, resource_type, resource_id, object_type, object_id, db=db) + recipients = await eligible_recipients(recipients, resource_type, resource_id, actor_id, db=db) + if not recipients: + return set() + dialect = db.get_bind().dialect.name + if dialect == 'postgresql': + from sqlalchemy.dialects.postgresql import insert + elif dialect == 'sqlite': + from sqlalchemy.dialects.sqlite import insert + else: + raise ValueError('Inbox requires SQLite or PostgreSQL') + try: + now = time.time_ns() + result = await db.execute( + insert(Notification) + .on_conflict_do_nothing(index_elements=['user_id', 'event_key']) + .returning(Notification.user_id), + [ + dict( + id=str(uuid4()), + user_id=user_id, + event_type=event_type, + event_key=event_key, + actor_type='user' if actor_id else 'system', + actor_id=actor_id, + resource_type=resource_type, + resource_id=resource_id, + object_type=object_type, + object_id=object_id, + data=payloads[event_type], + created_at=now, + updated_at=now, + ) + for user_id, event_type in recipients.items() + ], + ) + return set(result.scalars().all()) + except Exception: + log.exception('Inbox insertion failed for occurrence %s', event_key) + raise + + +async def delete_notifications(*, resource_type=None, resource_ids=None, object_type=None, object_ids=None, db): + condition = ( + (Notification.resource_type == resource_type) & Notification.resource_id.in_(resource_ids) + if resource_type + else (Notification.object_type == object_type) & Notification.object_id.in_(object_ids) + ) + recipients = ( + (await db.execute(delete(Notification).where(condition).returning(Notification.user_id))).scalars().all() + ) + return set(recipients) + + +async def delete_user_notifications(user_id, db): + await db.execute(delete(Notification).where(Notification.user_id == user_id)) + await db.execute(update(Notification).where(Notification.actor_id == user_id).values(actor_id=None)) diff --git a/backend/open_webui/models/prompts.py b/backend/open_webui/models/prompts.py index 797784b94b..c486e55e97 100644 --- a/backend/open_webui/models/prompts.py +++ b/backend/open_webui/models/prompts.py @@ -9,6 +9,7 @@ from typing import Optional log = logging.getLogger(__name__) +from open_webui.models.notifications import emit_changes from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, JSONField, get_async_db_context from open_webui.models.access_grants import AccessGrantModel, AccessGrants @@ -645,9 +646,10 @@ class PromptsTable: await PromptHistories.delete_history_by_prompt_id(prompt.id, db=session) await AccessGrants.revoke_all_access('prompt', prompt.id, db=session) - await delete_resource_comments(session, 'prompt', [prompt.id]) + recipients = await delete_resource_comments('prompt', [prompt.id], db=session) await session.delete(prompt) await session.commit() + await emit_changes(recipients) return True return False except Exception: @@ -663,9 +665,10 @@ class PromptsTable: await PromptHistories.delete_history_by_prompt_id(prompt.id, db=session) await AccessGrants.revoke_all_access('prompt', prompt.id, db=session) - await delete_resource_comments(session, 'prompt', [prompt.id]) + recipients = await delete_resource_comments('prompt', [prompt.id], db=session) await session.delete(prompt) await session.commit() + await emit_changes(recipients) return True return False except Exception as err: diff --git a/backend/open_webui/models/shared_chats.py b/backend/open_webui/models/shared_chats.py index 157c00d21d..8959d6c771 100644 --- a/backend/open_webui/models/shared_chats.py +++ b/backend/open_webui/models/shared_chats.py @@ -3,7 +3,6 @@ import time import uuid from typing import Literal, Optional -from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, JSONField, get_async_db_context from pydantic import BaseModel, ConfigDict from sqlalchemy import JSON, BigInteger, Column, ForeignKey, Text, delete, select @@ -215,7 +214,6 @@ class SharedChatsTable: """Delete a shared chat by its share token.""" try: async with get_async_db_context(db) as db: - await delete_resource_comments(db, 'shared_chat', select(SharedChat.id).filter_by(id=share_id)) await db.execute(delete(SharedChat).filter_by(id=share_id)) await db.commit() return True @@ -226,7 +224,6 @@ class SharedChatsTable: """Delete all shared chats for a given original chat.""" try: async with get_async_db_context(db) as db: - await delete_resource_comments(db, 'shared_chat', select(SharedChat.id).filter_by(chat_id=chat_id)) await db.execute(delete(SharedChat).filter_by(chat_id=chat_id)) await db.commit() return True @@ -237,7 +234,6 @@ class SharedChatsTable: """Delete all shared chats created by a user.""" try: async with get_async_db_context(db) as db: - await delete_resource_comments(db, 'shared_chat', select(SharedChat.id).filter_by(user_id=user_id)) await db.execute(delete(SharedChat).filter_by(user_id=user_id)) await db.commit() return True diff --git a/backend/open_webui/models/skills.py b/backend/open_webui/models/skills.py index 1a2a2e7ace..3dbc0f5514 100644 --- a/backend/open_webui/models/skills.py +++ b/backend/open_webui/models/skills.py @@ -3,6 +3,7 @@ import time from typing import Optional from fastapi import HTTPException +from open_webui.models.notifications import emit_changes from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, get_async_db_context from open_webui.models.access_grants import AccessGrant, AccessGrantModel, AccessGrants @@ -500,11 +501,12 @@ class SkillsTable: async def delete_skill_by_id(self, id: str, db: Optional[AsyncSession] = None) -> bool: try: async with get_async_db_context(db) as db: - await delete_resource_comments(db, 'skill', [id]) + recipients = await delete_resource_comments('skill', [id], db=db) await db.execute(delete(AccessGrant).filter_by(resource_type='skill', resource_id=id)) await db.execute(delete(SkillHistory).filter_by(skill_id=id)) await db.execute(delete(Skill).filter_by(id=id)) await db.commit() + await emit_changes(recipients) return True except Exception: diff --git a/backend/open_webui/models/tools.py b/backend/open_webui/models/tools.py index 163bd8d17e..8b8cb0f5c7 100644 --- a/backend/open_webui/models/tools.py +++ b/backend/open_webui/models/tools.py @@ -8,6 +8,7 @@ import time from fastapi import HTTPException # local imports +from open_webui.models.notifications import emit_changes from open_webui.models.comments import delete_resource_comments from open_webui.internal.db import Base, JSONField, get_async_db_context from open_webui.models.access_grants import AccessGrant, AccessGrantModel, AccessGrants @@ -425,12 +426,13 @@ class ToolsTable: async def delete_tool_by_id(self, id, db=None): async with get_async_db_context(db) as session: try: - await delete_resource_comments(session, 'tool', [id]) + recipients = await delete_resource_comments('tool', [id], db=session) await self._lock_tool(session, id) await session.execute(delete(AccessGrant).filter_by(resource_type='tool', resource_id=id)) await session.execute(delete(ToolHistory).filter_by(tool_id=id)) await session.execute(delete(Tool).filter_by(id=id)) await session.commit() + await emit_changes(recipients) return True except Exception: await session.rollback() diff --git a/backend/open_webui/models/users.py b/backend/open_webui/models/users.py index 355de84019..ee01cd7444 100644 --- a/backend/open_webui/models/users.py +++ b/backend/open_webui/models/users.py @@ -861,6 +861,9 @@ class UsersTable: from open_webui.models.comments import cleanup_comments_for_deleted_user await cleanup_comments_for_deleted_user(session, id) + from open_webui.models.notifications import delete_user_notifications + + await delete_user_notifications(id, db=session) await session.execute(delete(User).where(User.id == id)) await session.commit() return True diff --git a/backend/open_webui/routers/auths.py b/backend/open_webui/routers/auths.py index 700ce97f47..596a524268 100644 --- a/backend/open_webui/routers/auths.py +++ b/backend/open_webui/routers/auths.py @@ -129,6 +129,7 @@ ADMIN_CONFIG_KEYS = { 'ENABLE_AUTOMATIONS': 'automations.enable', 'ENABLE_CHANNELS': 'channels.enable', 'ENABLE_COMMENTS': 'comments.enable', + 'ENABLE_NOTIFICATIONS': 'notifications.enable', 'ENABLE_ENCRYPTED_CHANNELS': 'channels.encryption.enable', 'ENABLE_APPS': 'apps.enable', 'CHANNEL_MODEL_RESPONSE_MODE': 'channels.model_response_mode', @@ -1190,6 +1191,7 @@ class AdminConfig(MfaConfigForm): ENABLE_AUTOMATIONS: bool ENABLE_CHANNELS: bool ENABLE_COMMENTS: bool = False + ENABLE_NOTIFICATIONS: bool = True ENABLE_ENCRYPTED_CHANNELS: bool = False ENABLE_APPS: bool = False CHANNEL_MODEL_RESPONSE_MODE: str = 'thread' @@ -1245,6 +1247,8 @@ async def update_admin_config(request: Request, form_data: AdminConfig, user=Dep updates.pop('channels.encryption.enable', None) if 'ENABLE_COMMENTS' not in form_data.model_fields_set: updates.pop('comments.enable', None) + if 'ENABLE_NOTIFICATIONS' not in form_data.model_fields_set: + updates.pop('notifications.enable', None) if 'ENABLE_APPS' not in form_data.model_fields_set: updates.pop('apps.enable', None) if 'I18N' not in form_data.model_fields_set: diff --git a/backend/open_webui/routers/automations.py b/backend/open_webui/routers/automations.py index f5ab9b9536..a61490b0e0 100644 --- a/backend/open_webui/routers/automations.py +++ b/backend/open_webui/routers/automations.py @@ -399,3 +399,22 @@ async def get_automation_runs( automation = await Automations.get_by_id(id, db=db) check_automation_access(automation, user) return await AutomationRuns.get_by_automation(id, skip=skip, limit=limit, db=db) + + +@router.get('/{id}/runs/{run_id}', response_model=AutomationRunModel) +async def get_automation_run( + request: Request, + id: str, + run_id: str, + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): + from open_webui.models.automations import AutomationRun + + await check_automations_permission(request, user) + automation = await Automations.get_by_id(id, db=db) + check_automation_access(automation, user) + run = await db.get(AutomationRun, run_id) + if not run or run.automation_id != id: + raise HTTPException(404, 'Run not found') + return AutomationRunModel.model_validate(run) diff --git a/backend/open_webui/routers/channels.py b/backend/open_webui/routers/channels.py index f37668da90..b2bb710570 100644 --- a/backend/open_webui/routers/channels.py +++ b/backend/open_webui/routers/channels.py @@ -1428,7 +1428,9 @@ async def new_message_handler(request: Request, id: str, form_data: MessageForm, raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT()) try: - message = await Messages.insert_new_message(form_data, channel.id, user.id, db=db) + message = await Messages.insert_new_message( + form_data, channel.id, user.id, request=request, event_actor=user, db=db + ) if message: if channel.type in ['group', 'dm']: members = await Channels.get_members_by_channel_id(channel.id, db=db) @@ -1521,16 +1523,6 @@ 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: @@ -2308,6 +2300,9 @@ async def post_webhook_message( MessageForm(content=form_data.content, meta={'webhook': {'id': webhook.id}}), webhook.channel_id, webhook.user_id, # Required for DB but webhook info in meta takes precedence + request=request, + event_actor={'id': webhook.id, 'name': webhook.name, 'role': 'webhook', 'type': 'webhook'}, + event_source='channel_webhook', db=db, ) @@ -2351,12 +2346,4 @@ 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/comments.py b/backend/open_webui/routers/comments.py index 869c0fd581..6149c51489 100644 --- a/backend/open_webui/routers/comments.py +++ b/backend/open_webui/routers/comments.py @@ -3,6 +3,7 @@ import time from uuid import UUID, uuid4 from fastapi import APIRouter, Depends, HTTPException, Query, Request +from open_webui.models.notifications import emit_changes from open_webui.internal.db import get_async_session from open_webui.models.comments import ( Comment, @@ -22,6 +23,10 @@ from sqlalchemy import and_, delete, func, or_, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from open_webui.events import EVENTS, build_event, publish_event +from open_webui.utils.inbox import mentioned_users +from open_webui.models.notifications import delete_notifications + router = APIRouter() write_limiter = RateLimiter(30, 60) reaction_limiter = RateLimiter(120, 60) @@ -193,7 +198,21 @@ async def create_comment( db.add(comment) await db.flush() response = (await comment_responses(db, [comment], user, resource.user_id))[0] + from open_webui.utils.inbox import consume_event + from open_webui.models.notifications import emit_changes + + event = build_event( + request, + EVENTS.COMMENT_CREATED, + actor=user, + subject_id=comment.id, + subject_type='comment', + event_id=f'comment:{comment.id}:created', + ) + recipients = await consume_event(event, db=db) await db.commit() + await emit_changes(recipients) + await publish_event(request, event) await changed(form.resource_type, form.resource_id) return response @@ -262,7 +281,21 @@ async def reply( db.add(comment) await db.flush() response = (await comment_responses(db, [comment], user, resource.user_id))[0] + from open_webui.utils.inbox import consume_event + from open_webui.models.notifications import emit_changes + + event = build_event( + request, + EVENTS.COMMENT_CREATED, + actor=user, + subject_id=comment.id, + subject_type='comment', + event_id=f'comment:{comment.id}:created', + ) + recipients = await consume_event(event, db=db) await db.commit() + await emit_changes(recipients) + await publish_event(request, event) await changed(parent.resource_type, parent.resource_id) return response @@ -281,6 +314,7 @@ async def edit( raise HTTPException(403, 'Cannot edit this comment') if str(comment.updated_at) != form.etag: raise HTTPException(409, 'Comment changed. Reload before editing.') + previous_content = comment.content result = await db.execute( update(Comment) .where(Comment.id == id, Comment.updated_at == int(form.etag), Comment.deleted_at.is_(None)) @@ -291,7 +325,22 @@ async def edit( raise HTTPException(409, 'Comment changed. Reload before editing.') await db.refresh(comment) response = (await comment_responses(db, [comment], user, resource.user_id))[0] + from open_webui.utils.inbox import consume_event + from open_webui.models.notifications import emit_changes + + event = build_event( + request, + EVENTS.COMMENT_UPDATED, + actor=user, + subject_id=comment.id, + subject_type='comment', + event_id=f'comment:{comment.id}:{comment.updated_at}', + data={'previous_mentions': sorted(mentioned_users(previous_content))}, + ) + recipients = await consume_event(event, db=db) await db.commit() + await emit_changes(recipients) + await publish_event(request, event) await changed(comment.resource_type, comment.resource_id) return response @@ -321,8 +370,10 @@ async def remove( if result.rowcount != 1: raise HTTPException(409, 'Comment changed. Reload before deleting.') await db.execute(delete(CommentReaction).where(CommentReaction.comment_id == id)) + recipients = await delete_notifications(object_type='comment', object_ids=[id], db=db) # Keep thread and quoted-reply references intact, without retaining deleted text. await db.commit() + await emit_changes(recipients) await changed(kind, resource_id) return True diff --git a/backend/open_webui/routers/functions.py b/backend/open_webui/routers/functions.py index e5ceedcd66..86d8a4fff6 100644 --- a/backend/open_webui/routers/functions.py +++ b/backend/open_webui/routers/functions.py @@ -308,9 +308,8 @@ async def toggle_function_by_id( data={'type': function.type, 'name': function.name}, ) await dispatch_event_functions( - request.app, + request, lifecycle_event, - request=request, extra_function_ids=[function.id] if not function.is_active else None, ) schedule_webhook_dispatch(request.app, lifecycle_event) diff --git a/backend/open_webui/routers/notifications.py b/backend/open_webui/routers/notifications.py index 6cc098392f..894e29c9c5 100644 --- a/backend/open_webui/routers/notifications.py +++ b/backend/open_webui/routers/notifications.py @@ -1,8 +1,16 @@ from __future__ import annotations from typing import Any +import logging +import time +from uuid import UUID -from fastapi import APIRouter, Depends, HTTPException, Request, status +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status +from sqlalchemy import and_, func, or_, select, update +from sqlalchemy.ext.asyncio import AsyncSession +from open_webui.internal.db import get_async_session +from open_webui.models.notifications import Notification, emit_changes +from open_webui.utils.inbox import hydrate, visible_clause from pydantic import BaseModel from open_webui.constants import ERROR_MESSAGES @@ -21,6 +29,121 @@ from open_webui.utils.notifications import ( router = APIRouter() +log = logging.getLogger(__name__) + + +async def _check_notifications_enabled(): + if not await Config.get('notifications.enable'): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND) + + +async def inbox_page(user, cursor=None, limit=20, clause=None, *, db): + started = time.monotonic() + if clause is None: + clause = await visible_clause(user, db=db) + query = select(Notification).where(clause) + if cursor: + try: + timestamp, id = cursor.split(':', 1) + timestamp = int(timestamp) + UUID(id) + if not 0 <= timestamp <= 2**63 - 1: + raise ValueError() + except (ValueError, TypeError): + raise HTTPException(422, 'Invalid inbox cursor') + query = query.where( + or_(Notification.created_at < timestamp, and_(Notification.created_at == timestamp, Notification.id < id)) + ) + rows = ( + await db.scalars(query.order_by(Notification.created_at.desc(), Notification.id.desc()).limit(limit + 1)) + ).all() + more = len(rows) > limit + rows = rows[:limit] + result = { + 'items': await hydrate(rows, db=db), + 'next_cursor': f'{rows[-1].created_at}:{rows[-1].id}' if more else None, + } + log.debug('Inbox page loaded in %.3fs', time.monotonic() - started) + return result + + +@router.get('/inbox') +async def get_inbox( + cursor: str | None = Query(None, max_length=100), + limit: int = Query(20, ge=1, le=50), + user=Depends(get_verified_user), + db: AsyncSession = Depends(get_async_session), +): + await _check_notifications_enabled() + return await inbox_page(user, cursor, limit, db=db) + + +@router.get('/inbox/count') +async def get_inbox_count(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): + await _check_notifications_enabled() + started = time.monotonic() + clause = await visible_clause(user, db=db) + count = await db.scalar( + select(func.count()).select_from(Notification).where(clause, Notification.seen_at.is_(None)) + ) + log.debug('Inbox count loaded in %.3fs', time.monotonic() - started) + return {'count': count} + + +@router.post('/inbox/open') +async def open_inbox(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): + await _check_notifications_enabled() + clause = await visible_clause(user, db=db) + now = time.time_ns() + # A single UPDATE acknowledges only rows visible to this statement. A late + # commit with an older created_at must remain unseen (no timestamp watermark). + result = await db.execute( + update(Notification) + .where(clause, Notification.seen_at.is_(None)) + .values(seen_at=now, updated_at=now) + .returning(Notification.id) + .execution_options(synchronize_session=False) + ) + # SQLite reports rowcount=-1 for UPDATEs containing the recursive folder CTE. + acknowledged = len(result.scalars().all()) + await db.commit() + await emit_changes({user.id}) + page = await inbox_page(user, clause=clause, db=db) + for item in page['items']: + item['is_new'] = item['seen_at'] == str(now) or item['seen_at'] is None + page.update( + acknowledged_count=acknowledged, + acknowledged_at=str(now), + count=await db.scalar( + select(func.count()).select_from(Notification).where(clause, Notification.seen_at.is_(None)) + ), + ) + return page + + +@router.patch('/inbox/{id}') +async def read_inbox_item(id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): + await _check_notifications_enabled() + clause = await visible_clause(user, db=db) + now = time.time_ns() + row = await db.scalar( + update(Notification) + .where(clause, Notification.id == id) + .values( + seen_at=func.coalesce(Notification.seen_at, now), + read_at=func.coalesce(Notification.read_at, now), + updated_at=now, + ) + .returning(Notification) + .execution_options(populate_existing=True) + ) + if not row: + raise HTTPException(404, 'Notification not found') + response = (await hydrate([row], db=db))[0] + await db.commit() + await emit_changes({user.id}) + return response + class NotificationTargetForm(BaseModel): id: str | None = None diff --git a/backend/open_webui/routers/search.py b/backend/open_webui/routers/search.py index 2b50ddd2b7..02eac5515a 100644 --- a/backend/open_webui/routers/search.py +++ b/backend/open_webui/routers/search.py @@ -339,7 +339,7 @@ async def search_workspace( # noqa: C901 — keep permission scoping and source .exists() ) stmt = AccessGrants.has_permission_filter( - db, select(Note, pinned, note_shared), Note, {'user_id': user.id, 'group_ids': groups}, 'note', 'read' + select(Note, pinned, note_shared), Note, {'user_id': user.id, 'group_ids': groups}, 'note', 'read', db=db ) stmt = stmt.where(sharing_filter(Note.user_id)) stmt = apply_match( diff --git a/backend/open_webui/routers/terminals.py b/backend/open_webui/routers/terminals.py index d084710568..1d9a1da061 100644 --- a/backend/open_webui/routers/terminals.py +++ b/backend/open_webui/routers/terminals.py @@ -379,7 +379,7 @@ async def ws_terminal( if upstream_params: upstream_url += f'?{urllib.parse.urlencode(upstream_params)}' - app = ws.scope.get('app') + request = Request({**ws.scope, 'type': 'http', 'method': 'GET'}) opened = False session = aiohttp.ClientSession() try: @@ -401,7 +401,7 @@ async def ws_terminal( await upstream.send_str(_json.dumps({'type': 'auth', 'token': ''})) await publish_event( - app, + request, EVENTS.TERMINAL_SESSION_OPENED, actor=user, subject_id=session_id, @@ -470,7 +470,7 @@ async def ws_terminal( await session.close() if opened: await publish_event( - app, + request, EVENTS.TERMINAL_SESSION_CLOSED, actor=user, subject_id=session_id, diff --git a/backend/open_webui/socket/comments.py b/backend/open_webui/socket/comments.py index 1b3db65b5f..8722820d2c 100644 --- a/backend/open_webui/socket/comments.py +++ b/backend/open_webui/socket/comments.py @@ -36,10 +36,16 @@ def register_comment_events(sio, get_socket_session_user): except HTTPException: return {'error': 'Resource unavailable'} session = await sio.get_session(sid) - previous = session.get('comments_resource') - if previous: + subscription = data.get('subscription_id', 'default') + if not isinstance(subscription, str) or len(subscription) > 100: + return {'error': 'Invalid subscription'} + subscriptions = session.setdefault('comments_resources', {}) + if subscription not in subscriptions and len(subscriptions) >= 10: + return {'error': 'Too many comment panels'} + previous = subscriptions.get(subscription) + subscriptions[subscription] = [kind, id] + if previous and previous not in subscriptions.values(): await sio.leave_room(sid, comment_room(*previous)) - session['comments_resource'] = [kind, id] await sio.save_session(sid, session) await sio.enter_room(sid, comment_room(kind, id)) return {'ok': True} @@ -47,25 +53,32 @@ def register_comment_events(sio, get_socket_session_user): @sio.on('comments:leave') async def leave(sid, data=None): session = await sio.get_session(sid) - resource = session.pop('comments_resource', None) - if resource: + subscriptions = session.get('comments_resources', {}) + subscription = data.get('subscription_id', 'default') if isinstance(data, dict) else 'default' + if not isinstance(subscription, str): + return + resource = subscriptions.pop(subscription, None) + if resource and resource not in subscriptions.values(): await sio.leave_room(sid, comment_room(*resource)) - await sio.save_session(sid, session) + await sio.save_session(sid, session) async def revalidate_comment_subscription(sio, sid, session_user): session = await sio.get_session(sid) - resource = session.get('comments_resource') - if not resource: + subscriptions = session.get('comments_resources', {}) + if not subscriptions: return user = await Users.get_user_by_id(session_user['id']) - try: - if not user: - raise HTTPException(403) - async with get_async_db_context() as db: - await comment_resource(*resource, user, db, request=comment_connection(sio, sid)) - except HTTPException: - await sio.leave_room(sid, comment_room(*resource)) - session.pop('comments_resource', None) - await sio.save_session(sid, session) - await sio.emit('comments:unavailable', {}, to=sid) + for resource in set(tuple(value) for value in subscriptions.values()): + try: + if not user: + raise HTTPException(403) + async with get_async_db_context() as db: + await comment_resource(*resource, user, db, request=comment_connection(sio, sid)) + except HTTPException: + await sio.leave_room(sid, comment_room(*resource)) + session['comments_resources'] = { + key: value for key, value in session['comments_resources'].items() if tuple(value) != resource + } + await sio.save_session(sid, session) + await sio.emit('comments:unavailable', {'resource_type': resource[0], 'resource_id': resource[1]}, to=sid) diff --git a/backend/open_webui/utils/automations.py b/backend/open_webui/utils/automations.py index 292e435083..0d54e7aa1b 100644 --- a/backend/open_webui/utils/automations.py +++ b/backend/open_webui/utils/automations.py @@ -27,7 +27,7 @@ from uuid import uuid4 from fastapi import Request from fastapi.security import HTTPAuthorizationCredentials from open_webui.constants import ERROR_MESSAGES -from open_webui.events import EVENTS, publish_event +from open_webui.events import EVENTS, build_event, 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 @@ -340,9 +340,9 @@ async def _execute_channel_automation( room=f'user:{automation.user_id}', ) - await _record_run(automation.id, 'success', chat_id=f'channel:{channel.id}') + await _record_run(request, automation.id, 'success', chat_id=f'channel:{channel.id}') await publish_event( - app, + request, EVENTS.AUTOMATION_RUN_COMPLETED, actor=user, subject_id=automation.id, @@ -358,16 +358,11 @@ async def execute_automation(app, automation: AutomationModel) -> None: (filters, model params, knowledge/RAG, tools, DB saves, webhooks). """ chat = None + request = _build_request(app) try: 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'}, - ) + await _record_run(request, automation.id, 'error', error='User not found') return # Re-gate the rehydrated owner: a demoted/deactivated or de-permissioned owner must not run. @@ -378,14 +373,7 @@ async def execute_automation(app, automation: AutomationModel) -> None: and not await has_permission(user.id, 'features.automations', await Config.get('user.permissions')) ): 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, 'message': f'{automation.name}: {error}'}, - ) + await _record_run(request, automation.id, 'error', error=error, actor=user) return prompt = await prompt_template(automation.data['prompt'], user) @@ -464,14 +452,7 @@ async def execute_automation(app, automation: AutomationModel) -> None: if not 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, 'message': f'{automation.name}: {error}'}, - ) + await _record_run(request, automation.id, 'error', error=error, actor=user) return # Notify frontend to refresh chat list @@ -542,9 +523,9 @@ async def execute_automation(app, automation: AutomationModel) -> None: room=f'user:{automation.user_id}', ) - await _record_run(automation.id, 'success', chat_id=chat.id) + await _record_run(request, automation.id, 'success', chat_id=chat.id) await publish_event( - app, + request, EVENTS.AUTOMATION_RUN_COMPLETED, actor=user, subject_id=automation.id, @@ -554,19 +535,7 @@ async def execute_automation(app, automation: AutomationModel) -> None: except (Exception, asyncio.CancelledError) as e: log.exception(f'Automation {automation.id} failed') error = 'Automation reply was cancelled' if isinstance(e, asyncio.CancelledError) else str(e)[:4000] - await _record_run(automation.id, 'error', chat_id=chat.id if chat else None, error=error) - await publish_event( - app, - EVENTS.AUTOMATION_RUN_FAILED, - subject_id=automation.id, - data={ - 'name': automation.name, - 'error': error, - 'user_id': automation.user_id, - 'message': f'{automation.name}: {error}', - **({'chat_id': chat.id} if chat else {}), - }, - ) + await _record_run(request, automation.id, 'error', chat_id=chat.id if chat else None, error=error) if isinstance(e, asyncio.CancelledError): raise @@ -582,7 +551,7 @@ async def _check_calendar_alerts(app) -> None: De-duplication is DB-backed via meta.alerted_at — survives restarts and works across multiple instances. """ - from open_webui.models.calendar import CalendarEvents, CalendarEventUpdateForm + from open_webui.models.calendar import CalendarEvents from open_webui.socket.main import sio now_ns = int(time.time_ns()) @@ -597,6 +566,7 @@ async def _check_calendar_alerts(app) -> None: if not upcoming: return + request = _build_request(app) for event, user_tz in upcoming: # Skip if already alerted for this start time if event.meta and event.meta.get('alerted_at') == event.start_at: @@ -615,6 +585,45 @@ async def _check_calendar_alerts(app) -> None: 'location': event.location or '', } + from open_webui.models.calendar import CalendarEvent + from sqlalchemy import select + + async with get_async_db() as db: + current = await db.scalar(select(CalendarEvent).where(CalendarEvent.id == event.id).with_for_update()) + if ( + not current + or current.is_cancelled + or current.start_at != event.start_at + or (current.meta or {}).get('alerted_at') == event.start_at + ): + continue + time_str = f'in {minutes_until} min' if minutes_until > 0 else 'now' + from open_webui.utils.inbox import consume_event + from open_webui.models.notifications import emit_changes + + alert_event = build_event( + request, + EVENTS.CALENDAR_ALERT, + subject_id=event.id, + subject_type='calendar.event', + source='scheduler', + message=event.title, + event_id=f'calendar:{event.id}:{event.start_at}', + data={ + **alert_data, + 'user_id': event.user_id, + 'starts_in': time_str, + 'start_at': event.start_at, + 'message': f'{event.title}: starting {time_str}', + }, + ) + current.meta = {**(current.meta or {}), 'alerted_at': event.start_at} + current.updated_at = time.time_ns() + recipients = await consume_event(alert_event, db=db) + await db.commit() + await emit_changes(recipients) + await publish_event(request, alert_event) + await sio.emit( 'events', { @@ -626,42 +635,17 @@ async def _check_calendar_alerts(app) -> None: room=f'user:{event.user_id}', ) - # Mark as alerted in DB so it survives restarts / multi-instance - try: - await CalendarEvents.update_event_by_id( - event.id, - CalendarEventUpdateForm(meta={'alerted_at': event.start_at}), - ) - except Exception: - log.debug('Failed to mark event %s as alerted', event.id, exc_info=True) - - # Send target notification if user has one configured - try: - time_str = f'in {minutes_until} min' if minutes_until > 0 else 'now' - await publish_event( - app, - EVENTS.CALENDAR_ALERT, - subject_id=event.id, - subject_type='calendar.event', - source='scheduler', - data={ - **alert_data, - 'user_id': event.user_id, - 'starts_in': time_str, - 'message': f'{event.title}: starting {time_str}', - }, - message=event.title, - ) - except Exception: - log.debug('Failed to send notification for calendar alert %s', event.id, exc_info=True) - async def _record_run( + request: Request, automation_id: str, status: str, chat_id: str = None, error: str = None, + actor=None, ): """Insert a run record into automation_run.""" async with get_async_db() as db: - await AutomationRuns.insert(automation_id, status, chat_id=chat_id, error=error, db=db) + await AutomationRuns.insert( + automation_id, status, chat_id=chat_id, error=error, request=request, event_actor=actor, db=db + ) diff --git a/backend/open_webui/utils/inbox.py b/backend/open_webui/utils/inbox.py new file mode 100644 index 0000000000..da55d9755b --- /dev/null +++ b/backend/open_webui/utils/inbox.py @@ -0,0 +1,451 @@ +"""Resource-aware producers and reads for the general inbox.""" + +from collections import defaultdict + +from sqlalchemy import and_, false, or_, select, true +from sqlalchemy.orm import load_only + +from open_webui.models.notifications import Notification, insert_notifications + + +def source_models(): + from open_webui.utils.comments import resource_models + from open_webui.models.automations import Automation, AutomationRun + from open_webui.models.calendar import Calendar, CalendarEvent + from open_webui.models.channels import Channel + from open_webui.models.comments import Comment + from open_webui.models.messages import Message + + return { + **resource_models(), + 'automation': Automation, + 'automation_run': AutomationRun, + 'calendar': Calendar, + 'calendar_event': CalendarEvent, + 'channel': Channel, + 'comment': Comment, + 'message': Message, + } + + +async def validate_source(event_type, resource_type, resource_id, object_type, object_id, db): + if event_type == 'system.announcement': + if any((resource_type, resource_id, object_type, object_id)): + raise ValueError('System announcements cannot attach resources') + return + models = source_models() + if ( + not all((resource_type, resource_id, object_type, object_id)) + or resource_type not in models + or object_type not in models + ): + raise ValueError('Incomplete notification source') + obj = await db.get(models[object_type], object_id) + if obj is None: + raise ValueError('Notification object does not exist') + valid = False + if event_type.startswith('comment.'): + valid = ( + object_type == 'comment' + and not obj.deleted_at + and (obj.resource_type, obj.resource_id) == (resource_type, resource_id) + ) + elif event_type.startswith('channel.'): + valid = object_type == 'message' and resource_type == 'channel' and obj.channel_id == resource_id + elif event_type == 'automation.run_failed': + valid = ( + object_type == 'automation_run' + and resource_type == 'automation' + and obj.automation_id == resource_id + and obj.status == 'error' + ) + elif event_type == 'calendar.alert': + valid = ( + object_type == 'calendar_event' + and resource_type == 'calendar' + and obj.calendar_id == resource_id + and not obj.is_cancelled + ) + if not valid: + raise ValueError('Notification object does not belong to resource') + + +async def resource_queries(user, db): + """One set of indexed permission subqueries, shared by listing, counts and writes.""" + from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL + from open_webui.env import ENABLE_ADMIN_CHAT_ACCESS + from open_webui.models.access_grants import AccessGrants + from open_webui.models.channels import ChannelMember + from open_webui.models.config import Config + from open_webui.models.folders import Folder + from open_webui.models.groups import Groups + from open_webui.models.shared_chats import SharedChat + from open_webui.utils.access_control import has_permission + + if user.role not in {'user', 'admin'}: + return {} + groups = [g.id for g in await Groups.get_groups_by_member_id(user.id, db=db, include_inherited=True)] + config = await Config.get_many( + 'comments.enable', + 'notes.enable', + 'channels.enable', + 'calendar.enable', + 'automations.enable', + 'user.permissions', + ) + admin = user.role == 'admin' + + def grant(kind, column): + return AccessGrants.has_grant_expression(user.id, kind, column, user_group_ids=groups) + + async def enabled(feature): + return config.get(f'{feature}.enable') and ( + admin or await has_permission(user.id, f'features.{feature}', config.get('user.permissions'), db=db) + ) + + models = source_models() + queries = {} + for kind in ('model', 'knowledge', 'prompt', 'skill', 'tool', 'note'): + if not config.get('comments.enable') or (kind == 'note' and not await enabled('notes')): + continue + model = models[kind] + allowed = ( + true() + if admin and (kind == 'note' or BYPASS_ADMIN_ACCESS_CONTROL) + else or_(model.user_id == user.id, grant(kind, model.id)) + ) + queries[kind] = select(model.id).where(allowed) + if config.get('comments.enable'): + chat = models['chat'] + # UNION (not UNION ALL) terminates even if legacy folder data contains a cycle. + folders = ( + select(Folder.id) + .where(or_(Folder.user_id == user.id, grant('folder', Folder.id))) + .cte('inbox_folders', recursive=True) + ) + folders = folders.union(select(Folder.id).join(folders, Folder.parent_id == folders.c.id)) + shared = select(SharedChat.chat_id).where( + SharedChat.chat['share_mode'].as_string() == 'continue', grant('shared_chat', SharedChat.chat_id) + ) + queries['chat'] = select(chat.id).where( + or_( + chat.user_id == user.id, + true() if admin and ENABLE_ADMIN_CHAT_ACCESS else false(), + and_(true() if admin else false(), chat.meta['internal'].as_boolean().is_(True)), + chat.id.in_(shared), + chat.folder_id.in_(select(folders.c.id)), + ) + ) + if await enabled('channels'): + channel = models['channel'] + member = ( + select(ChannelMember.id) + .where( + ChannelMember.channel_id == channel.id, + ChannelMember.user_id == user.id, + ChannelMember.is_active.is_(True), + ) + .exists() + ) + queries['channel'] = select(channel.id).where( + channel.deleted_at.is_(None), + or_( + and_(channel.type.in_(['group', 'dm']), member), + and_( + or_(channel.type.is_(None), channel.type.not_in(['group', 'dm'])), + true() if admin else grant('channel', channel.id), + ), + ), + ) + if await enabled('calendar'): + model = models['calendar'] + queries['calendar'] = select(model.id).where( + true() if admin else or_(model.user_id == user.id, grant('calendar', model.id)) + ) + if await enabled('automations'): + model = models['automation'] + queries['automation'] = select(model.id).where(model.user_id == user.id) + return queries + + +async def eligible_recipients(recipients, resource_type, resource_id, actor_id, db): + from open_webui.models.users import User, UserModel + + users = (await db.scalars(select(User).where(User.id.in_(recipients), User.role.in_(['user', 'admin'])))).all() + eligible = {} + for row in users: + kind = recipients[row.id] + if row.id == actor_id and kind.startswith(('comment.', 'channel.')): + continue + if resource_type is not None: + queries = await resource_queries(UserModel.model_validate(row), db=db) + query = queries.get(resource_type) + if query is None or not await db.scalar( + select(query.where(source_models()[resource_type].id == resource_id).exists()) + ): + continue + eligible[row.id] = kind + return eligible + + +async def visible_clause(user, db): + queries = await resource_queries(user, db=db) + models = source_models() + resources = or_( + *[ + and_(Notification.resource_type == kind, Notification.resource_id.in_(query)) + for kind, query in queries.items() + ] + ) + objects = [] + for kind, parent_column, events in ( + ('comment', 'resource_id', ['comment.created', 'comment.replied', 'comment.mentioned']), + ('message', 'channel_id', ['channel.mentioned', 'channel.replied']), + ('automation_run', 'automation_id', ['automation.run_failed']), + ('calendar_event', 'calendar_id', ['calendar.alert']), + ): + model = models[kind] + conditions = [model.id == Notification.object_id, getattr(model, parent_column) == Notification.resource_id] + if kind == 'comment': + conditions += [model.resource_type == Notification.resource_type, model.deleted_at.is_(None)] + if kind == 'calendar_event': + conditions.append(model.is_cancelled.is_(False)) + objects.append( + and_( + Notification.event_type.in_(events), + Notification.object_type == kind, + select(model.id).where(*conditions).exists(), + ) + ) + return ( + and_( + Notification.user_id == user.id, + or_( + and_( + Notification.event_type == 'system.announcement', + Notification.resource_type.is_(None), + Notification.object_type.is_(None), + ), + and_(resources, or_(*objects)), + ), + ) + if user.role in {'user', 'admin'} + else false() + ) + + +async def hydrate(rows, db): + """Fetch each actor/resource/object type once, never once per notification.""" + from open_webui.models.users import User + from open_webui.utils.channels import replace_mentions + + models = source_models() + ids = defaultdict(set) + for row in rows: + if row.resource_type: + ids[row.resource_type].add(row.resource_id) + if row.object_type: + ids[row.object_type].add(row.object_id) + objects = {} + for kind, values in ids.items(): + model = models[kind] + fields = ['id', 'title', 'name'] + if kind in {'comment', 'message'}: + fields += ['content', 'target_type', 'target_id', 'parent_id'] + columns = [getattr(model, field) for field in fields if hasattr(model, field)] + objects[kind] = { + obj.id: obj + for obj in (await db.scalars(select(model).options(load_only(*columns)).where(model.id.in_(values)))).all() + } + actors = { + u.id: {'id': u.id, 'name': u.name} + for u in ( + await db.scalars( + select(User) + .options(load_only(User.id, User.name)) + .where(User.id.in_({r.actor_id for r in rows if r.actor_id})) + ) + ).all() + } + result = [] + for row in rows: + resource = objects.get(row.resource_type, {}).get(row.resource_id) + obj = objects.get(row.object_type, {}).get(row.object_id) + title = ( + getattr(resource, 'title', None) + or getattr(resource, 'name', None) + or '' + ) + preview = ( + replace_mentions(getattr(obj, 'content', '') or '')[:300] + if row.object_type in {'comment', 'message'} + else '' + ) + data = dict(row.data or {}) + if row.object_type == 'comment': + data.update(target_type=obj.target_type, target_id=obj.target_id, parent_id=obj.parent_id) + elif row.object_type == 'message': + data['parent_id'] = obj.parent_id + elif row.object_type == 'calendar_event': + title = obj.title + result.append( + { + **{c.name: getattr(row, c.name) for c in Notification.__table__.columns}, + 'actor': actors.get(row.actor_id), + 'title': title, + 'preview': preview, + 'data': data, + # Epoch nanoseconds must not become imprecise JavaScript cursor values. + 'created_at': str(row.created_at), + 'updated_at': str(row.updated_at), + 'seen_at': str(row.seen_at) if row.seen_at is not None else None, + 'read_at': str(row.read_at) if row.read_at is not None else None, + } + ) + return result + + +def mentioned_users(content): + from open_webui.utils.channels import extract_mentions + + return {m['id'] for m in extract_mentions(content or '') if m['id_type'] == 'U'} + + +async def comment_notifications(comment, owner_id, *, reply_to=None, previous_mentions=None, event_key=None, db): + mentions = mentioned_users(comment.content) + recipients = {} + if previous_mentions is None: + recipients = { + id: 'comment.replied' if comment.parent_id else 'comment.created' + for id in (owner_id, getattr(reply_to, 'user_id', None)) + if id + } + else: + mentions -= set(previous_mentions) + recipients.update({id: 'comment.mentioned' for id in mentions}) + return await insert_notifications( + recipients=recipients, + event_key=event_key + or f'comment:{comment.id}:{(comment.updated_at if previous_mentions is not None else "created")}', + actor_id=comment.user_id, + resource_type=comment.resource_type, + resource_id=comment.resource_id, + object_type='comment', + object_id=comment.id, + db=db, + ) + + +async def channel_notifications(message, *, event_key=None, actor_type='user', db): + from open_webui.models.channels import ChannelMember + from open_webui.models.messages import Message + + mentions = mentioned_users(message.content) + target = ( + await db.get(Message, message.reply_to_id or message.parent_id) + if message.reply_to_id or message.parent_id + else None + ) + recipients = ( + {target.user_id: 'channel.replied'} + if target and target.channel_id == message.channel_id and target.user_id + else {} + ) + recipients.update({id: 'channel.mentioned' for id in mentions}) + muted = ( + await db.scalars( + select(ChannelMember.user_id).where( + ChannelMember.channel_id == message.channel_id, + ChannelMember.user_id.in_(recipients), + ChannelMember.is_channel_muted.is_(True), + ) + ) + ).all() + for id in muted: + if id not in mentions: + recipients.pop(id, None) + return await insert_notifications( + recipients=recipients, + event_key=event_key or f'message:{message.id}:created', + actor_id=message.user_id if actor_type == 'user' else None, + resource_type='channel', + resource_id=message.channel_id, + object_type='message', + object_id=message.id, + db=db, + ) + + +async def consume_event(event, db): + """Transactional inbox consumer; source producers publish facts, not recipients.""" + from open_webui.events import EVENTS + from open_webui.models.config import Config + + if not await Config.get('notifications.enable'): + return set() + + subject_id = (event.subject or {}).get('id') + models = source_models() + if event.event in {EVENTS.COMMENT_CREATED.name, EVENTS.COMMENT_UPDATED.name}: + comment = await db.get(models['comment'], subject_id) + if not comment or comment.deleted_at: + raise ValueError('Comment event references an unavailable comment') + resource_model = models[comment.resource_type] + owner_id = await db.scalar(select(resource_model.user_id).where(resource_model.id == comment.resource_id)) + reply_to = ( + await db.get(models['comment'], comment.reply_to_id or comment.parent_id) if comment.parent_id else None + ) + recipients = await comment_notifications( + comment, + owner_id, + reply_to=reply_to, + previous_mentions=event.data.get('previous_mentions', []) + if event.event == EVENTS.COMMENT_UPDATED.name + else None, + event_key=event.id, + db=db, + ) + if event.event == EVENTS.COMMENT_UPDATED.name: + existing = await db.scalars( + select(Notification.user_id) + .where(Notification.object_type == 'comment', Notification.object_id == comment.id) + .distinct() + ) + recipients.update(existing.all()) + return recipients + elif event.event == EVENTS.MESSAGE_CREATED.name: + message = await db.get(models['message'], subject_id) + if message and message.channel_id: + return await channel_notifications( + message, event_key=event.id, actor_type=(event.actor or {}).get('type', 'system'), db=db + ) + elif event.event == EVENTS.AUTOMATION_RUN_FAILED.name: + run = await db.get(models['automation_run'], event.data.get('run_id')) + automation = await db.get(models['automation'], subject_id) + if not run or not automation or run.automation_id != automation.id: + raise ValueError('Automation failure event requires its persisted run') + return await insert_notifications( + recipients={automation.user_id: event.event}, + event_key=event.id, + resource_type='automation', + resource_id=automation.id, + object_type='automation_run', + object_id=run.id, + db=db, + ) + elif event.event == EVENTS.CALENDAR_ALERT.name: + calendar_event = await db.get(models['calendar_event'], subject_id) + if not calendar_event or event.data.get('start_at') != calendar_event.start_at: + raise ValueError('Calendar alert must reference the current occurrence') + return await insert_notifications( + recipients={calendar_event.user_id: event.event}, + event_key=event.id, + resource_type='calendar', + resource_id=calendar_event.calendar_id, + object_type='calendar_event', + object_id=calendar_event.id, + data={'start_at': calendar_event.start_at}, + db=db, + ) + + return set() diff --git a/src/app.d.ts b/src/app.d.ts index 80c9762761..d23728b866 100644 --- a/src/app.d.ts +++ b/src/app.d.ts @@ -11,6 +11,10 @@ declare global { const APP_BUILD_CHANNEL: 'main' | 'dev' | 'unknown'; namespace App { + interface PageState { + inbox?: boolean; + inboxUserId?: string; + } // interface Error {} // interface Locals {} // interface PageData {} diff --git a/src/lib/apis/notifications/inbox.ts b/src/lib/apis/notifications/inbox.ts new file mode 100644 index 0000000000..543e7e762b --- /dev/null +++ b/src/lib/apis/notifications/inbox.ts @@ -0,0 +1,56 @@ +import { WEBUI_API_BASE_URL } from '$lib/constants'; + +export type InboxItem = { + id: string; + event_type: string; + actor_type: 'user' | 'system'; + actor: { id: string; name: string } | null; + resource_type: string | null; + resource_id: string | null; + object_type: string | null; + object_id: string | null; + title: string; + preview: string; + data: { + message?: string; + parent_id?: string; + start_at?: number; + target_type?: string; + target_id?: string; + }; + created_at: string; + seen_at: string | null; + read_at: string | null; + is_new?: boolean; +}; + +export type InboxPage = { + items: InboxItem[]; + next_cursor: string | null; + count?: number; + acknowledged_count?: number; + acknowledged_at?: string; +}; + +export async function inboxRequest(path: string, method = 'GET'): Promise { + const response = await fetch(`${WEBUI_API_BASE_URL}/notifications/inbox${path}`, { + method, + headers: { Authorization: `Bearer ${localStorage.token}`, Accept: 'application/json' } + }); + if (!response.ok) { + const error = await response.json().catch(() => ({})); + throw new Error(error.detail || 'Could not load inbox'); + } + return response.json(); +} + +export const inboxLabels: Record = { + 'comment.created': '{{name}} commented', + 'comment.replied': '{{name}} replied', + 'comment.mentioned': '{{name}} mentioned you', + 'channel.mentioned': '{{name}} mentioned you', + 'channel.replied': '{{name}} replied', + 'calendar.alert': 'Upcoming event', + 'automation.run_failed': 'Automation failed', + 'system.announcement': 'Announcement' +}; diff --git a/src/lib/components/admin/Settings/General.svelte b/src/lib/components/admin/Settings/General.svelte index 3a245cf264..2ed26d6ebc 100644 --- a/src/lib/components/admin/Settings/General.svelte +++ b/src/lib/components/admin/Settings/General.svelte @@ -392,6 +392,13 @@ > + + + import { formatSchedule } from '$lib/utils/schedule'; - import { onMount, getContext } from 'svelte'; + import { onMount, onDestroy, getContext, tick } from 'svelte'; import { toast } from 'svelte-sonner'; import { goto } from '$app/navigation'; + import { page } from '$app/stores'; + import { WEBUI_API_BASE_URL } from '$lib/constants'; import dayjs from 'dayjs'; import localizedFormat from 'dayjs/plugin/localizedFormat'; @@ -42,6 +44,34 @@ let showEditModal = false; let runs: AutomationRunModel[] = []; + let linkedRun: AutomationRunModel | null = null; + let linkedRunId = ''; + let alive = true; + onDestroy(() => { + alive = false; + }); + $: if (($page.url.searchParams.get('run') ?? '') !== linkedRunId) { + linkedRunId = $page.url.searchParams.get('run') ?? ''; + void revealRun(linkedRunId); + } + async function revealRun(id: string) { + linkedRun = null; + if (!id) return; + try { + const response = await fetch( + `${WEBUI_API_BASE_URL}/automations/${encodeURIComponent(automation.id)}/runs/${encodeURIComponent(id)}`, + { headers: { Authorization: `Bearer ${localStorage.token}` } } + ); + if (!response.ok) throw new Error($i18n.t('Run not found')); + const run = await response.json(); + if (!alive || linkedRunId !== id) return; + linkedRun = run; + await tick(); + document.getElementById('inbox-automation-run')?.scrollIntoView({ block: 'center' }); + } catch (error) { + if (alive && linkedRunId === id) toast.error(String(error)); + } + } let runsLoading = false; let hasMoreRuns = true; let runsPage = 0; @@ -307,6 +337,15 @@
+ {#if linkedRun}
+
+ {$i18n.t('Automation failed')} · {formatTime(linkedRun.created_at)} +
+

{linkedRun.error}

+
{/if}
{$i18n.t('Runs')}
{#if runsLoading && runs.length === 0} diff --git a/src/lib/components/calendar/CalendarEventModal.svelte b/src/lib/components/calendar/CalendarEventModal.svelte index 3a896df900..04e65a46fe 100644 --- a/src/lib/components/calendar/CalendarEventModal.svelte +++ b/src/lib/components/calendar/CalendarEventModal.svelte @@ -22,6 +22,11 @@ export let defaultCalendarId: string = ''; export let defaultStartAt: number | null = null; + function close() { + show = false; + dispatch('close'); + } + let title = ''; let description = ''; let calendarId = ''; @@ -170,7 +175,7 @@ if (result) { toast.success($i18n.t('Event updated')); dispatch('save', result); - show = false; + close(); } } else { const form: CalendarEventForm = { @@ -188,7 +193,7 @@ if (result) { toast.success($i18n.t('Event created')); dispatch('save', result); - show = false; + close(); } } } catch (err) { @@ -205,7 +210,7 @@ await deleteCalendarEvent(localStorage.token, event.id); toast.success($i18n.t('Event deleted')); dispatch('delete', event); - show = false; + close(); } catch (err) { toast.error(`${err}`); } finally { @@ -214,7 +219,7 @@ }; - +
(show = false)} + on:click={close} > {$i18n.t('Cancel')} diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte index 9509270495..682b551402 100644 --- a/src/lib/components/chat/Chat.svelte +++ b/src/lib/components/chat/Chat.svelte @@ -7,7 +7,7 @@ import { fade } from 'svelte/transition'; const i18n: Writable = getContext('i18n'); - import { goto, beforeNavigate } from '$app/navigation'; + import { goto, beforeNavigate, afterNavigate } from '$app/navigation'; import { page } from '$app/stores'; import { get, writable, type Unsubscriber, type Writable } from 'svelte/store'; @@ -1929,15 +1929,6 @@ }); } - const pageSubscribe = page.subscribe(async (p) => { - if (p.url.pathname === '/' || p.url.pathname.startsWith('/folders/')) { - await tick(); - initNewChat(); - } - - stopAudio(); - }); - const showControlsSubscribe = showControls.subscribe((value) => { if (!value) { showCallOverlay.set(false); @@ -1999,7 +1990,6 @@ if (chatIdProp && !$temporaryChatEnabled) { updateLastReadAt(chatIdProp); } - pageSubscribe(); showControlsSubscribe(); showArtifactsSubscribe(); selectedFolderSubscribe(); @@ -2479,6 +2469,14 @@ }); return () => bridge?.end(); }); + // Only real navigation resets the chat; shallow sidebar state changes must preserve it. + afterNavigate(async ({ to }) => { + if (to?.url.pathname === '/' || to?.url.pathname.startsWith('/folders/')) { + await tick(); + initNewChat(); + } + stopAudio(); + }); beforeNavigate(({ from, to }) => { if (!embedded && to && from?.url.pathname !== to.url.pathname) loading = true; bridge?.end(); diff --git a/src/lib/components/comments/CommentsPanel.svelte b/src/lib/components/comments/CommentsPanel.svelte index a70b1d2cf9..0529297516 100644 --- a/src/lib/components/comments/CommentsPanel.svelte +++ b/src/lib/components/comments/CommentsPanel.svelte @@ -21,6 +21,7 @@ import CommentComposer from './CommentComposer.svelte'; import CommentReplyComposer from './CommentReplyComposer.svelte'; import { scrollToComment } from '$lib/utils'; + import { v4 as uuidv4 } from 'uuid'; export let view: CommentsRequest; export let embedded = false; export let modal = false; @@ -38,9 +39,11 @@ chat: 'Chat' }; let resource: CommentResource | null = null; + const subscriptionId = uuidv4(); let title = ''; $: if (resource && total !== null) window.dispatchEvent(new Event('comments:updated')); let composerVersion = 0; + let composer: CommentComposer; let loadedCount = 0; let items: ResourceComment[] = []; let cursor: string | null = null; @@ -113,9 +116,13 @@ }; const join = () => { if (resource) - connection?.emit('comments:join', resource, (result: { error?: string }) => { - if (result?.error) fail(new CommentError(403, result.error)); - }); + connection?.emit( + 'comments:join', + { ...resource, subscription_id: subscriptionId }, + (result: { error?: string }) => { + if (result?.error) fail(new CommentError(403, result.error)); + } + ); }; const changed = (data: CommentResource) => { if ( @@ -130,8 +137,13 @@ join(); refresh(); }; - const unavailable = () => - fail(new CommentError(403, $i18n.t('Comments are no longer available.'))); + const unavailable = (data: CommentResource) => { + if ( + data?.resource_type === resource?.resource_type && + data?.resource_id === resource?.resource_id + ) + fail(new CommentError(403, $i18n.t('Comments are no longer available.'))); + }; const revealLinkedComment = () => { if (!alive || loading || !revealPending || !linkedCommentId) return; if (commentList?.querySelector('[aria-busy="true"]')) return; @@ -193,6 +205,7 @@ setup(); }); $: if ($socket !== connection) { + connection?.emit('comments:leave', { subscription_id: subscriptionId }); connection?.off('comments:changed', changed); connection?.off('connect', reconnect); connection?.off('comments:unavailable', unavailable); @@ -205,7 +218,7 @@ onDestroy(() => { alive = false; clearTimeout(refreshTimer); - connection?.emit('comments:leave'); + connection?.emit('comments:leave', { subscription_id: subscriptionId }); connection?.off('comments:changed', changed); connection?.off('connect', reconnect); connection?.off('comments:unavailable', unavailable); @@ -309,9 +322,11 @@
{#key `${resourceQuery(resource)}:${composerVersion}`} { composerVersion++; await tick(); + if (alive) composer?.focus(); if (alive && linkedCommentId) document .getElementById(`comment-${linkedCommentId}`) diff --git a/src/lib/components/icons/Bell.svelte b/src/lib/components/icons/Bell.svelte new file mode 100644 index 0000000000..131fc54b1c --- /dev/null +++ b/src/lib/components/icons/Bell.svelte @@ -0,0 +1,20 @@ + + + diff --git a/src/lib/components/layout/InboxBadge.svelte b/src/lib/components/layout/InboxBadge.svelte new file mode 100644 index 0000000000..2687e3a04e --- /dev/null +++ b/src/lib/components/layout/InboxBadge.svelte @@ -0,0 +1,13 @@ + + +{#if $config?.features?.enable_notifications && $inboxCount > 0} + +{/if} diff --git a/src/lib/components/layout/InboxButton.svelte b/src/lib/components/layout/InboxButton.svelte new file mode 100644 index 0000000000..f12b4307ce --- /dev/null +++ b/src/lib/components/layout/InboxButton.svelte @@ -0,0 +1,42 @@ + + + + + diff --git a/src/lib/components/layout/InboxHost.svelte b/src/lib/components/layout/InboxHost.svelte new file mode 100644 index 0000000000..46931ab69c --- /dev/null +++ b/src/lib/components/layout/InboxHost.svelte @@ -0,0 +1,81 @@ + + + + diff --git a/src/lib/components/layout/InboxPanel.svelte b/src/lib/components/layout/InboxPanel.svelte new file mode 100644 index 0000000000..987e3bf700 --- /dev/null +++ b/src/lib/components/layout/InboxPanel.svelte @@ -0,0 +1,244 @@ + + + + + diff --git a/src/lib/components/layout/Sidebar.svelte b/src/lib/components/layout/Sidebar.svelte index fd8d7cb7a9..06d3ef7e8c 100644 --- a/src/lib/components/layout/Sidebar.svelte +++ b/src/lib/components/layout/Sidebar.svelte @@ -69,6 +69,17 @@ import { createNoteHandler } from '$lib/components/notes/utils'; import { WEBUI_API_BASE_URL, WEBUI_BASE_URL } from '$lib/constants'; + import InboxButton from './InboxButton.svelte'; + import InboxHost from './InboxHost.svelte'; + import InboxPanel from './InboxPanel.svelte'; + let inboxHost: InboxHost; + let inboxPanel: InboxPanel; + $: inboxOpened = + $config?.features?.enable_notifications && + !!$page.state.inbox && + $page.state.inboxUserId === $user?.id; + import InboxBadge from './InboxBadge.svelte'; + import { inboxCount } from '$lib/stores/inbox'; import UserMenu from './Sidebar/UserMenu.svelte'; import ChatItem from './Sidebar/ChatItem.svelte'; import Spinner from '../common/Spinner.svelte'; @@ -686,7 +697,7 @@ }; const onKeyDown = (e) => { - if (e.key === 'Escape' && $selectedChatIds.size > 0 && !showDeleteSelectedConfirm) { + if (e.key === 'Escape' && !inboxOpened && $selectedChatIds.size > 0 && !showDeleteSelectedConfirm) { e.preventDefault(); chatSelection.clear(); } @@ -716,6 +727,7 @@ const MIN_WIDTH = 220; const MAX_WIDTH = 480; + const MOBILE_WIDTH = 280; let isResizing = false; let activePointerId: number | null = null; @@ -1031,11 +1043,15 @@ on:pointercancel={resizeEndHandler} /> +{#if $config?.features?.enable_notifications} + {#key $user?.id}{/key} +{/if} + showSidebar.set(open)} let:visible let:progress @@ -1082,7 +1098,7 @@ class="w-[calc(42px*var(--app-text-scale,1))] shrink-0 py-[calc(0.25rem*var(--app-text-scale,1))] px-[calc(0.25rem*var(--app-text-scale,1))] flex flex-col justify-between text-gray-700 dark:text-gray-300 hover:bg-gray-50/30 dark:hover:bg-gray-800/30 h-full z-10 transition-all border-e-[0.5px] border-gray-50 dark:border-gray-850/30" id="sidebar" role="navigation" - aria-label={$i18n.t('Chat history')} + aria-label={$i18n.t(inboxOpened ? 'Notifications' : 'Chat history')} >
- + + +
0 ? 'visible' : 'invisible'} sidebar-bg-gradient-to-b bg-linear-to-b from-gray-50 dark:from-gray-950 to-transparent from-50% pointer-events-none absolute inset-0 -z-10 -mb-6" >
+ {#if $config?.features?.enable_notifications} + {#key $user?.id} + void inboxHost.close()} + onSelect={(item) => inboxHost.select(item)} + /> + {/key} + {/if} +
{ if (e.target.scrollTop === 0) { scrollTop = 0; @@ -1867,7 +1900,7 @@
- {#if $selectedChatIds.size > 0} + {#if !inboxOpened && $selectedChatIds.size > 0}
('/count'); + if (request === revision) inboxCount.set(result.count); + } catch { + /* Keep the last known count while disconnected. */ + } +} diff --git a/src/lib/stores/index.ts b/src/lib/stores/index.ts index 693ec5b442..174e9d7325 100644 --- a/src/lib/stores/index.ts +++ b/src/lib/stores/index.ts @@ -415,6 +415,7 @@ type Config = { enable_memories: boolean; enable_notes?: boolean; enable_comments?: boolean; + enable_notifications?: boolean; enable_channels?: boolean; enable_user_webhooks?: boolean; enable_automations?: boolean; diff --git a/src/routes/(app)/automations/[id]/+page.svelte b/src/routes/(app)/automations/[id]/+page.svelte index 6cb3769b84..03a39575e3 100644 --- a/src/routes/(app)/automations/[id]/+page.svelte +++ b/src/routes/(app)/automations/[id]/+page.svelte @@ -1,7 +1,7 @@ {#if loaded && automation} - + {#key automation.id}{/key} {:else}
diff --git a/src/routes/(app)/calendar/+page.svelte b/src/routes/(app)/calendar/+page.svelte index 51763d5ba9..66604204f5 100644 --- a/src/routes/(app)/calendar/+page.svelte +++ b/src/routes/(app)/calendar/+page.svelte @@ -1,7 +1,8 @@ @@ -229,6 +260,7 @@ {calendars} {defaultCalendarId} {defaultStartAt} + on:close={closeLinkedEvent} on:save={() => refresh()} on:delete={() => refresh()} />