Feat/Bulk and selective log deletion

api/controllers/console/app/conversation.py

@@ -129,6 +129,39 @@ def get(self, app_model):
 
         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):
+        current_user, _ = current_account_with_tenant()
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_ids", type=list, location="json", required=False)
+        args = parser.parse_args()
+
+        conversation_ids = [str(id) for id in args["conversation_ids"]] if args.get("conversation_ids") else None
+
+        result = ConversationService.clear_conversations(
+            app_model=app_model, user=current_user, conversation_ids=conversation_ids
+        )
+
+        return result, 202
+
 
 @console_ns.route("/apps/<uuid:app_id>/completion-conversations/<uuid:conversation_id>")
 class CompletionConversationDetailApi(Resource):
@@ -337,6 +370,39 @@ def get(self, app_model):
 
         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):
+        current_user, _ = current_account_with_tenant()
+        parser = reqparse.RequestParser()
+        parser.add_argument("conversation_ids", type=list, location="json", required=False)
+        args = parser.parse_args()
+
+        conversation_ids = [str(id) for id in args["conversation_ids"]] if args.get("conversation_ids") else None
+
+        result = ConversationService.clear_conversations(
+            app_model=app_model, user=current_user, conversation_ids=conversation_ids
+        )
+
+        return result, 202
+
 
 @console_ns.route("/apps/<uuid:app_id>/chat-conversations/<uuid:conversation_id>")
 class ChatConversationDetailApi(Resource):

api/services/conversation_service.py

@@ -23,6 +23,7 @@
     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 +129,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 +164,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 +217,67 @@ 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
+        """
+        # 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
+
+        # Queue the Celery task
+        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,
+        )
+
+        # 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",
+        }
+
     @classmethod
     def get_conversational_variable(
         cls,

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}