✨(backend): add command admin

we want to run the indexing from the admin.
in `dmin/core/runindexing/`is a form to do so.

Signed-off-by: charles <charles.englebert@protonmail.com>

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to
 
 ### Changed
 
+- ✨(backend) improve indexing command with checkpoint recovery and asynchronicity 
 - ♿️(frontend) improve version history list accessibility #2033
 - ♿(frontend) focus skip link on headings and skip grid dropzone #1983
 - ♿️(frontend) add sr-only format to export download button #2088

docs/commands/index.md

@@ -0,0 +1,52 @@
+# Index Command
+
+The `index` management command is used to index documents to the remote search indexer.
+
+It sends an asynchronous task to the Celery worker.
+
+## Usage
+
+### Command Line
+
+```bash
+python manage.py index \
+  --lower-time-bound "2024-01-01T00:00:00" \
+  --upper-time-bound "2024-01-31T23:59:59" \
+  --batch-size 200 \
+  --crash-safe-mode false
+```
+
+### Django Admin
+
+The command is available in the Django admin interface:
+
+1. Go to `/admin/`
+2. Click on **"Run Indexing"** in the CORE section
+3. Fill in the form with the desired parameters
+4. Click **"Run Indexing Command"**
+
+## Parameters
+
+### `--batch-size`
+- **type:** Integer
+- **default:** `50`
+- **description:** Number of documents to process per batch. Higher values may improve performance but use more memory.
+
+### `--lower-time-bound`
+- **optional**: true
+- **type:** ISO 8601 datetime string
+- **default:** `None`
+- **description:** Only documents updated after this date will be indexed.
+
+### `--upper-time-bound` (optional)
+- **optional**: true
+- **type:** ISO 8601 datetime string
+- **default:** `None`
+- **description:** Only documents updated before this date will be indexed.
+
+### `--crash-safe-mode`
+- **type:** Boolean flag
+- **default:** `True`
+- **description:** When enabled, the command orders documents by `updated_at` and stores the last indexed document's timestamp in cache. This allows resuming indexing from the last successful batch in case of a crash. However, it is more computationally expensive due to sorting.b
+
+

src/backend/core/admin.py

@@ -1,8 +1,10 @@
 """Admin classes and registrations for core app."""
 
+from django.conf import settings
 from django.contrib import admin, messages
 from django.contrib.auth import admin as auth_admin
-from django.shortcuts import redirect
+from django.core.management import call_command
+from django.shortcuts import redirect, render
 from django.utils.translation import gettext_lazy as _
 
 from treebeard.admin import TreeAdmin
@@ -227,3 +229,36 @@ class InvitationAdmin(admin.ModelAdmin):
     def save_model(self, request, obj, form, change):
         obj.issuer = request.user
         obj.save()
+
+
+@admin.register(models.RunIndexing)
+class RunIndexingAdmin(admin.ModelAdmin):
+    """Admin for running indexing commands."""
+
+    def changelist_view(self, request, extra_context=None):
+        """Override to avoid querying the database and handle form submission."""
+        if request.method == "POST":
+            try:
+                call_command(
+                    "index",
+                    batch_size=int(request.POST.get("batch_size")),
+                    lower_time_bound=request.POST.get("lower_time_bound"),
+                    upper_time_bound=request.POST.get("upper_time_bound"),
+                    crash_safe_mode=request.POST.get("crash_safe_mode"),
+                )
+                messages.success(request, _("Indexing triggered!"))
+            except Exception as e:
+                messages.error(
+                    request,
+                    _("Error running indexing command: %(error)s") % {"error": str(e)},
+                )
+
+        return render(
+            request=request,
+            template_name="runindexing.html",
+            context={
+                **self.admin_site.each_context(request),
+                "title": "Run Indexing Command",
+                "default_batch_size": settings.SEARCH_INDEXER_BATCH_SIZE,
+            },
+        )

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

