krz/hutch-stats
Server-side utility for calculating contributions for sourcehut users.
clone: git clone https://gitbay.org/krz/hutch-stats.git
acbff854f2da96bddcaede1385e7fefeba0fb34b
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-04-10T00:45:45Z
.env.example | 14 + .gitignore | 31 ++ README.md | 325 +++++++++++++ alembic.ini | 37 ++ alembic/env.py | 58 +++ alembic/script.py.mako | 18 + alembic/versions/20260409_0001_initial_schema.py | 148 ++++++ pyproject.toml | 35 ++ src/srht_contrib/__init__.py | 5 + src/srht_contrib/api/__init__.py | 1 + src/srht_contrib/api/dependencies.py | 45 ++ src/srht_contrib/api/routes_contributions.py | 78 ++++ src/srht_contrib/api/routes_health.py | 12 + src/srht_contrib/api/routes_repositories.py | 117 +++++ src/srht_contrib/config.py | 59 +++ src/srht_contrib/db.py | 50 ++ src/srht_contrib/jobs/__init__.py | 1 + src/srht_contrib/jobs/poller.py | 119 +++++ src/srht_contrib/logging.py | 10 + src/srht_contrib/main.py | 99 ++++ src/srht_contrib/models.py | 60 +++ src/srht_contrib/schemas.py | 78 ++++ src/srht_contrib/services/__init__.py | 1 + src/srht_contrib/services/aggregator.py | 100 ++++ src/srht_contrib/services/git.py | 202 ++++++++ src/srht_contrib/services/srht_client.py | 77 ++++ src/srht_contrib/services/todo.py | 559 +++++++++++++++++++++++ src/srht_contrib/utils/__init__.py | 1 + src/srht_contrib/utils/dates.py | 37 ++ src/srht_contrib/utils/identity.py | 27 ++ src/srht_contrib/utils/repositories.py | 31 ++ tests/conftest.py | 58 +++ tests/test_aggregator.py | 79 ++++ tests/test_contributions_api.py | 120 +++++ tests/test_health.py | 15 + tests/test_ingestion.py | 337 ++++++++++++++ tests/test_migrations.py | 117 +++++ tests/test_polling_api.py | 71 +++ tests/test_repositories_api.py | 77 ++++ tests/test_srht_client.py | 56 +++ 40 files changed, 3365 insertions(+) new file mode 100644 @@ -0,0 +1,14 @@ +API_KEY=replace-me +ENABLE_SCHEDULER=false +SRHT_TOKEN=replace-me +TODO_SRHT_ENDPOINT=https://todo.sr.ht/query +GIT_SRHT_ENDPOINT=https://git.sr.ht/query +DATABASE_URL=sqlite:///./srht_contrib.db +DEFAULT_ACTOR=~ccleberg +POLL_INTERVAL_SECONDS=900 +# Optional JSON object. Example: +# {"~ccleberg":["cmc@example.com","Chris Cleberg"]} +ACTOR_ALIASES_JSON={} +# Optional JSON array. Example: +# ["Hutch","~ccleberg/cleberg.net"] +GIT_TRACKED_REPOSITORIES=[] new file mode 100644 @@ -0,0 +1,31 @@ +.env +.env.* +!.env.example + +.venv/ +venv/ + +__pycache__/ +*.py[cod] +*.so + +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.hypothesis/ +.coverage +.coverage.* +htmlcov/ + +build/ +dist/ +*.egg-info/ +.eggs/ + +*.db +*.sqlite +*.sqlite3 + +.DS_Store +.idea/ +.vscode/ new file mode 100644 @@ -0,0 +1,325 @@ +# srht-contrib + +`srht-contrib` is a small Python service that polls SourceHut activity, normalizes it into one internal event model, stores it in SQLite, and exposes a contribution-calendar JSON API that an iOS app can render directly. + +The current V1 is intentionally narrow and production-oriented: + +- FastAPI JSON API only +- SQLite-backed persistence +- polling-based ingestion +- complete `todo.sr.ht` ingestion path +- practical `git.sr.ht` commit ingestion for tracked repositories +- API-key protection for `/api/*` +- Alembic-managed schema migrations + +## 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. + +Example use cases: + +- render a GitHub-style contribution grid in an iOS app +- show total score and streak stats for a SourceHut user +- poll recent activity on a schedule or trigger polling manually + +## Architecture Overview + +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/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 +- `src/srht_contrib/services/aggregator.py`: per-day aggregation and streak/stat calculations +- `src/srht_contrib/jobs/poller.py`: repeated-safe polling and idempotent persistence +- `src/srht_contrib/api/`: FastAPI routes, auth dependencies, and repository management +- `alembic/`: schema migration environment and versioned migrations + +## Supported sr.ht Services + +### Implemented + +- `todo.sr.ht` +- `git.sr.ht` + +Current normalized event types: + +- `ticket_created` +- `ticket_comment` +- `ticket_closed` +- `commit` + +`todo.sr.ht` uses a feed-first strategy and falls back to crawling the authenticated user’s trackers, tickets, and ticket events when the top-level activity feed is empty. `git.sr.ht` polls tracked repositories for recent commits on the default branch. + +## Canonical Event Model + +All ingestion services normalize external activity into this shape: + +- `service` +- `event_type` +- `actor` +- `repo_name` +- `resource_id` +- `external_uid` +- `occurred_at` +- `weight` +- `raw_payload_json` + +The database enforces uniqueness on `(service, external_uid)` so polling is safe to repeat. + +Tracked git repositories are persisted in the `tracked_repositories` table and stored in canonical `~owner/repo` form. The poller seeds that table from `GIT_TRACKED_REPOSITORIES`, and repositories can also be created, updated, and deleted through the API. + +## Configuration + +Environment variables: + +- `API_KEY`: required header token for all `/api/*` routes via `X-API-Key` +- `ENABLE_SCHEDULER`: defaults to `false`; enables in-process polling when set to `true` +- `SRHT_TOKEN`: bearer token for SourceHut GraphQL +- `TODO_SRHT_ENDPOINT`: defaults to `https://todo.sr.ht/query` +- `GIT_SRHT_ENDPOINT`: defaults to `https://git.sr.ht/query` +- `DATABASE_URL`: defaults to `sqlite:///./srht_contrib.db` +- `DEFAULT_ACTOR`: actor used by the scheduled poll job +- `POLL_INTERVAL_SECONDS`: scheduler interval in seconds +- `ACTOR_ALIASES_JSON`: optional JSON object for actor/email/display-name alias mapping +- `GIT_TRACKED_REPOSITORIES`: optional JSON array of repository names or `owner/repo` strings for git polling + +Example `.env`: + +```env +API_KEY=replace-me +ENABLE_SCHEDULER=false +SRHT_TOKEN=replace-me +TODO_SRHT_ENDPOINT=https://todo.sr.ht/query +GIT_SRHT_ENDPOINT=https://git.sr.ht/query +DATABASE_URL=sqlite:///./srht_contrib.db +DEFAULT_ACTOR=~ccleberg +POLL_INTERVAL_SECONDS=900 +ACTOR_ALIASES_JSON={"~ccleberg":["cmc@example.com","Chris Cleberg"]} +GIT_TRACKED_REPOSITORIES=["Hutch","~ccleberg/cleberg.net"] +``` + +## Local Run Instructions + +### 1. Create a virtual environment and install dependencies + +Using `uv`: + +```bash +uv venv +source .venv/bin/activate +uv pip install -e ".[dev]" +``` + +Using `pip`: + +```bash +python3.12 -m venv .venv +source .venv/bin/activate +pip install -e ".[dev]" +``` + +### 2. Configure environment + +```bash +cp .env.example .env +``` + +Set at least: + +- `API_KEY` +- `SRHT_TOKEN` +- `DEFAULT_ACTOR` +- `GIT_TRACKED_REPOSITORIES` if you want git commit ingestion + +### 3. Run database migrations + +```bash +alembic upgrade head +``` + +### 4. Run the API + +```bash +uvicorn srht_contrib.main:app --reload +``` + +## Manual Polling + +Manual polling is exposed as an API endpoint: + +```bash +curl -X POST "http://127.0.0.1:8000/api/contributions/poll?actor=~ccleberg" \ + -H "X-API-Key: replace-me" +``` + +Example response: + +```json +{ + "actor": "~ccleberg", + "inserted_events": 3, + "services": ["todo", "git"] +} +``` + +Scheduled polling only runs when `ENABLE_SCHEDULER=true` and uses `DEFAULT_ACTOR`. + +For `git.sr.ht`, tracked repositories are configured via `GIT_TRACKED_REPOSITORIES`. Entries may be either: + +- `"Hutch"` for a repository owned by `DEFAULT_ACTOR` +- `"~ccleberg/cleberg.net"` for an explicit owner/repository pair + +## API Endpoints + +### Health + +```bash +curl "http://127.0.0.1:8000/health" +``` + +Response: + +```json +{"status":"ok"} +``` + +### Contribution Calendar by Year + +```bash +curl "http://127.0.0.1:8000/api/contributions/~ccleberg?year=2026" \ + -H "X-API-Key: replace-me" +``` + +### Contribution Calendar by Date Range + +```bash +curl "http://127.0.0.1:8000/api/contributions/~ccleberg?from=2026-01-01&to=2026-03-30" \ + -H "X-API-Key: replace-me" +``` + +Example response: + +```json +{ + "actor": "~ccleberg", + "from": "2026-01-01", + "to": "2026-03-30", + "days": [ + {"date": "2026-03-28", "count": 3, "score": 3.5}, + {"date": "2026-03-29", "count": 0, "score": 0.0}, + {"date": "2026-03-30", "count": 7, "score": 8.25} + ] +} +``` + +### Contribution Stats + +```bash +curl "http://127.0.0.1:8000/api/contributions/~ccleberg/stats?year=2026" \ + -H "X-API-Key: replace-me" +``` + +Example response: + +```json +{ + "actor": "~ccleberg", + "from": "2026-01-01", + "to": "2026-12-31", + "total_events": 42, + "total_score": 37.5, + "active_days": 18, + "longest_streak": 5, + "current_streak": 2 +} +``` + +### Tracked Repositories + +List tracked repositories: + +```bash +curl "http://127.0.0.1:8000/api/repositories?actor=~ccleberg" \ + -H "X-API-Key: replace-me" +``` + +Create a tracked repository: + +```bash +curl -X POST "http://127.0.0.1:8000/api/repositories" \ + -H "X-API-Key: replace-me" \ + -H "Content-Type: application/json" \ + -d '{"actor":"~ccleberg","repo_name":"Hutch"}' +``` + +Get, update, and delete a tracked repository: + +```bash +curl "http://127.0.0.1:8000/api/repositories/1" \ + -H "X-API-Key: replace-me" + +curl -X PATCH "http://127.0.0.1:8000/api/repositories/1" \ + -H "X-API-Key: replace-me" \ + -H "Content-Type: application/json" \ + -d '{"repo_name":"~ccleberg/cleberg.net"}' + +curl -X DELETE "http://127.0.0.1:8000/api/repositories/1" \ + -H "X-API-Key: replace-me" +``` + +## Event Weighting + +Weights live in `src/srht_contrib/config.py` so they are easy to tune without touching aggregation code: + +- `commit`: `1.0` +- `ticket_created`: `1.0` +- `ticket_comment`: `0.5` +- `ticket_closed`: `0.75` +- `build_started`: `0.25` +- `build_passed`: `0.25` + +## Testing + +Run the test suite with: + +```bash +pytest +``` + +Covered areas: + +- health endpoint +- API key enforcement +- calendar aggregation +- zero-filled ranges +- stats calculations +- invalid date handling +- idempotent ingestion +- todo feed fallback traversal +- repository CRUD and normalization +- SourceHut error mapping +- git commit alias normalization +- Alembic upgrade path + +## SourceHut Schema Assumptions + +The SourceHut-specific assumptions are isolated to the service modules: + +- `src/srht_contrib/services/todo.py` uses the authenticated `events(cursor)` feed first, then falls back to tracker/ticket event traversal for reliable contribution discovery. +- `src/srht_contrib/services/git.py` uses the documented repository `log(cursor)` query against tracked repositories and attributes commits through the configured alias map. + +## Known Limitations + +- `git.sr.ht` polling is limited to repositories listed in `GIT_TRACKED_REPOSITORIES` +- scheduled polling runs in-process, so it is not a distributed scheduler +- 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 + +## Recommended Next Steps + +1. Add alias-management APIs or seed files for stronger actor identity mapping. +2. Add more SourceHut services such as `builds.sr.ht` and `lists.sr.ht`. +3. Move scheduled polling into an external worker if the deployment grows past a single process. new file mode 100644 @@ -0,0 +1,37 @@ +[alembic] +script_location = alembic +prepend_sys_path = . +path_separator = os +sqlalchemy.url = sqlite:///./srht_contrib.db + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = console +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s new file mode 100644 @@ -0,0 +1,58 @@ +from __future__ import annotations + +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +from srht_contrib.config import Settings +from srht_contrib.db import Base +from srht_contrib.models import ActorAlias, ContributionEvent, SyncState, TrackedRepository # noqa: F401 + +config = context.config + +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def get_database_url() -> str: + configured_url = config.get_main_option("sqlalchemy.url") + if configured_url: + return configured_url + return Settings().database_url + + +def run_migrations_offline() -> None: + context.configure( + url=get_database_url(), + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + section = config.get_section(config.config_ini_section) or {} + section["sqlalchemy.url"] = get_database_url() + connectable = engine_from_config( + section, + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() new file mode 100644 @@ -0,0 +1,18 @@ +"""${message}""" + +revision = ${repr(up_revision)} +down_revision = ${repr(down_revision)} +branch_labels = ${repr(branch_labels)} +depends_on = ${repr(depends_on)} + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} new file mode 100644 @@ -0,0 +1,148 @@ +"""initial schema""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa +from sqlalchemy import inspect + +from srht_contrib.config import Settings + + +revision = "20260409_0001" +down_revision = None +branch_labels = None +depends_on = None + + +def _table_names() -> set[str]: + return set(inspect(op.get_bind()).get_table_names()) + + +def _index_names(table_name: str) -> set[str]: + return {index["name"] for index in inspect(op.get_bind()).get_indexes(table_name)} + + +def _tracked_repositories_needs_upgrade() -> bool: + inspector = inspect(op.get_bind()) + columns = {column["name"]: column for column in inspector.get_columns("tracked_repositories")} + actor_column = columns.get("actor") + if actor_column is None or actor_column.get("nullable", True): + return True + + unique_constraints = {constraint["name"] for constraint in inspector.get_unique_constraints("tracked_repositories")} + return "uq_tracked_repository_service_actor_name" not in unique_constraints + + +def _upgrade_tracked_repositories() -> None: + default_actor = Settings().default_actor + op.execute( + sa.text( + """ + CREATE TABLE tracked_repositories__alembic_new ( + id INTEGER NOT NULL PRIMARY KEY, + service VARCHAR(32) NOT NULL, + repo_name VARCHAR(255) NOT NULL, + actor VARCHAR(255) NOT NULL, + CONSTRAINT uq_tracked_repository_service_actor_name UNIQUE (service, actor, repo_name) + ) + """ + ) + ) + op.execute( + sa.text( + """ + INSERT INTO tracked_repositories__alembic_new (id, service, repo_name, actor) + SELECT id, service, repo_name, COALESCE(actor, :default_actor) + FROM tracked_repositories + """ + ).bindparams(default_actor=default_actor) + ) + op.execute(sa.text("DROP TABLE tracked_repositories")) + op.execute(sa.text("ALTER TABLE tracked_repositories__alembic_new RENAME TO tracked_repositories")) + + +def upgrade() -> None: + table_names = _table_names() + + if "contribution_events" not in table_names: + op.create_table( + "contribution_events", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("service", sa.String(length=32), nullable=False), + sa.Column("event_type", sa.String(length=64), nullable=False), + sa.Column("actor", sa.String(length=255), nullable=False), + sa.Column("repo_name", sa.String(length=255), nullable=True), + sa.Column("resource_id", sa.String(length=255), nullable=False), + sa.Column("external_uid", sa.String(length=255), nullable=False), + sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("weight", sa.Float(), nullable=False), + sa.Column("raw_payload_json", sa.JSON(), nullable=True), + sa.UniqueConstraint("service", "external_uid", name="uq_contribution_event_service_uid"), + ) + + contribution_event_indexes = _index_names("contribution_events") + if "ix_contribution_events_actor_occurred_at" not in contribution_event_indexes: + op.create_index( + "ix_contribution_events_actor_occurred_at", + "contribution_events", + ["actor", "occurred_at"], + unique=False, + ) + if "ix_contribution_events_service_occurred_at" not in contribution_event_indexes: + op.create_index( + "ix_contribution_events_service_occurred_at", + "contribution_events", + ["service", "occurred_at"], + unique=False, + ) + + if "sync_states" not in table_names: + op.create_table( + "sync_states", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("service", sa.String(length=32), nullable=False), + sa.Column("actor", sa.String(length=255), nullable=False), + sa.Column("cursor_value", sa.Text(), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.UniqueConstraint("service", "actor", name="uq_sync_state_service_actor"), + ) + + if "tracked_repositories" not in table_names: + op.create_table( + "tracked_repositories", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("service", sa.String(length=32), nullable=False), + sa.Column("repo_name", sa.String(length=255), nullable=False), + sa.Column("actor", sa.String(length=255), nullable=False), + sa.UniqueConstraint("service", "actor", "repo_name", name="uq_tracked_repository_service_actor_name"), + ) + elif _tracked_repositories_needs_upgrade(): + _upgrade_tracked_repositories() + + if "actor_aliases" not in table_names: + op.create_table( + "actor_aliases", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("canonical_actor", sa.String(length=255), nullable=False), + sa.Column("alias", sa.String(length=255), nullable=False), + sa.UniqueConstraint("alias", name="uq_actor_alias_alias"), + ) + + +def downgrade() -> None: + table_names = _table_names() + + if "actor_aliases" in table_names: + op.drop_table("actor_aliases") + if "tracked_repositories" in table_names: + op.drop_table("tracked_repositories") + if "sync_states" in table_names: + op.drop_table("sync_states") + if "contribution_events" in table_names: + contribution_event_indexes = _index_names("contribution_events") + if "ix_contribution_events_service_occurred_at" in contribution_event_indexes: + op.drop_index("ix_contribution_events_service_occurred_at", table_name="contribution_events") + if "ix_contribution_events_actor_occurred_at" in contribution_event_indexes: + op.drop_index("ix_contribution_events_actor_occurred_at", table_name="contribution_events") + op.drop_table("contribution_events") new file mode 100644 @@ -0,0 +1,35 @@ +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "srht-contrib" +version = "0.1.0" +description = "SourceHut contribution calendar API built with FastAPI" +readme = "README.md" +requires-python = ">=3.12" +dependencies = [ + "fastapi>=0.115,<1.0", + "uvicorn[standard]>=0.30,<1.0", + "httpx>=0.27,<1.0", + "sqlalchemy>=2.0,<3.0", + "alembic>=1.13,<2.0", + "pydantic>=2.8,<3.0", + "pydantic-settings>=2.3,<3.0", + "apscheduler>=3.10,<4.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.2,<9.0", +] + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +pythonpath = ["src"] +testpaths = ["tests"] new file mode 100644 @@ -0,0 +1,5 @@ +"""srht_contrib package.""" + +__all__ = ["__version__"] + +__version__ = "0.1.0" new file mode 100644 @@ -0,0 +1 @@ +"""API routes.""" new file mode 100644 @@ -0,0 +1,45 @@ +from __future__ import annotations + +from fastapi import Depends, Header, HTTPException, Request, status +from sqlalchemy.orm import Session + +from srht_contrib.config import Settings +from srht_contrib.db import get_db_session +from srht_contrib.jobs.poller import PollerService +from srht_contrib.utils.identity import ActorIdentityResolver + + +def get_db(request: Request): + yield from get_db_session(request) + + +def get_settings(request: Request) -> Settings: + settings = getattr(request.app.state, "settings", None) + if settings is None: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Settings not configured.") + return settings + + +def get_actor_identity_resolver(request: Request) -> ActorIdentityResolver: + resolver = getattr(request.app.state, "actor_identity_resolver", None) + if resolver is None: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Actor identity resolver not configured.", + ) + return resolver + + +def get_poller(request: Request) -> PollerService: + poller = getattr(request.app.state, "poller", None) + if poller is None: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Poller not configured.") + return poller + + +def require_api_key( + x_api_key: str | None = Header(default=None, alias="X-API-Key"), + settings: Settings = Depends(get_settings), +) -> None: + if not settings.api_key or x_api_key != settings.api_key: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key.") new file mode 100644 @@ -0,0 +1,78 @@ +from __future__ import annotations + +from datetime import date + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session + +from srht_contrib.api.dependencies import get_actor_identity_resolver, get_db, get_poller, require_api_key +from srht_contrib.jobs.poller import PollerService +from srht_contrib.schemas import ContributionCalendarResponse, ContributionStatsResponse, PollResponse +from srht_contrib.services.aggregator import ContributionAggregator +from srht_contrib.services.srht_client import SourceHutClientError +from srht_contrib.utils.dates import parse_date, year_bounds +from srht_contrib.utils.identity import ActorIdentityResolver + +router = APIRouter(prefix="/api/contributions", tags=["contributions"], dependencies=[Depends(require_api_key)]) + + +def _resolve_range(year: int | None, from_date: str | None, to_date: str | None) -> tuple[date, date]: + if year is not None and (from_date or to_date): + raise HTTPException(status_code=400, detail="Use either `year` or `from`/`to`, not both.") + + try: + if year is not None: + return year_bounds(year) + if from_date and to_date: + start = parse_date(from_date) + end = parse_date(to_date) + if start > end: + raise HTTPException(status_code=400, detail="`from` must be on or before `to`.") + return start, end + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD.") from exc + + raise HTTPException(status_code=400, detail="Provide `year` or both `from` and `to`.") + + +@router.get("/{actor}", response_model=ContributionCalendarResponse) +def get_contributions( + actor: str, + year: int | None = Query(default=None, ge=1970, le=3000), + from_date: str | None = Query(default=None, alias="from"), + to_date: str | None = Query(default=None, alias="to"), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> ContributionCalendarResponse: + start, end = _resolve_range(year, from_date, to_date) + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + return ContributionAggregator().build_calendar(db, canonical_actor, start, end) + + +@router.get("/{actor}/stats", response_model=ContributionStatsResponse) +def get_contribution_stats( + actor: str, + year: int | None = Query(default=None, ge=1970, le=3000), + from_date: str | None = Query(default=None, alias="from"), + to_date: str | None = Query(default=None, alias="to"), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> ContributionStatsResponse: + start, end = _resolve_range(year, from_date, to_date) + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + return ContributionAggregator().build_stats(db, canonical_actor, start, end) + + +@router.post("/poll", response_model=PollResponse) +def trigger_manual_poll( + actor: str, + poller: PollerService = Depends(get_poller), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> PollResponse: + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + try: + inserted = poller.poll_all(db, canonical_actor) + except SourceHutClientError as exc: + raise HTTPException(status_code=502, detail=f"SourceHut polling failed: {exc}") from exc + return PollResponse(actor=canonical_actor, inserted_events=inserted, services=["todo", "git"]) new file mode 100644 @@ -0,0 +1,12 @@ +from __future__ import annotations + +from fastapi import APIRouter + +from srht_contrib.schemas import HealthResponse + +router = APIRouter(tags=["health"]) + + +@router.get("/health", response_model=HealthResponse) +def health() -> HealthResponse: + return HealthResponse(status="ok") new file mode 100644 @@ -0,0 +1,117 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from srht_contrib.api.dependencies import get_actor_identity_resolver, get_db, require_api_key +from srht_contrib.models import TrackedRepository +from srht_contrib.schemas import ( + TrackedRepositoryCreateRequest, + TrackedRepositoryResponse, + TrackedRepositoryUpdateRequest, +) +from srht_contrib.utils.identity import ActorIdentityResolver +from srht_contrib.utils.repositories import canonicalize_repository_name + +router = APIRouter(prefix="/api/repositories", tags=["repositories"], dependencies=[Depends(require_api_key)]) + + +def _to_response(repository: TrackedRepository) -> TrackedRepositoryResponse: + return TrackedRepositoryResponse( + id=repository.id, + service=repository.service, + actor=repository.actor, + repo_name=repository.repo_name, + ) + + +def _get_repository_or_404(db: Session, repository_id: int) -> TrackedRepository: + repository = db.scalar(select(TrackedRepository).where(TrackedRepository.id == repository_id)) + if repository is None: + raise HTTPException(status_code=404, detail="Tracked repository not found.") + return repository + + +@router.get("", response_model=list[TrackedRepositoryResponse]) +def list_tracked_repositories( + actor: str | None = Query(default=None), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> list[TrackedRepositoryResponse]: + stmt = select(TrackedRepository).where(TrackedRepository.service == "git").order_by(TrackedRepository.repo_name) + if actor: + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + stmt = stmt.where(TrackedRepository.actor == canonical_actor) + + repositories = db.scalars(stmt).all() + return [_to_response(repository) for repository in repositories] + + +@router.get("/{repository_id}", response_model=TrackedRepositoryResponse) +def get_tracked_repository( + repository_id: int, + db: Session = Depends(get_db), +) -> TrackedRepositoryResponse: + return _to_response(_get_repository_or_404(db, repository_id)) + + +@router.post("", response_model=TrackedRepositoryResponse, status_code=status.HTTP_201_CREATED) +def create_tracked_repository( + payload: TrackedRepositoryCreateRequest, + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> TrackedRepositoryResponse: + canonical_actor = actor_identity_resolver.canonicalize(payload.actor, db=db) + repository = TrackedRepository( + service="git", + actor=canonical_actor, + repo_name=canonicalize_repository_name(canonical_actor, payload.repo_name), + ) + db.add(repository) + try: + db.commit() + except IntegrityError as exc: + db.rollback() + raise HTTPException(status_code=409, detail="Tracked repository already exists.") from exc + + db.refresh(repository) + return _to_response(repository) + + +@router.patch("/{repository_id}", response_model=TrackedRepositoryResponse) +def update_tracked_repository( + repository_id: int, + payload: TrackedRepositoryUpdateRequest, + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> TrackedRepositoryResponse: + repository = _get_repository_or_404(db, repository_id) + + if payload.actor is not None: + repository.actor = actor_identity_resolver.canonicalize(payload.actor, db=db) + + if payload.repo_name is not None: + repository.repo_name = canonicalize_repository_name(repository.actor, payload.repo_name) + + db.add(repository) + try: + db.commit() + except IntegrityError as exc: + db.rollback() + raise HTTPException(status_code=409, detail="Tracked repository already exists.") from exc + + db.refresh(repository) + return _to_response(repository) + + +@router.delete("/{repository_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_tracked_repository( + repository_id: int, + db: Session = Depends(get_db), +) -> Response: + repository = _get_repository_or_404(db, repository_id) + db.delete(repository) + db.commit() + return Response(status_code=status.HTTP_204_NO_CONTENT) new file mode 100644 @@ -0,0 +1,59 @@ +from __future__ import annotations + +from functools import lru_cache + +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + case_sensitive=True, + ) + + app_name: str = "srht-contrib" + app_env: str = "development" + api_host: str = "127.0.0.1" + api_port: int = 8000 + enable_scheduler: bool = Field(default=False, alias="ENABLE_SCHEDULER") + api_key: str = Field(default="", alias="API_KEY") + + srht_token: str = Field(default="", alias="SRHT_TOKEN") + todo_srht_endpoint: str = Field( + default="https://todo.sr.ht/query", + alias="TODO_SRHT_ENDPOINT", + ) + git_srht_endpoint: str = Field( + default="https://git.sr.ht/query", + alias="GIT_SRHT_ENDPOINT", + ) + database_url: str = Field( + default="sqlite:///./srht_contrib.db", + alias="DATABASE_URL", + ) + default_actor: str = Field(default="~unknown", alias="DEFAULT_ACTOR") + poll_interval_seconds: int = Field(default=900, alias="POLL_INTERVAL_SECONDS") + actor_aliases_json: dict[str, list[str]] = Field( + default_factory=dict, + alias="ACTOR_ALIASES_JSON", + ) + git_tracked_repositories: list[str] = Field( + default_factory=list, + alias="GIT_TRACKED_REPOSITORIES", + ) + + event_weights: dict[str, float] = { + "commit": 1.0, + "ticket_created": 1.0, + "ticket_comment": 0.5, + "ticket_closed": 0.75, + "build_started": 0.25, + "build_passed": 0.25, + } + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + return Settings() new file mode 100644 @@ -0,0 +1,50 @@ +from __future__ import annotations + +from collections.abc import Generator + +from fastapi import HTTPException, Request, status +from sqlalchemy import Engine, create_engine, text +from sqlalchemy.pool import StaticPool +from sqlalchemy.orm import Session, declarative_base, sessionmaker + +from srht_contrib.config import Settings + +Base = declarative_base() + + +def make_engine(settings: Settings) -> Engine: + connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {} + engine_kwargs = {"future": True, "connect_args": connect_args} + if settings.database_url in {"sqlite://", "sqlite:///:memory:"}: + engine_kwargs["poolclass"] = StaticPool + return create_engine(settings.database_url, **engine_kwargs) + + +def make_session_factory(settings: Settings) -> sessionmaker[Session]: + engine = make_engine(settings) + return sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False) + + +def validate_db(bind: Engine) -> None: + with bind.connect() as connection: + connection.execute(text("SELECT 1")) + + +def get_db() -> Generator[Session, None, None]: + raise RuntimeError("Use get_db(request) dependency injection with a Request parameter.") + + +def get_session_factory(request: Request) -> sessionmaker[Session]: + session_factory = getattr(request.app.state, "session_factory", None) + if session_factory is None: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database not configured.") + return session_factory + + +def get_db_session(request: Request) -> Generator[Session, None, None]: + session_factory = get_session_factory(request) + db = session_factory() + try: + yield db + finally: + db.close() new file mode 100644 @@ -0,0 +1 @@ +"""Background jobs.""" new file mode 100644 @@ -0,0 +1,119 @@ +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from srht_contrib.models import ContributionEvent, SyncState, TrackedRepository +from srht_contrib.schemas import NormalizedEvent +from srht_contrib.services.git import GitIngestionService +from srht_contrib.services.todo import TodoIngestionService +from srht_contrib.utils.repositories import canonicalize_repository_name + + +logger = logging.getLogger(__name__) +SYNC_OVERLAP = timedelta(hours=24) + + +class PollerService: + def __init__(self, todo_service: TodoIngestionService, git_service: GitIngestionService) -> None: + self.todo_service = todo_service + self.git_service = git_service + + def poll_all(self, db: Session, actor: str) -> int: + inserted = 0 + inserted += self._poll_service(db, actor, self.todo_service.service_name, self.todo_service.fetch_recent_events) + self._sync_tracked_repositories(db, actor) + git_repositories = self._tracked_repositories_for_actor(db, actor) + inserted += self._poll_service( + db, + actor, + self.git_service.service_name, + lambda actor, since: self.git_service.fetch_recent_events( + actor=actor, + since=since, + repositories=git_repositories, + ), + ) + db.commit() + return inserted + + def _poll_service(self, db: Session, actor: str, service_name: str, fetcher) -> int: + state = db.scalar( + select(SyncState).where(SyncState.service == service_name).where(SyncState.actor == actor) + ) + since = datetime.now(tz=UTC) - timedelta(days=30) + if state and state.cursor_value: + since = datetime.fromisoformat(state.cursor_value.replace("Z", "+00:00")).astimezone(UTC) - SYNC_OVERLAP + logger.info( + "Using sync cursor for %s actor=%s with overlap; since=%s", + service_name, + actor, + since.isoformat(), + ) + + result = fetcher(actor=actor, since=since) + inserted = self._insert_events(db, result.events) + self._upsert_sync_state(db, service_name, actor, result.cursor) + logger.info("Polled %s for %s: inserted=%s", service_name, actor, inserted) + return inserted + + @staticmethod + def _insert_events(db: Session, events: list[NormalizedEvent]) -> int: + inserted = 0 + for event in events: + try: + with db.begin_nested(): + model = ContributionEvent(**event.model_dump()) + db.add(model) + db.flush() + inserted += 1 + except IntegrityError: + logger.info("Skipping duplicate event %s for service %s", event.external_uid, event.service) + return inserted + + @staticmethod + def _upsert_sync_state(db: Session, service: str, actor: str, cursor_value: str) -> None: + state = db.scalar(select(SyncState).where(SyncState.service == service).where(SyncState.actor == actor)) + now = datetime.now(tz=UTC) + if state is None: + db.add(SyncState(service=service, actor=actor, cursor_value=cursor_value, updated_at=now)) + db.flush() + return + + state.cursor_value = cursor_value + state.updated_at = now + db.add(state) + db.flush() + + def _sync_tracked_repositories(self, db: Session, actor: str) -> None: + configured = self.git_service.settings.git_tracked_repositories + for repo_name in configured: + canonical_repo_name = canonicalize_repository_name(actor, repo_name) + existing = db.scalar( + select(TrackedRepository) + .where(TrackedRepository.service == self.git_service.service_name) + .where(TrackedRepository.actor == actor) + .where(TrackedRepository.repo_name == canonical_repo_name) + ) + if existing is None: + db.add( + TrackedRepository( + service=self.git_service.service_name, + repo_name=canonical_repo_name, + actor=actor, + ) + ) + db.flush() + + def _tracked_repositories_for_actor(self, db: Session, actor: str) -> list[str]: + rows = db.scalars( + select(TrackedRepository.repo_name) + .where(TrackedRepository.service == self.git_service.service_name) + .where(TrackedRepository.actor == actor) + .order_by(TrackedRepository.repo_name) + ).all() + return list(rows) new file mode 100644 @@ -0,0 +1,10 @@ +from __future__ import annotations + +import logging + + +def configure_logging() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s [%(name)s] %(message)s", + ) new file mode 100644 @@ -0,0 +1,99 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from apscheduler.schedulers.background import BackgroundScheduler +from fastapi import FastAPI +from sqlalchemy import Engine +from sqlalchemy.orm import Session, sessionmaker + +from srht_contrib.api.routes_contributions import router as contributions_router +from srht_contrib.api.routes_health import router as health_router +from srht_contrib.api.routes_repositories import router as repositories_router +from srht_contrib.config import Settings, get_settings +from srht_contrib.db import make_engine, validate_db +from srht_contrib.jobs.poller import PollerService +from srht_contrib.logging import configure_logging +from srht_contrib.services.git import GitIngestionService +from srht_contrib.services.srht_client import SourceHutGraphQLClient +from srht_contrib.services.todo import TodoIngestionService +from srht_contrib.utils.identity import ActorIdentityResolver + + +def build_poller(settings: Settings) -> PollerService: + todo_client = SourceHutGraphQLClient(settings.todo_srht_endpoint, settings.srht_token) + git_client = SourceHutGraphQLClient(settings.git_srht_endpoint, settings.srht_token) + todo_service = TodoIngestionService(todo_client, settings) + git_service = GitIngestionService(git_client, settings) + return PollerService(todo_service=todo_service, git_service=git_service) + + +def create_app( + settings: Settings | None = None, + *, + engine: Engine | None = None, + session_factory: sessionmaker[Session] | None = None, + poller: PollerService | None = None, +) -> FastAPI: + app_settings = settings or get_settings() + configure_logging() + app_engine = engine or make_engine(app_settings) + app_session_factory = session_factory or sessionmaker( + bind=app_engine, + autoflush=False, + autocommit=False, + expire_on_commit=False, + ) + actor_identity_resolver = ActorIdentityResolver(app_settings.actor_aliases_json) + + @asynccontextmanager + async def lifespan(app: FastAPI) -> AsyncIterator[None]: + validate_db(app_engine) + app_poller = poller or build_poller(app_settings) + app.state.poller = app_poller + app.state.settings = app_settings + app.state.engine = app_engine + app.state.session_factory = app_session_factory + app.state.actor_identity_resolver = actor_identity_resolver + scheduler: BackgroundScheduler | None = None + if app_settings.enable_scheduler: + scheduler = BackgroundScheduler() + scheduler.add_job( + _scheduled_poll, + "interval", + seconds=app_settings.poll_interval_seconds, + args=[app], + id="srht-poller", + replace_existing=True, + ) + scheduler.start() + app.state.scheduler = scheduler + try: + yield + finally: + if scheduler is not None: + scheduler.shutdown(wait=False) + app_poller.todo_service.client.close() + app_poller.git_service.client.close() + + app = FastAPI(title=app_settings.app_name, lifespan=lifespan) + + app.include_router(health_router) + app.include_router(contributions_router) + app.include_router(repositories_router) + return app + + +def _scheduled_poll(app: FastAPI) -> None: + settings: Settings = app.state.settings + poller: PollerService = app.state.poller + session_factory: sessionmaker[Session] = app.state.session_factory + db = session_factory() + try: + poller.poll_all(db, settings.default_actor) + finally: + db.close() + + +app = create_app() new file mode 100644 @@ -0,0 +1,60 @@ +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import JSON, DateTime, Float, Index, Integer, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from srht_contrib.db import Base + + +class ContributionEvent(Base): + __tablename__ = "contribution_events" + __table_args__ = ( + UniqueConstraint("service", "external_uid", name="uq_contribution_event_service_uid"), + Index("ix_contribution_events_actor_occurred_at", "actor", "occurred_at"), + Index("ix_contribution_events_service_occurred_at", "service", "occurred_at"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + service: Mapped[str] = mapped_column(String(32), nullable=False) + event_type: Mapped[str] = mapped_column(String(64), nullable=False) + actor: Mapped[str] = mapped_column(String(255), nullable=False) + repo_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + resource_id: Mapped[str] = mapped_column(String(255), nullable=False) + external_uid: Mapped[str] = mapped_column(String(255), nullable=False) + occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + weight: Mapped[float] = mapped_column(Float, nullable=False) + raw_payload_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) + + +class SyncState(Base): + __tablename__ = "sync_states" + __table_args__ = (UniqueConstraint("service", "actor", name="uq_sync_state_service_actor"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + service: Mapped[str] = mapped_column(String(32), nullable=False) + actor: Mapped[str] = mapped_column(String(255), nullable=False) + cursor_value: Mapped[str | None] = mapped_column(Text, nullable=True) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + +class TrackedRepository(Base): + __tablename__ = "tracked_repositories" + __table_args__ = ( + UniqueConstraint("service", "actor", "repo_name", name="uq_tracked_repository_service_actor_name"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + service: Mapped[str] = mapped_column(String(32), nullable=False) + repo_name: Mapped[str] = mapped_column(String(255), nullable=False) + actor: Mapped[str] = mapped_column(String(255), nullable=False) + + +class ActorAlias(Base): + __tablename__ = "actor_aliases" + __table_args__ = (UniqueConstraint("alias", name="uq_actor_alias_alias"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + canonical_actor: Mapped[str] = mapped_column(String(255), nullable=False) + alias: Mapped[str] = mapped_column(String(255), nullable=False) new file mode 100644 @@ -0,0 +1,78 @@ +from __future__ import annotations + +from datetime import date, datetime + +from pydantic import BaseModel, Field, model_validator + + +class NormalizedEvent(BaseModel): + service: str + event_type: str + actor: str + repo_name: str | None = None + resource_id: str + external_uid: str + occurred_at: datetime + weight: float + raw_payload_json: dict | None = None + + +class ContributionDay(BaseModel): + date: date + count: int + score: float + + +class ContributionCalendarResponse(BaseModel): + actor: str + from_date: date = Field(alias="from") + to_date: date = Field(alias="to") + days: list[ContributionDay] + + model_config = {"populate_by_name": True} + + +class ContributionStatsResponse(BaseModel): + actor: str + from_date: date = Field(alias="from") + to_date: date = Field(alias="to") + total_events: int + total_score: float + active_days: int + longest_streak: int + current_streak: int + + model_config = {"populate_by_name": True} + + +class HealthResponse(BaseModel): + status: str + + +class PollResponse(BaseModel): + actor: str + inserted_events: int + services: list[str] + + +class TrackedRepositoryCreateRequest(BaseModel): + actor: str + repo_name: str + + +class TrackedRepositoryUpdateRequest(BaseModel): + actor: str | None = None + repo_name: str | None = None + + @model_validator(mode="after") + def validate_any_field_present(self) -> "TrackedRepositoryUpdateRequest": + if self.actor is None and self.repo_name is None: + raise ValueError("Provide `actor`, `repo_name`, or both.") + return self + + +class TrackedRepositoryResponse(BaseModel): + id: int + service: str + actor: str + repo_name: str new file mode 100644 @@ -0,0 +1 @@ +"""Service layer.""" new file mode 100644 @@ -0,0 +1,100 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from srht_contrib.models import ContributionEvent +from srht_contrib.schemas import ContributionCalendarResponse, ContributionDay, ContributionStatsResponse +from srht_contrib.utils.dates import date_range, date_to_utc_bounds + + +@dataclass(slots=True) +class DailyAggregate: + date: date + count: int + score: float + + +class ContributionAggregator: + def build_calendar(self, db: Session, actor: str, start: date, end: date) -> ContributionCalendarResponse: + aggregates = self._query_daily_aggregates(db, actor, start, end) + by_day = {row.date: row for row in aggregates} + days = [ + ContributionDay( + date=day, + count=by_day.get(day, DailyAggregate(date=day, count=0, score=0.0)).count, + score=by_day.get(day, DailyAggregate(date=day, count=0, score=0.0)).score, + ) + for day in date_range(start, end) + ] + return ContributionCalendarResponse(actor=actor, from_date=start, to_date=end, days=days) + + def build_stats(self, db: Session, actor: str, start: date, end: date) -> ContributionStatsResponse: + calendar = self.build_calendar(db, actor, start, end) + active_days = [day for day in calendar.days if day.count > 0] + streaks = self._streak_lengths(calendar.days) + current_streak = self._current_streak(calendar.days) + + return ContributionStatsResponse( + actor=actor, + from_date=start, + to_date=end, + total_events=sum(day.count for day in calendar.days), + total_score=round(sum(day.score for day in calendar.days), 2), + active_days=len(active_days), + longest_streak=max(streaks, default=0), + current_streak=current_streak, + ) + + def _query_daily_aggregates(self, db: Session, actor: str, start: date, end: date) -> list[DailyAggregate]: + start_dt, _ = date_to_utc_bounds(start) + _, end_dt = date_to_utc_bounds(end) + + stmt = ( + select( + func.date(ContributionEvent.occurred_at).label("day"), + func.count(ContributionEvent.id).label("count"), + func.coalesce(func.sum(ContributionEvent.weight), 0.0).label("score"), + ) + .where(ContributionEvent.actor == actor) + .where(ContributionEvent.occurred_at >= start_dt) + .where(ContributionEvent.occurred_at <= end_dt) + .group_by(func.date(ContributionEvent.occurred_at)) + .order_by(func.date(ContributionEvent.occurred_at)) + ) + rows = db.execute(stmt).all() + return [ + DailyAggregate( + date=date.fromisoformat(str(row.day)), + count=int(row.count), + score=round(float(row.score), 2), + ) + for row in rows + ] + + @staticmethod + def _streak_lengths(days: list[ContributionDay]) -> list[int]: + streaks: list[int] = [] + current = 0 + for day in days: + if day.count > 0: + current += 1 + elif current > 0: + streaks.append(current) + current = 0 + if current > 0: + streaks.append(current) + return streaks + + @staticmethod + def _current_streak(days: list[ContributionDay]) -> int: + streak = 0 + for day in reversed(days): + if day.count > 0: + streak += 1 + else: + break + return streak new file mode 100644 @@ -0,0 +1,202 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +import logging +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.utils.dates import ensure_utc, parse_datetime +from srht_contrib.utils.identity import ActorIdentityResolver + + +logger = logging.getLogger(__name__) + + +REPOSITORY_LOG_QUERY = """ +query RepositoryLog($username: String!, $repoName: String!, $cursor: Cursor) { + user(username: $username) { + repository(name: $repoName) { + name + owner { + canonicalName + } + log(cursor: $cursor) { + results { + id + shortId + author { + name + email + time + } + committer { + name + email + time + } + message + } + cursor + } + } + } +} +""".strip() + + +@dataclass(slots=True) +class GitPollResult: + events: list[NormalizedEvent] + cursor: str + + +class GitIngestionService: + """Polls tracked git.sr.ht repositories and normalizes commits for one actor.""" + + service_name = "git" + + def __init__(self, client: SourceHutGraphQLClient, settings: Settings) -> None: + self.client = client + self.settings = settings + self.identity_resolver = ActorIdentityResolver(settings.actor_aliases_json) + + def fetch_recent_events( + self, + actor: str, + since: datetime | None = None, + repositories: list[str] | None = None, + ) -> GitPollResult: + since_dt = ensure_utc(since or (datetime.now(tz=UTC) - timedelta(days=30))) + tracked_repositories = repositories or self._tracked_repositories(actor) + if not tracked_repositories: + logger.info("git poll skipped for actor=%s because no tracked repositories are configured", actor) + return GitPollResult(events=[], cursor=datetime.now(tz=UTC).isoformat()) + + events: list[NormalizedEvent] = [] + for repository in tracked_repositories: + owner, repo_name = self._split_repository(actor, repository) + repo_events = self._fetch_repository_commits(actor=actor, owner=owner, repo_name=repo_name, since=since_dt) + events.extend(repo_events) + + logger.info("git poll complete for actor=%s normalized_events=%s", actor, len(events)) + return GitPollResult(events=events, cursor=datetime.now(tz=UTC).isoformat()) + + def _tracked_repositories(self, actor: str) -> list[str]: + repositories = self.settings.git_tracked_repositories + return repositories + + @staticmethod + def _split_repository(default_actor: str, repository: str) -> tuple[str, str]: + if "/" in repository: + owner, repo_name = repository.split("/", 1) + canonical_owner = owner if owner.startswith("~") else f"~{owner}" + return canonical_owner.lstrip("~"), repo_name + return default_actor.lstrip("~"), repository + + def _fetch_repository_commits( + self, + *, + actor: str, + owner: str, + repo_name: str, + since: datetime, + ) -> list[NormalizedEvent]: + events: list[NormalizedEvent] = [] + cursor: str | None = None + + for _ in range(50): + data = self.client.execute( + REPOSITORY_LOG_QUERY, + {"username": owner, "repoName": repo_name, "cursor": cursor}, + ) + user = data.get("user") or {} + repository = user.get("repository") or {} + log_page = repository.get("log") or {} + commits = log_page.get("results") or [] + cursor = log_page.get("cursor") + logger.info( + "git repository=%s/%s commit page count=%s next_cursor=%s", + owner, + repo_name, + len(commits), + bool(cursor), + ) + + stop_paging = False + for commit in commits: + if not isinstance(commit, dict): + continue + commit_time = parse_datetime((commit.get("author") or {}).get("time")) + if commit_time < since: + stop_paging = True + logger.info( + "git commit %s skipped because commit_time=%s is before since=%s", + commit.get("shortId") or commit.get("id"), + commit_time.isoformat(), + since.isoformat(), + ) + continue + + normalized = self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit) + if normalized is not None: + logger.info( + "git commit accepted repo=%s shortId=%s author=%s email=%s", + repo_name, + commit.get("shortId"), + (commit.get("author") or {}).get("name"), + (commit.get("author") or {}).get("email"), + ) + events.append(normalized) + else: + logger.info( + "git commit skipped repo=%s shortId=%s author=%s email=%s", + repo_name, + commit.get("shortId"), + (commit.get("author") or {}).get("name"), + (commit.get("author") or {}).get("email"), + ) + + if stop_paging or not cursor: + break + + return events + + def _normalize_commit( + self, + *, + actor: str, + repo_name: str, + commit: dict[str, Any], + ) -> NormalizedEvent | None: + author = commit.get("author") or {} + candidate_aliases = [ + actor, + author.get("email", ""), + author.get("name", ""), + ] + matched_actor = None + for candidate in candidate_aliases: + canonical = self.identity_resolver.canonicalize(candidate) + if canonical == actor: + matched_actor = canonical + break + + if matched_actor is None: + return None + + commit_id = str(commit["id"]) + commit_time = parse_datetime(author["time"]) + return NormalizedEvent( + service=self.service_name, + event_type="commit", + actor=matched_actor, + repo_name=repo_name, + resource_id=commit_id, + external_uid=f"git:{repo_name}:{commit_id}", + occurred_at=commit_time, + weight=self.settings.event_weights["commit"], + raw_payload_json=commit, + ) new file mode 100644 @@ -0,0 +1,77 @@ +from __future__ import annotations + +import logging +from typing import Any + +import httpx + + +logger = logging.getLogger(__name__) + + +class SourceHutClientError(RuntimeError): + """Raised when a SourceHut GraphQL request fails.""" + + +class SourceHutGraphQLClient: + def __init__( + self, + endpoint: str, + token: str, + *, + timeout: float = 15.0, + max_retries: int = 2, + transport: httpx.BaseTransport | None = None, + ) -> None: + self.endpoint = endpoint + self.timeout = timeout + self.max_retries = max_retries + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + } + self._client = httpx.Client(headers=headers, timeout=timeout, transport=transport) + + def execute(self, query: str, variables: dict[str, Any] | None = None) -> dict[str, Any]: + payload = {"query": query, "variables": variables or {}} + attempts = self.max_retries + 1 + + for attempt in range(1, attempts + 1): + try: + response = self._client.post(self.endpoint, json=payload) + response.raise_for_status() + body = response.json() + except httpx.HTTPStatusError as exc: + response_text = exc.response.text[:500] + logger.warning( + "SourceHut HTTP failure from %s on attempt %s/%s: %s %s", + self.endpoint, + attempt, + attempts, + exc.response.status_code, + response_text, + ) + if exc.response.status_code >= 500 and attempt < attempts: + continue + raise SourceHutClientError( + f"HTTP error from SourceHut: {exc.response.status_code} {response_text}".strip() + ) from exc + except httpx.HTTPError as exc: + logger.warning( + "SourceHut network failure from %s on attempt %s/%s", + self.endpoint, + attempt, + attempts, + ) + if attempt < attempts: + continue + raise SourceHutClientError("Network error while contacting SourceHut") from exc + + if "errors" in body: + raise SourceHutClientError(f"GraphQL errors returned by SourceHut: {body['errors']}") + return body.get("data", {}) + + raise SourceHutClientError("SourceHut request exhausted retries") + + def close(self) -> None: + self._client.close() new file mode 100644 @@ -0,0 +1,559 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +import logging +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.utils.dates import ensure_utc, parse_datetime + + +logger = logging.getLogger(__name__) + + +TODO_ACTIVITY_QUERY = """ +query TodoActivity($cursor: Cursor) { + me { + canonicalName + } + events(cursor: $cursor) { + results { + id + created + ticket { + id + ref + status + resolution + tracker { + name + } + } + changes { + __typename + eventType + ticket { + id + } + ... on Created { + author { + canonicalName + } + } + ... on Comment { + author { + canonicalName + } + } + ... on StatusChange { + editor { + canonicalName + } + oldStatus + newStatus + oldResolution + newResolution + } + } + } + cursor + } +} +""".strip() + +TODO_TRACKERS_QUERY = """ +query TodoTrackers($cursor: Cursor) { + me { + canonicalName + trackers(cursor: $cursor) { + results { + id + rid + name + } + cursor + } + } +} +""".strip() + +TODO_TRACKER_TICKETS_QUERY = """ +query TodoTrackerTickets($trackerRid: ID!, $cursor: Cursor) { + tracker(rid: $trackerRid) { + id + name + tickets(cursor: $cursor) { + results { + id + ref + created + updated + status + resolution + submitter { + canonicalName + } + } + cursor + } + } +} +""".strip() + +TODO_TICKET_EVENTS_QUERY = """ +query TodoTicketEvents($trackerRid: ID!, $ticketId: Int!, $cursor: Cursor) { + tracker(rid: $trackerRid) { + id + name + ticket(id: $ticketId) { + id + ref + status + resolution + events(cursor: $cursor) { + results { + id + created + changes { + __typename + eventType + ticket { + id + } + ... on Created { + author { + canonicalName + } + } + ... on Comment { + author { + canonicalName + } + } + ... on StatusChange { + editor { + canonicalName + } + oldStatus + newStatus + oldResolution + newResolution + } + } + } + cursor + } + } + } +} +""".strip() + + +TICKET_CLOSED_STATUSES = {"RESOLVED"} +TICKET_CLOSED_RESOLUTIONS = { + "CLOSED", + "FIXED", + "IMPLEMENTED", + "WONT_FIX", + "BY_DESIGN", + "INVALID", + "DUPLICATE", + "NOT_OUR_BUG", +} + + +class TodoSchemaError(RuntimeError): + """Raised when SourceHut returns an unexpected todo event shape.""" + + +def _safe_nested_name(entity: dict[str, Any] | None) -> str | None: + if not entity: + return None + return entity.get("canonicalName") or entity.get("name") + + +def _repo_name_from_event(event: dict[str, Any]) -> str | None: + ticket = event.get("ticket") or {} + tracker = ticket.get("tracker") or {} + return tracker.get("name") + + +def _resource_id_from_event(event: dict[str, Any]) -> str: + ticket = event.get("ticket") or {} + return str(ticket.get("ref") or ticket.get("id") or event["id"]) + + +def _change_ticket_id(change: dict[str, Any], event: dict[str, Any]) -> str: + ticket = change.get("ticket") or event.get("ticket") or {} + return str(ticket.get("id") or event["id"]) + + +def _normalize_event_change( + *, + settings: Settings, + actor: str, + event: dict[str, Any], + change: dict[str, Any], + occurred_at: datetime, +) -> NormalizedEvent | None: + change_type = change.get("__typename") + event_id = str(event["id"]) + resource_id = _resource_id_from_event(event) + repo_name = _repo_name_from_event(event) + ticket_id = _change_ticket_id(change, event) + + if change_type == "Created" and _safe_nested_name(change.get("author")) == actor: + return NormalizedEvent( + service="todo", + event_type="ticket_created", + actor=actor, + repo_name=repo_name, + resource_id=resource_id, + external_uid=f"todo:event:{event_id}:created:{ticket_id}", + occurred_at=occurred_at, + weight=settings.event_weights["ticket_created"], + raw_payload_json={"event": event, "change": change}, + ) + + if change_type == "Comment" and _safe_nested_name(change.get("author")) == actor: + return NormalizedEvent( + service="todo", + event_type="ticket_comment", + actor=actor, + repo_name=repo_name, + resource_id=resource_id, + external_uid=f"todo:event:{event_id}:comment:{ticket_id}", + occurred_at=occurred_at, + weight=settings.event_weights["ticket_comment"], + raw_payload_json={"event": event, "change": change}, + ) + + if change_type == "StatusChange" and _safe_nested_name(change.get("editor")) == actor: + new_status = change.get("newStatus") + new_resolution = change.get("newResolution") + if new_status in TICKET_CLOSED_STATUSES or new_resolution in TICKET_CLOSED_RESOLUTIONS: + return NormalizedEvent( + service="todo", + event_type="ticket_closed", + actor=actor, + repo_name=repo_name, + resource_id=resource_id, + external_uid=f"todo:event:{event_id}:closed:{ticket_id}", + occurred_at=occurred_at, + weight=settings.event_weights["ticket_closed"], + raw_payload_json={"event": event, "change": change}, + ) + + return None + + +def _extract_event_cursor_page(data: dict[str, Any]) -> tuple[str | None, list[dict[str, Any]], str]: + me = data.get("me") or {} + canonical_actor = me.get("canonicalName") + if not canonical_actor: + raise TodoSchemaError("todo.sr.ht response did not include me.canonicalName") + + events = data.get("events") or {} + results = events.get("results") or [] + if not isinstance(results, list): + raise TodoSchemaError("todo.sr.ht response did not include events.results") + + return events.get("cursor"), [event for event in results if isinstance(event, dict)], canonical_actor + + +@dataclass(slots=True) +class TodoPollResult: + events: list[NormalizedEvent] + cursor: str + + +class TodoIngestionService: + """Fetches todo.sr.ht activity from the authenticated event feed and normalizes it.""" + + service_name = "todo" + + def __init__(self, client: SourceHutGraphQLClient, settings: Settings) -> None: + self.client = client + self.settings = settings + + def fetch_recent_events(self, actor: str, since: datetime | None = None) -> TodoPollResult: + since_dt = ensure_utc(since or (datetime.now(tz=UTC) - timedelta(days=30))) + cursor_time = datetime.now(tz=UTC).isoformat() + feed_result = self._fetch_from_activity_feed(actor=actor, since=since_dt) + if feed_result.events: + return TodoPollResult(events=feed_result.events, cursor=cursor_time) + + logger.info( + "todo activity feed returned no normalized events for actor=%s; falling back to tracker crawl", + actor, + ) + tracker_events = self._fetch_from_trackers(actor=actor, since=since_dt) + return TodoPollResult(events=tracker_events, cursor=cursor_time) + + def _fetch_from_activity_feed(self, actor: str, since: datetime) -> TodoPollResult: + events: list[NormalizedEvent] = [] + cursor: str | None = None + effective_actor = actor + + for _ in range(10): + data = self.client.execute(TODO_ACTIVITY_QUERY, {"cursor": cursor}) + cursor, page_events, canonical_actor = _extract_event_cursor_page(data) + effective_actor = actor or canonical_actor + logger.info( + "todo page fetched for actor=%s canonical_actor=%s events=%s next_cursor=%s since=%s", + actor, + canonical_actor, + len(page_events), + bool(cursor), + since.isoformat(), + ) + + stop_paging = False + for event in page_events: + occurred_at = parse_datetime(event["created"]) + event_id = str(event.get("id")) + resource_id = _resource_id_from_event(event) + repo_name = _repo_name_from_event(event) + change_list = event.get("changes") or [] + logger.info( + "todo event id=%s resource=%s repo=%s occurred_at=%s changes=%s", + event_id, + resource_id, + repo_name, + occurred_at.isoformat(), + len(change_list) if isinstance(change_list, list) else "unknown", + ) + if occurred_at < since: + stop_paging = True + logger.info( + "todo event id=%s skipped because occurred_at=%s is before since=%s", + event_id, + occurred_at.isoformat(), + since.isoformat(), + ) + continue + + for change in change_list: + if not isinstance(change, dict): + logger.info("todo event id=%s skipped non-dict change payload", event_id) + continue + change_type = change.get("__typename") + author = _safe_nested_name(change.get("author")) + editor = _safe_nested_name(change.get("editor")) + logger.info( + "todo change event_id=%s type=%s eventType=%s author=%s editor=%s newStatus=%s newResolution=%s", + event_id, + change_type, + change.get("eventType"), + author, + editor, + change.get("newStatus"), + change.get("newResolution"), + ) + normalized = _normalize_event_change( + settings=self.settings, + actor=effective_actor, + event=event, + change=change, + occurred_at=occurred_at, + ) + if normalized is not None: + logger.info( + "todo change accepted event_id=%s normalized_type=%s external_uid=%s", + event_id, + normalized.event_type, + normalized.external_uid, + ) + events.append(normalized) + else: + logger.info( + "todo change skipped event_id=%s for actor=%s", + event_id, + effective_actor, + ) + + if stop_paging or not cursor: + break + + logger.info("todo poll complete for actor=%s normalized_events=%s", effective_actor, len(events)) + return TodoPollResult(events=events, cursor=datetime.now(tz=UTC).isoformat()) + + def _fetch_from_trackers(self, actor: str, since: datetime) -> list[NormalizedEvent]: + data = self.client.execute(TODO_TRACKERS_QUERY, {"cursor": None}) + me = data.get("me") or {} + canonical_actor = me.get("canonicalName") or actor + trackers = ((me.get("trackers") or {}).get("results")) or [] + logger.info("todo tracker crawl actor=%s trackers=%s", canonical_actor, len(trackers)) + + events: list[NormalizedEvent] = [] + for tracker in trackers: + if not isinstance(tracker, dict): + continue + tracker_id = tracker.get("id") + tracker_rid = tracker.get("rid") + tracker_name = tracker.get("name") + tracker_events = self._fetch_tracker_tickets( + actor=canonical_actor, + tracker_id=str(tracker_id), + tracker_rid=str(tracker_rid), + tracker_name=tracker_name, + since=since, + ) + events.extend(tracker_events) + + logger.info("todo tracker crawl complete actor=%s normalized_events=%s", canonical_actor, len(events)) + return events + + def _fetch_tracker_tickets( + self, + *, + actor: str, + tracker_id: str, + tracker_rid: str, + tracker_name: str | None, + since: datetime, + ) -> list[NormalizedEvent]: + events: list[NormalizedEvent] = [] + cursor: str | None = None + + for _ in range(50): + data = self.client.execute( + TODO_TRACKER_TICKETS_QUERY, + {"trackerRid": tracker_rid, "cursor": cursor}, + ) + tracker = data.get("tracker") or {} + tickets_page = tracker.get("tickets") or {} + tickets = tickets_page.get("results") or [] + cursor = tickets_page.get("cursor") + logger.info( + "todo tracker=%s ticket page count=%s next_cursor=%s", + tracker_name or tracker_id, + len(tickets), + bool(cursor), + ) + + stop_paging = False + for ticket in tickets: + if not isinstance(ticket, dict): + continue + updated_at = parse_datetime(ticket["updated"]) + if updated_at < since: + stop_paging = True + logger.info( + "todo ticket ref=%s skipped because updated_at=%s is before since=%s", + ticket.get("ref"), + updated_at.isoformat(), + since.isoformat(), + ) + continue + events.extend( + self._fetch_ticket_events( + actor=actor, + tracker_id=tracker_id, + tracker_rid=tracker_rid, + tracker_name=tracker_name, + ticket=ticket, + since=since, + ) + ) + + if stop_paging or not cursor: + break + + return events + + def _fetch_ticket_events( + self, + *, + actor: str, + tracker_id: str, + tracker_rid: str, + tracker_name: str | None, + ticket: dict[str, Any], + since: datetime, + ) -> list[NormalizedEvent]: + events: list[NormalizedEvent] = [] + cursor: str | None = None + ticket_id = int(ticket["id"]) + ticket_ref = str(ticket.get("ref") or ticket_id) + + for _ in range(50): + data = self.client.execute( + TODO_TICKET_EVENTS_QUERY, + {"trackerRid": tracker_rid, "ticketId": ticket_id, "cursor": 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 [] + cursor = event_page.get("cursor") + logger.info( + "todo ticket events ref=%s tracker=%s count=%s next_cursor=%s", + ticket_ref, + tracker_name or tracker_id, + len(page_events), + bool(cursor), + ) + + stop_paging = False + for event in page_events: + if not isinstance(event, dict): + continue + event["ticket"] = { + "id": ticket_payload.get("id", ticket.get("id")), + "ref": ticket_payload.get("ref", ticket_ref), + "status": ticket_payload.get("status", ticket.get("status")), + "resolution": ticket_payload.get("resolution", ticket.get("resolution")), + "tracker": {"name": tracker_name}, + } + occurred_at = parse_datetime(event["created"]) + event_id = str(event.get("id")) + change_list = event.get("changes") or [] + logger.info( + "todo ticket event ref=%s id=%s occurred_at=%s changes=%s", + ticket_ref, + event_id, + occurred_at.isoformat(), + len(change_list) if isinstance(change_list, list) else "unknown", + ) + if occurred_at < since: + stop_paging = True + continue + + for change in change_list: + if not isinstance(change, dict): + continue + logger.info( + "todo ticket change ref=%s event_id=%s type=%s eventType=%s author=%s editor=%s newStatus=%s newResolution=%s", + ticket_ref, + event_id, + change.get("__typename"), + change.get("eventType"), + _safe_nested_name(change.get("author")), + _safe_nested_name(change.get("editor")), + change.get("newStatus"), + change.get("newResolution"), + ) + normalized = _normalize_event_change( + settings=self.settings, + actor=actor, + event=event, + change=change, + occurred_at=occurred_at, + ) + if normalized is not None: + logger.info( + "todo ticket change accepted ref=%s event_id=%s normalized_type=%s", + ticket_ref, + event_id, + normalized.event_type, + ) + events.append(normalized) + + if stop_paging or not cursor: + break + + return events new file mode 100644 @@ -0,0 +1 @@ +"""Utility helpers.""" new file mode 100644 @@ -0,0 +1,37 @@ +from __future__ import annotations + +from datetime import UTC, date, datetime, time, timedelta + + +def ensure_utc(dt: datetime) -> datetime: + if dt.tzinfo is None: + return dt.replace(tzinfo=UTC) + return dt.astimezone(UTC) + + +def parse_date(value: str) -> date: + return date.fromisoformat(value) + + +def parse_datetime(value: str) -> datetime: + normalized = value.replace("Z", "+00:00") + return ensure_utc(datetime.fromisoformat(normalized)) + + +def date_range(start: date, end: date) -> list[date]: + days: list[date] = [] + current = start + while current <= end: + days.append(current) + current += timedelta(days=1) + return days + + +def year_bounds(year: int) -> tuple[date, date]: + return date(year, 1, 1), date(year, 12, 31) + + +def date_to_utc_bounds(value: date) -> tuple[datetime, datetime]: + start = datetime.combine(value, time.min, tzinfo=UTC) + end = datetime.combine(value, time.max, tzinfo=UTC) + return start, end new file mode 100644 @@ -0,0 +1,27 @@ +from __future__ import annotations + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from srht_contrib.models import ActorAlias + + +class ActorIdentityResolver: + def __init__(self, configured_aliases: dict[str, list[str]] | None = None) -> None: + self.configured_aliases = configured_aliases or {} + + def canonicalize(self, actor: str, db: Session | None = None) -> str: + normalized = actor.strip() + if not normalized: + return normalized + + for canonical, aliases in self.configured_aliases.items(): + if normalized == canonical or normalized in aliases: + return canonical + + if db is not None: + alias = db.scalar(select(ActorAlias).where(ActorAlias.alias == normalized)) + if alias is not None: + return alias.canonical_actor + + return normalized new file mode 100644 @@ -0,0 +1,31 @@ +from __future__ import annotations + +from fastapi import HTTPException, status + + +def canonicalize_repository_name(actor: str, repo_name: str) -> str: + normalized_actor = actor.strip() + normalized_repo_name = repo_name.strip() + + if not normalized_actor: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Actor must not be blank.") + if not normalized_repo_name: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Repository name must not be blank.") + + if "/" in normalized_repo_name: + owner, name = normalized_repo_name.split("/", 1) + owner = owner.strip() + name = name.strip() + if not owner or not name or "/" in name: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail="Repository name must be `name` or `~owner/name`.", + ) + canonical_owner = owner if owner.startswith("~") else f"~{owner}" + return f"{canonical_owner}/{name}" + + if "/" in normalized_actor: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Actor must be a canonical sr.ht user.") + + canonical_actor = normalized_actor if normalized_actor.startswith("~") else f"~{normalized_actor}" + return f"{canonical_actor}/{normalized_repo_name}" new file mode 100644 @@ -0,0 +1,58 @@ +from __future__ import annotations + +from collections.abc import Generator + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session, sessionmaker + +from srht_contrib.config import Settings +from srht_contrib.db import Base, make_engine +from srht_contrib.main import create_app + + +@pytest.fixture() +def settings() -> Settings: + return Settings( + API_KEY="test-api-key", + ENABLE_SCHEDULER=False, + SRHT_TOKEN="test-token", + DATABASE_URL="sqlite://", + TODO_SRHT_ENDPOINT="https://todo.sr.ht/query", + GIT_SRHT_ENDPOINT="https://git.sr.ht/query", + DEFAULT_ACTOR="~ccleberg", + POLL_INTERVAL_SECONDS=3600, + GIT_TRACKED_REPOSITORIES=[], + ) + + +@pytest.fixture() +def db_engine(settings: Settings): + engine = make_engine(settings) + Base.metadata.create_all(bind=engine) + try: + yield engine + finally: + Base.metadata.drop_all(bind=engine) + + +@pytest.fixture() +def session_factory(db_engine) -> sessionmaker[Session]: + return sessionmaker(bind=db_engine, autoflush=False, autocommit=False, expire_on_commit=False) + + +@pytest.fixture() +def db_session(session_factory: sessionmaker[Session]) -> Generator[Session, None, None]: + session = session_factory() + try: + yield session + finally: + session.close() + + +@pytest.fixture() +def client(settings: Settings, db_engine, session_factory: sessionmaker[Session]) -> Generator[TestClient, None, None]: + app = create_app(settings, engine=db_engine, session_factory=session_factory) + with TestClient(app) as test_client: + test_client.headers.update({"X-API-Key": settings.api_key}) + yield test_client new file mode 100644 @@ -0,0 +1,79 @@ +from datetime import UTC, datetime, date + +from sqlalchemy.orm import Session + +from srht_contrib.models import ContributionEvent +from srht_contrib.services.aggregator import ContributionAggregator + + +def test_aggregator_zero_fills_days(db_session: Session) -> None: + db_session.add( + ContributionEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name=None, + resource_id="1", + external_uid="todo:1:created", + occurred_at=datetime(2026, 3, 28, 12, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ) + ) + db_session.commit() + + calendar = ContributionAggregator().build_calendar( + db_session, + "~ccleberg", + date(2026, 3, 28), + date(2026, 3, 30), + ) + + assert [day.model_dump() for day in calendar.days] == [ + {"date": date(2026, 3, 28), "count": 1, "score": 1.0}, + {"date": date(2026, 3, 29), "count": 0, "score": 0.0}, + {"date": date(2026, 3, 30), "count": 0, "score": 0.0}, + ] + + +def test_stats_calculation(db_session: Session) -> None: + db_session.add_all( + [ + ContributionEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name=None, + resource_id="1", + external_uid="todo:1:created", + occurred_at=datetime(2026, 3, 28, 12, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ), + ContributionEvent( + service="todo", + event_type="ticket_comment", + actor="~ccleberg", + repo_name=None, + resource_id="1", + external_uid="todo:1:comment:1", + occurred_at=datetime(2026, 3, 29, 12, 0, tzinfo=UTC), + weight=0.5, + raw_payload_json=None, + ), + ] + ) + db_session.commit() + + stats = ContributionAggregator().build_stats( + db_session, + "~ccleberg", + date(2026, 3, 28), + date(2026, 3, 30), + ) + + assert stats.total_events == 2 + assert stats.total_score == 1.5 + assert stats.active_days == 2 + assert stats.longest_streak == 2 + assert stats.current_streak == 0 new file mode 100644 @@ -0,0 +1,120 @@ +from datetime import UTC, datetime + +from fastapi.testclient import TestClient + +from srht_contrib.main import create_app +from srht_contrib.models import ContributionEvent + + +def test_api_routes_require_api_key(settings, db_engine, session_factory) -> None: + app = create_app(settings, engine=db_engine, session_factory=session_factory) + with TestClient(app) as open_client: + response = open_client.get("/health") + assert response.status_code == 200 + + with TestClient(app) as unauthorized: + unauthorized_response = unauthorized.get("/api/contributions/~ccleberg?from=2026-03-28&to=2026-03-30") + assert unauthorized_response.status_code == 401 + + with TestClient(app) as invalid: + invalid.headers.update({"X-API-Key": "wrong-key"}) + invalid_response = invalid.get("/api/repositories") + assert invalid_response.status_code == 401 + + +def test_contributions_api_returns_zero_filled_range(client: TestClient, db_session) -> None: + db_session.add( + ContributionEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name=None, + resource_id="1", + external_uid="todo:ticket:1:created", + occurred_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ) + ) + db_session.commit() + + response = client.get("/api/contributions/~ccleberg?from=2026-03-28&to=2026-03-30") + + assert response.status_code == 200 + assert response.json()["days"] == [ + {"date": "2026-03-28", "count": 0, "score": 0.0}, + {"date": "2026-03-29", "count": 0, "score": 0.0}, + {"date": "2026-03-30", "count": 1, "score": 1.0}, + ] + + +def test_contribution_stats_api(client: TestClient, db_session) -> None: + db_session.add_all( + [ + ContributionEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name=None, + resource_id="1", + external_uid="todo:ticket:1:created", + occurred_at=datetime(2026, 3, 29, 10, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ), + ContributionEvent( + service="todo", + event_type="ticket_comment", + actor="~ccleberg", + repo_name=None, + resource_id="1", + external_uid="todo:ticket:1:comment:2", + occurred_at=datetime(2026, 3, 30, 11, 0, tzinfo=UTC), + weight=0.5, + raw_payload_json=None, + ), + ] + ) + db_session.commit() + + response = client.get("/api/contributions/~ccleberg/stats?from=2026-03-28&to=2026-03-30") + + assert response.status_code == 200 + assert response.json()["total_events"] == 2 + assert response.json()["total_score"] == 1.5 + assert response.json()["longest_streak"] == 2 + assert response.json()["current_streak"] == 2 + + +def test_invalid_date_input_returns_400(client: TestClient) -> None: + response = client.get("/api/contributions/~ccleberg?from=2026-13-01&to=2026-03-30") + + assert response.status_code == 400 + assert "Invalid date format" in response.json()["detail"] + + +def test_contribution_routes_use_settings_backed_alias_resolution(settings, db_engine, session_factory) -> None: + alias_settings = settings.model_copy(update={"actor_aliases_json": {"~ccleberg": ["cmc@example.com"]}}) + app = create_app(alias_settings, engine=db_engine, session_factory=session_factory) + with session_factory() as session: + session.add( + ContributionEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name=None, + resource_id="1", + external_uid="todo:alias:1", + occurred_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ) + ) + session.commit() + + with TestClient(app) as client: + client.headers.update({"X-API-Key": alias_settings.api_key}) + response = client.get("/api/contributions/cmc@example.com?from=2026-03-30&to=2026-03-30") + + assert response.status_code == 200 + assert response.json()["actor"] == "~ccleberg" new file mode 100644 @@ -0,0 +1,15 @@ +from fastapi.testclient import TestClient + +from srht_contrib.main import create_app + + +def test_health_endpoint() -> None: + with TestClient(create_app()) as client: + response = client.get("/health") + + assert response.status_code == 200 + assert response.json() == {"status": "ok"} + + +def test_scheduler_is_disabled_by_default(client: TestClient) -> None: + assert client.app.state.scheduler is None new file mode 100644 @@ -0,0 +1,337 @@ +from datetime import UTC, datetime + +from sqlalchemy import select + +from srht_contrib.config import Settings +from srht_contrib.jobs.poller import PollerService +from srht_contrib.models import SyncState, TrackedRepository +from srht_contrib.schemas import NormalizedEvent +from srht_contrib.services.git import GitIngestionService, GitPollResult +from srht_contrib.services.todo import TodoIngestionService, TodoPollResult + + +class StubClient: + def __init__(self, payload: dict | None = None, payloads_by_query: dict[str, dict] | None = None) -> None: + self.payload = payload or {} + self.payloads_by_query = payloads_by_query or {} + self.calls: list[tuple[str, dict | None]] = [] + + def execute(self, query: str, variables: dict | None = None) -> dict: + self.calls.append((query, variables)) + for marker, payload in self.payloads_by_query.items(): + if marker in query: + return payload + return self.payload + + +class RecordingTodoService: + service_name = "todo" + + def __init__(self, events_by_call: list[list[NormalizedEvent]]) -> None: + self.events_by_call = events_by_call + self.calls: list[datetime] = [] + + def fetch_recent_events(self, actor: str, since: datetime | None = None) -> TodoPollResult: + assert since is not None + self.calls.append(since) + events = self.events_by_call.pop(0) + return TodoPollResult(events=events, cursor="2026-03-31T00:00:00+00:00") + + +class EmptyGitService: + service_name = "git" + + def __init__(self) -> None: + self.settings = Settings( + SRHT_TOKEN="x", + DATABASE_URL="sqlite://", + DEFAULT_ACTOR="~ccleberg", + TODO_SRHT_ENDPOINT="https://todo.sr.ht/query", + GIT_SRHT_ENDPOINT="https://git.sr.ht/query", + POLL_INTERVAL_SECONDS=60, + ) + + 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 make_settings(**overrides) -> Settings: + values = { + "API_KEY": "test-api-key", + "ENABLE_SCHEDULER": False, + "SRHT_TOKEN": "x", + "DATABASE_URL": "sqlite://", + "DEFAULT_ACTOR": "~ccleberg", + "TODO_SRHT_ENDPOINT": "https://todo.sr.ht/query", + "GIT_SRHT_ENDPOINT": "https://git.sr.ht/query", + "POLL_INTERVAL_SECONDS": 60, + } + values.update(overrides) + return Settings(**values) + + +def test_todo_ingestion_is_idempotent(db_session) -> None: + settings = make_settings() + payload = { + "me": {"canonicalName": "~ccleberg"}, + "events": { + "results": [ + { + "id": "1001", + "created": "2026-03-29T10:00:00Z", + "ticket": { + "id": "123", + "ref": "~ccleberg/todo/123", + "status": "RESOLVED", + "resolution": "CLOSED", + "tracker": {"name": "todo"}, + }, + "changes": [ + { + "__typename": "Created", + "eventType": "CREATED", + "ticket": {"id": "123"}, + "author": {"canonicalName": "~ccleberg"}, + } + ], + }, + { + "id": "1002", + "created": "2026-03-30T09:00:00Z", + "ticket": { + "id": "123", + "ref": "~ccleberg/todo/123", + "status": "RESOLVED", + "resolution": "CLOSED", + "tracker": {"name": "todo"}, + }, + "changes": [ + { + "__typename": "Comment", + "eventType": "COMMENT", + "ticket": {"id": "123"}, + "author": {"canonicalName": "~ccleberg"}, + } + ], + }, + { + "id": "1003", + "created": "2026-03-30T10:00:00Z", + "ticket": { + "id": "123", + "ref": "~ccleberg/todo/123", + "status": "RESOLVED", + "resolution": "CLOSED", + "tracker": {"name": "todo"}, + }, + "changes": [ + { + "__typename": "StatusChange", + "eventType": "STATUS_CHANGE", + "ticket": {"id": "123"}, + "editor": {"canonicalName": "~ccleberg"}, + "oldStatus": "IN_PROGRESS", + "newStatus": "RESOLVED", + "oldResolution": "UNRESOLVED", + "newResolution": "CLOSED", + } + ], + }, + ], + "cursor": None, + }, + } + + todo_service = TodoIngestionService(StubClient(payload), settings) + git_service = GitIngestionService(StubClient(payload={}), settings) + poller = PollerService(todo_service=todo_service, git_service=git_service) + + first_inserted = poller.poll_all(db_session, "~ccleberg") + second_inserted = poller.poll_all(db_session, "~ccleberg") + + assert first_inserted == 3 + assert second_inserted == 0 + + +def test_todo_ingestion_falls_back_to_tracker_crawl(db_session) -> None: + settings = make_settings() + client = StubClient( + payloads_by_query={ + "query TodoActivity": { + "me": {"canonicalName": "~ccleberg"}, + "events": {"results": [], "cursor": None}, + }, + "query TodoTrackers": { + "me": { + "canonicalName": "~ccleberg", + "trackers": {"results": [{"id": "1", "rid": "tracker-rid", "name": "todo"}], "cursor": None}, + } + }, + "query TodoTrackerTickets": { + "tracker": { + "id": "1", + "name": "todo", + "tickets": { + "results": [ + { + "id": 123, + "ref": "~ccleberg/todo/123", + "created": "2026-03-29T09:00:00Z", + "updated": "2026-03-30T09:00:00Z", + "status": "RESOLVED", + "resolution": "CLOSED", + "submitter": {"canonicalName": "~ccleberg"}, + } + ], + "cursor": None, + }, + } + }, + "query TodoTicketEvents": { + "tracker": { + "ticket": { + "id": 123, + "ref": "~ccleberg/todo/123", + "status": "RESOLVED", + "resolution": "CLOSED", + "events": { + "results": [ + { + "id": "evt-1", + "created": "2026-03-30T09:00:00Z", + "changes": [ + { + "__typename": "Comment", + "eventType": "COMMENT", + "ticket": {"id": "123"}, + "author": {"canonicalName": "~ccleberg"}, + } + ], + } + ], + "cursor": None, + }, + } + } + }, + } + ) + todo_service = TodoIngestionService(client, settings) + git_service = GitIngestionService(StubClient(payload={}), settings) + poller = PollerService(todo_service=todo_service, git_service=git_service) + + inserted = poller.poll_all(db_session, "~ccleberg") + + assert inserted == 1 + assert any("query TodoTrackers" in call[0] for call in client.calls) + + +def test_unsupported_todo_changes_are_ignored(db_session) -> None: + settings = make_settings() + payload = { + "me": {"canonicalName": "~ccleberg"}, + "events": { + "results": [ + { + "id": "1001", + "created": "2026-03-29T10:00:00Z", + "ticket": { + "id": "123", + "ref": "~ccleberg/todo/123", + "status": "OPEN", + "resolution": "UNRESOLVED", + "tracker": {"name": "todo"}, + }, + "changes": [ + {"__typename": "LabelUpdate", "eventType": "LABEL_UPDATE", "ticket": {"id": "123"}}, + {"__typename": "TicketMention", "eventType": "TICKET_MENTION", "ticket": {"id": "123"}}, + ], + } + ], + "cursor": None, + }, + } + + todo_service = TodoIngestionService(StubClient(payload), settings) + git_service = GitIngestionService(StubClient(payload={}), settings) + poller = PollerService(todo_service=todo_service, git_service=git_service) + + inserted = poller.poll_all(db_session, "~ccleberg") + + assert inserted == 0 + + +def test_git_ingestion_normalizes_commit_aliases_and_repository_names(db_session) -> None: + settings = make_settings( + ACTOR_ALIASES_JSON={"~ccleberg": ["cmc@example.com", "Chris Cleberg"]}, + GIT_TRACKED_REPOSITORIES=["Hutch"], + ) + git_payload = { + "user": { + "repository": { + "name": "Hutch", + "owner": {"canonicalName": "~ccleberg"}, + "log": { + "results": [ + { + "id": "abc123", + "shortId": "abc123", + "author": { + "name": "Chris Cleberg", + "email": "cmc@example.com", + "time": "2026-03-30T12:00:00Z", + }, + "committer": { + "name": "Chris Cleberg", + "email": "cmc@example.com", + "time": "2026-03-30T12:00:00Z", + }, + "message": "Add contribution calendar", + } + ], + "cursor": None, + }, + } + } + } + + todo_service = TodoIngestionService( + StubClient(payload={"me": {"canonicalName": "~ccleberg"}, "events": {"results": [], "cursor": None}}), + settings, + ) + git_service = GitIngestionService(StubClient(payloads_by_query={"query RepositoryLog": git_payload}), settings) + poller = PollerService(todo_service=todo_service, git_service=git_service) + + inserted = poller.poll_all(db_session, "~ccleberg") + + assert inserted == 1 + + tracked_repositories = db_session.scalars(select(TrackedRepository.repo_name)).all() + assert tracked_repositories == ["~ccleberg/Hutch"] + + +def test_sync_overlap_reuses_cursor_window_and_suppresses_duplicates(db_session) -> None: + event = NormalizedEvent( + service="todo", + event_type="ticket_created", + actor="~ccleberg", + repo_name="todo", + resource_id="123", + external_uid="todo:event:123:created:123", + occurred_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ) + todo_service = RecordingTodoService(events_by_call=[[event], [event]]) + poller = PollerService(todo_service=todo_service, git_service=EmptyGitService()) + + first_inserted = poller.poll_all(db_session, "~ccleberg") + second_inserted = poller.poll_all(db_session, "~ccleberg") + + state = db_session.scalar(select(SyncState).where(SyncState.service == "todo").where(SyncState.actor == "~ccleberg")) + + assert first_inserted == 1 + assert second_inserted == 0 + assert state is not None + assert len(todo_service.calls) == 2 + assert todo_service.calls[1].isoformat() == "2026-03-30T00:00:00+00:00" new file mode 100644 @@ -0,0 +1,117 @@ +from pathlib import Path + +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, inspect, text + +from srht_contrib.config import Settings + + +def test_alembic_upgrade_creates_schema(tmp_path) -> None: + database_path = tmp_path / "migrated.db" + database_url = f"sqlite:///{database_path}" + config = Config(str(Path(__file__).resolve().parent.parent / "alembic.ini")) + config.set_main_option("script_location", str(Path(__file__).resolve().parent.parent / "alembic")) + config.set_main_option("sqlalchemy.url", database_url) + + command.upgrade(config, "head") + + inspector = inspect(create_engine(database_url)) + assert sorted(inspector.get_table_names()) == [ + "actor_aliases", + "alembic_version", + "contribution_events", + "sync_states", + "tracked_repositories", + ] + + +def test_alembic_upgrade_adopts_legacy_schema(tmp_path) -> None: + database_path = tmp_path / "legacy.db" + database_url = f"sqlite:///{database_path}" + engine = create_engine(database_url) + with engine.begin() as connection: + connection.execute( + text( + """ + CREATE TABLE contribution_events ( + id INTEGER NOT NULL PRIMARY KEY, + service VARCHAR(32) NOT NULL, + event_type VARCHAR(64) NOT NULL, + actor VARCHAR(255) NOT NULL, + repo_name VARCHAR(255), + resource_id VARCHAR(255) NOT NULL, + external_uid VARCHAR(255) NOT NULL, + occurred_at DATETIME NOT NULL, + weight FLOAT NOT NULL, + raw_payload_json JSON, + CONSTRAINT uq_contribution_event_service_uid UNIQUE (service, external_uid) + ) + """ + ) + ) + connection.execute(text("CREATE INDEX ix_contribution_events_actor_occurred_at ON contribution_events (actor, occurred_at)")) + connection.execute(text("CREATE INDEX ix_contribution_events_service_occurred_at ON contribution_events (service, occurred_at)")) + connection.execute( + text( + """ + CREATE TABLE sync_states ( + id INTEGER NOT NULL PRIMARY KEY, + service VARCHAR(32) NOT NULL, + actor VARCHAR(255) NOT NULL, + cursor_value TEXT, + updated_at DATETIME NOT NULL, + CONSTRAINT uq_sync_state_service_actor UNIQUE (service, actor) + ) + """ + ) + ) + connection.execute( + text( + """ + CREATE TABLE tracked_repositories ( + id INTEGER NOT NULL PRIMARY KEY, + service VARCHAR(32) NOT NULL, + repo_name VARCHAR(255) NOT NULL, + actor VARCHAR(255), + CONSTRAINT uq_tracked_repository_service_name UNIQUE (service, repo_name) + ) + """ + ) + ) + connection.execute( + text( + """ + INSERT INTO tracked_repositories (id, service, repo_name, actor) + VALUES (1, 'git', 'Hutch', NULL) + """ + ) + ) + connection.execute( + text( + """ + CREATE TABLE actor_aliases ( + id INTEGER NOT NULL PRIMARY KEY, + canonical_actor VARCHAR(255) NOT NULL, + alias VARCHAR(255) NOT NULL, + CONSTRAINT uq_actor_alias_alias UNIQUE (alias) + ) + """ + ) + ) + + config = Config(str(Path(__file__).resolve().parent.parent / "alembic.ini")) + config.set_main_option("script_location", str(Path(__file__).resolve().parent.parent / "alembic")) + config.set_main_option("sqlalchemy.url", database_url) + + command.upgrade(config, "head") + + inspector = inspect(create_engine(database_url)) + columns = {column["name"]: column for column in inspector.get_columns("tracked_repositories")} + unique_constraints = {constraint["name"] for constraint in inspector.get_unique_constraints("tracked_repositories")} + with create_engine(database_url).connect() as connection: + actor = connection.execute(text("SELECT actor FROM tracked_repositories WHERE id = 1")).scalar_one() + + assert columns["actor"]["nullable"] is False + assert "uq_tracked_repository_service_actor_name" in unique_constraints + assert actor == Settings().default_actor new file mode 100644 @@ -0,0 +1,71 @@ +from datetime import UTC, datetime + +from fastapi.testclient import TestClient + +from srht_contrib.config import Settings +from srht_contrib.main import create_app +from srht_contrib.models import ContributionEvent +from srht_contrib.services.srht_client import SourceHutClientError + + +class _Closable: + def close(self) -> None: + return None + + +class InsertingPoller: + def __init__(self) -> None: + service = type("Service", (), {"client": _Closable()})() + self.todo_service = service + self.git_service = service + + def poll_all(self, db, actor: str) -> int: + db.add( + ContributionEvent( + service="todo", + event_type="ticket_created", + actor=actor, + repo_name="todo", + resource_id="1", + external_uid="todo:manual:1", + occurred_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC), + weight=1.0, + raw_payload_json=None, + ) + ) + db.commit() + return 1 + + +class FailingPoller: + def __init__(self) -> None: + service = type("Service", (), {"client": _Closable()})() + self.todo_service = service + self.git_service = service + + def poll_all(self, db, actor: str) -> int: + raise SourceHutClientError("boom") + + +def test_manual_poll_uses_same_database_session(settings: Settings, db_engine, session_factory) -> None: + app = create_app(settings, engine=db_engine, session_factory=session_factory, poller=InsertingPoller()) + with TestClient(app) as client: + client.headers.update({"X-API-Key": settings.api_key}) + + poll_response = client.post("/api/contributions/poll?actor=~ccleberg") + calendar_response = client.get("/api/contributions/~ccleberg?from=2026-03-30&to=2026-03-30") + + assert poll_response.status_code == 200 + assert poll_response.json()["inserted_events"] == 1 + assert calendar_response.status_code == 200 + assert calendar_response.json()["days"] == [{"date": "2026-03-30", "count": 1, "score": 1.0}] + + +def test_manual_poll_maps_sourcehut_failures_to_502(settings: Settings, db_engine, session_factory) -> None: + app = create_app(settings, engine=db_engine, session_factory=session_factory, poller=FailingPoller()) + with TestClient(app) as client: + client.headers.update({"X-API-Key": settings.api_key}) + response = client.post("/api/contributions/poll?actor=~ccleberg") + + assert response.status_code == 502 + assert "SourceHut polling failed" in response.json()["detail"] new file mode 100644 @@ -0,0 +1,77 @@ +from fastapi.testclient import TestClient + + +def test_create_list_get_update_delete_tracked_repositories(client: TestClient) -> None: + create_response = client.post( + "/api/repositories", + json={"actor": "~ccleberg", "repo_name": "Hutch"}, + ) + + assert create_response.status_code == 201 + assert create_response.json()["repo_name"] == "~ccleberg/Hutch" + + repository_id = create_response.json()["id"] + + get_response = client.get(f"/api/repositories/{repository_id}") + assert get_response.status_code == 200 + assert get_response.json()["repo_name"] == "~ccleberg/Hutch" + + update_response = client.patch( + f"/api/repositories/{repository_id}", + json={"repo_name": "~ccleberg/cleberg.net"}, + ) + assert update_response.status_code == 200 + assert update_response.json()["repo_name"] == "~ccleberg/cleberg.net" + + list_response = client.get("/api/repositories?actor=~ccleberg") + assert list_response.status_code == 200 + assert list_response.json() == [ + { + "id": repository_id, + "service": "git", + "actor": "~ccleberg", + "repo_name": "~ccleberg/cleberg.net", + } + ] + + delete_response = client.delete(f"/api/repositories/{repository_id}") + assert delete_response.status_code == 204 + + missing_response = client.get(f"/api/repositories/{repository_id}") + assert missing_response.status_code == 404 + + +def test_repository_validation_and_conflicts(client: TestClient) -> None: + invalid_response = client.post( + "/api/repositories", + json={"actor": "~ccleberg", "repo_name": " "}, + ) + assert invalid_response.status_code == 422 + + first_response = client.post( + "/api/repositories", + json={"actor": "~ccleberg", "repo_name": "Hutch"}, + ) + second_response = client.post( + "/api/repositories", + json={"actor": "~ccleberg", "repo_name": "~ccleberg/Hutch"}, + ) + + assert first_response.status_code == 201 + assert second_response.status_code == 409 + + +def test_same_repository_can_be_tracked_by_different_actors(client: TestClient) -> None: + first_response = client.post( + "/api/repositories", + json={"actor": "~ccleberg", "repo_name": "Hutch"}, + ) + second_response = client.post( + "/api/repositories", + json={"actor": "~other", "repo_name": "Hutch"}, + ) + + assert first_response.status_code == 201 + assert second_response.status_code == 201 + assert first_response.json()["repo_name"] == "~ccleberg/Hutch" + assert second_response.json()["repo_name"] == "~other/Hutch" new file mode 100644 @@ -0,0 +1,56 @@ +import httpx +import pytest + +from srht_contrib.services.srht_client import SourceHutClientError, SourceHutGraphQLClient + + +def test_graphql_client_retries_http_5xx_and_succeeds() -> None: + attempts = {"count": 0} + + def handler(request: httpx.Request) -> httpx.Response: + attempts["count"] += 1 + if attempts["count"] == 1: + return httpx.Response(502, json={"error": "bad gateway"}) + return httpx.Response(200, json={"data": {"ok": True}}) + + client = SourceHutGraphQLClient( + "https://todo.sr.ht/query", + "token", + transport=httpx.MockTransport(handler), + ) + + data = client.execute("query Ping { ping }") + + assert data == {"ok": True} + assert attempts["count"] == 2 + client.close() + + +def test_graphql_client_raises_for_graphql_errors() -> None: + client = SourceHutGraphQLClient( + "https://todo.sr.ht/query", + "token", + transport=httpx.MockTransport(lambda request: httpx.Response(200, json={"errors": [{"message": "nope"}]})), + ) + + with pytest.raises(SourceHutClientError): + client.execute("query Ping { ping }") + + client.close() + + +def test_graphql_client_raises_for_network_errors() -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("offline", request=request) + + client = SourceHutGraphQLClient( + "https://todo.sr.ht/query", + "token", + max_retries=0, + transport=httpx.MockTransport(handler), + ) + + with pytest.raises(SourceHutClientError): + client.execute("query Ping { ping }") + + client.close()