Feat/Bulk and selective log deletion

api/controllers/console/app/conversation.py

@@ -384,6 +384,47 @@
 
         return conversations
 
+    @api.doc("clear_completion_conversations")
+    @api.doc(description="Clear completion conversations and related data")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.expect(
+        api.parser().add_argument(
+            "conversation_ids",
+            type=list,
+            location="json",
+            required=False,
+            help="Optional list of conversation IDs to clear. If not provided, all conversations will be cleared.",
+        )
+    )
+    @api.response(202, "Clearing task queued successfully")
+    @api.response(403, "Insufficient permissions")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.COMPLETION)
+    @edit_permission_required
+    def delete(self, app_model):
+        from services.errors.conversation import ConversationClearInProgressError
+
+        current_user, _ = current_account_with_tenant()
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_ids", type=list, location="json", required=False, default=None)
+        args = parser.parse_args()
+
+        # Convert conversation IDs to strings if provided and non-empty
+        conversation_ids_raw = args.get("conversation_ids")
+        conversation_ids = (
+            [str(id) for id in conversation_ids_raw] if conversation_ids_raw and len(conversation_ids_raw) > 0 else None
+        )
+
+        try:
+            result = ConversationService.clear_conversations(
+                app_model=app_model, user=current_user, conversation_ids=conversation_ids
+            )
+            return result, 202
+        except ConversationClearInProgressError as e:
+            return {"message": str(e), "code": "task_in_progress"}, 409
+
 
 @console_ns.route("/apps/<uuid:app_id>/completion-conversations/<uuid:conversation_id>")
 class CompletionConversationDetailApi(Resource):
@@ -536,6 +577,47 @@
 
         return conversations
 
+    @api.doc("clear_chat_conversations")
+    @api.doc(description="Clear chat conversations and related data")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.expect(
+        api.parser().add_argument(
+            "conversation_ids",
+            type=list,
+            location="json",
+            required=False,
+            help="Optional list of conversation IDs to clear. If not provided, all conversations will be cleared.",
+        )
+    )
+    @api.response(202, "Clearing task queued successfully")
+    @api.response(403, "Insufficient permissions")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
+    @edit_permission_required
+    def delete(self, app_model):
+        from services.errors.conversation import ConversationClearInProgressError
+
+        current_user, _ = current_account_with_tenant()
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_ids", type=list, location="json", required=False, default=None)
+        args = parser.parse_args()
+
+        # Convert conversation IDs to strings if provided and non-empty
+        conversation_ids_raw = args.get("conversation_ids")
+        conversation_ids = (
+            [str(id) for id in conversation_ids_raw] if conversation_ids_raw and len(conversation_ids_raw) > 0 else None
+        )
+
+        try:
+            result = ConversationService.clear_conversations(
+                app_model=app_model, user=current_user, conversation_ids=conversation_ids
+            )
+            return result, 202
+        except ConversationClearInProgressError as e:
+            return {"message": str(e), "code": "task_in_progress"}, 409
+
 
 @console_ns.route("/apps/<uuid:app_id>/chat-conversations/<uuid:conversation_id>")
 class ChatConversationDetailApi(Resource):

api/services/conversation_service.py

@@ -1,4 +1,5 @@
 import contextlib
+import hashlib
 import logging
 from collections.abc import Callable, Sequence
 from typing import Any, Union
@@ -11,18 +12,21 @@
 from core.variables.types import SegmentType
 from core.workflow.nodes.variable_assigner.common.impl import conversation_variable_updater_factory
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from factories import variable_factory
 from libs.datetime_utils import naive_utc_now
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from models import Account, ConversationVariable
 from models.model import App, Conversation, EndUser, Message
 from services.errors.conversation import (
+    ConversationClearInProgressError,
     ConversationNotExistsError,
     ConversationVariableNotExistsError,
     ConversationVariableTypeMismatchError,
     LastConversationNotExistsError,
 )
 from services.errors.message import MessageNotExistsError
+from tasks.clear_conversation_task import clear_conversations_task
 from tasks.delete_conversation_task import delete_conversation_related_data
 
 logger = logging.getLogger(__name__)
@@ -128,7 +132,18 @@ def rename(
         else:
             conversation.name = name
             conversation.updated_at = naive_utc_now()
-            db.session.commit()
+            try:
+                db.session.commit()
+            except Exception as e:
+                # Handle case where conversation was deleted after we retrieved it
+                from sqlalchemy.orm.exc import StaleDataError
+
+                db.session.rollback()
+                if isinstance(e, StaleDataError):
+                    # Conversation was likely deleted, raise ConversationNotExistsError
+                    raise ConversationNotExistsError()
+                else:
+                    raise
 
         return conversation
 
@@ -152,7 +167,18 @@ def auto_generate_name(cls, app_model: App, conversation: Conversation):
             )
             conversation.name = name
 
-        db.session.commit()
+        try:
+            db.session.commit()
+        except Exception as e:
+            # Handle case where conversation was deleted after we retrieved it
+            from sqlalchemy.orm.exc import StaleDataError
+
+            db.session.rollback()
+            if isinstance(e, StaleDataError):
+                # Conversation was likely deleted, raise ConversationNotExistsError
+                raise ConversationNotExistsError()
+            else:
+                raise
 
         return conversation
 