@@ -3,11 +3,12 @@ Handle search setup that needs to be done at bootstrap time.
 """
 
 import logging
-import time
+from datetime import datetime
 
 from django.core.management.base import BaseCommand, CommandError
 
 from core.services.search_indexers import get_document_indexer
+from core.tasks.indexing import batch_document_indexer_task
 
 logger = logging.getLogger("docs.search.bootstrap_search")
 
@@ -27,6 +28,31 @@ class Command(BaseCommand):
             default=50,
             help="Indexation query batch size",
         )
+        parser.add_argument(
+            "--lower-time-bound",
+            action="store",
+            dest="lower_time_bound",
+            type=datetime.fromisoformat,
+            default=None,
+            help="DateTime in ISO format. Only documents updated after this date will be indexed",
+        )
+        parser.add_argument(
+            "--upper-time-bound",
+            action="store",
+            dest="upper_time_bound",
+            type=datetime.fromisoformat,
+            default=None,
+            help="DateTime in ISO format. Only documents updated before this date will be indexed",
+        )
+        parser.add_argument(
+            "--crash-safe-mode",
+            action="store_true",
+            dest="crash_safe_mode",
+            default=True,
+            help="If True, order documents by updated_at and store the last indexed document.updated_at in cache. "
+            "This allows resuming indexing from the last successful batch in case of a crash "
+            "but is more computationally expensive due to sorting.",
+        )
 
     def handle(self, *args, **options):
         """Launch and log search index generation."""
@@ -35,18 +61,18 @@ class Command(BaseCommand):
         if not indexer:
             raise CommandError("The indexer is not enabled or properly configured.")
 
-        logger.info("Starting to regenerate Find index...")
-        start = time.perf_counter()
-        batch_size = options["batch_size"]
-
         try:
-            count = indexer.index(batch_size=batch_size)
+            batch_document_indexer_task.apply_async(
+                kwargs={
+                    "lower_time_bound": options["lower_time_bound"],
+                    "upper_time_bound": options["upper_time_bound"],
+                    "batch_size": options["batch_size"],
+                    "crash_safe_mode": options["crash_safe_mode"],
+                },
+            )
         except Exception as err:
-            raise CommandError("Unable to regenerate index") from err
+            raise CommandError("Unable send indexing task to worker") from err
 
-        duration = time.perf_counter() - start
         logger.info(
-            "Search index regenerated from %d document(s) in %.2f seconds.",
+            "Document indexing task sent to worker",
-            count,
-            duration,
         )

src/backend/core/migrations/0032_runindexing.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.2.12 on 2026-03-26 16:34
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0031_clean_onboarding_accesses'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='RunIndexing',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+            ],
+            options={
+                'verbose_name': 'Run Indexing',
+                'managed': False,
+            },
+        ),
+    ]

src/backend/core/models.py

@@ -2029,3 +2029,13 @@ class Invitation(BaseModel):
             "partial_update": is_admin_or_owner,
             "retrieve": is_admin_or_owner,
         }
+
+
+class RunIndexing(models.Model):
+    """Proxy model for indexing management in admin."""
+
+    class Meta:
+        """Meta options."""
+
+        managed = False
+        verbose_name = _("Run Indexing")

src/backend/core/services/search_indexers.py

@@ -1,5 +1,6 @@
 """Document search index management utilities and indexers"""
 
+import itertools
 import logging
 from abc import ABC, abstractmethod
 from collections import defaultdict
@@ -7,6 +8,7 @@ from functools import cache
 
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
+from django.core.cache import cache as django_cache
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.module_loading import import_string
 
@@ -18,6 +20,9 @@ from core.enums import SearchType
 logger = logging.getLogger(__name__)
 
 
+BULK_INDEXER_CHECKPOINT = "bulk-indexer-checkpoint"
+
+
 @cache
 def get_document_indexer():
     """Returns an instance of indexer service if enabled and properly configured."""
@@ -125,7 +130,7 @@ class BaseDocumentIndexer(ABC):
         if not self.search_url:
             raise ImproperlyConfigured("SEARCH_URL must be set in Django settings.")
 
-    def index(self, queryset=None, batch_size=None):
+    def index(self, queryset=None, batch_size=None, crash_safe_mode=False):
         """
         Fetch documents in batches, serialize them, and push to the search backend.
 
@@ -134,35 +139,37 @@ class BaseDocumentIndexer(ABC):
                 Defaults to all documents without filter.
             batch_size (int, optional): Number of documents per batch.
                 Defaults to settings.SEARCH_INDEXER_BATCH_SIZE.
