✨ (backend) Add async triggers to enable document indexation with find.

Signed-off-by: Fabre Florian <ffabre@hybird.org>

Author: sampaccoud

src/backend/core/api/viewsets.py

@@ -987,6 +987,11 @@ def duplicate(self, request, *args, **kwargs):
             {"id": str(duplicated_document.id)}, status=status.HTTP_201_CREATED
         )
 
+    # TODO
+    # @drf.decorators.action(detail=False, methods=["get"])
+    # def search(self, request, *args, **kwargs):
+    #    index.search()
+
     @drf.decorators.action(detail=True, methods=["get"], url_path="versions")
     def versions_list(self, request, *args, **kwargs):
         """

src/backend/core/management/commands/index.py

@@ -25,4 +25,4 @@ def handle(self, *args, **options):
         FindDocumentIndexer().index()
 
         duration = time.perf_counter() - start
-        logger.info(f"Search index regenerated in {duration:.2f} seconds.")
+        logger.info("Search index regenerated in %.2f seconds.", duration)

src/backend/core/models.py

@@ -20,7 +20,9 @@
 from django.core.files.storage import default_storage
 from django.core.mail import send_mail
 from django.db import models, transaction
+from django.db.models import signals
 from django.db.models.functions import Left, Length
+from django.dispatch import receiver
 from django.template.loader import render_to_string
 from django.utils import timezone
 from django.utils.functional import cached_property
@@ -40,6 +42,7 @@
     get_equivalent_link_definition,
 )
 from .validators import sub_validator
+from .tasks.find import trigger_document_indexer
 
 logger = getLogger(__name__)
 
@@ -946,6 +949,16 @@ def restore(self):
             )
 
 
+@receiver(signals.post_save, sender=Document)
+def document_post_save(sender, instance, **kwargs):
+    """
+    Asynchronous call to the document indexer at the end of the transaction.
+    Note : Within the transaction we can have an empty content and a serialization
+    error.
+    """
+    trigger_document_indexer(instance, on_commit=True)
+
+
 class LinkTrace(BaseModel):
     """
     Relation model to trace accesses to a document via a link by a logged-in user.
@@ -1171,6 +1184,15 @@ def get_abilities(self, user):
         }
 
 
+@receiver(signals.post_save, sender=DocumentAccess)
+def document_access_post_save(sender, instance, created, **kwargs):
+    """
+    Asynchronous call to the document indexer at the end of the transaction.
+    """
+    if not created:
+        trigger_document_indexer(instance.document, on_commit=True)
+
+
 class DocumentAskForAccess(BaseModel):
     """Relation model to ask for access to a document."""
 

src/backend/core/services/search_indexers.py

@@ -5,6 +5,7 @@
 from collections import defaultdict
 
 from django.conf import settings
+from django.contrib.auth.models import AnonymousUser
 
 import requests
 
@@ -18,11 +19,13 @@ def get_batch_accesses_by_users_and_teams(paths):
     Get accesses related to a list of document paths,
     grouped by users and teams, including all ancestor paths.
     """
-    print("paths: ", paths)
-    ancestor_map = utils.get_ancestor_to_descendants_map(paths, steplen=models.Document.steplen)
+    # print("paths: ", paths)
+    ancestor_map = utils.get_ancestor_to_descendants_map(
+        paths, steplen=models.Document.steplen
+    )
     ancestor_paths = list(ancestor_map.keys())
-    print("ancestor map: ", ancestor_map)
-    print("ancestor paths: ", ancestor_paths)
+    # print("ancestor map: ", ancestor_map)
+    # print("ancestor paths: ", ancestor_paths)
 
     access_qs = models.DocumentAccess.objects.filter(
         document__path__in=ancestor_paths
@@ -44,6 +47,22 @@ def get_batch_accesses_by_users_and_teams(paths):
     return dict(access_by_document_path)
 
 
+def get_visited_document_ids_of(user):
+    if isinstance(user, AnonymousUser):
+        return []
+
+    # TODO : exclude links when user already have a specific access to the doc
+    qs = models.LinkTrace.objects.filter(
+        user=user
+    ).exclude(
+        document__accesses__user=user,
+    )
+
+    return list({
+        str(id) for id in qs.values_list("document_id", flat=True)
+    })
+
+
 class BaseDocumentIndexer(ABC):
     """
     Base class for document indexers.
@@ -84,6 +103,7 @@ def index(self):
             serialized_batch = [
                 self.serialize_document(document, accesses_by_document_path)
                 for document in documents_batch
+                if document.content
             ]
             self.push(serialized_batch)
 
@@ -103,6 +123,38 @@ def push(self, data):
         Must be implemented by subclasses.
         """
 
+    def search(self, text, user, token):
+        """
+        Search for documents in Find app.
+        """
+        visited_ids = get_visited_document_ids_of(user)
+
+        response = self.search_query(data={
+            "q": text,
+            "visited": visited_ids,
+            "services": ["docs"],
+        }, token=token)
+
+        print(response)
+
+        return self.format_response(response)
+
+    @abstractmethod
+    def search_query(self, data, token) -> dict:
+        """
+        Retreive documents from the Find app API.
+
+        Must be implemented by subclasses.
+        """
+
+    @abstractmethod
+    def format_response(self, data: dict):
+        """
+        Convert the JSON response from Find app as document queryset.
+
+        Must be implemented by subclasses.
+        """
+
 
 class FindDocumentIndexer(BaseDocumentIndexer):
     """
