From 7c9c69f03b623d5d35d03d3ab6125c1a0f81df7f Mon Sep 17 00:00:00 2001 From: TELVIN TEUM Date: Fri, 17 Jul 2026 02:48:26 +0300 Subject: [PATCH] feat(orchestration): event-driven webhook queue, trigger evaluation, PR/tag pipelines, Checks reporting, repo event timeline GitHub becomes purely the event source and repository provider; overup is the complete CI/CD control plane. The webhook receiver now does verification -> validation -> durable persistence -> 202 ack only, and every side effect runs asynchronously off the request path. Backend: - Durable webhook queue: webhook_deliveries gains a normalized server-built payload (never the raw body), pending|processing|processed|ignored|failed lifecycle, and retry_count (migration 20260717000001) - services/webhook_processor.rs: sequential FOR UPDATE SKIP LOCKED consumer (poke + 2s tick) preserving per-repo ordering; <=3 retries, 5-min stuck-row revert, payload nulling on terminal states; manual GitHub redelivery revives a terminally failed row - services/trigger_eval.rs: pure/bounded GitHub-flavored trigger evaluation - parser v3 extracts on.push / on.pull_request filters into workflows.metadata.triggerFilters; hand-rolled glob (*, **, ?, +, [a-z], \ escapes, ordered ! negation) with no regex or recursion; branch+path dimensions compose, paths never apply to tags, truncated changed-path sets fail open on path filters only - PR + tag pipelines: trigger vocabulary push|manual|pull_request|tag, pipelines.pr_number; PR head SHAs build under refs/pull/{n}/head; fork PRs are skipped with static fork_pr_skipped (secrets never flow to fork code); refs/tags/ pushes run on:push workflows as trigger tag - services/github_checks.rs: best-effort Checks API reporting for event-triggered pipelines (queued -> in_progress -> completed with a details link); a SEPARATE checks:write installation token keeps the sync token read-only; 403/422 parks the installation for 1h with one edge-triggered warn; GITHUB_CHECKS_ENABLED gate; fast pipelines that finish before the async create fall back to creating the check run directly in completed status - Immutable repository_events timeline (outcome CHECK, static ignored_reason, pipeline_ids[], capped summary) + keyset GET .../repositories/{id}/events + a health object on the repo detail response; terminal delivery failures always surface a failed timeline row Frontend (same design system throughout): - RepositorySyncPanel: KPI strip (Auto-sync, Last event, Pending, Failed 24h, GitHub checks) between the meta strip and tabs - RepositoryEventsList: Events tab with outcome badges, ref/PR/tag chips, skip tooltips, pipeline links, IntersectionObserver infinite scroll - Pipelines UI: pull_request/tag trigger filters + icons, PR # link on the Metadata tab, branchOfRef renders tag and PR refs Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 96 +- backend/.env.example | 12 +- ...17000001_webhook_async_and_repo_events.sql | 76 ++ backend/src/config.rs | 7 + backend/src/db/mod.rs | 1 + backend/src/db/pipelines.rs | 57 +- backend/src/db/repository_events.rs | 146 +++ backend/src/db/webhook_deliveries.rs | 147 ++- backend/src/db/workflows.rs | 23 +- backend/src/handlers/github_webhooks.rs | 495 ++++++---- backend/src/handlers/pipelines.rs | 12 +- backend/src/handlers/repositories.rs | 65 +- backend/src/main.rs | 5 + backend/src/models/pipeline.rs | 9 + backend/src/models/repository.rs | 54 ++ backend/src/routes/mod.rs | 4 + backend/src/services/github_app.rs | 54 +- backend/src/services/github_checks.rs | 366 ++++++++ backend/src/services/mod.rs | 3 + backend/src/services/pipeline_run.rs | 9 + backend/src/services/repo_sync.rs | 15 +- backend/src/services/trigger_eval.rs | 695 ++++++++++++++ backend/src/services/webhook_processor.rs | 859 ++++++++++++++++++ backend/src/services/workflow_parse.rs | 305 ++++++- backend/src/state.rs | 8 + .../pipelines/components/JobIdentityBar.tsx | 19 +- .../pipelines/components/PipelineFilters.tsx | 2 + .../pipelines/components/PipelinesTable.tsx | 22 +- .../components/panels/MetadataPanel.tsx | 15 +- src/features/pipelines/lib/format.ts | 11 +- .../pipelines/pages/PipelinesPage.tsx | 2 +- .../repositories/api/repositoriesApi.ts | 14 + .../components/RepositoryEventsList.tsx | 225 +++++ .../components/RepositorySyncPanel.tsx | 104 +++ .../repositories/hooks/useRepositories.ts | 21 +- .../pages/RepositoryDetailPage.tsx | 11 +- src/types/pipeline.ts | 4 +- src/types/repository.ts | 48 + 38 files changed, 3768 insertions(+), 253 deletions(-) create mode 100644 backend/migrations/20260717000001_webhook_async_and_repo_events.sql create mode 100644 backend/src/db/repository_events.rs create mode 100644 backend/src/services/github_checks.rs create mode 100644 backend/src/services/trigger_eval.rs create mode 100644 backend/src/services/webhook_processor.rs create mode 100644 src/features/repositories/components/RepositoryEventsList.tsx create mode 100644 src/features/repositories/components/RepositorySyncPanel.tsx diff --git a/CLAUDE.md b/CLAUDE.md index e192318..cf3c013 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,8 +15,11 @@ module** (named deployment targets bound from workflow YAML `environment:`, acti highest-precedence secrets scope with `environments.manage` RBAC and a catalog + detail UI), and the **Notification Center** (a per-user, actionable projection of the audit ledger: bell + popover/bottom-sheet in the shell, per-user WebSocket delivery, dedup -grouping, preferences, and a history page — see below). Matrix expansion and PR/cron -triggers build on this foundation. +grouping, preferences, and a history page — see below), and the **Event-driven +orchestration layer** (durable async webhook queue + background processor, GitHub-style +trigger evaluation with branch/tag/path glob filters, pull_request + tag pipelines, +Checks API status reporting back to GitHub, and a per-repository event timeline + sync +health panel — see below). Matrix expansion and cron triggers build on this foundation. ## Architecture @@ -48,7 +51,9 @@ app JWT must succeed AND the account login must match the user — or an `installation.created` webhook recorded them as installer) before linking it to the workspace. `POST /webhooks/github` (outside `/api`, no CSRF/CORS) authenticates every delivery with constant-time HMAC-SHA256 over the raw body, dedupes on `X-GitHub-Delivery`, -and drives incremental sync on push/repository/installation events. Sync fetches repo +persists a normalized capped payload into the durable `webhook_deliveries` queue, and +acks immediately — `services/webhook_processor.rs` drives incremental sync and pipeline +triggering asynchronously (see "Event-driven orchestration" below). Sync fetches repo metadata, branches, and `.github/workflows` via the Contents API (no cloning, no git2), diffs blob shas, parses YAML with `services/workflow_parse.rs` (node budget 20k, depth 32, 512 KB cap, ≤100 jobs — parse errors become diagnostics, never 500s; note YAML 1.1 parses @@ -293,6 +298,55 @@ read/unread/archive, CSV export) routed WITHOUT a nav item, like `search`. Link server-built `{kind, …Id}` objects resolved client-side through an allow-list (`lib/notificationPresentation.ts`) — never URLs. +**Event-driven orchestration (async webhooks + trigger evaluation + Checks reporting).** +GitHub is purely the event source and repository provider; overup is the control plane. +The webhook receiver does verification → validation → durable persistence → 202 ack ONLY +(GitHub's 10-second budget): each delivery lands in `webhook_deliveries` (now a durable +queue — normalized server-built payload JSONB, `pending|processing|processed|ignored|failed`, +retry_count) and `services/webhook_processor.rs` (the notification-projector loop shape: +poke + 2 s tick, but per-row `FOR UPDATE SKIP LOCKED` claims drained SEQUENTIALLY — the +per-repo ordering guarantee) performs every side effect off the request path: repo lookup +(unconnected repos are discarded before any processing), default-branch sync scheduling, +trigger evaluation, pipeline creation, and the timeline write. Deliveries retry ≤3 times +(5-min stuck-row revert on the tick; a manual GitHub redelivery revives a terminally +failed row), and completion commits the `repository_events` timeline row + the status +flip in one transaction. **Trigger evaluation** (`services/trigger_eval.rs`, pure/bounded): +parser v3 extracts `on.push`/`on.pull_request` filter lists into +`workflows.metadata.triggerFilters` (branches/branchesIgnore/tags/tagsIgnore/paths/ +pathsIgnore + PR `types`, ≤50 patterns × 256 B, branches+branches-ignore on one event is +an Error like GitHub); a hand-rolled GitHub-flavored glob (`*` no-slash, `**`, `?`/`+` +quantify the preceding char, `[a-z]`, `\` escapes; ordered `!` negation, last match wins) +evaluates each workflow against the event before `create_pipeline` — only-tags blocks +branch pushes and vice versa, branch AND path dimensions must both pass, paths never +apply to tag pushes, changed paths come from the push payload commits (truncated set ⇒ +path filters fail OPEN), PR `types` default `opened|synchronize|reopened` and PR branch +filters match the BASE branch. Pre-v3 metadata evaluates as unfiltered. Skips are +recorded per-workflow with static reasons in the event summary. **PR + tag pipelines**: +pipeline trigger vocabulary is now `push|manual|pull_request|tag` (+ `pipelines.pr_number`); +`pull_request` events build the PR head SHA under `refs/pull/{n}/head` (fork PRs are +skipped with static `fork_pr_skipped` — secrets never flow to fork code; `closed` never +builds — the merge commit's push event drives push workflows); `refs/tags/` pushes run +`on: push` workflows as trigger `tag`. Cron stays parse/display-only. **Checks API +reporting** (`services/github_checks.rs`): event-triggered pipelines (never manual) +surface as GitHub check runs — created queued after `create_pipeline` (processor), +in_progress in `on_job_started`, completed in `maybe_finalize` (conclusion map: partial → +failure with a summary note; `details_url` → the pipeline page; output is static +templates + job counts, never runner text). A second, separately cached installation +token is minted with `checks:write` only (the sync token stays read-only); 403/422 marks +the installation unavailable for 1 h with ONE edge-triggered warn (operator action: +grant the App Checks Read & write + approve on installations). `GITHUB_CHECKS_ENABLED` +(default true) gates it. **Repository timeline + health**: immutable `repository_events` +(outcome CHECK `pipelines_created|sync_scheduled|pipelines_and_sync|ignored|failed`, +static `ignored_reason`, `pipeline_ids[]`, `sync_run_id`, capped summary) feeds +`GET …/repositories/{id}/events` (keyset, `content.read`) and a `health` object on the +detail response (last event at/outcome, failed 24 h, pending deliveries, checksEnabled). +Frontend: `RepositorySyncPanel` (KPI strip: Auto-sync, Last event, Pending, Failed 24 h, +GitHub checks) between the meta strip and the tabs, plus an Events tab +(`RepositoryEventsList`, ActivityFeed-compact rows with outcome badges, ref/PR chips, +skip tooltips, pipeline links, infinite scroll); pipelines UI gained the +pull_request/tag trigger filter options, icons (`GitPullRequest`/`Tag`), PR # link on the +Metadata tab, and `branchOfRef` now renders tag and `PR #n` refs. + **Dev networking.** CRA's `"proxy": "http://localhost:8080"` forwards XHR (`/api/*`, `/auth/logout`) to the backend. Full-page navigations are NOT proxied (CRA serves index.html for `Accept: text/html`), so the login redirect uses the absolute `REACT_APP_API_ORIGIN`. @@ -367,7 +421,9 @@ overup/ │ # RBAC backfill), secret value_set_at rotation clock, │ # notifications (+preferences/projector cursor), │ # storage_backend markers (artifacts/pipeline_jobs/ - │ # workspaces — MinIO/R2 object routing) + │ # workspaces — MinIO/R2 object routing), + │ # webhook queue + repository_events + PR/tag triggers + │ # (payload/retry columns, pr_number, check_run_id) └── src/ ├── main.rs # bootstrap: env, tracing, pool, migrate, orphan recovery, │ # scheduler spawn, janitor, serve @@ -386,7 +442,10 @@ overup/ │ # notifications, notification_ws ├── middleware/ # security_headers, csrf, auth (CurrentUser extractor) └── services/ # session, github, github_app (JWT + token cache), - # auth_flow, workspace, authz (RBAC), repo_sync, + # github_checks (Checks API reporter), auth_flow, + # workspace, authz (RBAC), repo_sync, + # webhook_processor (async delivery queue consumer), + # trigger_eval (GitHub-flavored filter globs), # workflow_parse, pipeline_plan, pipeline_run (state # machine), scheduler, log_hub (mask+cap+broadcast), # runner_hub, object_store (generic S3Store + Storage @@ -533,9 +592,29 @@ into a traversal-safe tar streamed into the container via the Docker archive API (`contents:read + metadata:read`), cached in memory only — never logged, never in responses, never in Postgres; the private key PEM loads once at startup - Webhooks: constant-time HMAC-SHA256 over the raw body (`X-Hub-Signature-256`) before any - parsing; `X-GitHub-Delivery` primary key makes redeliveries no-ops; payloads are parsed + parsing; `X-GitHub-Delivery` primary key makes redeliveries no-ops (a redelivery only + revives a terminally FAILED row for another processing round); payloads are parsed into minimal typed envelopes and never logged; `GITHUB_WEBHOOK_SECRET` must be ≥ 16 bytes at boot (signing-key parity — a guessable secret would let anyone forge deliveries) +- **Webhook processing is async and durable**: the HTTP handler only verifies, validates, + persists a NORMALIZED server-built payload (strings ≤512 B, changed paths deduped/capped + at 300 with a truncation flag, avatars sanitized — never the raw body), and acks 2xx; + every side effect runs in `services/webhook_processor.rs` off the request path. Events + from repositories not connected to any workspace are discarded before any further + processing. A single sequential consumer preserves per-repo ordering; retries cap at 3; + stuck `processing` rows revert after 5 min; consumed payloads are nulled so the queue + table stays bounded. Trigger evaluation (`trigger_eval.rs`) is pure and bounded + (patterns ≤50×256 B re-capped at read time, values ≤1 KB, iterative matcher — no regex, + no recursion); unknown changed-file sets fail OPEN on path filters only. Fork PRs never + execute (static `fork_pr_skipped`) — secrets must not flow to fork code +- **Checks API reporting is best-effort and permission-isolated**: a SEPARATE installation + token is minted with `checks:write` only (the sync token stays contents/metadata + read-only); reporting failures never fail/delay a pipeline; 403/422 (App lacks the + Checks permission) parks the installation for 1 h with one edge-triggered warn; check + bodies are static templates + job counts with owner/repo re-validated against the + segment allow-list before URL interpolation; manual dispatches never report; + `repository_events.outcome`/`ignored_reason` and every skip reason are static category + strings only - Request tracing records method + PATH only (custom `MakeSpan` in routes) — the query string carries secrets on some routes (`?code=`/`?state=` on the OAuth callback, `?ticket=` on WS upgrades) and must never land in spans, even at debug level @@ -819,7 +898,10 @@ GitHub Apps → New GitHub App. configuration avoids the detour entirely - Webhook URL: needs a public tunnel in dev — `smee.io` or `cloudflared tunnel` forwarding to `http://localhost:8080/webhooks/github`; set a strong webhook secret -- Repository permissions: **Metadata (read)** + **Contents (read)** — least privilege; +- Repository permissions: **Metadata (read)** + **Contents (read)** + **Checks (read & + write)** (for reporting pipeline status back to commits/PRs via the Checks API — + existing installations must approve the permission update; until then reporting + degrades to one warn per installation and `GITHUB_CHECKS_ENABLED=false` silences it); `Workflows (write)` is only needed when editor write-back ships - Subscribe to events: Push, Repository, Installation target, Pull request (installation/installation_repositories events arrive automatically) diff --git a/backend/.env.example b/backend/.env.example index 0e785b3..a29debd 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -24,7 +24,10 @@ OAUTH_REDIRECT_URL=http://localhost:8080/auth/github/callback # UNCHECKED — enabling it makes GitHub redirect installs to the # OAuth callback without a state parameter and the install fails. # Webhook URL: https:///webhooks/github (smee.io/cloudflared for local dev) -# Permissions: Repository metadata (read), Contents (read) +# Permissions: Repository metadata (read), Contents (read), +# Checks (read & write — for reporting pipeline status back to +# commits/PRs via the Checks API; existing installations must +# approve the permission update before reporting works) # Events: push, repository, installation, installation repositories, pull request GITHUB_APP_CLIENT_ID= # Path to the downloaded private key PEM (or set GITHUB_APP_PRIVATE_KEY_B64 @@ -35,6 +38,13 @@ GITHUB_WEBHOOK_SECRET= # Public slug of the app (from its URL: github.com/apps/). GITHUB_APP_SLUG= +# Report pipeline status back to GitHub as check runs (queued → in_progress +# → completed with a details link to the pipeline page) for push/PR/tag +# pipelines. Needs the App's Checks (Read & write) permission — without it +# the backend logs one warning per installation and keeps running normally. +# Manual dispatches never report. +GITHUB_CHECKS_ENABLED=true + # Where the React app is served; used for CORS and post-login redirects. FRONTEND_URL=http://localhost:3000 diff --git a/backend/migrations/20260717000001_webhook_async_and_repo_events.sql b/backend/migrations/20260717000001_webhook_async_and_repo_events.sql new file mode 100644 index 0000000..722e999 --- /dev/null +++ b/backend/migrations/20260717000001_webhook_async_and_repo_events.sql @@ -0,0 +1,76 @@ +-- Async webhook processing + per-repository event timeline + PR/tag triggers. +-- +-- 1. webhook_deliveries grows from a pure idempotency ledger into a durable +-- work queue: the HTTP handler persists a normalized (server-built, capped) +-- payload and acks GitHub immediately; a background worker claims rows and +-- performs sync/trigger-evaluation/pipeline creation asynchronously. +-- 2. repository_events is the immutable per-repo timeline the UI renders: +-- what arrived, what it caused (pipelines/sync), or why it was ignored — +-- outcome/ignored_reason hold STATIC category strings only, never upstream +-- text. +-- 3. pipelines learn the pull_request/tag trigger vocabulary, the PR number, +-- and the GitHub check-run linkage for Checks API reporting. + +-- 1. Durable webhook queue ------------------------------------------------- + +ALTER TABLE webhook_deliveries + ADD COLUMN payload JSONB, + ADD COLUMN github_repo_id BIGINT, + ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0, + ADD COLUMN last_attempt_at TIMESTAMPTZ, + ADD COLUMN processed_at TIMESTAMPTZ; + +ALTER TABLE webhook_deliveries + DROP CONSTRAINT webhook_deliveries_status_check; +ALTER TABLE webhook_deliveries + ADD CONSTRAINT webhook_deliveries_status_check + CHECK (status IN ('pending', 'processing', 'processed', 'ignored', 'failed')); + +-- Worker drain scans only live rows, oldest first (per-repo ordering). +CREATE INDEX webhook_deliveries_pending_idx + ON webhook_deliveries (received_at) + WHERE status IN ('pending', 'processing'); + +-- Per-repo webhook health (pending counts) without a workspace join. +CREATE INDEX webhook_deliveries_repo_idx + ON webhook_deliveries (github_repo_id, received_at DESC) + WHERE github_repo_id IS NOT NULL; + +-- 2. Immutable per-repository event timeline -------------------------------- + +CREATE TABLE repository_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + repository_id UUID NOT NULL REFERENCES repositories (id) ON DELETE CASCADE, + delivery_id TEXT NOT NULL, + event TEXT NOT NULL, + action TEXT, + git_ref TEXT, + head_sha TEXT, + actor_login TEXT, + actor_avatar_url TEXT, + -- Static categories only, enforced by consts in services/webhook_processor.rs. + outcome TEXT NOT NULL CHECK (outcome IN + ('pipelines_created', 'sync_scheduled', 'pipelines_and_sync', 'ignored', 'failed')), + ignored_reason TEXT, + pipeline_ids UUID[] NOT NULL DEFAULT '{}', + sync_run_id UUID, + -- Server-built, capped summary (skipped workflows, PR number, merged flag). + summary JSONB NOT NULL DEFAULT '{}'::jsonb, + received_at TIMESTAMPTZ NOT NULL, + processed_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX repository_events_repo_idx + ON repository_events (repository_id, processed_at DESC, id DESC); + +-- 3. Pipelines: PR/tag triggers + Checks API linkage ------------------------- + +ALTER TABLE pipelines + DROP CONSTRAINT pipelines_trigger_check; +ALTER TABLE pipelines + ADD CONSTRAINT pipelines_trigger_check + CHECK (trigger IN ('push', 'manual', 'pull_request', 'tag')); + +ALTER TABLE pipelines + ADD COLUMN pr_number INTEGER, + ADD COLUMN check_run_id BIGINT; diff --git a/backend/src/config.rs b/backend/src/config.rs index 762dea6..422cbab 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -36,6 +36,10 @@ pub struct Config { /// Public slug of the GitHub App; builds the install URL /// https://github.com/apps/{slug}/installations/new. pub github_app_slug: String, + /// Report pipeline status back to GitHub via the Checks API (queued → + /// in_progress → completed). Requires the App's Checks (Read & write) + /// permission; without it reporting degrades to an edge-triggered warn. + pub github_checks_enabled: bool, /// Shared key (>= 32 bytes) for HMAC-SHA256 signatures over job payloads /// sent to runners. Never logged. pub runner_job_signing_key: Vec, @@ -424,6 +428,9 @@ impl Config { github_app_private_key_pem, github_webhook_secret, github_app_slug: required("GITHUB_APP_SLUG")?, + github_checks_enabled: optional("GITHUB_CHECKS_ENABLED", "true") + .parse() + .context("GITHUB_CHECKS_ENABLED must be true or false")?, runner_job_signing_key, default_job_image, job_timeout_seconds: optional("JOB_TIMEOUT_SECONDS", "3600") diff --git a/backend/src/db/mod.rs b/backend/src/db/mod.rs index b8746af..9ba4c6c 100644 --- a/backend/src/db/mod.rs +++ b/backend/src/db/mod.rs @@ -11,6 +11,7 @@ pub mod pipeline_jobs; pub mod pipeline_logs; pub mod pipelines; pub mod repositories; +pub mod repository_events; pub mod runners; pub mod search; pub mod secrets; diff --git a/backend/src/db/pipelines.rs b/backend/src/db/pipelines.rs index ad007e4..2a6cbd4 100644 --- a/backend/src/db/pipelines.rs +++ b/backend/src/db/pipelines.rs @@ -22,6 +22,8 @@ pub struct NewPipeline<'a> { pub actor_avatar_url: Option<&'a str>, pub git_ref: &'a str, pub trigger_inputs: Option<&'a serde_json::Value>, + /// Pull request number when trigger = 'pull_request'. + pub pr_number: Option, pub timeout_seconds: i32, pub job_timeout_seconds: i32, pub request_id: Option<&'a str>, @@ -55,8 +57,9 @@ pub async fn create( INSERT INTO pipelines (workspace_id, repository_id, workflow_id, workflow_name, workflow_path, number, trigger, triggered_by, commit_sha, commit_message, commit_author, - actor_login, actor_avatar_url, git_ref, trigger_inputs, timeout_seconds) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) + actor_login, actor_avatar_url, git_ref, trigger_inputs, pr_number, + timeout_seconds) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) RETURNING * "#, ) @@ -75,6 +78,7 @@ pub async fn create( .bind(new.actor_avatar_url) .bind(new.git_ref) .bind(new.trigger_inputs) + .bind(new.pr_number) .bind(new.timeout_seconds) .fetch_one(&mut *tx) .await?; @@ -291,6 +295,55 @@ pub async fn finalize( .await } +/// Everything the Checks API reporter needs to address GitHub for one +/// pipeline, resolved in a single join. Returns None when the pipeline (or +/// its repository/installation) is gone — the reporter then no-ops. +#[derive(Debug, sqlx::FromRow)] +pub struct ChecksContext { + pub workflow_name: String, + pub commit_sha: String, + pub trigger: String, + pub conclusion: Option, + pub check_run_id: Option, + pub repo_owner: String, + pub repo_name: String, + pub installation_id: i64, + pub workspace_slug: String, +} + +pub async fn checks_context(pool: &PgPool, id: Uuid) -> sqlx::Result> { + sqlx::query_as::<_, ChecksContext>( + r#" + SELECT p.workflow_name, p.commit_sha, p.trigger, p.conclusion, + p.check_run_id, + r.owner AS repo_owner, r.name AS repo_name, + gi.installation_id, + w.slug AS workspace_slug + FROM pipelines p + JOIN repositories r ON r.id = p.repository_id + JOIN github_installations gi ON gi.id = r.installation_id + JOIN workspaces w ON w.id = p.workspace_id + WHERE p.id = $1 + "#, + ) + .bind(id) + .fetch_optional(pool) + .await +} + +/// Record the GitHub check-run id after a successful create. Guarded so a +/// duplicate create (retry race) never overwrites the first linkage. +pub async fn set_check_run_id(pool: &PgPool, id: Uuid, check_run_id: i64) -> sqlx::Result<()> { + sqlx::query( + "UPDATE pipelines SET check_run_id = $2 WHERE id = $1 AND check_run_id IS NULL", + ) + .bind(id) + .bind(check_run_id) + .execute(pool) + .await?; + Ok(()) +} + /// Pipelines whose wall-clock budget lapsed; the scheduler sweep times /// them out. pub async fn find_timed_out(pool: &PgPool) -> sqlx::Result> { diff --git a/backend/src/db/repository_events.rs b/backend/src/db/repository_events.rs new file mode 100644 index 0000000..87ee412 --- /dev/null +++ b/backend/src/db/repository_events.rs @@ -0,0 +1,146 @@ +//! The immutable per-repository event timeline: one row per processed +//! webhook delivery that concerned a connected repository, recording what +//! arrived and what it caused. `outcome` / `ignored_reason` hold STATIC +//! category strings only (consts in `services/webhook_processor.rs`) — +//! never upstream text. + +use chrono::{DateTime, Utc}; +use sqlx::PgPool; +use uuid::Uuid; + +#[derive(Debug)] +pub struct NewRepositoryEvent<'a> { + pub repository_id: Uuid, + pub delivery_id: &'a str, + pub event: &'a str, + pub action: Option<&'a str>, + pub git_ref: Option<&'a str>, + pub head_sha: Option<&'a str>, + pub actor_login: Option<&'a str>, + pub actor_avatar_url: Option<&'a str>, + /// Static category: pipelines_created | sync_scheduled | + /// pipelines_and_sync | ignored | failed. + pub outcome: &'a str, + /// Static category, only when outcome = 'ignored'. + pub ignored_reason: Option<&'a str>, + pub pipeline_ids: &'a [Uuid], + pub sync_run_id: Option, + /// Server-built, capped JSON (skipped workflows, PR number, merged flag). + pub summary: &'a serde_json::Value, + pub received_at: DateTime, +} + +#[derive(Debug, sqlx::FromRow)] +pub struct RepositoryEventRow { + pub id: Uuid, + pub event: String, + pub action: Option, + pub git_ref: Option, + pub head_sha: Option, + pub actor_login: Option, + pub actor_avatar_url: Option, + pub outcome: String, + pub ignored_reason: Option, + pub pipeline_ids: Vec, + pub sync_run_id: Option, + pub summary: serde_json::Value, + pub received_at: DateTime, + pub processed_at: DateTime, +} + +/// Insert one timeline row inside the delivery-completion transaction. +pub async fn insert( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + ev: &NewRepositoryEvent<'_>, +) -> sqlx::Result { + let (id,): (Uuid,) = sqlx::query_as( + r#" + INSERT INTO repository_events + (repository_id, delivery_id, event, action, git_ref, head_sha, + actor_login, actor_avatar_url, outcome, ignored_reason, + pipeline_ids, sync_run_id, summary, received_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) + RETURNING id + "#, + ) + .bind(ev.repository_id) + .bind(ev.delivery_id) + .bind(ev.event) + .bind(ev.action) + .bind(ev.git_ref) + .bind(ev.head_sha) + .bind(ev.actor_login) + .bind(ev.actor_avatar_url) + .bind(ev.outcome) + .bind(ev.ignored_reason) + .bind(ev.pipeline_ids) + .bind(ev.sync_run_id) + .bind(ev.summary) + .bind(ev.received_at) + .fetch_one(&mut **tx) + .await?; + Ok(id) +} + +/// Keyset-paginated timeline for one repository, newest first. The cursor is +/// the (processed_at, id) tuple of the last row seen (the pipelines-list +/// pattern). +pub async fn list_for_repo( + pool: &PgPool, + repository_id: Uuid, + cursor: Option<(DateTime, Uuid)>, + limit: i64, +) -> sqlx::Result> { + let (cursor_at, cursor_id) = match cursor { + Some((at, id)) => (Some(at), Some(id)), + None => (None, None), + }; + sqlx::query_as::<_, RepositoryEventRow>( + r#" + SELECT id, event, action, git_ref, head_sha, actor_login, actor_avatar_url, + outcome, ignored_reason, pipeline_ids, sync_run_id, summary, + received_at, processed_at + FROM repository_events + WHERE repository_id = $1 + AND ($2::timestamptz IS NULL OR (processed_at, id) < ($2, $3)) + ORDER BY processed_at DESC, id DESC + LIMIT $4 + "#, + ) + .bind(repository_id) + .bind(cursor_at) + .bind(cursor_id) + .bind(limit.clamp(1, 100)) + .fetch_all(pool) + .await +} + +/// Webhook-health aggregate for the repository sync panel: the latest +/// event's timestamp/outcome plus the failure count over the last 24 h. +#[derive(Debug, sqlx::FromRow)] +pub struct EventsOverviewRow { + pub last_event_at: Option>, + pub last_event_outcome: Option, + pub failed_events_24h: i64, +} + +pub async fn overview(pool: &PgPool, repository_id: Uuid) -> sqlx::Result { + sqlx::query_as::<_, EventsOverviewRow>( + r#" + SELECT + (SELECT processed_at FROM repository_events + WHERE repository_id = $1 + ORDER BY processed_at DESC, id DESC LIMIT 1) AS last_event_at, + (SELECT outcome FROM repository_events + WHERE repository_id = $1 + ORDER BY processed_at DESC, id DESC LIMIT 1) AS last_event_outcome, + (SELECT COUNT(*) FROM repository_events + WHERE repository_id = $1 + AND outcome = 'failed' + AND processed_at > now() - interval '24 hours') AS failed_events_24h + "#, + ) + .bind(repository_id) + .fetch_one(pool) + .await +} diff --git a/backend/src/db/webhook_deliveries.rs b/backend/src/db/webhook_deliveries.rs index 5575f35..8fe4023 100644 --- a/backend/src/db/webhook_deliveries.rs +++ b/backend/src/db/webhook_deliveries.rs @@ -1,35 +1,160 @@ +//! The durable webhook queue. The HTTP handler persists a normalized, +//! server-built payload (never the raw body) and acks GitHub immediately; +//! `services/webhook_processor.rs` claims rows one at a time and processes +//! them asynchronously. `delivery_id` (GitHub's `X-GitHub-Delivery`) is the +//! idempotency key — redeliveries keep the original id, so replays collapse. + +use chrono::{DateTime, Utc}; use sqlx::PgPool; -/// Record a delivery id (the idempotency gate). Returns false when this -/// delivery was already processed — GitHub redeliveries become no-ops. +/// One claimed delivery, ready for processing. +#[derive(Debug, sqlx::FromRow)] +pub struct ClaimedDelivery { + pub delivery_id: String, + pub event: String, + pub action: Option, + pub installation_id: Option, + pub github_repo_id: Option, + pub payload: Option, + pub retry_count: i32, + pub received_at: DateTime, +} + +/// Record a delivery (the idempotency gate). Returns false when this +/// delivery was already recorded — GitHub redeliveries become no-ops, EXCEPT +/// that a manual redelivery of a terminally `failed` row revives it for +/// another processing round (GitHub's redelivery keeps the original id). +#[allow(clippy::too_many_arguments)] pub async fn insert( pool: &PgPool, delivery_id: &str, event: &str, action: Option<&str>, installation_id: Option, + github_repo_id: Option, + status: &str, + payload: Option<&serde_json::Value>, ) -> sqlx::Result { let result = sqlx::query( r#" - INSERT INTO webhook_deliveries (delivery_id, event, action, installation_id) - VALUES ($1, $2, $3, $4) - ON CONFLICT (delivery_id) DO NOTHING + INSERT INTO webhook_deliveries + (delivery_id, event, action, installation_id, github_repo_id, status, payload) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (delivery_id) DO UPDATE + SET status = 'pending', retry_count = 0, payload = EXCLUDED.payload + WHERE webhook_deliveries.status = 'failed' "#, ) .bind(delivery_id) .bind(event) .bind(action) .bind(installation_id) + .bind(github_repo_id) + .bind(status) + .bind(payload) .execute(pool) .await?; Ok(result.rows_affected() == 1) } -pub async fn set_status(pool: &PgPool, delivery_id: &str, status: &str) -> sqlx::Result<()> { - sqlx::query("UPDATE webhook_deliveries SET status = $2 WHERE delivery_id = $1") - .bind(delivery_id) - .bind(status) - .execute(pool) - .await?; +/// Claim the oldest pending delivery for processing. `FOR UPDATE SKIP +/// LOCKED` keeps concurrent claimers (multiple backend instances) from +/// blocking each other; the single in-process consumer drains sequentially, +/// which preserves per-repo event ordering. +pub async fn claim_next(pool: &PgPool) -> sqlx::Result> { + sqlx::query_as::<_, ClaimedDelivery>( + r#" + UPDATE webhook_deliveries + SET status = 'processing', last_attempt_at = now() + WHERE delivery_id = ( + SELECT delivery_id FROM webhook_deliveries + WHERE status = 'pending' + ORDER BY received_at, delivery_id + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + RETURNING delivery_id, event, action, installation_id, github_repo_id, + payload, retry_count, received_at + "#, + ) + .fetch_optional(pool) + .await +} + +/// Terminal success ('processed') or deliberate drop ('ignored'). The +/// consumed payload is nulled so the table stays bounded. +pub async fn finish( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + delivery_id: &str, + status: &str, +) -> sqlx::Result<()> { + sqlx::query( + r#" + UPDATE webhook_deliveries + SET status = $2, processed_at = now(), payload = NULL + WHERE delivery_id = $1 + "#, + ) + .bind(delivery_id) + .bind(status) + .execute(&mut **tx) + .await?; Ok(()) } + +/// Processing failed: requeue with a bumped retry count, or park the row as +/// terminally 'failed' once the budget is spent (payload dropped — a manual +/// GitHub redelivery revives it with a fresh payload). +pub async fn record_failure( + pool: &PgPool, + delivery_id: &str, + max_retries: i32, +) -> sqlx::Result<()> { + sqlx::query( + r#" + UPDATE webhook_deliveries + SET retry_count = retry_count + 1, + status = CASE WHEN retry_count + 1 >= $2 THEN 'failed' ELSE 'pending' END, + payload = CASE WHEN retry_count + 1 >= $2 THEN NULL ELSE payload END, + processed_at = CASE WHEN retry_count + 1 >= $2 THEN now() ELSE processed_at END + WHERE delivery_id = $1 + "#, + ) + .bind(delivery_id) + .bind(max_retries) + .execute(pool) + .await?; + Ok(()) +} + +/// Crash recovery: a row stuck in 'processing' past the window (its worker +/// died mid-flight) goes back to 'pending' for the next drain. +pub async fn revert_stuck(pool: &PgPool, older_than_secs: i64) -> sqlx::Result { + let result = sqlx::query( + r#" + UPDATE webhook_deliveries + SET status = 'pending' + WHERE status = 'processing' + AND last_attempt_at < now() - make_interval(secs => $1::double precision) + "#, + ) + .bind(older_than_secs) + .execute(pool) + .await?; + Ok(result.rows_affected()) +} + +/// Live (unprocessed) deliveries attributed to one GitHub repository — the +/// repository health panel's "pending events" figure. +pub async fn pending_count_for_repo(pool: &PgPool, github_repo_id: i64) -> sqlx::Result { + let (count,): (i64,) = sqlx::query_as( + r#" + SELECT COUNT(*) FROM webhook_deliveries + WHERE github_repo_id = $1 AND status IN ('pending', 'processing') + "#, + ) + .bind(github_repo_id) + .fetch_one(pool) + .await?; + Ok(count) +} diff --git a/backend/src/db/workflows.rs b/backend/src/db/workflows.rs index 7d043d8..8f23dbd 100644 --- a/backend/src/db/workflows.rs +++ b/backend/src/db/workflows.rs @@ -213,33 +213,40 @@ pub async fn find_detail_for_workspace( .await } -/// Workflows of one repository that a push event can trigger: they declare -/// the `push` trigger and are not in an error state. -pub async fn push_runnable_for_repo( +/// Workflows of one repository that an event of the given kind can trigger: +/// they declare the event in `triggers` and are not in an error state. +/// Callers pass static event names only ("push", "pull_request"); the +/// per-workflow filter conditions in `metadata.triggerFilters` are evaluated +/// afterwards by `services/trigger_eval.rs`. +pub async fn runnable_for_repo( pool: &PgPool, repository_id: Uuid, -) -> sqlx::Result> { - sqlx::query_as::<_, PushRunnableWorkflow>( + event: &str, +) -> sqlx::Result> { + sqlx::query_as::<_, RunnableWorkflow>( r#" - SELECT id, name, path, raw_content + SELECT id, name, path, raw_content, triggers, metadata FROM workflows WHERE repository_id = $1 - AND 'push' = ANY(triggers) + AND $2 = ANY(triggers) AND validation_status <> 'errors' ORDER BY path "#, ) .bind(repository_id) + .bind(event) .fetch_all(pool) .await } #[derive(Debug, sqlx::FromRow)] -pub struct PushRunnableWorkflow { +pub struct RunnableWorkflow { pub id: Uuid, pub name: String, pub path: String, pub raw_content: String, + pub triggers: Vec, + pub metadata: serde_json::Value, } /// One workflow-YAML reference to a secret/var/environment name, with the diff --git a/backend/src/handlers/github_webhooks.rs b/backend/src/handlers/github_webhooks.rs index 1bf37ac..4e1f38e 100644 --- a/backend/src/handlers/github_webhooks.rs +++ b/backend/src/handlers/github_webhooks.rs @@ -2,6 +2,13 @@ //! CSRF header, no CORS — authentication is the HMAC-SHA256 signature over //! the raw body, verified in constant time before anything is parsed. //! Payload contents and signatures are never logged. +//! +//! The handler does verification, validation, persistence, acknowledgement, +//! and a wake-up poke ONLY (GitHub's 10-second ack budget): the delivery is +//! stored as a normalized, server-built, capped payload in the durable +//! `webhook_deliveries` queue and every side effect — sync scheduling, +//! trigger evaluation, pipeline creation — happens asynchronously in +//! `services/webhook_processor.rs`. use axum::body::Bytes; use axum::extract::State; @@ -11,14 +18,30 @@ use serde::Deserialize; use sha2::Sha256; use crate::db; -use crate::services::{github_app, pipeline_run, repo_sync}; +use crate::services::github_app; use crate::state::AppState; type HmacSha256 = Hmac; const MAX_HEADER_LEN: usize = 256; +/// Caps applied while normalizing the payload: everything stored is +/// server-built and bounded, never the raw body. +const MAX_FIELD_LEN: usize = 512; +const MAX_TITLE_LEN: usize = 200; +const MAX_CHANGED_PATHS: usize = 300; +const MAX_REMOVED_REPOS: usize = 100; -/// Loose envelope: only the fields the dispatcher needs, everything else is +/// Events the background processor handles; everything else is recorded as +/// 'ignored' for observability and dropped. +const PROCESSED_EVENTS: &[&str] = &[ + "push", + "pull_request", + "repository", + "installation", + "installation_repositories", +]; + +/// Loose envelope: only the fields the normalizer needs, everything else is /// ignored by serde. Strong typing per event keeps parsing unambiguous. #[derive(Debug, Deserialize)] struct Envelope { @@ -34,6 +57,12 @@ struct Envelope { after: Option, head_commit: Option, pusher: Option, + /// Push events: per-commit changed files feed the path-filter evaluation. + commits: Option>, + /// Push events: total commit count — when it exceeds `commits.len()` + /// GitHub truncated the list and path filters must fail open. + size: Option, + pull_request: Option, } #[derive(Debug, Deserialize)] @@ -53,6 +82,32 @@ struct PusherRef { name: Option, } +#[derive(Debug, Deserialize)] +struct PushCommitRef { + added: Option>, + removed: Option>, + modified: Option>, +} + +#[derive(Debug, Deserialize)] +struct PullRequestRef { + number: i64, + title: Option, + merged: Option, + draft: Option, + head: Option, + base: Option, + user: Option, +} + +#[derive(Debug, Deserialize)] +struct PrSideRef { + sha: Option, + #[serde(rename = "ref")] + git_ref: Option, + repo: Option, +} + #[derive(Debug, Deserialize)] struct InstallationRef { id: i64, @@ -106,14 +161,25 @@ pub async fn receive( return StatusCode::BAD_REQUEST; }; - // 3. Idempotency gate: replays are acknowledged and dropped. + // 3. Durable persistence — the idempotency gate AND the work queue. + // Replayed deliveries are acknowledged and dropped; events the processor + // handles are stored 'pending' with a normalized, capped payload. let installation_id = envelope.installation.as_ref().map(|i| i.id); + let github_repo_id = envelope.repository.as_ref().map(|r| r.id); + let (status, payload) = if PROCESSED_EVENTS.contains(&event.as_str()) { + ("pending", Some(normalize_payload(&envelope))) + } else { + ("ignored", None) + }; match db::webhook_deliveries::insert( &state.pool, &delivery_id, &event, envelope.action.as_deref(), installation_id, + github_repo_id, + status, + payload.as_ref(), ) .await { @@ -125,16 +191,147 @@ pub async fn receive( } } - // 4. Dispatch. Failures are recorded but still acknowledged with 2xx so - // GitHub does not retry a delivery we have already claimed. - if let Err(error) = dispatch(&state, &event, &envelope).await { - tracing::error!(error = ?error, event, "webhook processing failed"); - let _ = db::webhook_deliveries::set_status(&state.pool, &delivery_id, "failed").await; + // 4. Acknowledge immediately; the background worker does the rest. + if status == "pending" { + state.webhook_processor.poke(); } - StatusCode::ACCEPTED } +/// Build the normalized payload the processor consumes: server-built JSON +/// with every field capped, avatar URLs sanitized, and changed paths +/// aggregated/deduped. The raw body is never stored. +fn normalize_payload(envelope: &Envelope) -> serde_json::Value { + let cap = |s: &str| s.chars().take(MAX_FIELD_LEN).collect::(); + + let mut payload = serde_json::Map::new(); + + if let Some(git_ref) = envelope.git_ref.as_deref() { + payload.insert("ref".into(), cap(git_ref).into()); + } + if let Some(after) = envelope.after.as_deref() { + payload.insert("after".into(), cap(after).into()); + } + if let Some(head) = &envelope.head_commit { + if let Some(id) = head.id.as_deref() { + payload.insert("headCommitSha".into(), cap(id).into()); + } + if let Some(message) = head.message.as_deref() { + // First line is enough, and caps stored size. + payload.insert( + "commitMessage".into(), + cap(message.lines().next().unwrap_or("")).into(), + ); + } + if let Some(name) = head.author.as_ref().and_then(|a| a.name.as_deref()) { + payload.insert("commitAuthor".into(), cap(name).into()); + } + } + if let Some(name) = envelope.pusher.as_ref().and_then(|p| p.name.as_deref()) { + payload.insert("pusherName".into(), cap(name).into()); + } + if let Some(sender) = &envelope.sender { + if !sender.login.is_empty() && sender.login.len() <= 200 { + payload.insert("senderLogin".into(), sender.login.clone().into()); + } + if let Some(avatar) = github_app::sanitize_avatar_url(sender.avatar_url.as_deref()) { + payload.insert("senderAvatarUrl".into(), avatar.into()); + } + } + if let Some(account) = envelope + .installation + .as_ref() + .and_then(|i| i.account.as_ref()) + { + payload.insert("installationAccountLogin".into(), cap(&account.login).into()); + } + if let Some(removed) = &envelope.repositories_removed { + let ids: Vec = removed.iter().take(MAX_REMOVED_REPOS).map(|r| r.id).collect(); + payload.insert("repositoriesRemoved".into(), serde_json::json!(ids)); + } + + // Changed paths for push path-filter evaluation: aggregated across + // commits, deduped, capped. `pathsTruncated` marks an incomplete set — + // the trigger evaluator then fails open. + if let Some(commits) = &envelope.commits { + let mut paths: Vec = Vec::new(); + let mut truncated = envelope.size.is_some_and(|s| s > commits.len() as i64); + 'outer: for commit in commits { + for list in [&commit.added, &commit.removed, &commit.modified] + .into_iter() + .flatten() + { + for path in list { + if path.is_empty() || path.len() > MAX_FIELD_LEN { + continue; + } + if !paths.contains(path) { + if paths.len() >= MAX_CHANGED_PATHS { + truncated = true; + break 'outer; + } + paths.push(path.clone()); + } + } + } + } + payload.insert("changedPaths".into(), serde_json::json!(paths)); + payload.insert("pathsTruncated".into(), truncated.into()); + } + + if let Some(pr) = &envelope.pull_request { + let mut pr_json = serde_json::Map::new(); + pr_json.insert("number".into(), pr.number.into()); + if let Some(title) = pr.title.as_deref() { + pr_json.insert( + "title".into(), + title + .lines() + .next() + .unwrap_or("") + .chars() + .take(MAX_TITLE_LEN) + .collect::() + .into(), + ); + } + if let Some(merged) = pr.merged { + pr_json.insert("merged".into(), merged.into()); + } + if let Some(draft) = pr.draft { + pr_json.insert("draft".into(), draft.into()); + } + if let Some(head) = &pr.head { + if let Some(sha) = head.sha.as_deref() { + pr_json.insert("headSha".into(), cap(sha).into()); + } + if let Some(git_ref) = head.git_ref.as_deref() { + pr_json.insert("headRef".into(), cap(git_ref).into()); + } + if let Some(repo) = &head.repo { + pr_json.insert("headRepoId".into(), repo.id.into()); + } + } + if let Some(base) = &pr.base { + if let Some(git_ref) = base.git_ref.as_deref() { + pr_json.insert("baseRef".into(), cap(git_ref).into()); + } + if let Some(repo) = &base.repo { + pr_json.insert("baseRepoId".into(), repo.id.into()); + } + } + if let Some(user) = &pr.user + && !user.login.is_empty() + && user.login.len() <= 200 + { + pr_json.insert("userLogin".into(), user.login.clone().into()); + } + payload.insert("pullRequest".into(), serde_json::Value::Object(pr_json)); + } + + serde_json::Value::Object(payload) +} + fn header_str(headers: &HeaderMap, name: &str) -> Option { let value = headers.get(name)?.to_str().ok()?; if value.is_empty() || value.len() > MAX_HEADER_LEN { @@ -166,198 +363,104 @@ fn verify_signature(secret: &str, headers: &HeaderMap, body: &[u8]) -> bool { mac.verify_slice(&expected).is_ok() } -async fn dispatch(state: &AppState, event: &str, envelope: &Envelope) -> anyhow::Result<()> { - let action = envelope.action.as_deref().unwrap_or(""); - match event { - // A push to a connected repository's default branch re-syncs it. - // The sync itself diffs blob shas, so this stays cheap even when - // no workflow file changed. - "push" => { - let (Some(repository), Some(git_ref)) = (&envelope.repository, &envelope.git_ref) - else { - return Ok(()); - }; - let Some(repo) = - db::repositories::find_by_github_id(&state.pool, repository.id).await? - else { - return Ok(()); - }; - if *git_ref == format!("refs/heads/{}", repo.default_branch) { - repo_sync::schedule(state, repo.id, "webhook").await?; - } - // Pipelines trigger for any branch push carrying a real head - // commit (branch deletions send after = 0000...). - if git_ref.starts_with("refs/heads/") { - trigger_push_pipelines(state, &repo, envelope, git_ref).await?; - } - } - "repository" => { - let Some(repository) = &envelope.repository else { - return Ok(()); - }; - match action { - "deleted" => { - db::repositories::mark_failed( - &state.pool, - &[repository.id], - "repository deleted on github", - ) - .await?; - } - "renamed" | "edited" | "privatized" | "publicized" | "transferred" => { - if let Some(repo) = - db::repositories::find_by_github_id(&state.pool, repository.id).await? - { - repo_sync::schedule(state, repo.id, "webhook").await?; - } - } - _ => {} - } - } - "installation" => { - let Some(installation) = &envelope.installation else { - return Ok(()); - }; - match action { - // Recorded so an org install can be claimed via the setup - // redirect by the member who performed it. - "created" => { - let account = installation - .account - .as_ref() - .map(|a| a.login.as_str()) - .unwrap_or(""); - let sender = envelope - .sender - .as_ref() - .map(|s| s.login.as_str()) - .unwrap_or(""); - if !account.is_empty() && !sender.is_empty() { - db::github_installations::record_created_event( - &state.pool, - installation.id, - account, - sender, - ) - .await?; - } - } - "deleted" => { - state.github_app.evict_token(installation.id).await; - db::github_installations::delete_by_installation_id( - &state.pool, - installation.id, - ) - .await?; - } - "suspend" => { - state.github_app.evict_token(installation.id).await; - db::github_installations::set_suspended(&state.pool, installation.id, true) - .await?; - } - "unsuspend" => { - db::github_installations::set_suspended(&state.pool, installation.id, false) - .await?; - } - _ => {} - } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn event_routing_table() { + for event in PROCESSED_EVENTS { + assert!(PROCESSED_EVENTS.contains(event)); } - "installation_repositories" => { - if action == "removed" - && let Some(removed) = &envelope.repositories_removed - { - let ids: Vec = removed.iter().map(|r| r.id).collect(); - if !ids.is_empty() { - db::repositories::mark_failed(&state.pool, &ids, "access revoked").await?; - } - } + for event in ["create", "delete", "check_suite", "workflow_run", "star"] { + assert!(!PROCESSED_EVENTS.contains(&event)); } - // Recorded (step 3) for observability. Extension points: pipeline - // triggers for pull_request / tag events land here. - "pull_request" | "create" | "delete" => {} - _ => {} } - Ok(()) -} -/// Create one pipeline per push-triggered workflow of this repository. -/// Failures are per-workflow: one bad workflow never blocks the others. -async fn trigger_push_pipelines( - state: &AppState, - repo: &crate::models::repository::Repository, - envelope: &Envelope, - git_ref: &str, -) -> anyhow::Result<()> { - let commit_sha = envelope - .after - .as_deref() - .or(envelope - .head_commit - .as_ref() - .and_then(|c| c.id.as_deref())) - .unwrap_or(""); - // Branch deletions push an all-zero SHA; nothing to build. - if commit_sha.is_empty() || commit_sha.chars().all(|c| c == '0') { - return Ok(()); + #[test] + fn normalize_caps_and_aggregates_changed_paths() { + let body = serde_json::json!({ + "ref": "refs/heads/main", + "after": "abc123", + "size": 2, + "head_commit": { + "id": "abc123", + "message": "first line\nsecond line", + "author": { "name": "Dev" } + }, + "sender": { "login": "octocat", "avatar_url": "https://avatars.githubusercontent.com/u/1" }, + "commits": [ + { "added": ["src/a.rs"], "modified": ["src/a.rs", "README.md"], "removed": [] }, + { "added": [], "modified": ["docs/x.md"], "removed": ["old.txt"] } + ] + }); + let envelope: Envelope = serde_json::from_value(body).unwrap(); + let payload = normalize_payload(&envelope); + assert_eq!(payload["commitMessage"], "first line"); + assert_eq!(payload["senderLogin"], "octocat"); + let paths: Vec<&str> = payload["changedPaths"] + .as_array() + .unwrap() + .iter() + .map(|p| p.as_str().unwrap()) + .collect(); + // Aggregation order: per commit, added → removed → modified. + assert_eq!(paths, vec!["src/a.rs", "README.md", "old.txt", "docs/x.md"]); + assert_eq!(payload["pathsTruncated"], false); } - let commit_message = envelope - .head_commit - .as_ref() - .and_then(|c| c.message.as_deref()) - // First line is enough, and caps stored size. - .map(|m| m.lines().next().unwrap_or("").to_string()); - let commit_author = envelope - .head_commit - .as_ref() - .and_then(|c| c.author.as_ref()) - .and_then(|a| a.name.as_deref()) - .or(envelope.pusher.as_ref().and_then(|p| p.name.as_deref())); - - // Actor identity snapshot: the webhook sender is who pushed. Untrusted - // upstream text — cap the login, https-check the avatar, else NULL. - let actor_login = envelope - .sender - .as_ref() - .map(|s| s.login.as_str()) - .filter(|l| !l.is_empty() && l.len() <= 200); - let actor_avatar_url = github_app::sanitize_avatar_url( - envelope.sender.as_ref().and_then(|s| s.avatar_url.as_deref()), - ); - - let workflows = db::workflows::push_runnable_for_repo(&state.pool, repo.id).await?; - for workflow in workflows { - let ctx = pipeline_run::TriggerContext { - trigger: "push", - triggered_by: None, - commit_sha, - commit_message: commit_message.as_deref(), - commit_author, - actor_login, - actor_avatar_url, - git_ref, - inputs: None, - request_id: None, - }; - if let Err(error) = pipeline_run::create_pipeline( - state, - repo, - workflow.id, - &workflow.name, - &workflow.path, - &workflow.raw_content, - &ctx, - ) - .await - { - // Validation failures (e.g. uses:-only workflows) are expected; - // they must not fail the whole delivery. - tracing::debug!( - workflow = %workflow.path, - error = ?error, - "push did not trigger a pipeline for this workflow" - ); - } + #[test] + fn normalize_flags_truncated_pushes() { + // size > commits.len(): GitHub truncated the commit list. + let body = serde_json::json!({ + "ref": "refs/heads/main", + "size": 50, + "commits": [ { "added": ["a.txt"] } ] + }); + let envelope: Envelope = serde_json::from_value(body).unwrap(); + let payload = normalize_payload(&envelope); + assert_eq!(payload["pathsTruncated"], true); + + // Path-cap overflow also flags truncation. + let many: Vec = (0..(MAX_CHANGED_PATHS + 10)) + .map(|i| format!("file-{i}.rs")) + .collect(); + let body = serde_json::json!({ + "ref": "refs/heads/main", + "size": 1, + "commits": [ { "added": many } ] + }); + let envelope: Envelope = serde_json::from_value(body).unwrap(); + let payload = normalize_payload(&envelope); + assert_eq!( + payload["changedPaths"].as_array().unwrap().len(), + MAX_CHANGED_PATHS + ); + assert_eq!(payload["pathsTruncated"], true); + } + + #[test] + fn normalize_extracts_pull_request_context() { + let body = serde_json::json!({ + "action": "opened", + "pull_request": { + "number": 42, + "title": "Add feature\nwith details", + "merged": false, + "draft": false, + "head": { "sha": "headsha", "ref": "feature/x", "repo": { "id": 7 } }, + "base": { "sha": "basesha", "ref": "main", "repo": { "id": 7 } }, + "user": { "login": "contributor" } + } + }); + let envelope: Envelope = serde_json::from_value(body).unwrap(); + let payload = normalize_payload(&envelope); + let pr = &payload["pullRequest"]; + assert_eq!(pr["number"], 42); + assert_eq!(pr["title"], "Add feature"); + assert_eq!(pr["headSha"], "headsha"); + assert_eq!(pr["headRepoId"], 7); + assert_eq!(pr["baseRef"], "main"); + assert_eq!(pr["userLogin"], "contributor"); } - Ok(()) } diff --git a/backend/src/handlers/pipelines.rs b/backend/src/handlers/pipelines.rs index 33a2a19..60caf93 100644 --- a/backend/src/handlers/pipelines.rs +++ b/backend/src/handlers/pipelines.rs @@ -114,7 +114,7 @@ fn build_list_filter( let trigger = match query.trigger.as_deref() { None | Some("") => None, - Some(t @ ("push" | "manual")) => Some(t.to_string()), + Some(t @ ("push" | "manual" | "pull_request" | "tag")) => Some(t.to_string()), Some(_) => return Err(AppError::Validation("invalid trigger filter".into())), }; @@ -260,6 +260,8 @@ pub async fn rerun( git_ref: &original.git_ref, // Reruns reproduce the original run, inputs included. inputs: original.trigger_inputs.as_ref(), + // A rerun of a PR pipeline keeps its PR association visible. + pr_number: original.pr_number, request_id, }; let pipeline = pipeline_run::create_pipeline( @@ -492,6 +494,7 @@ pub async fn dispatch( actor_avatar_url: github_app::sanitize_avatar_url(user.avatar_url.as_deref()), git_ref: &git_ref, inputs: inputs.as_ref(), + pr_number: None, request_id, }; let pipeline = pipeline_run::create_pipeline( @@ -794,6 +797,13 @@ mod tests { let mut query = empty_query(); query.trigger = Some("cron".into()); assert!(build_list_filter(&query, None, 50).is_err()); + + for trigger in ["push", "manual", "pull_request", "tag"] { + let mut query = empty_query(); + query.trigger = Some(trigger.into()); + let filter = build_list_filter(&query, None, 50).unwrap(); + assert_eq!(filter.trigger.as_deref(), Some(trigger)); + } } #[test] diff --git a/backend/src/handlers/repositories.rs b/backend/src/handlers/repositories.rs index 0eb9656..e775104 100644 --- a/backend/src/handlers/repositories.rs +++ b/backend/src/handlers/repositories.rs @@ -12,7 +12,8 @@ use crate::db; use crate::error::{AppError, AppResult}; use crate::middleware::auth::CurrentUser; use crate::models::repository::{ - AvailableRepoResponse, BranchResponse, RepositoryResponse, SyncRunResponse, + AvailableRepoResponse, BranchResponse, RepositoryEventResponse, RepositoryHealthResponse, + RepositoryResponse, SyncRunResponse, }; use crate::models::workflow::WorkflowSummaryResponse; use crate::services::workspace_hub::WorkspaceEvent; @@ -217,15 +218,72 @@ pub async fn detail( .map(SyncRunResponse::from) .collect::>(); + // Webhook/sync health for the sync status panel: latest event, recent + // failures, and the live queue depth for this repository. + let overview = db::repository_events::overview(&state.pool, repository.id).await?; + let pending_deliveries = + db::webhook_deliveries::pending_count_for_repo(&state.pool, repository.github_repo_id) + .await?; + let health = RepositoryHealthResponse { + last_event_at: overview.last_event_at, + last_event_outcome: overview.last_event_outcome, + failed_events_24h: overview.failed_events_24h, + pending_deliveries, + checks_enabled: state.config.github_checks_enabled, + }; + let workflow_count = workflows.len() as i64; Ok(Json(json!({ "repository": RepositoryResponse::from_row(repository, workflow_count), "branches": branches, "workflows": workflows, "syncRuns": sync_runs, + "health": health, }))) } +#[derive(Debug, Deserialize)] +pub struct EventsQuery { + cursor: Option, + limit: Option, +} + +/// GET /api/workspaces/{workspace_id}/repositories/{repository_id}/events +/// +/// Keyset-paginated repository event timeline, newest first. The cursor is +/// the shared `~` shape from the pipelines list. +pub async fn events( + State(state): State, + CurrentUser(user): CurrentUser, + Path((workspace_id, repository_id)): Path<(Uuid, Uuid)>, + axum::extract::Query(query): axum::extract::Query, +) -> AppResult> { + authz::require_permission(&state.pool, user.id, workspace_id, authz::CONTENT_READ).await?; + + let repository = db::repositories::find_for_workspace(&state.pool, workspace_id, repository_id) + .await? + .ok_or(AppError::NotFound)?; + + let cursor = match query.cursor.as_deref() { + None | Some("") => None, + Some(raw) => Some(super::pipelines::parse_cursor(raw)?), + }; + let limit = query.limit.unwrap_or(30).clamp(1, 100); + + let rows = + db::repository_events::list_for_repo(&state.pool, repository.id, cursor, limit).await?; + let next_cursor = (rows.len() as i64 == limit) + .then(|| rows.last()) + .flatten() + .map(|row| super::pipelines::format_cursor(row.processed_at, row.id)); + let events = rows + .into_iter() + .map(RepositoryEventResponse::from) + .collect::>(); + + Ok(Json(json!({ "events": events, "nextCursor": next_cursor }))) +} + /// POST /api/workspaces/{workspace_id}/repositories/{repository_id}/sync pub async fn sync( State(state): State, @@ -239,7 +297,10 @@ pub async fn sync( .await? .ok_or(AppError::NotFound)?; - if !repo_sync::schedule(&state, repository_id, "manual").await? { + if repo_sync::schedule(&state, repository_id, "manual") + .await? + .is_none() + { return Err(AppError::Conflict("a sync is already running")); } diff --git a/backend/src/main.rs b/backend/src/main.rs index 8a10abf..a2c2672 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -99,6 +99,11 @@ async fn main() -> anyhow::Result<()> { // cursor and materializes per-user Notification Center rows. tokio::spawn(services::notification_projector::run(state.clone())); + // Webhook processor: drains the durable webhook_deliveries queue — + // sync scheduling, trigger evaluation, and pipeline creation all happen + // here, off the HTTP ack path. + tokio::spawn(services::webhook_processor::run(state.clone())); + let router = routes::build_router(state)?; let listener = tokio::net::TcpListener::bind(&config.bind_addr) diff --git a/backend/src/models/pipeline.rs b/backend/src/models/pipeline.rs index 8c73ffc..420bc95 100644 --- a/backend/src/models/pipeline.rs +++ b/backend/src/models/pipeline.rs @@ -22,6 +22,13 @@ pub struct Pipeline { pub actor_avatar_url: Option, pub git_ref: String, pub trigger_inputs: Option, + /// Pull request number when trigger = 'pull_request'. + pub pr_number: Option, + /// GitHub check-run id once Checks reporting created one. Internal + /// routing state — never serialized into API responses (the checks + /// reporter reads it through db::pipelines::ChecksContext). + #[allow(dead_code)] + pub check_run_id: Option, pub status: String, pub conclusion: Option, #[allow(dead_code)] // enforced in SQL sweeps, mapped for completeness @@ -58,6 +65,7 @@ pub struct PipelineResponse { pub git_ref: String, #[serde(skip_serializing_if = "Option::is_none")] pub trigger_inputs: Option, + pub pr_number: Option, pub status: String, pub conclusion: Option, pub created_at: DateTime, @@ -83,6 +91,7 @@ impl PipelineResponse { actor_avatar_url: pipeline.actor_avatar_url, git_ref: pipeline.git_ref, trigger_inputs: pipeline.trigger_inputs, + pr_number: pipeline.pr_number, status: pipeline.status, conclusion: pipeline.conclusion, created_at: pipeline.created_at, diff --git a/backend/src/models/repository.rs b/backend/src/models/repository.rs index c541038..571e29e 100644 --- a/backend/src/models/repository.rs +++ b/backend/src/models/repository.rs @@ -163,3 +163,57 @@ impl From for SyncRunResponse { } } } + +/// One entry of the repository event timeline. `outcome`/`ignoredReason` +/// are static category strings; `summary` is server-built, capped JSON. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct RepositoryEventResponse { + pub id: Uuid, + pub event: String, + pub action: Option, + pub git_ref: Option, + pub head_sha: Option, + pub actor_login: Option, + pub actor_avatar_url: Option, + pub outcome: String, + pub ignored_reason: Option, + pub pipeline_ids: Vec, + pub sync_run_id: Option, + pub summary: serde_json::Value, + pub received_at: DateTime, + pub processed_at: DateTime, +} + +impl From for RepositoryEventResponse { + fn from(row: crate::db::repository_events::RepositoryEventRow) -> Self { + Self { + id: row.id, + event: row.event, + action: row.action, + git_ref: row.git_ref, + head_sha: row.head_sha, + actor_login: row.actor_login, + actor_avatar_url: row.actor_avatar_url, + outcome: row.outcome, + ignored_reason: row.ignored_reason, + pipeline_ids: row.pipeline_ids, + sync_run_id: row.sync_run_id, + summary: row.summary, + received_at: row.received_at, + processed_at: row.processed_at, + } + } +} + +/// Webhook/sync health for the repository detail sync panel. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct RepositoryHealthResponse { + pub last_event_at: Option>, + /// Static outcome category of the latest processed event. + pub last_event_outcome: Option, + pub failed_events_24h: i64, + pub pending_deliveries: i64, + pub checks_enabled: bool, +} diff --git a/backend/src/routes/mod.rs b/backend/src/routes/mod.rs index 1499cc7..96e38a2 100644 --- a/backend/src/routes/mod.rs +++ b/backend/src/routes/mod.rs @@ -249,6 +249,10 @@ pub fn build_router(state: AppState) -> anyhow::Result { "/workspaces/{workspace_id}/repositories/{repository_id}/sync", post(repositories::sync).layer(GovernorLayer::new(sync_governor)), ) + .route( + "/workspaces/{workspace_id}/repositories/{repository_id}/events", + get(repositories::events), + ) .route( "/workspaces/{workspace_id}/workflows", get(workflows::list), diff --git a/backend/src/services/github_app.rs b/backend/src/services/github_app.rs index 5f824ee..e709d99 100644 --- a/backend/src/services/github_app.rs +++ b/backend/src/services/github_app.rs @@ -29,6 +29,9 @@ pub struct GitHubApp { client_id: String, encoding_key: EncodingKey, token_cache: RwLock>, + /// Separate cache for checks-scoped tokens: the sync token stays + /// read-only (least privilege) and never grows write permissions. + checks_token_cache: RwLock>, } struct CachedToken { @@ -75,6 +78,7 @@ impl GitHubApp { client_id, encoding_key, token_cache: RwLock::new(HashMap::new()), + checks_token_cache: RwLock::new(HashMap::new()), }) } @@ -97,10 +101,47 @@ impl GitHubApp { &self, http: &reqwest::Client, installation_id: i64, + ) -> anyhow::Result { + self.mint_scoped_token( + http, + installation_id, + &self.token_cache, + // Least privilege: the token can never do more than read. + &serde_json::json!({ + "permissions": { "contents": "read", "metadata": "read" } + }), + ) + .await + } + + /// Checks-scoped installation token for the Checks API reporter. Minted + /// separately so the sync token stays read-only; requires the GitHub App + /// to have the Checks (Read & write) permission — GitHub answers 422 + /// when it doesn't, which the reporter treats as "checks unavailable". + pub async fn checks_token( + &self, + http: &reqwest::Client, + installation_id: i64, + ) -> anyhow::Result { + self.mint_scoped_token( + http, + installation_id, + &self.checks_token_cache, + &serde_json::json!({ "permissions": { "checks": "write" } }), + ) + .await + } + + async fn mint_scoped_token( + &self, + http: &reqwest::Client, + installation_id: i64, + cache: &RwLock>, + permissions: &serde_json::Value, ) -> anyhow::Result { let refresh_after = Utc::now() + Duration::minutes(TOKEN_REFRESH_MARGIN_MINUTES); { - let cache = self.token_cache.read().await; + let cache = cache.read().await; if let Some(cached) = cache.get(&installation_id) && cached.expires_at > refresh_after { @@ -116,10 +157,7 @@ impl GitHubApp { .bearer_auth(&jwt) .header("Accept", "application/vnd.github+json") .header("X-GitHub-Api-Version", "2022-11-28") - // Least privilege: the token can never do more than read. - .json(&serde_json::json!({ - "permissions": { "contents": "read", "metadata": "read" } - })) + .json(permissions) .send() .await .context("installation token request failed")?; @@ -137,7 +175,7 @@ impl GitHubApp { .context("installation token response was malformed")?; let token = minted.token.clone(); - self.token_cache.write().await.insert( + cache.write().await.insert( installation_id, CachedToken { token: minted.token, @@ -166,6 +204,10 @@ impl GitHubApp { pub async fn evict_token(&self, installation_id: i64) { self.token_cache.write().await.remove(&installation_id); + self.checks_token_cache + .write() + .await + .remove(&installation_id); } } diff --git a/backend/src/services/github_checks.rs b/backend/src/services/github_checks.rs new file mode 100644 index 0000000..1371a9c --- /dev/null +++ b/backend/src/services/github_checks.rs @@ -0,0 +1,366 @@ +//! GitHub Checks API reporting: pipelines triggered by repository events +//! (push / pull_request / tag) surface as check runs on their commit — +//! queued at creation, in_progress on first job start, completed with a +//! mapped conclusion and a details link back to the pipeline page. +//! +//! Everything here is best-effort and spawned off the caller's path: a +//! reporting failure can never fail, delay, or retry a pipeline. The check +//! payloads are server-built from static templates and job counts — never +//! runner or upstream text. Manual dispatches deliberately do not report +//! (only event-triggered pipelines represent repository state on GitHub). +//! +//! Requires the GitHub App's Checks (Read & write) permission. When the app +//! lacks it (403/422 from GitHub) the installation is marked unavailable +//! for an hour with a single edge-triggered warning — the operator action +//! (grant the permission, approve on installations) is logged once, not +//! sprayed per pipeline. + +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +use tokio::sync::RwLock; +use uuid::Uuid; + +use crate::db; +use crate::services::github_app; +use crate::state::AppState; + +const GITHUB_API: &str = "https://api.github.com"; +/// A permission-denied installation is retried after this long. +const UNAVAILABLE_RETRY: Duration = Duration::from_secs(3600); +/// Triggers that report to GitHub. Manual dispatch/rerun never does. +const REPORTED_TRIGGERS: &[&str] = &["push", "pull_request", "tag"]; + +/// Per-installation availability cache; held in AppState. +#[derive(Default)] +pub struct GithubChecks { + unavailable: RwLock>, +} + +impl GithubChecks { + async fn is_unavailable(&self, installation_id: i64) -> bool { + let mut map = self.unavailable.write().await; + match map.get(&installation_id) { + Some(marked) if marked.elapsed() < UNAVAILABLE_RETRY => true, + Some(_) => { + map.remove(&installation_id); + false + } + None => false, + } + } + + /// Returns true when this installation was newly marked — the caller + /// warns exactly once per outage edge. + async fn mark_unavailable(&self, installation_id: i64) -> bool { + self.unavailable + .write() + .await + .insert(installation_id, Instant::now()) + .is_none() + } +} + +#[derive(Clone, Copy)] +enum Phase { + Create, + Started, + Completed, +} + +/// Report a queued check run for a freshly created pipeline. +pub fn spawn_create(state: &AppState, pipeline_id: Uuid) { + spawn(state, pipeline_id, Phase::Create); +} + +/// Flip the check run to in_progress (first job started). +pub fn spawn_started(state: &AppState, pipeline_id: Uuid) { + spawn(state, pipeline_id, Phase::Started); +} + +/// Complete the check run with the pipeline's conclusion. +pub fn spawn_completed(state: &AppState, pipeline_id: Uuid) { + spawn(state, pipeline_id, Phase::Completed); +} + +fn spawn(state: &AppState, pipeline_id: Uuid, phase: Phase) { + if !state.config.github_checks_enabled { + return; + } + let state = state.clone(); + tokio::spawn(async move { + report(&state, pipeline_id, phase).await; + }); +} + +/// Map an overup pipeline conclusion onto GitHub's check-run vocabulary. +/// `partial` (some jobs succeeded, some failed) reads as failure on GitHub, +/// with the summary carrying the nuance. +pub fn map_conclusion(conclusion: &str) -> &'static str { + match conclusion { + "success" => "success", + "cancelled" => "cancelled", + "timed_out" => "timed_out", + // failure, partial, and anything unexpected fail closed. + _ => "failure", + } +} + +async fn report(state: &AppState, pipeline_id: Uuid, phase: Phase) { + let ctx = match db::pipelines::checks_context(&state.pool, pipeline_id).await { + Ok(Some(ctx)) => ctx, + Ok(None) => return, // pipeline/repository gone — nothing to report + Err(error) => { + tracing::debug!(error = ?error, %pipeline_id, "checks context lookup failed"); + return; + } + }; + if !REPORTED_TRIGGERS.contains(&ctx.trigger.as_str()) { + return; + } + // Defense-in-depth before URL interpolation: owner/name are our own + // synced metadata, but they still must pass the segment allow-list. + if !github_app::is_safe_name_segment(&ctx.repo_owner) + || !github_app::is_safe_name_segment(&ctx.repo_name) + { + return; + } + // The in_progress flip needs the created check run; completion instead + // falls back to creating the run directly in `completed` status — a fast + // pipeline can finish before the async create round-trip persists the id, + // and skipping would leave the GitHub check stuck at `queued` forever. + if matches!(phase, Phase::Started) && ctx.check_run_id.is_none() { + return; + } + if state.github_checks.is_unavailable(ctx.installation_id).await { + return; + } + + let token = match state + .github_app + .checks_token(&state.http, ctx.installation_id) + .await + { + Ok(token) => token, + Err(error) => { + // Minting fails with 422 when the app lacks the Checks + // permission; the error text carries the status only. + let newly = state + .github_checks + .mark_unavailable(ctx.installation_id) + .await; + if newly { + tracing::warn!( + installation_id = ctx.installation_id, + error = ?error, + "GitHub checks reporting unavailable — grant the App the \ + Checks (Read & write) permission and approve it on the \ + installation; retrying hourly" + ); + } + return; + } + }; + + let details_url = format!( + "{}/w/{}/pipelines/{}", + state.config.frontend_url, ctx.workspace_slug, pipeline_id + ); + let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + + let (method_url, body) = match phase { + Phase::Create => ( + ( + reqwest::Method::POST, + format!( + "{GITHUB_API}/repos/{}/{}/check-runs", + ctx.repo_owner, ctx.repo_name + ), + ), + serde_json::json!({ + "name": format!("overup / {}", ctx.workflow_name), + "head_sha": ctx.commit_sha, + "status": "queued", + "external_id": pipeline_id.to_string(), + "details_url": details_url, + "output": { + "title": "Pipeline queued", + "summary": "The pipeline is queued on overup. Follow the details link for live logs.", + }, + }), + ), + Phase::Started => ( + ( + reqwest::Method::PATCH, + format!( + "{GITHUB_API}/repos/{}/{}/check-runs/{}", + ctx.repo_owner, + ctx.repo_name, + ctx.check_run_id.unwrap_or_default() + ), + ), + serde_json::json!({ + "status": "in_progress", + "started_at": now, + "details_url": details_url, + "output": { + "title": "Pipeline running", + "summary": "The pipeline is executing on overup. Follow the details link for live logs.", + }, + }), + ), + Phase::Completed => { + let conclusion = ctx.conclusion.as_deref().unwrap_or("failure"); + let (title, summary) = match completion_output(state, pipeline_id, conclusion).await { + Some(output) => output, + None => return, + }; + let mut body = serde_json::json!({ + "status": "completed", + "conclusion": map_conclusion(conclusion), + "completed_at": now, + "details_url": details_url, + "output": { "title": title, "summary": summary }, + }); + match ctx.check_run_id { + Some(check_run_id) => ( + ( + reqwest::Method::PATCH, + format!( + "{GITHUB_API}/repos/{}/{}/check-runs/{}", + ctx.repo_owner, ctx.repo_name, check_run_id + ), + ), + body, + ), + // Fallback create: same name/head_sha/external_id as the + // Create phase, so a late-landing queued run is superseded. + None => { + body["name"] = + serde_json::Value::from(format!("overup / {}", ctx.workflow_name)); + body["head_sha"] = serde_json::Value::from(ctx.commit_sha.clone()); + body["external_id"] = serde_json::Value::from(pipeline_id.to_string()); + ( + ( + reqwest::Method::POST, + format!( + "{GITHUB_API}/repos/{}/{}/check-runs", + ctx.repo_owner, ctx.repo_name + ), + ), + body, + ) + } + } + } + }; + + let response = state + .http + .request(method_url.0, &method_url.1) + .bearer_auth(&token) + .header("Accept", "application/vnd.github+json") + .header("X-GitHub-Api-Version", "2022-11-28") + .json(&body) + .send() + .await; + + match response { + Ok(response) if response.status().is_success() => { + // Both create paths (queued create and the completed fallback) + // persist the returned id; the UPDATE is guarded on NULL so a + // racing late create can never clobber it. + if ctx.check_run_id.is_none() { + #[derive(serde::Deserialize)] + struct CreatedCheckRun { + id: i64, + } + if let Ok(created) = response.json::().await + && let Err(error) = + db::pipelines::set_check_run_id(&state.pool, pipeline_id, created.id) + .await + { + tracing::debug!(error = ?error, %pipeline_id, "failed to store check run id"); + } + } + } + Ok(response) + if response.status() == reqwest::StatusCode::FORBIDDEN + || response.status() == reqwest::StatusCode::UNPROCESSABLE_ENTITY => + { + let newly = state + .github_checks + .mark_unavailable(ctx.installation_id) + .await; + if newly { + tracing::warn!( + installation_id = ctx.installation_id, + status = %response.status(), + "GitHub checks reporting unavailable — grant the App the \ + Checks (Read & write) permission and approve it on the \ + installation; retrying hourly" + ); + } + } + Ok(response) => { + // Status only — the body could echo request details. + tracing::debug!(status = %response.status(), %pipeline_id, "check run report rejected"); + } + Err(error) => { + tracing::debug!(error = ?error, %pipeline_id, "check run report failed"); + } + } +} + +/// Static title + summary from server-side job counts only. +async fn completion_output( + state: &AppState, + pipeline_id: Uuid, + conclusion: &str, +) -> Option<(String, String)> { + let statuses = db::pipeline_jobs::statuses_for_pipeline(&state.pool, pipeline_id) + .await + .ok()?; + let total = statuses.len(); + let count = |c: &str| { + statuses + .iter() + .filter(|(_, conclusion)| conclusion.as_deref() == Some(c)) + .count() + }; + let succeeded = count("success"); + let failed = count("failure") + count("timed_out"); + let title = match conclusion { + "success" => "Pipeline succeeded".to_string(), + "cancelled" => "Pipeline cancelled".to_string(), + "timed_out" => "Pipeline timed out".to_string(), + "partial" => "Pipeline partially succeeded".to_string(), + _ => "Pipeline failed".to_string(), + }; + let mut summary = format!("{succeeded} of {total} jobs succeeded."); + if failed > 0 { + summary.push_str(&format!(" {failed} failed.")); + } + if conclusion == "partial" { + summary.push_str(" Some jobs succeeded while others failed (reported as failure)."); + } + summary.push_str(" Full logs and artifacts are on overup."); + Some((title, summary)) +} + +#[cfg(test)] +mod tests { + use super::map_conclusion; + + #[test] + fn conclusion_mapping() { + assert_eq!(map_conclusion("success"), "success"); + assert_eq!(map_conclusion("failure"), "failure"); + assert_eq!(map_conclusion("cancelled"), "cancelled"); + assert_eq!(map_conclusion("timed_out"), "timed_out"); + // partial reads as failure on GitHub; the summary carries the nuance. + assert_eq!(map_conclusion("partial"), "failure"); + // Anything unexpected fails closed. + assert_eq!(map_conclusion("bogus"), "failure"); + } +} diff --git a/backend/src/services/mod.rs b/backend/src/services/mod.rs index 9e63fe5..fe52036 100644 --- a/backend/src/services/mod.rs +++ b/backend/src/services/mod.rs @@ -3,6 +3,7 @@ pub mod auth_flow; pub mod authz; pub mod github; pub mod github_app; +pub mod github_checks; pub mod image_sniff; pub mod janitor; pub mod log_archive; @@ -24,6 +25,8 @@ pub mod scheduler; pub mod search_indexer; pub mod secrets_crypto; pub mod session; +pub mod trigger_eval; +pub mod webhook_processor; pub mod workflow_parse; pub mod workspace; pub mod workspace_hub; diff --git a/backend/src/services/pipeline_run.rs b/backend/src/services/pipeline_run.rs index 6ce51ad..8d434b7 100644 --- a/backend/src/services/pipeline_run.rs +++ b/backend/src/services/pipeline_run.rs @@ -34,6 +34,8 @@ pub struct TriggerContext<'a> { /// Always an object; values are strings/numbers/booleans, already checked /// against the workflow's parsed input definitions by the handler. pub inputs: Option<&'a serde_json::Value>, + /// Pull request number (pull_request trigger; reruns copy the original's). + pub pr_number: Option, pub request_id: Option<&'a str>, } @@ -138,6 +140,7 @@ pub async fn create_pipeline( actor_avatar_url: ctx.actor_avatar_url, git_ref: ctx.git_ref, trigger_inputs: ctx.inputs, + pr_number: ctx.pr_number, timeout_seconds: state.config.pipeline_timeout_seconds, job_timeout_seconds: state.config.job_timeout_seconds, request_id: ctx.request_id, @@ -265,6 +268,8 @@ pub async fn on_job_started(state: &AppState, job: &PipelineJob) -> sqlx::Result ) .await?; publish_pipeline(state, &pipeline); + // Report in_progress to GitHub (best-effort, spawned). + crate::services::github_checks::spawn_started(state, pipeline.id); } record_event( state, @@ -536,6 +541,10 @@ pub async fn maybe_finalize(state: &AppState, pipeline_id: Uuid) -> sqlx::Result ) .await?; publish_pipeline(state, &pipeline); + // Report the terminal conclusion to GitHub (best-effort, spawned). This + // is the single terminal convergence point, so cancel/timeout/partial all + // report through here. + crate::services::github_checks::spawn_completed(state, pipeline.id); sqlx::query( r#" diff --git a/backend/src/services/repo_sync.rs b/backend/src/services/repo_sync.rs index 94ba7c4..c001d8e 100644 --- a/backend/src/services/repo_sync.rs +++ b/backend/src/services/repo_sync.rs @@ -17,12 +17,17 @@ use crate::state::AppState; /// Never parse more workflow files than this per repository. const MAX_WORKFLOW_FILES: usize = 50; -/// Claim the repository and spawn a background sync. Returns false when a -/// sync is already running (callers map that to 409). -pub async fn schedule(state: &AppState, repository_id: Uuid, trigger: &str) -> sqlx::Result { +/// Claim the repository and spawn a background sync. Returns the sync-run id, +/// or None when a sync is already running (callers map that to 409; the +/// webhook processor links the id into the repository event timeline). +pub async fn schedule( + state: &AppState, + repository_id: Uuid, + trigger: &str, +) -> sqlx::Result> { let Some(repository) = db::repositories::claim_for_sync(&state.pool, repository_id).await? else { - return Ok(false); + return Ok(None); }; let run_id = db::repositories::insert_sync_run(&state.pool, repository_id, trigger).await?; @@ -65,7 +70,7 @@ pub async fn schedule(state: &AppState, repository_id: Uuid, trigger: &str) -> s } }); - Ok(true) + Ok(Some(run_id)) } /// Static failure category + full server-side detail. diff --git a/backend/src/services/trigger_eval.rs b/backend/src/services/trigger_eval.rs new file mode 100644 index 0000000..ae578fe --- /dev/null +++ b/backend/src/services/trigger_eval.rs @@ -0,0 +1,695 @@ +//! Dispatch-time trigger evaluation: does a repository event satisfy a +//! workflow's `on:` filter conditions? Pure functions over the parser's +//! stored `metadata.triggerFilters` — no I/O, no expression evaluation. +//! Semantics mirror GitHub Actions' documented filter rules: +//! +//! - glob patterns support `*` (no `/`), `**` (crosses `/`), `?` / `+` +//! (zero-or-one / one-or-more of the preceding character), `[abc]` / +//! `[a-z]` classes, and `\` escapes; +//! - filter lists are ORDERED: a matching `!`-negated pattern after a +//! positive match excludes the value, a later positive match re-includes +//! it; +//! - with only `tags` defined a branch push never runs (and vice versa); +//! - branch AND path dimensions must BOTH pass; paths never apply to tag +//! pushes; +//! - `pull_request` defaults to the `opened|synchronize|reopened` activity +//! types, and its branch filters match the BASE branch; +//! - unavailable changed-file data (`changed_paths: None`) fails OPEN: path +//! filters pass, matching GitHub's behavior when a diff can't be computed. +//! +//! Everything here is bounded: patterns are capped at parse time (50 per +//! list, 256 bytes each) and re-capped defensively on read, values at 1 KB, +//! and the matcher is iterative (position-set DP, no recursion, no regex). + +use serde_json::Value; + +/// Defensive read-time caps mirroring the parser's write-time caps +/// (`workflow_parse::MAX_FILTER_PATTERNS`) — hand-edited metadata must not +/// widen the budget. +const MAX_PATTERNS: usize = 50; +const MAX_PATTERN_LEN: usize = 256; +const MAX_VALUE_LEN: usize = 1024; + +/// GitHub's default `pull_request` activity types when `types:` is absent. +const DEFAULT_PR_TYPES: &[&str] = &["opened", "synchronize", "reopened"]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Decision { + Run, + /// Static skip category for the repository-event timeline. + Skip(&'static str), +} + +pub const SKIP_EVENT_NOT_DECLARED: &str = "event_not_declared"; +pub const SKIP_BRANCH_FILTERED: &str = "branch_filtered"; +pub const SKIP_TAG_FILTERED: &str = "tag_filtered"; +pub const SKIP_PATH_FILTERED: &str = "path_filtered"; +pub const SKIP_TYPE_FILTERED: &str = "type_filtered"; + +#[derive(Debug, Clone, Copy)] +pub enum EventKind<'a> { + Push { branch: &'a str }, + TagPush { tag: &'a str }, + PullRequest { action: &'a str, base_branch: &'a str }, +} + +#[derive(Debug, Clone, Copy)] +pub struct EventContext<'a> { + pub kind: EventKind<'a>, + /// `None` means the changed-file set is unknown (truncated push payload, + /// pull_request event) — path filters then pass (fail-open). + pub changed_paths: Option<&'a [String]>, +} + +/// Evaluate one workflow against one repository event. `triggers` is the +/// stored `workflows.triggers` event-name array; `metadata` the stored +/// `workflows.metadata` JSONB (any missing/malformed `triggerFilters` shape — +/// pre-v3 rows — degrades to "declared event runs unfiltered"). +pub fn evaluate(triggers: &[String], metadata: &Value, ctx: &EventContext) -> Decision { + let declared_event = match ctx.kind { + EventKind::Push { .. } | EventKind::TagPush { .. } => "push", + EventKind::PullRequest { .. } => "pull_request", + }; + if !triggers.iter().any(|t| t == declared_event) { + return Decision::Skip(SKIP_EVENT_NOT_DECLARED); + } + + let filters = &metadata["triggerFilters"]; + + match ctx.kind { + EventKind::Push { branch } => { + let push = &filters["push"]; + let branches = patterns(push, "branches"); + let branches_ignore = patterns(push, "branchesIgnore"); + let tags = patterns(push, "tags"); + let tags_ignore = patterns(push, "tagsIgnore"); + + if !branches.is_empty() { + // Defense-in-depth: when both forms are somehow present + // (the parser flags it as an error), `branches` wins. + if !match_ordered(&branches, branch) { + return Decision::Skip(SKIP_BRANCH_FILTERED); + } + } else if !branches_ignore.is_empty() { + if match_ordered(&branches_ignore, branch) { + return Decision::Skip(SKIP_BRANCH_FILTERED); + } + } else if !tags.is_empty() || !tags_ignore.is_empty() { + // Only the tag dimension is filtered: branch pushes never run. + return Decision::Skip(SKIP_BRANCH_FILTERED); + } + + evaluate_paths(push, ctx.changed_paths) + } + EventKind::TagPush { tag } => { + let push = &filters["push"]; + let branches = patterns(push, "branches"); + let branches_ignore = patterns(push, "branchesIgnore"); + let tags = patterns(push, "tags"); + let tags_ignore = patterns(push, "tagsIgnore"); + + if !tags.is_empty() { + if !match_ordered(&tags, tag) { + return Decision::Skip(SKIP_TAG_FILTERED); + } + } else if !tags_ignore.is_empty() { + if match_ordered(&tags_ignore, tag) { + return Decision::Skip(SKIP_TAG_FILTERED); + } + } else if !branches.is_empty() || !branches_ignore.is_empty() { + // Only the branch dimension is filtered: tag pushes never run. + return Decision::Skip(SKIP_TAG_FILTERED); + } + + // Path filters deliberately never apply to tag pushes. + Decision::Run + } + EventKind::PullRequest { action, base_branch } => { + let pr = &filters["pullRequest"]; + + let types = patterns(pr, "types"); + let type_matches = if types.is_empty() { + DEFAULT_PR_TYPES.contains(&action) + } else { + types.iter().any(|t| t == action) + }; + if !type_matches { + return Decision::Skip(SKIP_TYPE_FILTERED); + } + + let branches = patterns(pr, "branches"); + let branches_ignore = patterns(pr, "branchesIgnore"); + if !branches.is_empty() { + if !match_ordered(&branches, base_branch) { + return Decision::Skip(SKIP_BRANCH_FILTERED); + } + } else if !branches_ignore.is_empty() && match_ordered(&branches_ignore, base_branch) { + return Decision::Skip(SKIP_BRANCH_FILTERED); + } + + evaluate_paths(pr, ctx.changed_paths) + } + } +} + +/// The path dimension shared by branch pushes and pull requests: with +/// `paths`, at least one changed file must match; with `paths-ignore`, at +/// least one changed file must NOT match. Unknown changed files pass. +fn evaluate_paths(filters: &Value, changed_paths: Option<&[String]>) -> Decision { + let paths = patterns(filters, "paths"); + let paths_ignore = patterns(filters, "pathsIgnore"); + if paths.is_empty() && paths_ignore.is_empty() { + return Decision::Run; + } + let Some(changed) = changed_paths else { + // Fail-open: the diff is unknown (truncated payload / PR event). + return Decision::Run; + }; + let run = if !paths.is_empty() { + // Defense-in-depth: `paths` wins when both forms are present. + changed.iter().any(|p| match_ordered(&paths, p)) + } else { + changed.iter().any(|p| !match_ordered(&paths_ignore, p)) + }; + if run { + Decision::Run + } else { + Decision::Skip(SKIP_PATH_FILTERED) + } +} + +/// One filter list out of stored metadata: strings only, re-capped. +fn patterns(filters: &Value, key: &str) -> Vec { + filters[key] + .as_array() + .map(|list| { + list.iter() + .filter_map(Value::as_str) + .filter(|p| !p.is_empty() && p.len() <= MAX_PATTERN_LEN) + .map(str::to_string) + .take(MAX_PATTERNS) + .collect() + }) + .unwrap_or_default() +} + +/// Ordered filter-list matching with `!` negation: patterns apply in order +/// and the LAST matching pattern decides. A list with no positive pattern +/// matches nothing (GitHub rejects such lists outright; failing closed here +/// is the safe mirror). +pub fn match_ordered(patterns: &[String], value: &str) -> bool { + let mut matched = false; + for pattern in patterns { + if let Some(negated) = pattern.strip_prefix('!') { + if matched && glob_match(negated, value) { + matched = false; + } + } else if glob_match(pattern, value) { + matched = true; + } + } + matched +} + +#[derive(Debug, Clone, PartialEq)] +enum Tok { + Lit(char), + /// `[abc]` / `[a-z]`: inclusive ranges (single chars are (c, c)). + Class(Vec<(char, char)>), + /// `*`: any run not crossing `/`. + AnyNoSlash, + /// `**`: any run. + AnyAll, +} + +#[derive(Debug, Clone, Copy, PartialEq)] +enum Quant { + One, + /// `?`: zero or one of the preceding token. + ZeroOrOne, + /// `+`: one or more of the preceding token. + OneOrMore, +} + +/// GitHub-flavored filter glob. Iterative position-set matching (the classic +/// NFA-simulation shape): O(pattern × value), bounded by the caps above — +/// pathological patterns cannot blow up. +pub fn glob_match(pattern: &str, value: &str) -> bool { + if pattern.len() > MAX_PATTERN_LEN || value.len() > MAX_VALUE_LEN { + return false; + } + let tokens = tokenize(pattern); + let value: Vec = value.chars().collect(); + + // positions[i] == true → the tokens consumed so far can end at value[i]. + let mut positions = vec![false; value.len() + 1]; + positions[0] = true; + + for (tok, quant) in &tokens { + let mut next = vec![false; value.len() + 1]; + match tok { + Tok::AnyNoSlash | Tok::AnyAll => { + // Quantifiers on a star are meaningless (a star is already a + // closure) — treat as the star itself. + let cross_slash = *tok == Tok::AnyAll; + for start in 0..positions.len() { + if !positions[start] { + continue; + } + next[start] = true; + for (offset, c) in value[start..].iter().enumerate() { + if !cross_slash && *c == '/' { + break; + } + next[start + offset + 1] = true; + } + } + } + Tok::Lit(_) | Tok::Class(_) => { + let matches_at = |pos: usize| -> bool { + value.get(pos).is_some_and(|c| match tok { + Tok::Lit(l) => c == l, + Tok::Class(ranges) => ranges.iter().any(|(lo, hi)| *lo <= *c && *c <= *hi), + _ => unreachable!(), + }) + }; + for start in 0..positions.len() { + if !positions[start] { + continue; + } + match quant { + Quant::One => { + if matches_at(start) { + next[start + 1] = true; + } + } + Quant::ZeroOrOne => { + next[start] = true; + if matches_at(start) { + next[start + 1] = true; + } + } + Quant::OneOrMore => { + let mut pos = start; + while matches_at(pos) { + pos += 1; + next[pos] = true; + } + } + } + } + } + } + positions = next; + if !positions.iter().any(|p| *p) { + return false; + } + } + + positions[value.len()] +} + +/// Pattern → token stream. Malformed constructs degrade to literals (an +/// unclosed `[` matches a literal `[`), matching glob conventions — never a +/// panic, never an error. +fn tokenize(pattern: &str) -> Vec<(Tok, Quant)> { + let chars: Vec = pattern.chars().collect(); + let mut tokens: Vec<(Tok, Quant)> = Vec::new(); + let mut i = 0; + while i < chars.len() { + match chars[i] { + '\\' => { + // Escape: next char is a literal; a trailing `\` is itself. + if i + 1 < chars.len() { + tokens.push((Tok::Lit(chars[i + 1]), Quant::One)); + i += 2; + } else { + tokens.push((Tok::Lit('\\'), Quant::One)); + i += 1; + } + } + '*' => { + if chars.get(i + 1) == Some(&'*') { + tokens.push((Tok::AnyAll, Quant::One)); + i += 2; + } else { + tokens.push((Tok::AnyNoSlash, Quant::One)); + i += 1; + } + } + '[' => match parse_class(&chars[i + 1..]) { + Some((ranges, consumed)) => { + tokens.push((Tok::Class(ranges), Quant::One)); + i += consumed + 1; + } + None => { + tokens.push((Tok::Lit('['), Quant::One)); + i += 1; + } + }, + '?' => { + // Quantifies the preceding token; leading `?` is a literal. + match tokens.last_mut() { + Some(last) if last.1 == Quant::One => last.1 = Quant::ZeroOrOne, + _ => tokens.push((Tok::Lit('?'), Quant::One)), + } + i += 1; + } + '+' => { + match tokens.last_mut() { + Some(last) if last.1 == Quant::One => last.1 = Quant::OneOrMore, + _ => tokens.push((Tok::Lit('+'), Quant::One)), + } + i += 1; + } + c => { + tokens.push((Tok::Lit(c), Quant::One)); + i += 1; + } + } + } + tokens +} + +/// `[...]` class body: single chars and `a-z` ranges up to the closing `]`. +/// Returns the ranges and how many chars (incl. `]`) were consumed, or None +/// when unclosed/empty. +fn parse_class(rest: &[char]) -> Option<(Vec<(char, char)>, usize)> { + let mut ranges = Vec::new(); + let mut i = 0; + while i < rest.len() { + match rest[i] { + ']' => { + return if ranges.is_empty() { + None + } else { + Some((ranges, i + 1)) + }; + } + c => { + if rest.get(i + 1) == Some(&'-') + && rest.get(i + 2).is_some_and(|end| *end != ']') + { + let end = rest[i + 2]; + if c <= end { + ranges.push((c, end)); + } + i += 3; + } else { + ranges.push((c, c)); + i += 1; + } + } + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + fn strs(patterns: &[&str]) -> Vec { + patterns.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn glob_matches_github_documented_examples() { + // feature/* — `*` does not cross `/`. + assert!(glob_match("feature/*", "feature/my-branch")); + assert!(glob_match("feature/*", "feature/your-branch")); + assert!(!glob_match("feature/*", "feature/beta-a/my-branch")); + // feature/** — `**` crosses `/`. + assert!(glob_match("feature/**", "feature/beta-a/my-branch")); + assert!(glob_match("feature/**", "feature/my-branch")); + // v2.* — dot is literal. + assert!(glob_match("v2.*", "v2.0")); + assert!(glob_match("v2.*", "v2.9")); + assert!(!glob_match("v2.*", "v3.0")); + // Character classes with `+`. + assert!(glob_match("v[12].[0-9]+.[0-9]+", "v1.10.1")); + assert!(glob_match("v[12].[0-9]+.[0-9]+", "v2.0.0")); + assert!(!glob_match("v[12].[0-9]+.[0-9]+", "v3.0.0")); + assert!(!glob_match("v[12].[0-9]+.[0-9]+", "v1..0")); + // Bare stars. + assert!(glob_match("*", "main")); + assert!(!glob_match("*", "releases/x")); + assert!(glob_match("**", "releases/x/y")); + // Path globs. + assert!(glob_match("*.js", "app.js")); + assert!(!glob_match("*.js", "src/app.js")); + assert!(glob_match("**.js", "src/app.js")); + assert!(glob_match("src/**", "src/a/b.rs")); + } + + #[test] + fn glob_quantifiers_apply_to_preceding_char() { + // `?`: zero or one of the preceding character. + assert!(glob_match("releases/v?", "releases/v")); + assert!(glob_match("releases/v?", "releases/")); + assert!(!glob_match("releases/v?", "releases/vv")); + // `+`: one or more of the preceding character. + assert!(glob_match("ab+", "ab")); + assert!(glob_match("ab+", "abbb")); + assert!(!glob_match("ab+", "a")); + // Leading `?` / `+` degrade to literals. + assert!(glob_match("?x", "?x")); + assert!(glob_match("+y", "+y")); + } + + #[test] + fn glob_escapes_and_malformed_classes() { + assert!(glob_match("\\*literal", "*literal")); + assert!(!glob_match("\\*literal", "xliteral")); + assert!(glob_match("a\\[b", "a[b")); + // Unclosed class is a literal `[`. + assert!(glob_match("a[bc", "a[bc")); + // Trailing escape is a literal backslash. + assert!(glob_match("end\\", "end\\")); + } + + #[test] + fn glob_rejects_oversized_inputs() { + let long_value = "x".repeat(MAX_VALUE_LEN + 1); + assert!(!glob_match("**", &long_value)); + let long_pattern = "x".repeat(MAX_PATTERN_LEN + 1); + assert!(!glob_match(&long_pattern, "x")); + } + + #[test] + fn ordered_negation_last_match_wins() { + let patterns = strs(&["releases/**", "!releases/**-alpha"]); + assert!(match_ordered(&patterns, "releases/10")); + assert!(match_ordered(&patterns, "releases/beta/mona")); + assert!(!match_ordered(&patterns, "releases/10-alpha")); + // A later positive re-includes. + let patterns = strs(&["releases/**", "!releases/**-alpha", "releases/v1-alpha"]); + assert!(match_ordered(&patterns, "releases/v1-alpha")); + // Only-negative lists match nothing (GitHub rejects them; fail closed). + assert!(!match_ordered(&strs(&["!main"]), "dev")); + } + + fn meta(filters: serde_json::Value) -> serde_json::Value { + serde_json::json!({ "triggerFilters": filters }) + } + + fn push_ctx<'a>(branch: &'a str, changed: Option<&'a [String]>) -> EventContext<'a> { + EventContext { + kind: EventKind::Push { branch }, + changed_paths: changed, + } + } + + #[test] + fn evaluate_requires_declared_event() { + let triggers = vec!["pull_request".to_string()]; + assert_eq!( + evaluate(&triggers, &meta(serde_json::json!({})), &push_ctx("main", None)), + Decision::Skip(SKIP_EVENT_NOT_DECLARED) + ); + } + + #[test] + fn evaluate_pre_v3_metadata_runs_unfiltered() { + let triggers = vec!["push".to_string()]; + // No triggerFilters key at all (pre-v3 rows). + assert_eq!( + evaluate(&triggers, &serde_json::json!({}), &push_ctx("anything", None)), + Decision::Run + ); + // Malformed shape. + assert_eq!( + evaluate( + &triggers, + &serde_json::json!({ "triggerFilters": "bogus" }), + &push_ctx("anything", None) + ), + Decision::Run + ); + } + + #[test] + fn evaluate_push_branch_filters() { + let triggers = vec!["push".to_string()]; + let m = meta(serde_json::json!({ + "push": { "branches": ["main", "releases/**", "!releases/**-alpha"] } + })); + assert_eq!(evaluate(&triggers, &m, &push_ctx("main", None)), Decision::Run); + assert_eq!( + evaluate(&triggers, &m, &push_ctx("releases/10", None)), + Decision::Run + ); + assert_eq!( + evaluate(&triggers, &m, &push_ctx("releases/10-alpha", None)), + Decision::Skip(SKIP_BRANCH_FILTERED) + ); + assert_eq!( + evaluate(&triggers, &m, &push_ctx("dev", None)), + Decision::Skip(SKIP_BRANCH_FILTERED) + ); + + let m = meta(serde_json::json!({ "push": { "branchesIgnore": ["dev/*"] } })); + assert_eq!( + evaluate(&triggers, &m, &push_ctx("dev/x", None)), + Decision::Skip(SKIP_BRANCH_FILTERED) + ); + assert_eq!(evaluate(&triggers, &m, &push_ctx("main", None)), Decision::Run); + } + + #[test] + fn evaluate_only_tags_blocks_branch_pushes_and_vice_versa() { + let triggers = vec!["push".to_string()]; + let tags_only = meta(serde_json::json!({ "push": { "tags": ["v*"] } })); + assert_eq!( + evaluate(&triggers, &tags_only, &push_ctx("main", None)), + Decision::Skip(SKIP_BRANCH_FILTERED) + ); + let tag_ctx = EventContext { + kind: EventKind::TagPush { tag: "v1.2" }, + changed_paths: None, + }; + assert_eq!(evaluate(&triggers, &tags_only, &tag_ctx), Decision::Run); + + let branches_only = meta(serde_json::json!({ "push": { "branches": ["main"] } })); + assert_eq!( + evaluate(&triggers, &branches_only, &tag_ctx), + Decision::Skip(SKIP_TAG_FILTERED) + ); + // Neither dimension filtered: both run. + let unfiltered = meta(serde_json::json!({ "push": {} })); + assert_eq!(evaluate(&triggers, &unfiltered, &tag_ctx), Decision::Run); + assert_eq!( + evaluate(&triggers, &unfiltered, &push_ctx("main", None)), + Decision::Run + ); + } + + #[test] + fn evaluate_tag_filters() { + let triggers = vec!["push".to_string()]; + let m = meta(serde_json::json!({ "push": { "tags": ["v[0-9]+.*"] } })); + let ctx = |tag| EventContext { + kind: EventKind::TagPush { tag }, + changed_paths: None, + }; + assert_eq!(evaluate(&triggers, &m, &ctx("v1.0")), Decision::Run); + assert_eq!( + evaluate(&triggers, &m, &ctx("latest")), + Decision::Skip(SKIP_TAG_FILTERED) + ); + // tags-ignore. + let m = meta(serde_json::json!({ "push": { "tagsIgnore": ["nightly-*"] } })); + assert_eq!( + evaluate(&triggers, &m, &ctx("nightly-2026")), + Decision::Skip(SKIP_TAG_FILTERED) + ); + assert_eq!(evaluate(&triggers, &m, &ctx("v1.0")), Decision::Run); + } + + #[test] + fn evaluate_paths_require_a_matching_changed_file() { + let triggers = vec!["push".to_string()]; + let m = meta(serde_json::json!({ + "push": { "branches": ["main"], "paths": ["src/**", "*.toml"] } + })); + let changed = vec!["src/lib.rs".to_string(), "README.md".to_string()]; + assert_eq!( + evaluate(&triggers, &m, &push_ctx("main", Some(&changed))), + Decision::Run + ); + let docs_only = vec!["docs/guide.md".to_string()]; + assert_eq!( + evaluate(&triggers, &m, &push_ctx("main", Some(&docs_only))), + Decision::Skip(SKIP_PATH_FILTERED) + ); + // Branch AND path must both pass. + assert_eq!( + evaluate(&triggers, &m, &push_ctx("dev", Some(&changed))), + Decision::Skip(SKIP_BRANCH_FILTERED) + ); + // Unknown diff fails open. + assert_eq!( + evaluate(&triggers, &m, &push_ctx("main", None)), + Decision::Run + ); + // Empty diff with a paths filter skips. + let empty: Vec = Vec::new(); + assert_eq!( + evaluate(&triggers, &m, &push_ctx("main", Some(&empty))), + Decision::Skip(SKIP_PATH_FILTERED) + ); + } + + #[test] + fn evaluate_paths_ignore_needs_one_unignored_file() { + let triggers = vec!["push".to_string()]; + let m = meta(serde_json::json!({ "push": { "pathsIgnore": ["docs/**"] } })); + let docs_only = vec!["docs/a.md".to_string(), "docs/b.md".to_string()]; + assert_eq!( + evaluate(&triggers, &m, &push_ctx("main", Some(&docs_only))), + Decision::Skip(SKIP_PATH_FILTERED) + ); + let mixed = vec!["docs/a.md".to_string(), "src/main.rs".to_string()]; + assert_eq!( + evaluate(&triggers, &m, &push_ctx("main", Some(&mixed))), + Decision::Run + ); + } + + #[test] + fn evaluate_pull_request_types_and_base_branch() { + let triggers = vec!["pull_request".to_string()]; + let pr_ctx = |action, base| EventContext { + kind: EventKind::PullRequest { + action, + base_branch: base, + }, + changed_paths: None, + }; + // Default types. + let m = meta(serde_json::json!({ "pullRequest": {} })); + assert_eq!(evaluate(&triggers, &m, &pr_ctx("opened", "main")), Decision::Run); + assert_eq!( + evaluate(&triggers, &m, &pr_ctx("synchronize", "main")), + Decision::Run + ); + assert_eq!( + evaluate(&triggers, &m, &pr_ctx("labeled", "main")), + Decision::Skip(SKIP_TYPE_FILTERED) + ); + // Explicit types replace the defaults. + let m = meta(serde_json::json!({ "pullRequest": { "types": ["closed"] } })); + assert_eq!( + evaluate(&triggers, &m, &pr_ctx("opened", "main")), + Decision::Skip(SKIP_TYPE_FILTERED) + ); + assert_eq!(evaluate(&triggers, &m, &pr_ctx("closed", "main")), Decision::Run); + // Branch filters match the BASE branch. + let m = meta(serde_json::json!({ "pullRequest": { "branches": ["main"] } })); + assert_eq!(evaluate(&triggers, &m, &pr_ctx("opened", "main")), Decision::Run); + assert_eq!( + evaluate(&triggers, &m, &pr_ctx("opened", "dev")), + Decision::Skip(SKIP_BRANCH_FILTERED) + ); + } +} diff --git a/backend/src/services/webhook_processor.rs b/backend/src/services/webhook_processor.rs new file mode 100644 index 0000000..cc47c6c --- /dev/null +++ b/backend/src/services/webhook_processor.rs @@ -0,0 +1,859 @@ +//! Asynchronous webhook processor. +//! +//! The HTTP receiver (`handlers/github_webhooks.rs`) only verifies, +//! validates, persists, and acks — every side effect happens here, off the +//! request path (GitHub's 10-second ack budget). The loop mirrors the +//! notification projector (Notify poke + tick backstop) but claims queue +//! rows individually with guarded status UPDATEs: a single sequential +//! consumer per process, so two pushes to the same repository can never be +//! processed out of order. Failures retry up to `MAX_RETRIES`, stuck +//! 'processing' rows revert on the tick (crash recovery), and every +//! completion commits the repository-event timeline row and the delivery +//! status flip in ONE transaction. +//! +//! Everything persisted here is a static category string or server-built +//! JSON — never upstream text. + +use std::collections::HashSet; +use std::time::Duration; + +use tokio::sync::Notify; +use uuid::Uuid; + +use crate::db; +use crate::db::webhook_deliveries::ClaimedDelivery; +use crate::models::repository::Repository; +use crate::services::{github_checks, pipeline_run, repo_sync, trigger_eval}; +use crate::state::AppState; + +/// Coalesce webhook bursts before draining. +const DEBOUNCE: Duration = Duration::from_millis(200); +/// Safety tick: drains pending rows even if every poke was lost. +const TICK: Duration = Duration::from_secs(2); +/// A delivery whose processing fails re-queues this many times before the +/// row parks as terminally 'failed'. +const MAX_RETRIES: i32 = 3; +/// 'processing' rows older than this revert to 'pending' (crash recovery). +const STUCK_SECS: i64 = 300; +/// Cap on per-workflow skip entries recorded into the event summary. +const MAX_SKIPPED_SUMMARY: usize = 50; + +// Outcome vocabulary (CHECK-constrained in SQL). +const OUTCOME_PIPELINES: &str = "pipelines_created"; +const OUTCOME_SYNC: &str = "sync_scheduled"; +const OUTCOME_BOTH: &str = "pipelines_and_sync"; +const OUTCOME_IGNORED: &str = "ignored"; +const OUTCOME_FAILED: &str = "failed"; + +// Static ignored_reason categories (never upstream text). +const REASON_NO_MATCHING_WORKFLOWS: &str = "no_matching_workflows"; +const REASON_FILTERS_NOT_MATCHED: &str = "filters_not_matched"; +const REASON_BRANCH_DELETED: &str = "branch_deleted"; +const REASON_TAG_DELETED: &str = "tag_deleted"; +const REASON_NO_HEAD_COMMIT: &str = "no_head_commit"; +const REASON_FORK_PR_SKIPPED: &str = "fork_pr_skipped"; +const REASON_PR_ACTION_IGNORED: &str = "pr_action_ignored"; +const REASON_PR_CLOSED: &str = "pr_closed"; +const REASON_EVENT_NOT_SUPPORTED: &str = "event_not_supported"; +const REASON_REPOSITORY_DELETED: &str = "repository_deleted"; +const REASON_ACCESS_REVOKED: &str = "access_revoked"; + +/// Wake handle held in `AppState`; pure signal, no payload — the queue in +/// Postgres is the only source of truth for what needs processing. +#[derive(Default)] +pub struct WebhookProcessor { + notify: Notify, +} + +impl WebhookProcessor { + /// Queue a drain; cheap and callable from anywhere. + pub fn poke(&self) { + self.notify.notify_one(); + } +} + +/// The processor loop, spawned once from main. +pub async fn run(state: AppState) { + let mut tick = tokio::time::interval(TICK); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + loop { + tokio::select! { + _ = state.webhook_processor.notify.notified() => { + tokio::time::sleep(DEBOUNCE).await; + } + _ = tick.tick() => { + // Crash recovery rides the tick, never the hot poke path. + if let Err(error) = + db::webhook_deliveries::revert_stuck(&state.pool, STUCK_SECS).await + { + tracing::warn!(error = ?error, "failed to revert stuck webhook deliveries"); + } + } + } + // Drain sequentially until the queue is empty. Sequential processing + // is the per-repo ordering guarantee. + loop { + match db::webhook_deliveries::claim_next(&state.pool).await { + Ok(Some(delivery)) => process(&state, delivery).await, + Ok(None) => break, + Err(error) => { + tracing::warn!(error = ?error, "webhook delivery claim failed; will retry"); + break; + } + } + } + } +} + +/// One repository-event timeline row to be committed with the delivery. +struct EventRow { + repository_id: Uuid, + git_ref: Option, + head_sha: Option, + outcome: &'static str, + ignored_reason: Option<&'static str>, + pipeline_ids: Vec, + sync_run_id: Option, + summary: serde_json::Value, +} + +/// What processing one delivery concluded: the terminal delivery status and +/// any timeline rows for connected repositories. +struct Completion { + delivery_status: &'static str, // 'processed' | 'ignored' + events: Vec, +} + +impl Completion { + fn processed(events: Vec) -> Self { + Self { + delivery_status: "processed", + events, + } + } + + fn ignored() -> Self { + Self { + delivery_status: "ignored", + events: Vec::new(), + } + } +} + +/// Process one claimed delivery end to end, then commit its completion. +async fn process(state: &AppState, delivery: ClaimedDelivery) { + match process_delivery(state, &delivery).await { + Ok(completion) => { + if let Err(error) = commit_completion(state, &delivery, &completion).await { + tracing::warn!( + error = ?error, + event = %delivery.event, + "failed to commit webhook completion; delivery will retry" + ); + let _ = db::webhook_deliveries::record_failure( + &state.pool, + &delivery.delivery_id, + MAX_RETRIES, + ) + .await; + if delivery.retry_count + 1 >= MAX_RETRIES { + record_failed_event(state, &delivery).await; + } + } + } + Err(error) => { + tracing::warn!( + error = ?error, + event = %delivery.event, + retry = delivery.retry_count, + "webhook processing failed" + ); + let _ = db::webhook_deliveries::record_failure( + &state.pool, + &delivery.delivery_id, + MAX_RETRIES, + ) + .await; + // Terminal failure: surface it on the repository timeline + // (best-effort — the delivery row is the authority). + if delivery.retry_count + 1 >= MAX_RETRIES { + record_failed_event(state, &delivery).await; + } + } + } +} + +/// Timeline rows + delivery status flip in one transaction, then post-commit +/// check-run creation for any new pipelines. +async fn commit_completion( + state: &AppState, + delivery: &ClaimedDelivery, + completion: &Completion, +) -> anyhow::Result<()> { + let actor_login = payload_str(delivery, "senderLogin"); + let actor_avatar = payload_str(delivery, "senderAvatarUrl"); + + let mut tx = state.pool.begin().await?; + for event in &completion.events { + db::repository_events::insert( + &mut tx, + &db::repository_events::NewRepositoryEvent { + repository_id: event.repository_id, + delivery_id: &delivery.delivery_id, + event: &delivery.event, + action: delivery.action.as_deref(), + git_ref: event.git_ref.as_deref(), + head_sha: event.head_sha.as_deref(), + actor_login, + actor_avatar_url: actor_avatar, + outcome: event.outcome, + ignored_reason: event.ignored_reason, + pipeline_ids: &event.pipeline_ids, + sync_run_id: event.sync_run_id, + summary: &event.summary, + received_at: delivery.received_at, + }, + ) + .await?; + } + db::webhook_deliveries::finish(&mut tx, &delivery.delivery_id, completion.delivery_status) + .await?; + tx.commit().await?; + Ok(()) +} + +/// Best-effort 'failed' timeline row for a terminally failed delivery. +async fn record_failed_event(state: &AppState, delivery: &ClaimedDelivery) { + let Some(github_repo_id) = delivery.github_repo_id else { + return; + }; + let Ok(Some(repo)) = db::repositories::find_by_github_id(&state.pool, github_repo_id).await + else { + return; + }; + let mut tx = match state.pool.begin().await { + Ok(tx) => tx, + Err(_) => return, + }; + let inserted = db::repository_events::insert( + &mut tx, + &db::repository_events::NewRepositoryEvent { + repository_id: repo.id, + delivery_id: &delivery.delivery_id, + event: &delivery.event, + action: delivery.action.as_deref(), + git_ref: payload_str(delivery, "ref"), + head_sha: payload_str(delivery, "after"), + actor_login: payload_str(delivery, "senderLogin"), + actor_avatar_url: payload_str(delivery, "senderAvatarUrl"), + outcome: OUTCOME_FAILED, + ignored_reason: None, + pipeline_ids: &[], + sync_run_id: None, + summary: &serde_json::json!({}), + received_at: delivery.received_at, + }, + ) + .await; + if inserted.is_ok() { + let _ = tx.commit().await; + } +} + +fn payload_str<'a>(delivery: &'a ClaimedDelivery, key: &str) -> Option<&'a str> { + delivery.payload.as_ref()?.get(key)?.as_str() +} + +/// Route one delivery. Unconnected repositories are dropped before any +/// further processing (the delivery records as 'ignored', no timeline row). +async fn process_delivery( + state: &AppState, + delivery: &ClaimedDelivery, +) -> anyhow::Result { + let action = delivery.action.as_deref().unwrap_or(""); + match delivery.event.as_str() { + "push" => process_push(state, delivery).await, + "pull_request" => process_pull_request(state, delivery, action).await, + "repository" => process_repository(state, delivery, action).await, + "installation" => process_installation(state, delivery, action).await, + "installation_repositories" => { + process_installation_repositories(state, delivery, action).await + } + _ => Ok(Completion::ignored()), + } +} + +/// An all-zero SHA marks a ref deletion push. +fn is_zero_sha(sha: &str) -> bool { + sha.is_empty() || sha.chars().all(|c| c == '0') +} + +async fn process_push( + state: &AppState, + delivery: &ClaimedDelivery, +) -> anyhow::Result { + let Some(github_repo_id) = delivery.github_repo_id else { + return Ok(Completion::ignored()); + }; + let Some(repo) = db::repositories::find_by_github_id(&state.pool, github_repo_id).await? + else { + // Not connected to any workspace: discard before further processing. + return Ok(Completion::ignored()); + }; + let Some(git_ref) = payload_str(delivery, "ref").map(str::to_string) else { + return Ok(Completion::ignored()); + }; + + let commit_sha = payload_str(delivery, "after") + .or(payload_str(delivery, "headCommitSha")) + .unwrap_or("") + .to_string(); + + if let Some(branch) = git_ref.strip_prefix("refs/heads/") { + // A push to the default branch re-syncs the repository (the sync + // diffs blob shas, so this stays cheap when nothing changed). + let mut sync_run_id = None; + let mut sync_collapsed = false; + if branch == repo.default_branch { + match repo_sync::schedule(state, repo.id, "webhook").await? { + Some(run_id) => sync_run_id = Some(run_id), + // A sync was already running; the intent still counts. + None => sync_collapsed = true, + } + } + let synced = sync_run_id.is_some() || sync_collapsed; + + // Branch deletions push an all-zero SHA; nothing to build. + if is_zero_sha(&commit_sha) { + return Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: Some(git_ref), + head_sha: None, + outcome: if synced { OUTCOME_SYNC } else { OUTCOME_IGNORED }, + ignored_reason: (!synced).then_some(REASON_BRANCH_DELETED), + pipeline_ids: Vec::new(), + sync_run_id, + summary: serde_json::json!({ "branchDeleted": true }), + }])); + } + + // Changed paths for filter evaluation; a truncated set fails open. + let truncated = delivery + .payload + .as_ref() + .and_then(|p| p.get("pathsTruncated")) + .and_then(serde_json::Value::as_bool) + .unwrap_or(true); + let changed_paths: Option> = if truncated { + None + } else { + delivery + .payload + .as_ref() + .and_then(|p| p.get("changedPaths")) + .and_then(serde_json::Value::as_array) + .map(|paths| { + paths + .iter() + .filter_map(serde_json::Value::as_str) + .map(str::to_string) + .collect() + }) + }; + + let ctx = trigger_eval::EventContext { + kind: trigger_eval::EventKind::Push { branch }, + changed_paths: changed_paths.as_deref(), + }; + let (pipeline_ids, skipped, considered) = + run_matching_workflows(state, delivery, &repo, "push", &commit_sha, &git_ref, None, &ctx) + .await?; + + let mut summary = serde_json::json!({ "skipped": skipped }); + if sync_collapsed { + summary["syncCollapsed"] = serde_json::json!(true); + } + let (outcome, reason) = push_outcome(&pipeline_ids, synced, considered); + return Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: Some(git_ref), + head_sha: Some(commit_sha), + outcome, + ignored_reason: reason, + pipeline_ids, + sync_run_id, + summary, + }])); + } + + if let Some(tag) = git_ref.strip_prefix("refs/tags/") { + if is_zero_sha(&commit_sha) { + return Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: Some(git_ref), + head_sha: None, + outcome: OUTCOME_IGNORED, + ignored_reason: Some(REASON_TAG_DELETED), + pipeline_ids: Vec::new(), + sync_run_id: None, + summary: serde_json::json!({ "tagDeleted": true }), + }])); + } + + // Path filters deliberately never apply to tag pushes. + let ctx = trigger_eval::EventContext { + kind: trigger_eval::EventKind::TagPush { tag }, + changed_paths: None, + }; + let (pipeline_ids, skipped, considered) = + run_matching_workflows(state, delivery, &repo, "tag", &commit_sha, &git_ref, None, &ctx) + .await?; + + let (outcome, reason) = push_outcome(&pipeline_ids, false, considered); + return Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: Some(git_ref), + head_sha: Some(commit_sha), + outcome, + ignored_reason: reason, + pipeline_ids, + sync_run_id: None, + summary: serde_json::json!({ "skipped": skipped }), + }])); + } + + // Neither a branch nor a tag ref — record it, run nothing. + Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: Some(git_ref), + head_sha: None, + outcome: OUTCOME_IGNORED, + ignored_reason: Some(REASON_EVENT_NOT_SUPPORTED), + pipeline_ids: Vec::new(), + sync_run_id: None, + summary: serde_json::json!({}), + }])) +} + +fn push_outcome( + pipeline_ids: &[Uuid], + synced: bool, + considered: usize, +) -> (&'static str, Option<&'static str>) { + match (pipeline_ids.is_empty(), synced) { + (false, true) => (OUTCOME_BOTH, None), + (false, false) => (OUTCOME_PIPELINES, None), + (true, true) => (OUTCOME_SYNC, None), + (true, false) => ( + OUTCOME_IGNORED, + Some(if considered == 0 { + REASON_NO_MATCHING_WORKFLOWS + } else { + REASON_FILTERS_NOT_MATCHED + }), + ), + } +} + +/// Evaluate every runnable workflow of the repo against the event and create +/// pipelines for the matches. Returns (created ids, skip summary entries, +/// how many workflows were considered). Per-workflow creation failures are +/// expected (uses:-only workflows) and never block the others. +#[allow(clippy::too_many_arguments)] +async fn run_matching_workflows( + state: &AppState, + delivery: &ClaimedDelivery, + repo: &Repository, + trigger: &'static str, + commit_sha: &str, + git_ref: &str, + pr: Option<&PrContext>, + ctx: &trigger_eval::EventContext<'_>, +) -> anyhow::Result<(Vec, Vec, usize)> { + // Both "push" and "tag" pipelines run workflows declaring `on: push`. + let event = match trigger { + "pull_request" => "pull_request", + _ => "push", + }; + let workflows = db::workflows::runnable_for_repo(&state.pool, repo.id, event).await?; + let considered = workflows.len(); + + let commit_message = match pr { + Some(pr) => pr.title.clone(), + None => payload_str(delivery, "commitMessage").map(str::to_string), + }; + let commit_author = match pr { + Some(pr) => pr.user_login.clone(), + None => payload_str(delivery, "commitAuthor") + .or(payload_str(delivery, "pusherName")) + .map(str::to_string), + }; + let actor_login = payload_str(delivery, "senderLogin"); + let actor_avatar_url = payload_str(delivery, "senderAvatarUrl"); + + let mut pipeline_ids = Vec::new(); + let mut skipped = Vec::new(); + for workflow in workflows { + match trigger_eval::evaluate(&workflow.triggers, &workflow.metadata, ctx) { + trigger_eval::Decision::Skip(reason) => { + if skipped.len() < MAX_SKIPPED_SUMMARY { + skipped.push(serde_json::json!({ "path": workflow.path, "reason": reason })); + } + } + trigger_eval::Decision::Run => { + let trigger_ctx = pipeline_run::TriggerContext { + trigger, + triggered_by: None, + commit_sha, + commit_message: commit_message.as_deref(), + commit_author: commit_author.as_deref(), + actor_login, + actor_avatar_url, + git_ref, + inputs: None, + pr_number: pr.map(|pr| pr.number), + request_id: None, + }; + match pipeline_run::create_pipeline( + state, + repo, + workflow.id, + &workflow.name, + &workflow.path, + &workflow.raw_content, + &trigger_ctx, + ) + .await + { + Ok(pipeline) => { + // Report a queued check run to GitHub (best-effort). + github_checks::spawn_create(state, pipeline.id); + pipeline_ids.push(pipeline.id); + } + Err(error) => { + // Validation failures (e.g. uses:-only workflows) are + // expected; they must not fail the whole delivery. + tracing::debug!( + workflow = %workflow.path, + error = ?error, + "event did not trigger a pipeline for this workflow" + ); + } + } + } + } + } + Ok((pipeline_ids, skipped, considered)) +} + +/// PR context lifted out of the normalized payload. +struct PrContext { + number: i32, + title: Option, + user_login: Option, +} + +async fn process_pull_request( + state: &AppState, + delivery: &ClaimedDelivery, + action: &str, +) -> anyhow::Result { + let Some(github_repo_id) = delivery.github_repo_id else { + return Ok(Completion::ignored()); + }; + let Some(repo) = db::repositories::find_by_github_id(&state.pool, github_repo_id).await? + else { + return Ok(Completion::ignored()); + }; + let Some(pr) = delivery.payload.as_ref().and_then(|p| p.get("pullRequest")) else { + return Ok(Completion::ignored()); + }; + + let number = pr.get("number").and_then(serde_json::Value::as_i64); + let head_sha = pr.get("headSha").and_then(serde_json::Value::as_str); + let base_ref = pr.get("baseRef").and_then(serde_json::Value::as_str); + let head_repo = pr.get("headRepoId").and_then(serde_json::Value::as_i64); + let base_repo = pr.get("baseRepoId").and_then(serde_json::Value::as_i64); + let merged = pr + .get("merged") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + + let event_row = |outcome: &'static str, + reason: Option<&'static str>, + pipeline_ids: Vec, + skipped: Vec| { + let mut summary = serde_json::json!({ "skipped": skipped }); + if let Some(number) = number { + summary["prNumber"] = serde_json::json!(number); + } + if merged { + summary["merged"] = serde_json::json!(true); + } + EventRow { + repository_id: repo.id, + git_ref: number.map(|n| format!("refs/pull/{n}/head")), + head_sha: head_sha.map(str::to_string), + outcome, + ignored_reason: reason, + pipeline_ids, + sync_run_id: None, + summary, + } + }; + + // A closed PR never creates a pipeline here: a merge produces a push + // event for the merge commit, which flows through push triggers. + if action == "closed" { + return Ok(Completion::processed(vec![event_row( + OUTCOME_IGNORED, + Some(REASON_PR_CLOSED), + Vec::new(), + Vec::new(), + )])); + } + if action.is_empty() { + return Ok(Completion::processed(vec![event_row( + OUTCOME_IGNORED, + Some(REASON_PR_ACTION_IGNORED), + Vec::new(), + Vec::new(), + )])); + } + + // Fork PRs never execute: workspace secrets must not flow to fork code, + // and the checkout token could not fetch the fork anyway. Missing repo + // ids fail safe (treated as a fork). + let same_repo = matches!((head_repo, base_repo), (Some(h), Some(b)) if h == b); + if !same_repo { + return Ok(Completion::processed(vec![event_row( + OUTCOME_IGNORED, + Some(REASON_FORK_PR_SKIPPED), + Vec::new(), + Vec::new(), + )])); + } + + let (Some(number), Some(head_sha), Some(base_ref)) = (number, head_sha, base_ref) else { + return Ok(Completion::processed(vec![event_row( + OUTCOME_IGNORED, + Some(REASON_NO_HEAD_COMMIT), + Vec::new(), + Vec::new(), + )])); + }; + let Ok(pr_number) = i32::try_from(number) else { + return Ok(Completion::processed(vec![event_row( + OUTCOME_IGNORED, + Some(REASON_PR_ACTION_IGNORED), + Vec::new(), + Vec::new(), + )])); + }; + + let pr_ctx = PrContext { + number: pr_number, + title: pr + .get("title") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + user_login: pr + .get("userLogin") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + }; + let git_ref = format!("refs/pull/{pr_number}/head"); + // PR changed files are not in the webhook payload; path filters fail + // open (deliberate — no List-PR-files API call this phase). + let ctx = trigger_eval::EventContext { + kind: trigger_eval::EventKind::PullRequest { + action, + base_branch: base_ref, + }, + changed_paths: None, + }; + let (pipeline_ids, skipped, considered) = run_matching_workflows( + state, + delivery, + &repo, + "pull_request", + head_sha, + &git_ref, + Some(&pr_ctx), + &ctx, + ) + .await?; + + let (outcome, reason) = if pipeline_ids.is_empty() { + ( + OUTCOME_IGNORED, + Some(if considered == 0 { + REASON_NO_MATCHING_WORKFLOWS + } else { + REASON_FILTERS_NOT_MATCHED + }), + ) + } else { + (OUTCOME_PIPELINES, None) + }; + Ok(Completion::processed(vec![event_row( + outcome, + reason, + pipeline_ids, + skipped, + )])) +} + +async fn process_repository( + state: &AppState, + delivery: &ClaimedDelivery, + action: &str, +) -> anyhow::Result { + let Some(github_repo_id) = delivery.github_repo_id else { + return Ok(Completion::ignored()); + }; + let Some(repo) = db::repositories::find_by_github_id(&state.pool, github_repo_id).await? + else { + return Ok(Completion::ignored()); + }; + match action { + "deleted" => { + db::repositories::mark_failed( + &state.pool, + &[github_repo_id], + "repository deleted on github", + ) + .await?; + Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: None, + head_sha: None, + outcome: OUTCOME_IGNORED, + ignored_reason: Some(REASON_REPOSITORY_DELETED), + pipeline_ids: Vec::new(), + sync_run_id: None, + summary: serde_json::json!({}), + }])) + } + "renamed" | "edited" | "privatized" | "publicized" | "transferred" => { + let sync_run_id = repo_sync::schedule(state, repo.id, "webhook").await?; + Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: None, + head_sha: None, + outcome: OUTCOME_SYNC, + ignored_reason: None, + pipeline_ids: Vec::new(), + sync_run_id, + summary: if sync_run_id.is_none() { + serde_json::json!({ "syncCollapsed": true }) + } else { + serde_json::json!({}) + }, + }])) + } + _ => Ok(Completion::processed(vec![EventRow { + repository_id: repo.id, + git_ref: None, + head_sha: None, + outcome: OUTCOME_IGNORED, + ignored_reason: Some(REASON_EVENT_NOT_SUPPORTED), + pipeline_ids: Vec::new(), + sync_run_id: None, + summary: serde_json::json!({}), + }])), + } +} + +async fn process_installation( + state: &AppState, + delivery: &ClaimedDelivery, + action: &str, +) -> anyhow::Result { + let Some(installation_id) = delivery.installation_id else { + return Ok(Completion::ignored()); + }; + match action { + // Recorded so an org install can be claimed via the setup redirect + // by the member who performed it. + "created" => { + let account = payload_str(delivery, "installationAccountLogin").unwrap_or(""); + let sender = payload_str(delivery, "senderLogin").unwrap_or(""); + if !account.is_empty() && !sender.is_empty() { + db::github_installations::record_created_event( + &state.pool, + installation_id, + account, + sender, + ) + .await?; + } + } + "deleted" => { + state.github_app.evict_token(installation_id).await; + db::github_installations::delete_by_installation_id(&state.pool, installation_id) + .await?; + } + "suspend" => { + state.github_app.evict_token(installation_id).await; + db::github_installations::set_suspended(&state.pool, installation_id, true).await?; + } + "unsuspend" => { + db::github_installations::set_suspended(&state.pool, installation_id, false).await?; + } + _ => {} + } + // Installation lifecycle targets the installation, not a repository — + // no timeline row. + Ok(Completion::processed(Vec::new())) +} + +async fn process_installation_repositories( + state: &AppState, + delivery: &ClaimedDelivery, + action: &str, +) -> anyhow::Result { + if action != "removed" { + return Ok(Completion::processed(Vec::new())); + } + let ids: Vec = delivery + .payload + .as_ref() + .and_then(|p| p.get("repositoriesRemoved")) + .and_then(serde_json::Value::as_array) + .map(|list| { + list.iter() + .filter_map(serde_json::Value::as_i64) + .collect::>() + }) + .unwrap_or_default(); + if ids.is_empty() { + return Ok(Completion::processed(Vec::new())); + } + + // Timeline rows only for repositories actually connected here. + let mut events = Vec::new(); + let mut seen = HashSet::new(); + for github_repo_id in &ids { + if !seen.insert(*github_repo_id) { + continue; + } + if let Some(repo) = + db::repositories::find_by_github_id(&state.pool, *github_repo_id).await? + { + events.push(EventRow { + repository_id: repo.id, + git_ref: None, + head_sha: None, + outcome: OUTCOME_IGNORED, + ignored_reason: Some(REASON_ACCESS_REVOKED), + pipeline_ids: Vec::new(), + sync_run_id: None, + summary: serde_json::json!({}), + }); + } + } + db::repositories::mark_failed(&state.pool, &ids, "access revoked").await?; + Ok(Completion::processed(events)) +} diff --git a/backend/src/services/workflow_parse.rs b/backend/src/services/workflow_parse.rs index ee2fb2c..22cd3c0 100644 --- a/backend/src/services/workflow_parse.rs +++ b/backend/src/services/workflow_parse.rs @@ -18,7 +18,7 @@ use super::github_app::MAX_WORKFLOW_FILE_BYTES; /// whose stamped version is older, even when its blob sha is unchanged — /// otherwise new metadata (e.g. the detected-requirements ref arrays) would /// never materialize for files that don't change on GitHub. -pub const PARSER_VERSION: i64 = 2; +pub const PARSER_VERSION: i64 = 3; /// Resource budgets: a workflow file that exceeds these is hostile or /// broken, not "large". They bound both memory and walk time. @@ -226,6 +226,7 @@ pub fn parse_and_validate(content: &str) -> ParsedWorkflow { "varRefs": scan_var_refs(content), "environments": distinct_environments(&jobs), "dispatchInputs": extract_dispatch_inputs(root, &mut diagnostics), + "triggerFilters": extract_trigger_filters(root, &mut diagnostics), }); ParsedWorkflow { @@ -308,6 +309,185 @@ fn extract_triggers(root: &Mapping, diagnostics: &mut Vec) -> Vec` filter lists consumed by the dispatch-time trigger +/// evaluation engine (`services/trigger_eval.rs`). Only the mapping form of +/// `on:` carries filters; string/sequence forms yield the all-empty shape +/// (= unfiltered). Every list is capped and length-bounded — patterns are +/// stored verbatim (they are matched, never executed or interpolated). +/// Combining `branches` with `branches-ignore` (or `tags` with `tags-ignore`) +/// on the same event is an Error, matching GitHub's own rejection. +fn extract_trigger_filters(root: &Mapping, diagnostics: &mut Vec) -> serde_json::Value { + // Same YAML 1.1 quirk as extract_triggers: `on` may be the bool key. + let events = root + .get(Value::String("on".into())) + .or_else(|| root.get(Value::Bool(true))) + .and_then(Value::as_mapping); + + let mut filters = serde_json::json!({ + "push": { + "branches": [], "branchesIgnore": [], + "tags": [], "tagsIgnore": [], + "paths": [], "pathsIgnore": [], + }, + "pullRequest": { + "types": [], + "branches": [], "branchesIgnore": [], + "paths": [], "pathsIgnore": [], + }, + }); + let Some(events) = events else { + return filters; + }; + + for (event, yaml_key, keys) in [ + ( + "push", + "push", + &[ + ("branches", "branches"), + ("branches-ignore", "branchesIgnore"), + ("tags", "tags"), + ("tags-ignore", "tagsIgnore"), + ("paths", "paths"), + ("paths-ignore", "pathsIgnore"), + ][..], + ), + ( + "pullRequest", + "pull_request", + &[ + ("branches", "branches"), + ("branches-ignore", "branchesIgnore"), + ("paths", "paths"), + ("paths-ignore", "pathsIgnore"), + ][..], + ), + ] { + let Some(body) = get(events, yaml_key).and_then(Value::as_mapping) else { + continue; + }; + for (yaml_name, json_name) in keys { + filters[event][*json_name] = serde_json::json!(filter_patterns( + get(body, yaml_name), + &format!("on.{yaml_key}.{yaml_name}"), + diagnostics, + )); + } + // GitHub rejects a workflow that combines the include and ignore + // forms of the same dimension on one event. + for (include, ignore) in [("branches", "branches-ignore"), ("tags", "tags-ignore")] { + if get(body, include).is_some() && get(body, ignore).is_some() { + diagnostics.push(Diagnostic::error( + format!("`on.{yaml_key}` cannot combine `{include}` with `{ignore}`"), + Some(format!("on.{yaml_key}.{ignore}")), + )); + } + } + if get(body, "paths").is_some() && get(body, "paths-ignore").is_some() { + diagnostics.push(Diagnostic::error( + format!("`on.{yaml_key}` cannot combine `paths` with `paths-ignore`"), + Some(format!("on.{yaml_key}.paths-ignore")), + )); + } + } + + // pull_request activity types: allow-listed, deduped, capped. + if let Some(body) = get(events, "pull_request").and_then(Value::as_mapping) { + let mut types: Vec = Vec::new(); + for value in string_or_list(get(body, "types")) { + if !PR_ACTIVITY_TYPES.contains(&value.as_str()) { + diagnostics.push(Diagnostic::warning( + format!( + "unknown pull_request activity type `{}`", + value.chars().take(50).collect::() + ), + Some("on.pull_request.types".into()), + )); + continue; + } + if !types.contains(&value) { + types.push(value); + } + } + filters["pullRequest"]["types"] = serde_json::json!(types); + } + + filters +} + +/// One filter pattern list: string-or-sequence, non-empty, byte-capped, +/// count-capped. Oversized patterns and overflow are dropped with a warning +/// — a filter that can't be stored must never silently match everything. +fn filter_patterns( + value: Option<&Value>, + path: &str, + diagnostics: &mut Vec, +) -> Vec { + let raw = match value { + None => return Vec::new(), + Some(Value::String(s)) => vec![s.clone()], + Some(Value::Sequence(seq)) => seq + .iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect(), + Some(_) => { + diagnostics.push(Diagnostic::warning( + format!("`{path}` must be a string or a list of patterns"), + Some(path.to_string()), + )); + return Vec::new(); + } + }; + if raw.len() > MAX_FILTER_PATTERNS { + diagnostics.push(Diagnostic::warning( + format!("`{path}` lists more than {MAX_FILTER_PATTERNS} patterns; extras are ignored"), + Some(path.to_string()), + )); + } + let mut out = Vec::new(); + for pattern in raw.into_iter().take(MAX_FILTER_PATTERNS) { + if pattern.is_empty() { + continue; + } + if pattern.len() > MAX_FILTER_PATTERN_LEN { + diagnostics.push(Diagnostic::warning( + format!("`{path}` contains a pattern longer than {MAX_FILTER_PATTERN_LEN} bytes; it was ignored"), + Some(path.to_string()), + )); + continue; + } + out.push(pattern); + } + out +} + /// Caps for `on.workflow_dispatch.inputs` extraction. GitHub itself allows /// at most 25 top-level inputs; the string caps bound stored metadata. const MAX_DISPATCH_INPUTS: usize = 25; @@ -1232,6 +1412,129 @@ jobs: ); } + #[test] + fn extracts_trigger_filters_from_mapping_form() { + let parsed = parse_and_validate( + r#" +on: + push: + branches: [main, 'releases/**'] + paths-ignore: + - 'docs/**' + pull_request: + types: [opened, labeled] + branches: [main] +jobs: + a: + runs-on: x + steps: [] +"#, + ); + assert_eq!(parsed.status(), "valid"); + let filters = &parsed.metadata["triggerFilters"]; + assert_eq!( + filters["push"]["branches"], + serde_json::json!(["main", "releases/**"]) + ); + assert_eq!( + filters["push"]["pathsIgnore"], + serde_json::json!(["docs/**"]) + ); + assert_eq!(filters["push"]["tags"], serde_json::json!([])); + assert_eq!( + filters["pullRequest"]["types"], + serde_json::json!(["opened", "labeled"]) + ); + assert_eq!( + filters["pullRequest"]["branches"], + serde_json::json!(["main"]) + ); + } + + #[test] + fn string_and_sequence_on_forms_yield_empty_filters() { + for yaml in [ + "on: push\njobs:\n a:\n runs-on: x\n steps: []\n", + "on: [push, pull_request]\njobs:\n a:\n runs-on: x\n steps: []\n", + ] { + let parsed = parse_and_validate(yaml); + let filters = &parsed.metadata["triggerFilters"]; + assert_eq!(filters["push"]["branches"], serde_json::json!([])); + assert_eq!(filters["pullRequest"]["types"], serde_json::json!([])); + } + } + + #[test] + fn combining_branches_with_branches_ignore_is_an_error() { + let parsed = parse_and_validate( + "on:\n push:\n branches: [main]\n branches-ignore: [dev]\n\ + jobs:\n a:\n runs-on: x\n steps: []\n", + ); + assert_eq!(parsed.status(), "errors"); + assert!( + parsed + .diagnostics + .iter() + .any(|d| d.message.contains("cannot combine `branches` with `branches-ignore`")) + ); + } + + #[test] + fn combining_paths_with_paths_ignore_is_an_error() { + let parsed = parse_and_validate( + "on:\n pull_request:\n paths: [src/**]\n paths-ignore: [docs/**]\n\ + jobs:\n a:\n runs-on: x\n steps: []\n", + ); + assert_eq!(parsed.status(), "errors"); + } + + #[test] + fn unknown_pr_types_are_dropped_with_a_warning() { + let parsed = parse_and_validate( + "on:\n pull_request:\n types: [opened, bogus_type]\n\ + jobs:\n a:\n runs-on: x\n steps: []\n", + ); + assert_eq!(parsed.status(), "warnings"); + assert_eq!( + parsed.metadata["triggerFilters"]["pullRequest"]["types"], + serde_json::json!(["opened"]) + ); + } + + #[test] + fn filter_patterns_enforce_caps() { + let mut yaml = String::from("on:\n push:\n branches:\n"); + for i in 0..60 { + yaml.push_str(&format!(" - branch-{i}\n")); + } + yaml.push_str(&format!(" - '{}'\n", "x".repeat(300))); + yaml.push_str("jobs:\n a:\n runs-on: x\n steps: []\n"); + let parsed = parse_and_validate(&yaml); + let branches = parsed.metadata["triggerFilters"]["push"]["branches"] + .as_array() + .unwrap(); + assert_eq!(branches.len(), MAX_FILTER_PATTERNS); + assert!( + parsed + .diagnostics + .iter() + .any(|d| d.message.contains("more than")) + ); + } + + #[test] + fn bool_on_key_still_extracts_filters() { + // Unquoted `on` resolves to boolean true in YAML 1.1; the mapping + // form must still be found under the bool key. + let parsed = parse_and_validate( + "on:\n push:\n branches: [main]\njobs:\n a:\n runs-on: x\n steps: []\n", + ); + assert_eq!( + parsed.metadata["triggerFilters"]["push"]["branches"], + serde_json::json!(["main"]) + ); + } + #[test] fn metadata_is_parser_version_stamped_on_success_and_failure() { let parsed = parse_and_validate( diff --git a/backend/src/state.rs b/backend/src/state.rs index 08697a0..cf062c6 100644 --- a/backend/src/state.rs +++ b/backend/src/state.rs @@ -8,6 +8,7 @@ use sqlx::PgPool; use crate::config::Config; use crate::services::github_app::GitHubApp; +use crate::services::github_checks::GithubChecks; use crate::services::log_hub::LogHub; use crate::services::notification_hub::NotificationHub; use crate::services::notification_projector::NotificationProjector; @@ -17,6 +18,7 @@ use crate::services::runner_provisioner::RunnerProvisioner; use crate::services::scheduler::Scheduler; use crate::services::search_indexer::SearchIndexer; use crate::services::secrets_crypto::SecretsCrypto; +use crate::services::webhook_processor::WebhookProcessor; use crate::services::workspace_hub::WorkspaceHub; use crate::services::ws_ticket::WsTicketStore; @@ -48,6 +50,10 @@ pub struct AppState { pub notification_hub: Arc, /// Wake handle for the audit-tail notification projector loop. pub notification_projector: Arc, + /// Wake handle for the async webhook-delivery processor loop. + pub webhook_processor: Arc, + /// Per-installation availability cache for GitHub Checks reporting. + pub github_checks: Arc, /// Object storage router (MinIO primary / R2 fallback when both are /// configured); None disables artifact grants cleanly. pub storage: Option>, @@ -133,6 +139,8 @@ impl AppState { search_indexer, notification_hub: Arc::new(NotificationHub::default()), notification_projector, + webhook_processor: Arc::new(WebhookProcessor::default()), + github_checks: Arc::new(GithubChecks::default()), storage, ws_tickets: Arc::new(WsTicketStore::default()), // Requires async Docker probing; main fills it in right after. diff --git a/src/features/pipelines/components/JobIdentityBar.tsx b/src/features/pipelines/components/JobIdentityBar.tsx index 3b7b255..8daa609 100644 --- a/src/features/pipelines/components/JobIdentityBar.tsx +++ b/src/features/pipelines/components/JobIdentityBar.tsx @@ -1,4 +1,12 @@ -import { GitBranch, GitCommitHorizontal, MousePointerClick, Server, Webhook } from 'lucide-react'; +import { + GitBranch, + GitCommitHorizontal, + GitPullRequest, + MousePointerClick, + Server, + Tag, + Webhook, +} from 'lucide-react'; import { Link } from 'react-router-dom'; import { workspacePath } from '../../../app/navigation'; import { Avatar } from '../../../components/ui/Avatar'; @@ -31,7 +39,14 @@ function Item({ label, children }: { label: string; children: React.ReactNode }) */ export function JobIdentityBar({ slug, pipeline, job, runner }: JobIdentityBarProps) { useNow(job.status !== 'completed'); - const TriggerIcon = pipeline.trigger === 'push' ? Webhook : MousePointerClick; + const TriggerIcon = + pipeline.trigger === 'pull_request' + ? GitPullRequest + : pipeline.trigger === 'tag' + ? Tag + : pipeline.trigger === 'push' + ? Webhook + : MousePointerClick; return (
diff --git a/src/features/pipelines/components/PipelineFilters.tsx b/src/features/pipelines/components/PipelineFilters.tsx index ccd5b1c..84773ce 100644 --- a/src/features/pipelines/components/PipelineFilters.tsx +++ b/src/features/pipelines/components/PipelineFilters.tsx @@ -135,6 +135,8 @@ export function PipelineFilters({ value, onChange }: PipelineFiltersProps) { > + + diff --git a/src/features/pipelines/components/PipelinesTable.tsx b/src/features/pipelines/components/PipelinesTable.tsx index f7d86b1..182202b 100644 --- a/src/features/pipelines/components/PipelinesTable.tsx +++ b/src/features/pipelines/components/PipelinesTable.tsx @@ -1,5 +1,12 @@ import { format, formatDistanceToNow } from 'date-fns'; -import { GitBranch, GitCommitHorizontal, MousePointerClick, Webhook } from 'lucide-react'; +import { + GitBranch, + GitCommitHorizontal, + GitPullRequest, + MousePointerClick, + Tag, + Webhook, +} from 'lucide-react'; import { useNavigate } from 'react-router-dom'; import { workspacePath } from '../../../app/navigation'; import { Avatar } from '../../../components/ui/Avatar'; @@ -9,9 +16,18 @@ import { branchOfRef, formatDuration, shortSha } from '../lib/format'; import { PipelineStatusBadge } from './PipelineStatusBadge'; /** Trigger source icon + label, matching the API trigger vocabulary. */ +const TRIGGER_PRESENTATION: Record< + Pipeline['trigger'], + { icon: typeof Webhook; label: string } +> = { + push: { icon: Webhook, label: 'Push' }, + manual: { icon: MousePointerClick, label: 'Manual' }, + pull_request: { icon: GitPullRequest, label: 'Pull request' }, + tag: { icon: Tag, label: 'Tag' }, +}; + function TriggerCell({ trigger }: { trigger: Pipeline['trigger'] }) { - const Icon = trigger === 'push' ? Webhook : MousePointerClick; - const label = trigger === 'push' ? 'Push' : 'Manual'; + const { icon: Icon, label } = TRIGGER_PRESENTATION[trigger] ?? TRIGGER_PRESENTATION.push; return (
+ + @@ -217,6 +222,8 @@ export function RepositoryDetailPage() { ))} + {tab === 'events' && repoId && } + {tab === 'history' && } | null; + /** Pull request number when trigger = 'pull_request' (reruns keep it). */ + prNumber: number | null; status: PipelineStatus; conclusion: PipelineConclusion | null; createdAt: string; diff --git a/src/types/repository.ts b/src/types/repository.ts index 1fe886e..79b1771 100644 --- a/src/types/repository.ts +++ b/src/types/repository.ts @@ -63,3 +63,51 @@ export interface SyncRun { startedAt: string; finishedAt: string | null; } + +/** What processing one repository event caused (static categories). */ +export type RepositoryEventOutcome = + | 'pipelines_created' + | 'sync_scheduled' + | 'pipelines_and_sync' + | 'ignored' + | 'failed'; + +/** One entry of the chronological repository event timeline. */ +export interface RepositoryEvent { + id: string; + event: string; + action: string | null; + gitRef: string | null; + headSha: string | null; + actorLogin: string | null; + actorAvatarUrl: string | null; + outcome: RepositoryEventOutcome; + /** Static category (e.g. filters_not_matched) when outcome = ignored. */ + ignoredReason: string | null; + pipelineIds: string[]; + syncRunId: string | null; + summary: { + skipped?: { path: string; reason: string }[]; + prNumber?: number; + merged?: boolean; + syncCollapsed?: boolean; + branchDeleted?: boolean; + tagDeleted?: boolean; + } & Record; + receivedAt: string; + processedAt: string; +} + +export interface RepositoryEventsPage { + events: RepositoryEvent[]; + nextCursor: string | null; +} + +/** Webhook/sync health figures for the repository sync status panel. */ +export interface RepositoryHealth { + lastEventAt: string | null; + lastEventOutcome: RepositoryEventOutcome | null; + failedEvents24h: number; + pendingDeliveries: number; + checksEnabled: boolean; +}