+            crash_safe_mode (bool, optional): If True, order documents by updated_at
+                and store the last indexed document.updated_at in cache.
+                This allows resuming indexing from the last successful batch in case of a crash
+                but is more computationally expensive due to sorting.
         """
-        last_id = 0
         count = 0
         queryset = queryset or models.Document.objects.all()
         batch_size = batch_size or self.batch_size
 
-        while True:
+        if crash_safe_mode:
-            documents_batch = list(
+            queryset = queryset.order_by("updated_at")
-                queryset.filter(
-                    id__gt=last_id,
-                ).order_by("id")[:batch_size]
-            )
-
-            if not documents_batch:
-                break
 
+        for documents_batch in itertools.batched(queryset.iterator(), batch_size):
             doc_paths = [doc.path for doc in documents_batch]
-            last_id = documents_batch[-1].id
             accesses_by_document_path = get_batch_accesses_by_users_and_teams(doc_paths)
-
             serialized_batch = [
                 self.serialize_document(document, accesses_by_document_path)
                 for document in documents_batch
                 if document.content or document.title
             ]
 
-            if serialized_batch:
+            if not serialized_batch:
-                self.push(serialized_batch)
+                continue
-                count += len(serialized_batch)
+
+            self.push(serialized_batch)
+            count += len(serialized_batch)
+
+            if crash_safe_mode:
+                django_cache.set(
+                    BULK_INDEXER_CHECKPOINT, serialized_batch[-1]["updated_at"]
+                )
 
         return count
 

src/backend/core/signals.py

@@ -10,7 +10,7 @@ from django.db.models import signals
 from django.dispatch import receiver
 
 from core import models
-from core.tasks.search import trigger_batch_document_indexer
+from core.tasks.indexing import trigger_batch_document_indexer
 from core.utils import get_users_sharing_documents_with_cache_key
 
 

src/backend/core/tasks/indexing.py

@@ -0,0 +1,130 @@
+"""Trigger document indexation using celery task."""
+
+from logging import getLogger
+
+from django.conf import settings
+from django.core.cache import cache
+
+from django_redis.cache import RedisCache
+
+from core import models
+from core.services.search_indexers import (
+    get_document_indexer,
+)
+from core.utils import build_indexable_documents_queryset
+
+from impress.celery_app import app
+
+logger = getLogger(__file__)
+
+
+@app.task
+def document_indexer_task(document_id):
+    """
+    Celery Task: Indexes a single document by its ID.
+
+    Args:
+        document_id: Primary key of the document to index.
+    """
+    indexer = get_document_indexer()
+
+    if indexer:
+        logger.info("Start document %s indexation", document_id)
+        indexer.index(models.Document.objects.filter(pk=document_id))
+
+
+def batch_indexer_throttle_acquire(timeout: int = 0, atomic: bool = True):
+    """
+    Acquire a throttle lock to prevent multiple batch indexation tasks during countdown.
+
+    implements a debouncing pattern: only the first call during the timeout period
+    will succeed, subsequent calls are skipped until the timeout expires.
+
+    Args:
+        timeout (int): Lock duration in seconds (countdown period).
+        atomic (bool): Use Redis locks for atomic operations if available.
+
+    Returns:
+        bool: True if lock acquired (first call), False if already held (subsequent calls).
+    """
+    key = "document-batch-indexer-throttle"
+
+    # Redis is used as cache database (not in tests). Use the lock feature here
+    # to ensure atomicity of changes to the throttle flag.
+    if isinstance(cache, RedisCache) and atomic:
+        with cache.locks(key):
+            return batch_indexer_throttle_acquire(timeout, atomic=False)
+
+    # cache.add() is atomic test-and-set operation:
+    #   - If key doesn't exist: creates it with timeout and returns True
+    #   - If key already exists: does nothing and returns False
+    # The key expires after timeout seconds, releasing the lock.
+    # The value 1 is irrelevant, only the key presence/absence matters.
+    return cache.add(key, 1, timeout=timeout)
+
+
+@app.task
+def batch_document_indexer_task(lower_time_bound=None, upper_time_bound=None, **kwargs):
+    """
+    Celery Task: Batch indexes all documents modified since timestamp.
+
+    Args:
+        lower_time_bound (datetime|isoformat, optional):
+            indexes documents updated or deleted after this timestamp.
+        upper_time_bound (datetime|isoformat, optional):
+            indexes documents updated or deleted before this timestamp.
+    """
+    indexer = get_document_indexer()
+
+    if not indexer:
+        logger.warning("Indexing task triggered but no indexer configured: skipping")
+        return
+
+    count = indexer.index(
+        queryset=build_indexable_documents_queryset(
+            lower_time_bound=lower_time_bound, upper_time_bound=upper_time_bound
+        ),
+        **kwargs,
+    )
+    logger.info("Indexed %d documents", count)
+
+
+def trigger_batch_document_indexer(document):
+    """
+    Trigger document indexation with optional debounce mechanism.
+
+    behavior depends on SEARCH_INDEXER_COUNTDOWN setting:
+    - if countdown > 0 sec (async batch mode):
+      * schedules a batch indexation task after countdown in seconds
+      * uses throttle mechanism to ensure only ONE batch task runs per countdown period
+      * all documents modified since first trigger are indexed together
+    - if countdown == 0 sec (sync mode):
+      * executes indexation synchronously in the current thread
+      * no batching, no throttling, no Celery task queuing
+
+    Args:
+        document (Document): the document instance that triggered the indexation.
+    """
+    countdown = int(settings.SEARCH_INDEXER_COUNTDOWN)
+
+    # DO NOT create a task if indexation is disabled
+    if not settings.SEARCH_INDEXER_CLASS:
+        return
+
+    if countdown > 0:
+        # use throttle to ensure only one task is scheduled per countdown period.
+        # if throttle acquired, schedule batch task; otherwise skip.
+        if batch_indexer_throttle_acquire(timeout=countdown):
+            logger.info(
+                "Add task for batch document indexation from updated_at=%s in %d seconds",
+                document.updated_at.isoformat(),
+                countdown,
+            )
+
+            batch_document_indexer_task.apply_async(
+                kwargs={"lower_time_bound": document.updated_at}, countdown=countdown
+            )
+        else:
+            logger.info("Skip task for batch document %s indexation", document.pk)
+    else:
+        document_indexer_task.apply(args=[document.pk])

src/backend/core/tasks/search.py

@@ -1,95 +0,0 @@
-"""Trigger document indexation using celery task."""
-
-from logging import getLogger
-
-from django.conf import settings
-from django.core.cache import cache
-from django.db.models import Q
-
-from django_redis.cache import RedisCache
-
-from core import models
-from core.services.search_indexers import (
-    get_document_indexer,
-)
-
-from impress.celery_app import app
-
-logger = getLogger(__file__)
-
-
-@app.task
-def document_indexer_task(document_id):
-    """Celery Task : Sends indexation query for a document."""
-    indexer = get_document_indexer()
-
-    if indexer:
-        logger.info("Start document %s indexation", document_id)
-        indexer.index(models.Document.objects.filter(pk=document_id))
-
-
-def batch_indexer_throttle_acquire(timeout: int = 0, atomic: bool = True):
-    """
-    Enable the task throttle flag for a delay.
-    Uses redis locks if available to ensure atomic changes
-    """
-    key = "document-batch-indexer-throttle"
-
-    # Redis is used as cache database (not in tests). Use the lock feature here
-    # to ensure atomicity of changes to the throttle flag.
-    if isinstance(cache, RedisCache) and atomic:
-        with cache.locks(key):
-            return batch_indexer_throttle_acquire(timeout, atomic=False)
-
-    # Use add() here :
-    #   - set the flag and returns true if not exist
-    #   - do nothing and return false if exist
-    return cache.add(key, 1, timeout=timeout)
-
-
-@app.task
-def batch_document_indexer_task(timestamp):
-    """Celery Task : Sends indexation query for a batch of documents."""
-    indexer = get_document_indexer()
-
-    if indexer:
-        queryset = models.Document.objects.filter(
-            Q(updated_at__gte=timestamp)
-            | Q(deleted_at__gte=timestamp)
-            | Q(ancestors_deleted_at__gte=timestamp)
-        )
-
-        count = indexer.index(queryset)
-        logger.info("Indexed %d documents", count)
-
-
-def trigger_batch_document_indexer(document):
-    """
-    Trigger indexation task with debounce a delay set by the SEARCH_INDEXER_COUNTDOWN setting.
-
-    Args:
-        document (Document): The document instance.
-    """
-    countdown = int(settings.SEARCH_INDEXER_COUNTDOWN)
-
-    # DO NOT create a task if indexation if disabled
-    if not settings.SEARCH_INDEXER_CLASS:
-        return
-
-    if countdown > 0:
-        # Each time this method is called during a countdown, we increment the
-        # counter and each task decrease it, so the index be run only once.
-        if batch_indexer_throttle_acquire(timeout=countdown):
-            logger.info(
-                "Add task for batch document indexation from updated_at=%s in %d seconds",
-                document.updated_at.isoformat(),
-                countdown,
-            )
-
-            batch_document_indexer_task.apply_async(
-                args=[document.updated_at], countdown=countdown
-            )
-        else:
-            logger.info("Skip task for batch document %s indexation", document.pk)
-    else:
-        document_indexer_task.apply(args=[document.pk])

src/backend/core/templates/runindexing.html

@@ -0,0 +1,43 @@
+{% extends "admin/base_site.html" %}
+{% load i18n %}
+
+{% block content %}
+
+<form method="POST" >
+    {% csrf_token %}
+
+    <hr style="margin-bottom: 10px;">
+
+    <p>
+        This command triggers the indexing of all documents within the specified time bound.
+    </p>
+    <p>
+        See <b>docs/commands/index.md</b> for more details.
+    </p>
+
+    <hr style="margin-bottom: 20px;">
+
+    <div>
+        <label for="batch_size" style="margin-right: 10px">Batch size</label>
+        <input id="batch_size" type="number" name="batch_size" value={{ default_batch_size }} style="width: 80px;" >
+    </div>
+
+    <div>
+        <label for="lower_time_bound" style="margin-right: 10px">Lower time bound (optional)</label>
+        <input id="lower_time_bound" type="datetime-local" name="lower_time_bound" >
+    </div>
+
+    <div>
+        <label for="upper_time_bound" style="margin-right: 10px">Upper time bound (optional)</label>
+        <input id="upper_time_bound" type="datetime-local" name="upper_time_bound" >
+    </div>
+
+    <div>
+        <label for="crash_safe_mode" style="margin-right: 10px">Crash save mode</label>
+        <input id="crash_safe_mode" type="checkbox" name="crash_safe_mode" value="true" checked>
+    </div>
+
+    <input type="submit" value="{% translate 'Run Indexing' %}" style="margin-top: 20px;">
+</form>
+
+{% endblock %}

src/backend/core/tests/commands/test_index.py

@@ -2,21 +2,24 @@
 Unit test for `index` command.
 """
 