@@ -194,6 +220,93 @@ def delete(cls, app_model: App, conversation_id: str, user: Union[Account, EndUs
             db.session.rollback()
             raise e
 
+    @classmethod
+    def clear_conversations(
+        cls, app_model: App, user: Union[Account, EndUser] | None, conversation_ids: list[str] | None = None
+    ) -> dict[str, Any]:
+        """
+        Clear conversations and related data, optionally for specific conversation IDs.
+        Uses Celery task for handling large datasets.
+
+        Args:
+            app_model: The app model
+            user: The user (Account or EndUser)
+            conversation_ids: Optional list of specific conversation IDs to clear
+
+        Returns:
+            dict with task info and estimated counts
+
+        Raises:
+            ConversationClearInProgressError: If a clearing task is already in progress
+        """
+        # Validate conversation ownership if specific IDs provided
+        if conversation_ids:
+            for conversation_id in conversation_ids:
+                cls.get_conversation(app_model, conversation_id, user)
+
+        # Get conversation mode for task
+        if app_model.mode == "completion":
+            mode = "completion"
+        else:
+            mode = "chat"  # covers chat, agent-chat, advanced-chat
+
+        # Generate unique lock key to prevent duplicate tasks
+        # Key format: clear_conversations:{app_id}:{mode}[:{conversation_ids_hash}]
+        lock_key = f"clear_conversations:{app_model.id}:{mode}"
+        if conversation_ids:
+            # For selective deletion, include hash of conversation IDs
+            ids_hash = hashlib.md5(",".join(sorted(conversation_ids)).encode()).hexdigest()[:8]
+            lock_key += f":{ids_hash}"
+
+        # Try to acquire lock (expires in 10 minutes)
+        # nx=True means set only if key doesn't exist
+        lock_acquired = redis_client.set(lock_key, "1", ex=600, nx=True)
+        if not lock_acquired:
+            # Another clearing task is already in progress
+            raise ConversationClearInProgressError(
+                "A conversation clearing task is already in progress for this app. Please wait for it to complete."
+            )
+
+        try:
+            # Queue the Celery task with lock_key for cleanup
+            task = clear_conversations_task.delay(
+                app_id=app_model.id,
+                conversation_mode=mode,
+                conversation_ids=conversation_ids,
+                user_id=user.id if user else None,
+                user_type="account" if isinstance(user, Account) else "end_user" if isinstance(user, EndUser) else None,
+                lock_key=lock_key,
+            )
+
+            # Get estimated counts for response
+            if conversation_ids:
+                conversation_count = len(conversation_ids)
+            else:
+                # Estimate total conversations for this app
+                conversation_count = (
+                    db.session.query(Conversation)
+                    .filter(
+                        Conversation.app_id == app_model.id,
+                        Conversation.mode == mode,
+                        Conversation.from_source == ("api" if isinstance(user, EndUser) else "console"),
+                        Conversation.from_end_user_id == (user.id if isinstance(user, EndUser) else None),
+                        Conversation.from_account_id == (user.id if isinstance(user, Account) else None),
+                        Conversation.is_deleted == False,
+                    )
+                    .count()
+                )
+
+            return {
+                "task_id": task.id,
+                "status": "queued",
+                "estimated_conversations": conversation_count,
+                "mode": "selective" if conversation_ids else "all",
+            }
+        except Exception:
+            # Release lock if task queueing failed
+            redis_client.delete(lock_key)
+            raise
+
     @classmethod
     def get_conversational_variable(
         cls,

api/services/errors/conversation.py

@@ -19,3 +19,9 @@ class ConversationVariableNotExistsError(BaseServiceError):
 
 class ConversationVariableTypeMismatchError(BaseServiceError):
     pass
+
+
+class ConversationClearInProgressError(BaseServiceError):
+    """Raised when a conversation clear operation is already in progress for the same app/mode."""
+
+    pass

api/tasks/clean_uploaded_files_task.py

@@ -0,0 +1,39 @@
+import logging
+
+from celery import shared_task
+
+from extensions.ext_database import db
+from extensions.ext_storage import storage
+from models.model import UploadFile
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(queue="dataset")
+def clean_uploaded_files_task(upload_file_ids: list[str]):
+    """
+    Asynchronously clean uploaded files from storage.
+    This task is called after database records have been successfully deleted.
+
+    Args:
+        upload_file_ids: List of upload file IDs to delete from storage
+    """
+    if not upload_file_ids:
+        return
+
+    success_count = 0
+    failed_count = 0
+
+    for upload_file_id in upload_file_ids:
+        try:
+            upload_file = db.session.query(UploadFile).where(UploadFile.id == upload_file_id).first()
+
+            if upload_file and upload_file.key:
+                storage.delete(upload_file.key)
+                success_count += 1
+
+        except Exception:
+            logger.exception("Failed to delete upload file %s", upload_file_id)
+            failed_count += 1
+
+    return {"total": len(upload_file_ids), "success": success_count, "failed": failed_count}