@@ -121,10 +173,12 @@ def serialize_document(self, document, accesses):
             dict: A JSON-serializable dictionary.
         """
         doc_path = document.path
-        text_content = utils.base64_yjs_to_text(document.content)
+        doc_content = document.content
+        text_content = utils.base64_yjs_to_text(doc_content) if doc_content else ""
+
         return {
             "id": str(document.id),
-            "title": document.title,
+            "title": document.title or "",
             "content": text_content,
             "depth": document.depth,
             "path": document.path,
@@ -138,6 +192,46 @@ def serialize_document(self, document, accesses):
             "is_active": not bool(document.ancestors_deleted_at),
         }
 
+    def search_query(self, data, token) -> requests.Response:
+        """
+        Retrieve documents from the Find app API.
+
+        Args:
+            data (dict): search data
+            token (str): OICD token
+
+        Returns:
+            dict: A JSON-serializable dictionary.
+        """
+        url = getattr(settings, "SEARCH_INDEXER_QUERY_URL", None)
+
+        if not url:
+            raise RuntimeError(
+                "SEARCH_INDEXER_QUERY_URL must be set in Django settings before indexing."
+            )
+
+        try:
+            response = requests.post(
+                url,
+                json=data,
+                headers={"Authorization": f"Bearer {token}"},
+                timeout=10,
+            )
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.HTTPError as e:
+            logger.error("HTTPError: %s", e)
+            logger.error("Response content: %s", response.text)  # type: ignore
+            raise
+
+    def format_response(self, data: dict):
+        """
+        Retrieve documents ids from Find app response and return a queryset.
+        """
+        return models.Document.objects.filter(pk__in=[
+            d['_id'] for d in data
+        ])
+
     def push(self, data):
         """
         Push a batch of documents to the Find backend.
@@ -156,6 +250,7 @@ def push(self, data):
             raise RuntimeError(
                 "SEARCH_INDEXER_SECRET must be set in Django settings before indexing."
             )
+
         try:
             response = requests.post(
                 url,

src/backend/core/tasks/find.py

@@ -0,0 +1,96 @@
+"""Trigger document indexation using celery task."""
+
+from logging import getLogger
+
+from django.conf import settings
+from django.core.cache import cache
+from django.db import transaction
+
+from core import models
+from core.services.search_indexers import (
+    FindDocumentIndexer,
+    get_batch_accesses_by_users_and_teams,
+)
+
+from impress.celery_app import app
+
+logger = getLogger(__file__)
+
+
+def document_indexer_debounce_key(document_id):
+    """Returns debounce cache key"""
+    return f"doc-indexer-debounce-{document_id}"
+
+
+def incr_counter(key):
+    """Increase or reset counter"""
+    try:
+        return cache.incr(key)
+    except ValueError:
+        cache.set(key, 1)
+        return 1
+
+
+def decr_counter(key):
+    """Decrease or reset counter"""
+    try:
+        return cache.decr(key)
+    except ValueError:
+        cache.set(key, 0)
+        return 0
+
+
+@app.task
+def document_indexer_task(document_id):
+    """Send indexation query for a document using celery task."""
+    key = document_indexer_debounce_key(document_id)
+
+    # check if the counter : if still up, skip the task. only the last one
+    # within the countdown delay will do the query.
+    if decr_counter(key) > 0:
+        logger.info("Skip document %s indexation", document_id)
+        return
+
+    doc = models.Document.objects.get(pk=document_id)
+    indexer = FindDocumentIndexer()
+    accesses = get_batch_accesses_by_users_and_teams((doc.path,))
+
+    data = indexer.serialize_document(document=doc, accesses=accesses)
+
+    logger.info("Start document %s indexation", document_id)
+    indexer.push(data)
+
+
+def trigger_document_indexer(document, on_commit=False):
+    """
+    Trigger indexation task with debounce a delay set by the SEARCH_INDEXER_COUNTDOWN setting.
+
+    Args:
+        document (Document): The document instance.
+        on_commit (bool): Wait for the end of the transaction before starting the task
+            (some fields may be in wrong state within the transaction)
+    """
+
+    if document.deleted_at or document.ancestors_deleted_at:
+        pass
+
+    if on_commit:
+
+        def _aux():
+            trigger_document_indexer(document, on_commit=False)
+
+        transaction.on_commit(_aux)
+    else:
+        key = document_indexer_debounce_key(document.pk)
+        countdown = getattr(settings, "SEARCH_INDEXER_COUNTDOWN", 1)
+
+        logger.info(
+            "Add task for document %s indexation in %.2f seconds",
+            document.pk, countdown
+        )
+
+        # Each time this method is called during the countdown, we increment the
+        # counter and each task decrease it, so the index be run only once.
+        incr_counter(key)
+
+        document_indexer_task.apply_async(args=[document.pk], countdown=countdown)