+from datetime import datetime, timedelta, timezone
 from operator import itemgetter
 from unittest import mock
 
+from django.core.cache import cache
 from django.core.management import CommandError, call_command
 from django.db import transaction
 
 import pytest
 
 from core import factories
-from core.services.search_indexers import FindDocumentIndexer
+from core.services.search_indexers import BULK_INDEXER_CHECKPOINT, FindDocumentIndexer
+from core.tests.conftest import create_document_with_updated_at
 
 
 @pytest.mark.django_db
 @pytest.mark.usefixtures("indexer_settings")
-def test_index():
+def test_index_without_bound_success():
     """Test the command `index` that run the Find app indexer for all the available documents."""
     user = factories.UserFactory()
     indexer = FindDocumentIndexer()
@@ -37,20 +40,152 @@ def test_index():
     }
 
     with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
-        call_command("index")
+        call_command("index", crash_safe_mode=False)
 
-        push_call_args = [call.args[0] for call in mock_push.call_args_list]
+    push_call_args = [call.args[0] for call in mock_push.call_args_list]
 
-        # called once but with a batch of docs
+    # called once but with a batch of docs
-        mock_push.assert_called_once()
+    mock_push.assert_called_once()
 
-        assert sorted(push_call_args[0], key=itemgetter("id")) == sorted(
+    assert sorted(push_call_args[0], key=itemgetter("id")) == sorted(
-            [
+        [
-                indexer.serialize_document(doc, accesses),
+            indexer.serialize_document(doc, accesses),
-                indexer.serialize_document(no_title_doc, accesses),
+            indexer.serialize_document(no_title_doc, accesses),
-            ],
+        ],
-            key=itemgetter("id"),
+        key=itemgetter("id"),
+    )
+
+    # crash_safe_mode deactivated -> no checkpoint stored
+    assert cache.get(BULK_INDEXER_CHECKPOINT) is None
+
+
+@pytest.mark.django_db
+@pytest.mark.usefixtures("indexer_settings")
+def test_index_with_both_bounds_success():
+    """Test the command `index` for all documents within time bound."""
+    cache.clear()
+    lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
+    upper_time_bound = lower_time_bound + timedelta(days=30)
+
+    document_too_early = create_document_with_updated_at(
+        lower_time_bound - timedelta(days=10)
+    )
+    document_in_window_1 = create_document_with_updated_at(
+        lower_time_bound + timedelta(days=5)
+    )
+    document_in_window_2 = create_document_with_updated_at(
+        lower_time_bound + timedelta(days=15)
+    )
+    document_too_late = create_document_with_updated_at(
+        upper_time_bound + timedelta(days=10)
+    )
+
+    with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
+        call_command(
+            "index",
+            lower_time_bound=lower_time_bound.isoformat(),
+            upper_time_bound=upper_time_bound.isoformat(),
         )
+    all_push_call_args = [
+        document["id"]
+        for call_arg_list in mock_push.call_args_list
+        for document in call_arg_list.args[0]
+    ]
+
+    # Only documents in window should be indexed
+    assert str(document_too_early.id) not in all_push_call_args
+    assert str(document_in_window_1.id) in all_push_call_args
+    assert str(document_in_window_2.id) in all_push_call_args
+    assert str(document_too_late.id) not in all_push_call_args
+
+    # Checkpoint should be set to last indexed document's updated_at
+    assert (
+        cache.get(BULK_INDEXER_CHECKPOINT)
+        == document_in_window_2.updated_at.isoformat()
+    )
+
+
+@pytest.mark.django_db
+@pytest.mark.usefixtures("indexer_settings")
+def test_index_with_crash_recovery():
+    """Test resuming indexing from checkpoint after a crash."""
+    cache.clear()
+    lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
+    upper_time_bound = lower_time_bound + timedelta(days=60)
+
+    batch_size = 2
+    documents = [
+        # batch 0
+        create_document_with_updated_at(lower_time_bound + timedelta(days=5)),
+        create_document_with_updated_at(lower_time_bound + timedelta(days=10)),
+        # batch 1
+        create_document_with_updated_at(lower_time_bound + timedelta(days=20)),
+        create_document_with_updated_at(lower_time_bound + timedelta(days=25)),
+        # batch 2 - will crash here
+        create_document_with_updated_at(lower_time_bound + timedelta(days=30)),
+        create_document_with_updated_at(lower_time_bound + timedelta(days=35)),
+        # batch 3
+        create_document_with_updated_at(lower_time_bound + timedelta(days=40)),
+        create_document_with_updated_at(lower_time_bound + timedelta(days=45)),
+    ]
+
+    # First run: simulate crash on batch 3
+    with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
+
+        def push_with_failure(data):
+            # Crash when we encounter document at index 4 (batch 2 with batch_size=2)
+            if str(documents[4].id) in [document["id"] for document in data]:
+                raise ConnectionError("Simulated indexing error")
+
+        mock_push.side_effect = push_with_failure
+
+        call_command(
+            "index",
+            batch_size=batch_size,
+            lower_time_bound=lower_time_bound.isoformat(),
+            upper_time_bound=upper_time_bound.isoformat(),
+        )
+    all_push_call_args = [
+        document["id"]
+        for call_arg_list in mock_push.call_args_list
+        for document in call_arg_list.args[0]
+    ]
+
+    # First 2 batches should have been indexed and checkpoint saved
+    checkpoint = cache.get(BULK_INDEXER_CHECKPOINT)
+    assert checkpoint == documents[3].updated_at.isoformat()
+    # first 2 batches should be indexed successfully
+    for i in range(0, 4):
+        assert str(documents[i].id) in all_push_call_args
+    # next batch should have been attempted but failed
+    for i in range(4, 6):
+        assert str(documents[i].id) in all_push_call_args
+    # last batches indexing should not have been attempted
+    for i in range(6, 8):
+        assert str(documents[i].id) not in all_push_call_args
+
+    # Second run: resume from checkpoint
+    with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
+        call_command(
+            "index",
+            batch_size=batch_size,
+            lower_time_bound=checkpoint,
+            upper_time_bound=upper_time_bound.isoformat(),
+        )
+        all_push_call_args = [
+            document["id"]
+            for call_arg_list in mock_push.call_args_list
+            for document in call_arg_list.args[0]
+        ]
+
+        # first 2 batches should NOT be re-indexed
+        # except the last document of the last batch which is on the checkpoint boundary
+        for i in range(0, 3):
+            assert str(documents[i].id) not in all_push_call_args
+        # next batches should be indexed including the document at the checkpoint boundary
+        # which has already been indexed and is re-indexed
+        for i in range(4, 8):
+            assert str(documents[i].id) in all_push_call_args
 
 
 @pytest.mark.django_db

src/backend/core/tests/conftest.py

@@ -8,7 +8,7 @@ from django.core.cache import cache
 import pytest
 import responses
 
-from core import factories
+from core import factories, models
 from core.tests.utils.urls import reload_urls
 
 USER = "user"
@@ -143,3 +143,10 @@ def user_token():
     A fixture to create a user token for testing.
     """
     return build_authorization_bearer("some_token")
+
+
+def create_document_with_updated_at(updated_at, **kwargs):
+    """Helper to create a document with a specific updated_at, bypassing auto_now=True."""
+    document = factories.DocumentFactory(**kwargs)
+    models.Document.objects.filter(pk=document.pk).update(updated_at=updated_at)
+    return models.Document.objects.get(pk=document.pk)

src/backend/core/tests/test_utils.py

@@ -2,6 +2,7 @@
 
 import base64
 import uuid
+from datetime import datetime, timedelta, timezone
 
 from django.core.cache import cache
 
@@ -9,6 +10,7 @@ import pycrdt
 import pytest
 
 from core import factories, utils
+from core.tests.conftest import create_document_with_updated_at
 
 pytestmark = pytest.mark.django_db
 
@@ -240,3 +242,127 @@ def test_utils_get_value_by_pattern_no_match():
     result = utils.get_value_by_pattern(data, r"^title\.")
 
     assert result == []
+
+
+def test_utils_build_indexable_documents_queryset_no_filters():
+    """Test build_indexable_documents_queryset with no filters returns all documents."""
+    factories.DocumentFactory.create_batch(3)
+
+    queryset = utils.build_indexable_documents_queryset()
+
+    assert queryset.count() == 3
+
+
+def test_utils_build_indexable_documents_queryset_oldest_updated_at():
+    """Test filtering by oldest_updated_at includes documents updated after timestamp."""
+    lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
+
+    document_too_old = create_document_with_updated_at(
+        updated_at=lower_time_bound - timedelta(days=30)
+    )
+    document_at_boundary = create_document_with_updated_at(lower_time_bound)
+    document_recent = create_document_with_updated_at(
+        updated_at=lower_time_bound + timedelta(days=15)
+    )
+
+    queryset = utils.build_indexable_documents_queryset(
+        lower_time_bound=lower_time_bound
+    )
+
+    assert queryset.count() == 2
+    assert document_too_old not in queryset
+    assert document_at_boundary in queryset
+    assert document_recent in queryset
+
+
+def test_utils_build_indexable_documents_queryset_newest_updated_at():
+    """Test filtering by newest_updated_at includes documents updated before timestamp."""
+    upper_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
+
+    document_old = create_document_with_updated_at(
+        updated_at=upper_time_bound - timedelta(days=30)
+    )
+    document_at_boundary = create_document_with_updated_at(upper_time_bound)
+    document_too_recent = create_document_with_updated_at(
+        updated_at=upper_time_bound + timedelta(days=15)
+    )
+
+    queryset = utils.build_indexable_documents_queryset(
+        upper_time_bound=upper_time_bound
+    )
+
+    assert queryset.count() == 2
+    assert document_old in queryset
+    assert document_at_boundary in queryset
+    assert document_too_recent not in queryset
+
+
+def test_utils_build_indexable_documents_queryset_both_bounds():
+    """Test filtering with both oldest and newest bounds."""
+    lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
+    upper_time_bound = lower_time_bound + timedelta(days=30)
+
+    document_too_early = create_document_with_updated_at(
+        updated_at=lower_time_bound - timedelta(days=10)
+    )
+    document_in_window = create_document_with_updated_at(
+        updated_at=lower_time_bound + timedelta(days=5)
+    )
+    other_document_in_window = create_document_with_updated_at(
+        updated_at=lower_time_bound + timedelta(days=15)
+    )
+    document_too_late = create_document_with_updated_at(
+        updated_at=upper_time_bound + timedelta(days=10)
+    )
+
+    queryset = utils.build_indexable_documents_queryset(
+        lower_time_bound=lower_time_bound, upper_time_bound=upper_time_bound
+    )
+
+    assert queryset.count() == 2
+    assert document_too_early not in queryset
+    assert document_in_window in queryset
+    assert other_document_in_window in queryset
+    assert document_too_late not in queryset
+
+
+def test_utils_build_indexable_documents_queryset_deleted_at():
+    """Test filtering includes documents with deleted_at in range."""
+    lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
+
+    document_deleted_in_range = create_document_with_updated_at(
+        updated_at=lower_time_bound - timedelta(days=30),
+        deleted_at=lower_time_bound + timedelta(days=15),
+    )
+    document_updated_before_range = create_document_with_updated_at(
+        updated_at=lower_time_bound - timedelta(days=30)
+    )
+
+    queryset = utils.build_indexable_documents_queryset(
+        lower_time_bound=lower_time_bound
+    )
+
+    assert queryset.count() == 1
+    assert document_deleted_in_range in queryset
+    assert document_updated_before_range not in queryset
+
+
+def test_utils_build_indexable_documents_queryset_ancestors_deleted_at():
+    """Test filtering includes documents with ancestors_deleted_at in range."""
+    lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
+
+    document_ancestor_deleted_in_range = create_document_with_updated_at(
+        updated_at=lower_time_bound - timedelta(days=30),
+        ancestors_deleted_at=lower_time_bound + timedelta(days=15),
+    )
+    document_updated_before_range = create_document_with_updated_at(
+        updated_at=lower_time_bound - timedelta(days=30)
+    )
+
+    queryset = utils.build_indexable_documents_queryset(
+        lower_time_bound=lower_time_bound
+    )
+
+    assert queryset.count() == 1
+    assert document_ancestor_deleted_in_range in queryset
+    assert document_updated_before_range not in queryset

src/backend/core/utils.py

@@ -8,7 +8,7 @@ from collections import defaultdict
 
 from django.core.cache import cache
 from django.db import models as db
-from django.db.models import Subquery
+from django.db.models import Q, Subquery
 
 import pycrdt
 from bs4 import BeautifulSoup
@@ -168,3 +168,47 @@ def users_sharing_documents_with(user):
         elapsed,
     )
     return result
+
+
+def build_indexable_documents_queryset(lower_time_bound=None, upper_time_bound=None):
+    """
+    Build a queryset of documents to index filtered by timestamp range.
+
+    filters out documents that have been updated, or deleted outside the time range.
+
+    Args:
+        lower_time_bound (datetime, optional):
+            keeps documents updated or deleted after this timestamp.
+        upper_time_bound (datetime, optional):
+            keeps documents updated or deleted before this timestamp.
+
+    Returns:
+        QuerySet: Filtered Document queryset ready for indexation.
+    """
+
+    queryset = models.Document.objects.all()
+
+    conditions = Q()
+    if lower_time_bound and upper_time_bound:
+        conditions = (
+            Q(updated_at__gte=lower_time_bound, updated_at__lte=upper_time_bound)
+            | Q(deleted_at__gte=lower_time_bound, deleted_at__lte=upper_time_bound)
+            | Q(
+                ancestors_deleted_at__gte=lower_time_bound,
+                ancestors_deleted_at__lte=upper_time_bound,
+            )
+        )
+    elif lower_time_bound:
+        conditions = (
+            Q(updated_at__gte=lower_time_bound)
+            | Q(deleted_at__gte=lower_time_bound)
+            | Q(ancestors_deleted_at__gte=lower_time_bound)
+        )
+    elif upper_time_bound:
+        conditions = (
+            Q(updated_at__lte=upper_time_bound)
+            | Q(deleted_at__lte=upper_time_bound)
+            | Q(ancestors_deleted_at__lte=upper_time_bound)
+        )
+
+    return queryset.filter(conditions)