krz/hutch-stats
Server-side utility for calculating contributions for sourcehut users.
clone: git clone https://gitbay.org/krz/hutch-stats.git
e1a3233ec2204055faa8dadb18cb2efa12cf0e17
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-04-11T22:39:21Z
API.md | 25 ++++- README.md | 16 ++- alembic/versions/20260411_0003_backfill_state.py | 55 +++++++++ src/srht_contrib/jobs/poller.py | 91 ++++++++++++++- src/srht_contrib/models.py | 19 ++++ src/srht_contrib/schemas.py | 3 + src/srht_contrib/services/aggregator.py | 3 + src/srht_contrib/services/git.py | 90 +++++++++++++++ src/srht_contrib/services/todo.py | 136 +++++++++++++++++++++++ src/srht_contrib/services/types.py | 12 ++ tests/test_contributions_api.py | 4 + tests/test_ingestion.py | 48 +++++++- tests/test_migrations.py | 3 + 13 files changed, 498 insertions(+), 7 deletions(-) @@ -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: @@ -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 new file mode 100644 @@ -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") @@ -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 @@ -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) @@ -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): @@ -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]: @@ -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] = [] @@ -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 new file mode 100644 @@ -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 @@ -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: @@ -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) @@ -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()