feat: prioritize recent-window backfill before full history

3 files changed, 77 insertions, 13 deletions

diff --git a/src/srht_contrib/services/aggregator.py b/src/srht_contrib/services/aggregator.py
index 1c8af86..4973f16 100644
--- a/src/srht_contrib/services/aggregator.py
+++ b/src/srht_contrib/services/aggregator.py
@@ -62,6 +62,12 @@ class ContributionAggregator:
             is_indexed=calendar.is_indexed,
             last_polled_at=calendar.last_polled_at,
             indexing_state=calendar.indexing_state,
+            is_recent_window_backfilled=calendar.is_recent_window_backfilled,
+            recent_backfill_state=calendar.recent_backfill_state,
+            recent_backfill_completed_at=calendar.recent_backfill_completed_at,
+            is_backfilled=calendar.is_backfilled,
+            backfill_state=calendar.backfill_state,
+            backfill_completed_at=calendar.backfill_completed_at,
         )
 
     def _index_metadata(self, db: Session, actor: str) -> ContributionIndexMetadata:
@@ -81,6 +87,9 @@ class ContributionAggregator:
             is_indexed=is_indexed,
             last_polled_at=tracked_actor.last_polled_at if tracked_actor is not None else None,
             indexing_state=indexing_state,
+            is_recent_window_backfilled=(tracked_actor.recent_backfill_status == "completed") if tracked_actor is not None else False,
+            recent_backfill_state=(tracked_actor.recent_backfill_status if tracked_actor is not None else "pending"),
+            recent_backfill_completed_at=tracked_actor.recent_backfill_completed_at if tracked_actor is not None else None,
             is_backfilled=(tracked_actor.backfill_status == "completed") if tracked_actor is not None else False,
             backfill_state=(tracked_actor.backfill_status if tracked_actor is not None else "pending"),
             backfill_completed_at=tracked_actor.backfill_completed_at if tracked_actor is not None else None,
diff --git a/src/srht_contrib/services/git.py b/src/srht_contrib/services/git.py
index ebb823e..bf7493b 100644
--- a/src/srht_contrib/services/git.py
+++ b/src/srht_contrib/services/git.py
@@ -120,6 +120,24 @@ class GitIngestionService:
         return repositories
 
     def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=None)
+
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=since)
+
+    def _fetch_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None,
+        *,
+        since: datetime | None,
+    ) -> BackfillBatchResult:
         state = {
             "discovery_cursor": None,
             "discovery_complete": False,
@@ -186,13 +204,18 @@ class GitIngestionService:
         log_page = repository.get("log") or {}
         commits = log_page.get("results") or []
         next_cursor = log_page.get("cursor")
-        events = [
-            normalized
-            for commit in commits
-            if isinstance(commit, dict)
-            for normalized in [self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)]
-            if normalized is not None
-        ]
+        events: list[NormalizedEvent] = []
+        stop_repository = False
+        for commit in commits:
+            if not isinstance(commit, dict):
+                continue
+            commit_time = parse_datetime((commit.get("author") or {}).get("time"))
+            if since is not None and commit_time < since:
+                stop_repository = True
+                break
+            normalized = self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)
+            if normalized is not None:
+                events.append(normalized)
         logger.info(
             "git backfill actor=%s repository=%s commits=%s next_cursor=%s",
             actor,
@@ -200,7 +223,7 @@ class GitIngestionService:
             len(commits),
             bool(next_cursor),
         )
-        if next_cursor:
+        if next_cursor and not stop_repository:
             state["current_repository"]["cursor"] = next_cursor
         else:
             state["current_repository"] = None
diff --git a/src/srht_contrib/services/todo.py b/src/srht_contrib/services/todo.py
index b7209b1..9f2ece6 100644
--- a/src/srht_contrib/services/todo.py
+++ b/src/srht_contrib/services/todo.py
@@ -296,6 +296,24 @@ class TodoIngestionService:
         return TodoPollResult(events=tracker_events, cursor=cursor_time)
 
     def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=None)
+
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=since)
+
+    def _fetch_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None,
+        *,
+        since: datetime | None,
+    ) -> BackfillBatchResult:
         state = {
             "trackers_cursor": None,
             "tracker_queue": [],
@@ -349,10 +367,20 @@ class TodoIngestionService:
             tracker = data.get("tracker") or {}
             tickets_page = tracker.get("tickets") or {}
             tickets = tickets_page.get("results") or []
-            current_tracker["tickets_cursor"] = tickets_page.get("cursor")
-            current_tracker["pending_tickets"].extend(
-                [{"id": int(ticket["id"]), "ref": str(ticket.get("ref") or ticket["id"])} for ticket in tickets if isinstance(ticket, dict)]
-            )
+            next_tickets_cursor = tickets_page.get("cursor")
+            stop_tracker_paging = False
+            for ticket in tickets:
+                if not isinstance(ticket, dict):
+                    continue
+                if since is not None:
+                    updated = parse_datetime(ticket["updated"])
+                    if updated < since:
+                        stop_tracker_paging = True
+                        continue
+                current_tracker["pending_tickets"].append(
+                    {"id": int(ticket["id"]), "ref": str(ticket.get("ref") or ticket["id"])}
+                )
+            current_tracker["tickets_cursor"] = None if stop_tracker_paging else next_tickets_cursor
             logger.info(
                 "todo backfill tracker=%s ticket page count=%s next_cursor=%s pending_tickets=%s",
                 current_tracker.get("name") or current_tracker.get("id"),
@@ -391,6 +419,7 @@ class TodoIngestionService:
         )
 
         events: list[NormalizedEvent] = []
+        stop_ticket_paging = False
         for event in page_events:
             if not isinstance(event, dict):
                 continue
@@ -402,6 +431,9 @@ class TodoIngestionService:
                 "tracker": {"name": current_tracker.get("name")},
             }
             occurred_at = parse_datetime(event["created"])
+            if since is not None and occurred_at < since:
+                stop_ticket_paging = True
+                continue
             for change in event.get("changes") or []:
                 if not isinstance(change, dict):
                     continue
@@ -415,7 +447,7 @@ class TodoIngestionService:
                 if normalized is not None:
                     events.append(normalized)
 
-        if next_cursor:
+        if next_cursor and not stop_ticket_paging:
             state["current_ticket"]["cursor"] = next_cursor
         else:
             state["current_ticket"] = None