aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--API.md25
-rw-r--r--README.md16
-rw-r--r--alembic/versions/20260411_0003_backfill_state.py55
-rw-r--r--src/srht_contrib/jobs/poller.py91
-rw-r--r--src/srht_contrib/models.py19
-rw-r--r--src/srht_contrib/schemas.py3
-rw-r--r--src/srht_contrib/services/aggregator.py3
-rw-r--r--src/srht_contrib/services/git.py90
-rw-r--r--src/srht_contrib/services/todo.py136
-rw-r--r--src/srht_contrib/services/types.py12
-rw-r--r--tests/test_contributions_api.py4
-rw-r--r--tests/test_ingestion.py48
-rw-r--r--tests/test_migrations.py3
13 files changed, 498 insertions, 7 deletions
diff --git a/API.md b/API.md
index 6fcf40e..5b8f0e9 100644
--- a/API.md
+++ b/API.md
@@ -44,6 +44,7 @@ Contribution ranges:
- Contribution read endpoints return zero-filled days, so clients do not need to patch missing dates.
- Public contribution reads also register the actor for background indexing. A first lookup may therefore return an empty graph while the scheduler catches up.
+- Incremental indexing and historical backfill are separate. An actor can be recently indexed without being fully backfilled yet.
Background polling:
@@ -110,6 +111,7 @@ Behavior notes:
- This endpoint resolves aliases to a canonical actor before querying data.
- This endpoint also registers the actor for background indexing and updates the actor's `last_requested_at` timestamp.
- The response is always immediate; it does not wait for SourceHut polling to finish.
+- Historical backfill runs in bounded background batches and may take multiple scheduler passes to complete.
Example by year:
@@ -133,6 +135,9 @@ Response `200 OK`:
"is_indexed": false,
"last_polled_at": null,
"indexing_state": "pending",
+ "is_backfilled": false,
+ "backfill_state": "in_progress",
+ "backfill_completed_at": null,
"days": [
{ "date": "2026-03-01", "count": 0, "score": 0.0 },
{ "date": "2026-03-02", "count": 3, "score": 2.5 }
@@ -145,9 +150,12 @@ Response fields:
- `actor` string: canonical actor after alias resolution
- `from` string: inclusive start date
- `to` string: inclusive end date
-- `is_indexed` boolean: whether the service has already indexed activity for this actor
+- `is_indexed` boolean: whether the service has already completed at least one successful recent/incremental poll for this actor
- `last_polled_at` string or `null`: most recent successful poll time, if any
- `indexing_state` string: one of `pending`, `indexed`, or `error`
+- `is_backfilled` boolean: whether historical backfill has completed for this actor
+- `backfill_state` string: one of `pending`, `in_progress`, `completed`, or `error`
+- `backfill_completed_at` string or `null`: when full historical backfill completed, if it has
- `days` array:
- `date` string `YYYY-MM-DD`
- `count` integer contribution count for the day
@@ -159,6 +167,13 @@ Indexing state semantics:
- `indexed`: at least one successful poll has completed for the actor
- `error`: the most recent poll attempt for the actor failed
+Backfill state semantics:
+
+- `pending`: the actor has not started historical backfill yet
+- `in_progress`: historical backfill is actively progressing in bounded background batches
+- `completed`: historical backfill has completed for all supported services
+- `error`: the most recent backfill attempt failed
+
Possible errors:
- `400 Bad Request` for invalid or conflicting date input
@@ -193,6 +208,7 @@ Behavior notes:
- This endpoint has the same actor-registration and alias-resolution behavior as the calendar endpoint.
- This endpoint returns immediately and does not block on SourceHut polling.
+- This endpoint also reflects historical backfill state so clients can distinguish recent indexing from complete history.
Example:
@@ -210,6 +226,9 @@ Response `200 OK`:
"is_indexed": true,
"last_polled_at": "2026-04-11T18:05:00Z",
"indexing_state": "indexed",
+ "is_backfilled": false,
+ "backfill_state": "in_progress",
+ "backfill_completed_at": null,
"total_events": 126,
"total_score": 116.75,
"active_days": 14,
@@ -226,6 +245,9 @@ Response fields:
- `is_indexed` boolean
- `last_polled_at` string or `null`
- `indexing_state` string
+- `is_backfilled` boolean
+- `backfill_state` string
+- `backfill_completed_at` string or `null`
- `total_events` integer
- `total_score` float
- `active_days` integer
@@ -275,6 +297,7 @@ Response fields:
Behavior notes:
- Manual polling also updates the actor's indexing metadata.
+- Manual polling also advances historical backfill by one bounded batch per supported service.
- Git polling auto-discovers the actor's owned repositories and unions in any configured tracked repositories.
Possible errors:
diff --git a/README.md b/README.md
index 8e4d32a..b5939f6 100644
--- a/README.md
+++ b/README.md
@@ -14,7 +14,7 @@ The current V1 is intentionally narrow and production-oriented:
## What It Does
-The service collects SourceHut activity from one or more sr.ht GraphQL services, turns those records into a canonical event shape, aggregates activity by day, and returns zero-filled calendar ranges so the client never has to patch missing dates.
+The service collects SourceHut activity from one or more sr.ht GraphQL services, turns those records into a canonical event shape, aggregates activity by day, and returns zero-filled calendar ranges so the client never has to patch missing dates. It performs both recent incremental polling and bounded historical backfill.
Example use cases:
@@ -28,7 +28,7 @@ The code is split into small, testable layers:
- `src/srht_contrib/config.py`: environment-driven settings and event weights
- `src/srht_contrib/db.py`: SQLAlchemy engine/session setup and app-scoped DB access
-- `src/srht_contrib/models.py`: ORM models for normalized events, sync state, aliases, and tracked repos
+- `src/srht_contrib/models.py`: ORM models for normalized events, sync state, aliases, tracked actors, and backfill state
- `src/srht_contrib/services/srht_client.py`: generic SourceHut GraphQL client with error handling and simple retries
- `src/srht_contrib/services/todo.py`: `todo.sr.ht` ingestion and normalization
- `src/srht_contrib/services/git.py`: `git.sr.ht` tracked-repository commit ingestion
@@ -165,7 +165,7 @@ Example response:
}
```
-Scheduled polling only runs when `ENABLE_SCHEDULER=true`. The scheduler seeds `DEFAULT_ACTOR` as an initial known actor, and public contribution reads register additional actors for later background polling.
+Scheduled polling only runs when `ENABLE_SCHEDULER=true`. The scheduler seeds `DEFAULT_ACTOR` as an initial known actor, runs one poll immediately at startup, and public contribution reads register additional actors for later background polling and historical backfill.
For `git.sr.ht`, owned repositories are auto-discovered for the actor. `GIT_TRACKED_REPOSITORIES` can still be used to union in extra repositories. Entries may be either:
@@ -208,6 +208,9 @@ Example response:
"is_indexed": true,
"last_polled_at": "2026-04-11T18:05:00Z",
"indexing_state": "indexed",
+ "is_backfilled": false,
+ "backfill_state": "in_progress",
+ "backfill_completed_at": null,
"days": [
{"date": "2026-03-28", "count": 3, "score": 3.5},
{"date": "2026-03-29", "count": 0, "score": 0.0},
@@ -229,6 +232,12 @@ Example response:
"actor": "~your-user",
"from": "2026-01-01",
"to": "2026-12-31",
+ "is_indexed": true,
+ "last_polled_at": "2026-04-11T18:05:00Z",
+ "indexing_state": "indexed",
+ "is_backfilled": false,
+ "backfill_state": "in_progress",
+ "backfill_completed_at": null,
"total_events": 42,
"total_score": 37.5,
"active_days": 18,
@@ -317,6 +326,7 @@ The SourceHut-specific assumptions are isolated to the service modules:
- `git.sr.ht` polling assumes the actor's repositories are discoverable through the SourceHut GraphQL API
- scheduled polling runs in-process, so it is not a distributed scheduler
- newly requested actors are indexed asynchronously, so the first public read may be empty until a scheduler or manual poll runs
+- full historical backfill can take many scheduler passes for active users because it runs in bounded batches
- alias management is config-driven; there is no alias CRUD API yet
- current deployment model is trusted-operator V1, not a public multi-tenant service
diff --git a/alembic/versions/20260411_0003_backfill_state.py b/alembic/versions/20260411_0003_backfill_state.py
new file mode 100644
index 0000000..371c739
--- /dev/null
+++ b/alembic/versions/20260411_0003_backfill_state.py
@@ -0,0 +1,55 @@
+"""actor and service backfill state"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+
+revision = "20260411_0003"
+down_revision = "20260411_0002"
+branch_labels = None
+depends_on = None
+
+
+def _table_names() -> set[str]:
+ return set(inspect(op.get_bind()).get_table_names())
+
+
+def _column_names(table_name: str) -> set[str]:
+ return {column["name"] for column in inspect(op.get_bind()).get_columns(table_name)}
+
+
+def upgrade() -> None:
+ if "tracked_actors" in _table_names():
+ columns = _column_names("tracked_actors")
+ if "backfill_status" not in columns:
+ op.add_column("tracked_actors", sa.Column("backfill_status", sa.String(length=32), nullable=True))
+ op.execute(sa.text("UPDATE tracked_actors SET backfill_status = 'pending' WHERE backfill_status IS NULL"))
+ if "backfill_started_at" not in columns:
+ op.add_column("tracked_actors", sa.Column("backfill_started_at", sa.DateTime(timezone=True), nullable=True))
+ if "backfill_completed_at" not in columns:
+ op.add_column("tracked_actors", sa.Column("backfill_completed_at", sa.DateTime(timezone=True), nullable=True))
+ if "last_backfill_error" not in columns:
+ op.add_column("tracked_actors", sa.Column("last_backfill_error", sa.Text(), nullable=True))
+
+ if "service_backfill_states" not in _table_names():
+ op.create_table(
+ "service_backfill_states",
+ sa.Column("id", sa.Integer(), primary_key=True),
+ sa.Column("actor", sa.String(length=255), nullable=False),
+ sa.Column("service", sa.String(length=32), nullable=False),
+ sa.Column("cursor_json", sa.JSON(), nullable=True),
+ sa.Column("status", sa.String(length=32), nullable=False),
+ sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+ sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+ sa.Column("last_error", sa.Text(), nullable=True),
+ sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+ sa.UniqueConstraint("actor", "service", name="uq_service_backfill_state_actor_service"),
+ )
+
+
+def downgrade() -> None:
+ if "service_backfill_states" in _table_names():
+ op.drop_table("service_backfill_states")
diff --git a/src/srht_contrib/jobs/poller.py b/src/srht_contrib/jobs/poller.py
index 910f1ab..2b00090 100644
--- a/src/srht_contrib/jobs/poller.py
+++ b/src/srht_contrib/jobs/poller.py
@@ -7,7 +7,7 @@ from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
-from srht_contrib.models import ContributionEvent, SyncState, TrackedActor, TrackedRepository
+from srht_contrib.models import ContributionEvent, ServiceBackfillState, SyncState, TrackedActor, TrackedRepository
from srht_contrib.schemas import NormalizedEvent
from srht_contrib.services.git import GitIngestionService
from srht_contrib.services.srht_client import SourceHutClientError
@@ -35,6 +35,7 @@ class PollerService:
raise
self._update_tracked_actor_poll_state(db, actor, status="indexed", error=None)
+ inserted += self._run_backfill_batches(db, actor)
db.commit()
return inserted
@@ -61,7 +62,7 @@ class PollerService:
def track_actor_request(self, db: Session, actor: str, *, update_last_requested: bool = True) -> TrackedActor:
tracked_actor = db.scalar(select(TrackedActor).where(TrackedActor.actor == actor))
if tracked_actor is None:
- tracked_actor = TrackedActor(actor=actor, is_active=True)
+ tracked_actor = TrackedActor(actor=actor, is_active=True, backfill_status="pending")
db.add(tracked_actor)
tracked_actor.is_active = True
@@ -172,3 +173,89 @@ class PollerService:
tracked_actor.last_polled_at = datetime.now(tz=UTC)
db.add(tracked_actor)
db.flush()
+
+ def _run_backfill_batches(self, db: Session, actor: str) -> int:
+ tracked_actor = self.track_actor_request(db, actor, update_last_requested=False)
+ if tracked_actor.backfill_status == "completed":
+ return 0
+
+ if tracked_actor.backfill_started_at is None:
+ tracked_actor.backfill_started_at = datetime.now(tz=UTC)
+ tracked_actor.backfill_status = "in_progress"
+ tracked_actor.last_backfill_error = None
+ db.add(tracked_actor)
+ db.flush()
+ total_inserted = 0
+
+ services = [
+ (self.todo_service.service_name, self.todo_service.fetch_backfill_batch),
+ (self.git_service.service_name, self.git_service.fetch_backfill_batch),
+ ]
+ all_complete = True
+ for service_name, fetcher in services:
+ state = db.scalar(
+ select(ServiceBackfillState)
+ .where(ServiceBackfillState.actor == actor)
+ .where(ServiceBackfillState.service == service_name)
+ )
+ if state is None:
+ state = ServiceBackfillState(
+ actor=actor,
+ service=service_name,
+ cursor_json=None,
+ status="pending",
+ started_at=None,
+ completed_at=None,
+ last_error=None,
+ updated_at=datetime.now(tz=UTC),
+ )
+ db.add(state)
+ db.flush()
+
+ if state.status == "completed":
+ continue
+
+ all_complete = False
+ if state.started_at is None:
+ state.started_at = datetime.now(tz=UTC)
+ state.status = "in_progress"
+ state.updated_at = datetime.now(tz=UTC)
+ try:
+ result = fetcher(actor=actor, cursor_state=state.cursor_json)
+ inserted = self._insert_events(db, result.events)
+ total_inserted += inserted
+ state.cursor_json = result.cursor_state
+ state.last_error = None
+ state.updated_at = datetime.now(tz=UTC)
+ if result.complete:
+ state.status = "completed"
+ state.completed_at = datetime.now(tz=UTC)
+ logger.info("Backfill complete for service=%s actor=%s inserted=%s", service_name, actor, inserted)
+ else:
+ logger.info("Backfill batch complete for service=%s actor=%s inserted=%s", service_name, actor, inserted)
+ except Exception as exc:
+ state.status = "error"
+ state.last_error = str(exc)
+ state.updated_at = datetime.now(tz=UTC)
+ tracked_actor.backfill_status = "error"
+ tracked_actor.last_backfill_error = str(exc)
+ db.add(state)
+ db.add(tracked_actor)
+ db.flush()
+ raise
+
+ db.add(state)
+ db.flush()
+
+ completed = db.scalars(
+ select(ServiceBackfillState.status).where(ServiceBackfillState.actor == actor)
+ ).all()
+ if completed and all(status == "completed" for status in completed):
+ tracked_actor.backfill_status = "completed"
+ tracked_actor.backfill_completed_at = datetime.now(tz=UTC)
+ tracked_actor.last_backfill_error = None
+ elif tracked_actor.backfill_status != "error":
+ tracked_actor.backfill_status = "in_progress"
+ db.add(tracked_actor)
+ db.flush()
+ return total_inserted
diff --git a/src/srht_contrib/models.py b/src/srht_contrib/models.py
index 2f33482..a99a306 100644
--- a/src/srht_contrib/models.py
+++ b/src/srht_contrib/models.py
@@ -71,3 +71,22 @@ class TrackedActor(Base):
last_polled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
last_poll_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
last_poll_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+ backfill_status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending")
+ backfill_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+ backfill_completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+ last_backfill_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+
+class ServiceBackfillState(Base):
+ __tablename__ = "service_backfill_states"
+ __table_args__ = (UniqueConstraint("actor", "service", name="uq_service_backfill_state_actor_service"),)
+
+ id: Mapped[int] = mapped_column(Integer, primary_key=True)
+ actor: Mapped[str] = mapped_column(String(255), nullable=False)
+ service: Mapped[str] = mapped_column(String(32), nullable=False)
+ cursor_json: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+ status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending")
+ started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+ completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+ last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+ updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
diff --git a/src/srht_contrib/schemas.py b/src/srht_contrib/schemas.py
index bebf93d..6c77f6b 100644
--- a/src/srht_contrib/schemas.py
+++ b/src/srht_contrib/schemas.py
@@ -28,6 +28,9 @@ class ContributionIndexMetadata(BaseModel):
is_indexed: bool
last_polled_at: datetime | None = None
indexing_state: Literal["pending", "indexed", "error"]
+ is_backfilled: bool = False
+ backfill_state: Literal["pending", "in_progress", "completed", "error"] = "pending"
+ backfill_completed_at: datetime | None = None
class ContributionCalendarResponse(ContributionIndexMetadata):
diff --git a/src/srht_contrib/services/aggregator.py b/src/srht_contrib/services/aggregator.py
index 90ef93b..1c8af86 100644
--- a/src/srht_contrib/services/aggregator.py
+++ b/src/srht_contrib/services/aggregator.py
@@ -81,6 +81,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_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,
)
def _query_daily_aggregates(self, db: Session, actor: str, start: date, end: date) -> list[DailyAggregate]:
diff --git a/src/srht_contrib/services/git.py b/src/srht_contrib/services/git.py
index 9531709..69a6c1c 100644
--- a/src/srht_contrib/services/git.py
+++ b/src/srht_contrib/services/git.py
@@ -8,6 +8,7 @@ from typing import Any
from srht_contrib.config import Settings
from srht_contrib.schemas import NormalizedEvent
from srht_contrib.services.srht_client import SourceHutGraphQLClient
+from srht_contrib.services.types import BackfillBatchResult
from srht_contrib.utils.dates import ensure_utc, parse_datetime
from srht_contrib.utils.identity import ActorIdentityResolver
@@ -117,6 +118,95 @@ class GitIngestionService:
)
return repositories
+ def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+ state = {
+ "discovery_cursor": None,
+ "discovery_complete": False,
+ "repository_queue": sorted(
+ {
+ self._canonical_repository_name(actor, repository)
+ for repository in self.settings.git_tracked_repositories
+ }
+ ),
+ "current_repository": None,
+ }
+ if cursor_state:
+ state.update(cursor_state)
+
+ if not state["discovery_complete"]:
+ data = self.client.execute(
+ USER_REPOSITORIES_QUERY,
+ {"username": actor.lstrip("~"), "cursor": state["discovery_cursor"]},
+ )
+ user = data.get("user") or {}
+ repositories_page = user.get("repositories") or {}
+ results = repositories_page.get("results") or []
+ state["discovery_cursor"] = repositories_page.get("cursor")
+ state["discovery_complete"] = not bool(state["discovery_cursor"])
+ known = set(state["repository_queue"])
+ current_repository = state.get("current_repository")
+ if current_repository:
+ known.add(current_repository["name"])
+ for repository in results:
+ if not isinstance(repository, dict):
+ continue
+ name = repository.get("name")
+ repository_owner = ((repository.get("owner") or {}).get("canonicalName") or actor).strip()
+ if not name or not repository_owner:
+ continue
+ canonical_name = f"{repository_owner}/{name}"
+ if canonical_name not in known:
+ state["repository_queue"].append(canonical_name)
+ known.add(canonical_name)
+ state["repository_queue"] = sorted(state["repository_queue"])
+ logger.info(
+ "git backfill discovery actor=%s page_count=%s queue=%s next_cursor=%s",
+ actor,
+ len(results),
+ len(state["repository_queue"]),
+ bool(state["discovery_cursor"]),
+ )
+ complete = state["discovery_complete"] and not state["repository_queue"] and not state["current_repository"]
+ return BackfillBatchResult(events=[], cursor_state=state, complete=complete)
+
+ if state["current_repository"] is None:
+ if not state["repository_queue"]:
+ return BackfillBatchResult(events=[], cursor_state=None, complete=True)
+ state["current_repository"] = {"name": state["repository_queue"].pop(0), "cursor": None}
+
+ repository_name = state["current_repository"]["name"]
+ owner, repo_name = self._split_repository(actor, repository_name)
+ data = self.client.execute(
+ REPOSITORY_LOG_QUERY,
+ {"username": owner, "repoName": repo_name, "cursor": state["current_repository"]["cursor"]},
+ )
+ user = data.get("user") or {}
+ repository = user.get("repository") or {}
+ 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
+ ]
+ logger.info(
+ "git backfill actor=%s repository=%s commits=%s next_cursor=%s",
+ actor,
+ repository_name,
+ len(commits),
+ bool(next_cursor),
+ )
+ if next_cursor:
+ state["current_repository"]["cursor"] = next_cursor
+ else:
+ state["current_repository"] = None
+
+ complete = state["discovery_complete"] and not state["repository_queue"] and not state["current_repository"]
+ return BackfillBatchResult(events=events, cursor_state=None if complete else state, complete=complete)
+
def _discover_owned_repositories(self, actor: str) -> list[str]:
owner = actor.lstrip("~")
repositories: list[str] = []
diff --git a/src/srht_contrib/services/todo.py b/src/srht_contrib/services/todo.py
index 2d25acf..0142005 100644
--- a/src/srht_contrib/services/todo.py
+++ b/src/srht_contrib/services/todo.py
@@ -8,6 +8,7 @@ from typing import Any
from srht_contrib.config import Settings
from srht_contrib.schemas import NormalizedEvent
from srht_contrib.services.srht_client import SourceHutGraphQLClient
+from srht_contrib.services.types import BackfillBatchResult
from srht_contrib.utils.dates import ensure_utc, parse_datetime
@@ -293,6 +294,141 @@ class TodoIngestionService:
tracker_events = self._fetch_from_trackers(actor=actor, since=since_dt)
return TodoPollResult(events=tracker_events, cursor=cursor_time)
+ def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+ state = {
+ "trackers_cursor": None,
+ "tracker_queue": [],
+ "current_tracker": None,
+ "current_ticket": None,
+ "trackers_loaded": False,
+ }
+ if cursor_state:
+ state.update(cursor_state)
+
+ if not state["trackers_loaded"]:
+ data = self.client.execute(TODO_TRACKERS_QUERY, {"cursor": state["trackers_cursor"]})
+ me = data.get("me") or {}
+ trackers_page = me.get("trackers") or {}
+ results = trackers_page.get("results") or []
+ state["trackers_cursor"] = trackers_page.get("cursor")
+ for tracker in results:
+ if not isinstance(tracker, dict):
+ continue
+ state["tracker_queue"].append(
+ {
+ "id": str(tracker.get("id")),
+ "rid": str(tracker.get("rid")),
+ "name": tracker.get("name"),
+ "tickets_cursor": None,
+ "pending_tickets": [],
+ }
+ )
+ state["trackers_loaded"] = not bool(state["trackers_cursor"])
+ logger.info(
+ "todo backfill tracker discovery actor=%s trackers=%s next_cursor=%s queue=%s",
+ actor,
+ len(results),
+ bool(state["trackers_cursor"]),
+ len(state["tracker_queue"]),
+ )
+ complete = state["trackers_loaded"] and not state["tracker_queue"]
+ return BackfillBatchResult(events=[], cursor_state=None if complete else state, complete=complete)
+
+ if state["current_tracker"] is None:
+ if not state["tracker_queue"]:
+ return BackfillBatchResult(events=[], cursor_state=None, complete=True)
+ state["current_tracker"] = state["tracker_queue"].pop(0)
+
+ current_tracker = state["current_tracker"]
+ if state["current_ticket"] is None and not current_tracker["pending_tickets"]:
+ data = self.client.execute(
+ TODO_TRACKER_TICKETS_QUERY,
+ {"trackerRid": current_tracker["rid"], "cursor": current_tracker["tickets_cursor"]},
+ )
+ 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)]
+ )
+ 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"),
+ len(tickets),
+ bool(current_tracker["tickets_cursor"]),
+ len(current_tracker["pending_tickets"]),
+ )
+ if not current_tracker["pending_tickets"] and not current_tracker["tickets_cursor"]:
+ state["current_tracker"] = None
+ return BackfillBatchResult(events=[], cursor_state=state, complete=False)
+
+ if state["current_ticket"] is None:
+ if current_tracker["pending_tickets"]:
+ next_ticket = current_tracker["pending_tickets"].pop(0)
+ state["current_ticket"] = {**next_ticket, "cursor": None}
+ else:
+ state["current_tracker"] = None
+ return BackfillBatchResult(events=[], cursor_state=state, complete=False)
+
+ current_ticket = state["current_ticket"]
+ data = self.client.execute(
+ TODO_TICKET_EVENTS_QUERY,
+ {"trackerRid": current_tracker["rid"], "ticketId": current_ticket["id"], "cursor": current_ticket["cursor"]},
+ )
+ tracker = data.get("tracker") or {}
+ ticket_payload = (tracker.get("ticket") or {}) if isinstance(tracker, dict) else {}
+ event_page = ticket_payload.get("events") or {}
+ page_events = event_page.get("results") or []
+ next_cursor = event_page.get("cursor")
+ logger.info(
+ "todo backfill ticket events ref=%s tracker=%s count=%s next_cursor=%s",
+ current_ticket["ref"],
+ current_tracker.get("name") or current_tracker.get("id"),
+ len(page_events),
+ bool(next_cursor),
+ )
+
+ events: list[NormalizedEvent] = []
+ for event in page_events:
+ if not isinstance(event, dict):
+ continue
+ event["ticket"] = {
+ "id": ticket_payload.get("id", current_ticket["id"]),
+ "ref": ticket_payload.get("ref", current_ticket["ref"]),
+ "status": ticket_payload.get("status"),
+ "resolution": ticket_payload.get("resolution"),
+ "tracker": {"name": current_tracker.get("name")},
+ }
+ occurred_at = parse_datetime(event["created"])
+ for change in event.get("changes") or []:
+ if not isinstance(change, dict):
+ continue
+ normalized = _normalize_event_change(
+ settings=self.settings,
+ actor=actor,
+ event=event,
+ change=change,
+ occurred_at=occurred_at,
+ )
+ if normalized is not None:
+ events.append(normalized)
+
+ if next_cursor:
+ state["current_ticket"]["cursor"] = next_cursor
+ else:
+ state["current_ticket"] = None
+ if not current_tracker["pending_tickets"] and not current_tracker["tickets_cursor"]:
+ state["current_tracker"] = None
+
+ complete = (
+ state["trackers_loaded"]
+ and not state["tracker_queue"]
+ and state["current_tracker"] is None
+ and state["current_ticket"] is None
+ )
+ return BackfillBatchResult(events=events, cursor_state=None if complete else state, complete=complete)
+
def _fetch_from_activity_feed(self, actor: str, since: datetime) -> TodoPollResult:
events: list[NormalizedEvent] = []
cursor: str | None = None
diff --git a/src/srht_contrib/services/types.py b/src/srht_contrib/services/types.py
new file mode 100644
index 0000000..69e64af
--- /dev/null
+++ b/src/srht_contrib/services/types.py
@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from srht_contrib.schemas import NormalizedEvent
+
+
+@dataclass(slots=True)
+class BackfillBatchResult:
+ events: list[NormalizedEvent]
+ cursor_state: dict | None
+ complete: bool
diff --git a/tests/test_contributions_api.py b/tests/test_contributions_api.py
index 22d0af3..f2b8564 100644
--- a/tests/test_contributions_api.py
+++ b/tests/test_contributions_api.py
@@ -63,6 +63,9 @@ def test_public_read_registers_actor_for_lazy_indexing(client: TestClient, db_se
assert response.status_code == 200
assert response.json()["is_indexed"] is False
assert response.json()["indexing_state"] == "pending"
+ assert response.json()["is_backfilled"] is False
+ assert response.json()["backfill_state"] == "pending"
+ assert response.json()["backfill_completed_at"] is None
assert response.json()["last_polled_at"] is None
assert tracked_actor is not None
assert tracked_actor.is_active is True
@@ -107,6 +110,7 @@ def test_contribution_stats_api(client: TestClient, db_session) -> None:
assert response.json()["current_streak"] == 2
assert response.json()["is_indexed"] is True
assert response.json()["indexing_state"] == "indexed"
+ assert response.json()["is_backfilled"] is False
def test_invalid_date_input_returns_400(client: TestClient) -> None:
diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py
index 3b7db16..857f1aa 100644
--- a/tests/test_ingestion.py
+++ b/tests/test_ingestion.py
@@ -4,10 +4,11 @@ from sqlalchemy import select
from srht_contrib.config import Settings
from srht_contrib.jobs.poller import PollerService
-from srht_contrib.models import SyncState, TrackedActor, TrackedRepository
+from srht_contrib.models import ServiceBackfillState, SyncState, TrackedActor, TrackedRepository
from srht_contrib.schemas import NormalizedEvent
from srht_contrib.services.git import GitIngestionService, GitPollResult
from srht_contrib.services.todo import TodoIngestionService, TodoPollResult
+from srht_contrib.services.types import BackfillBatchResult
class StubClient:
@@ -37,6 +38,9 @@ class RecordingTodoService:
events = self.events_by_call.pop(0)
return TodoPollResult(events=events, cursor="2026-03-31T00:00:00+00:00")
+ def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+ return BackfillBatchResult(events=[], cursor_state=None, complete=True)
+
class EmptyGitService:
service_name = "git"
@@ -54,6 +58,30 @@ class EmptyGitService:
def fetch_recent_events(self, actor: str, since: datetime | None = None, repositories=None) -> GitPollResult:
return GitPollResult(events=[], cursor="2026-03-31T00:00:00+00:00")
+ def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+ return BackfillBatchResult(events=[], cursor_state=None, complete=True)
+
+
+class BackfillingTodoService:
+ service_name = "todo"
+
+ def fetch_recent_events(self, actor: str, since: datetime | None = None) -> TodoPollResult:
+ return TodoPollResult(events=[], cursor="2026-03-31T00:00:00+00:00")
+
+ def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+ event = NormalizedEvent(
+ service="todo",
+ event_type="ticket_created",
+ actor=actor,
+ repo_name="todo",
+ resource_id="backfill-ticket",
+ external_uid=f"todo:backfill:{actor}",
+ occurred_at=datetime(2024, 1, 1, 12, 0, tzinfo=UTC),
+ weight=1.0,
+ raw_payload_json=None,
+ )
+ return BackfillBatchResult(events=[event], cursor_state=None, complete=True)
+
def make_settings(**overrides) -> Settings:
values = {
@@ -424,3 +452,21 @@ def test_scheduled_poll_polls_known_actors_and_seeds_default_actor(db_session) -
assert [actor.actor for actor in tracked_actors] == ["~default", "~known"]
assert all(actor.last_poll_status == "indexed" for actor in tracked_actors)
assert all(actor.last_polled_at is not None for actor in tracked_actors)
+
+
+def test_poll_marks_backfill_complete_and_persists_service_state(db_session) -> None:
+ poller = PollerService(todo_service=BackfillingTodoService(), git_service=EmptyGitService())
+
+ inserted = poller.poll_all(db_session, "~ccleberg")
+
+ tracked_actor = db_session.scalar(select(TrackedActor).where(TrackedActor.actor == "~ccleberg"))
+ service_states = db_session.scalars(
+ select(ServiceBackfillState).where(ServiceBackfillState.actor == "~ccleberg").order_by(ServiceBackfillState.service)
+ ).all()
+
+ assert inserted == 1
+ assert tracked_actor is not None
+ assert tracked_actor.backfill_status == "completed"
+ assert tracked_actor.backfill_completed_at is not None
+ assert [state.service for state in service_states] == ["git", "todo"]
+ assert all(state.status == "completed" for state in service_states)
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 505ddb3..6173693 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -21,6 +21,7 @@ def test_alembic_upgrade_creates_schema(tmp_path) -> None:
"actor_aliases",
"alembic_version",
"contribution_events",
+ "service_backfill_states",
"sync_states",
"tracked_actors",
"tracked_repositories",
@@ -117,6 +118,7 @@ def test_alembic_upgrade_adopts_legacy_schema(tmp_path) -> None:
assert "uq_tracked_repository_service_actor_name" in unique_constraints
assert actor == Settings().default_actor
assert "tracked_actors" in inspector.get_table_names()
+ assert "service_backfill_states" in inspector.get_table_names()
def test_alembic_prefers_database_url_from_environment(tmp_path, monkeypatch) -> None:
@@ -131,3 +133,4 @@ def test_alembic_prefers_database_url_from_environment(tmp_path, monkeypatch) ->
inspector = inspect(create_engine(database_url))
assert "actor_aliases" in inspector.get_table_names()
assert "tracked_actors" in inspector.get_table_names()
+ assert "service_backfill_states" in inspector.get_table_names()
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-04 13:54:18 +0000