mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-29 14:37:44 +02:00
feat(github): API-snapshot PR cards — CI status + mergeability (MUL-5265)
Fetch each linked PR's CI checks and mergeability from the GitHub GraphQL API as the single source of truth (Plan C). Webhooks, page visits and a bounded TTL sweep are refresh triggers only; nothing is inferred from webhook payloads anymore. Backend (server/internal/integrations/ghsnapshot): - installation-token cache + GraphQL client (private key / tokens never logged) - one paginated pullRequest query -> normalized per-check snapshot - outbound queue: (installation,repo,PR) dedup + single in-flight per PR, bounded worker pool, Retry-After / rate-limit backoff, jitter - head-SHA-guarded atomic batch replace (a slow response for an old head can never overwrite a newer head's snapshot) - bounded chase window (30s->5m, stops on terminal/closed) + page-visit + TTL refresh; clean degradation when no App private key is configured Removes the old suite-level webhook aggregation display path (query + handlers + tests). check_suite / check_run / status are now pure triggers. Frontend: PR card shows two independent tri-state elements (CI status + mergeability). "Ready to merge" only when merge state is clean; no-checks and unknown-mergeable never assert a positive verdict; progress strip removed; four locales; stale marker. Docs: github-integration + environment-variables (four languages) — now required App private key, read-only Checks/Commit-statuses permissions, new event subscriptions, capability boundaries and troubleshooting. Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
Bohan-J
@@ -75,20 +75,47 @@ type GitHubPullRequestResponse struct {
|
||||
ClosedAt *string `json:"closed_at"`
|
||||
PRCreatedAt string `json:"pr_created_at"`
|
||||
PRUpdatedAt string `json:"pr_updated_at"`
|
||||
// Mergeable state mirrors GitHub's `mergeable_state` field. We only
|
||||
// surface `clean`/`dirty` in the UI today; other values (`blocked`,
|
||||
// `behind`, `unstable`, `unknown`) round-trip but render as unknown.
|
||||
// Mergeable state mirrors GitHub's REST `mergeable_state` field, retained
|
||||
// for compatibility. The card now reads the richer GraphQL fields below.
|
||||
MergeableState *string `json:"mergeable_state"`
|
||||
// ChecksConclusion is the aggregated state of the latest CI check
|
||||
// suites for the PR's current head SHA. One of "passed", "failed",
|
||||
// "pending", or nil when no completed suite has been observed.
|
||||
// ── GitHub API snapshot (MUL-5265, Plan C) ──────────────────────────────
|
||||
// These come from an authenticated GraphQL query, the single source of
|
||||
// truth. All are null / empty / 0 when no snapshot has landed (or the
|
||||
// GitHub App private key is unconfigured), so the card hides the CI / merge
|
||||
// region and degrades cleanly.
|
||||
//
|
||||
// Mergeable answers ONLY "is there a conflict": "mergeable" | "conflicting"
|
||||
// | "unknown" | null. A false "conflicting" must never be reported as
|
||||
// "not mergeable" — that verdict is MergeStateStatus's job.
|
||||
Mergeable *string `json:"mergeable"`
|
||||
// MergeStateStatus is GitHub's merge-state verdict, lowercased: "clean" |
|
||||
// "dirty" | "blocked" | "behind" | "unstable" | "draft" | "has_hooks" |
|
||||
// "unknown" | null. "Ready to merge" is derived ONLY from "clean".
|
||||
MergeStateStatus *string `json:"merge_state_status"`
|
||||
// ChecksRollup is GitHub's overall CI verdict, lowercased: "success" |
|
||||
// "failure" | "pending" | "error" | "expected" | null. null means
|
||||
// statusCheckRollup was null (no checks yet) and must NEVER render as
|
||||
// passed.
|
||||
ChecksRollup *string `json:"checks_rollup"`
|
||||
// ChecksConclusion is a coarse compat alias derived from the snapshot:
|
||||
// "passed" | "failed" | "pending" | null.
|
||||
ChecksConclusion *string `json:"checks_conclusion"`
|
||||
// Per-suite counts that drive the card's segmented progress bar.
|
||||
// Always present on list rows; bare upsert broadcasts default to 0
|
||||
// and the frontend hides the bar when total == 0.
|
||||
// Run-level counts for the PR's snapshot head. ChecksPending mirrors
|
||||
// ChecksRunning for older clients that still read the old key.
|
||||
ChecksTotal int64 `json:"checks_total"`
|
||||
ChecksPassed int64 `json:"checks_passed"`
|
||||
ChecksFailed int64 `json:"checks_failed"`
|
||||
ChecksRunning int64 `json:"checks_running"`
|
||||
ChecksPending int64 `json:"checks_pending"`
|
||||
// FailedCheckNames names the failing checks so the card can point at them
|
||||
// (e.g. "✗ 2/7 · backend, e2e").
|
||||
FailedCheckNames []string `json:"failed_check_names"`
|
||||
// SnapshotStale is true when an open PR's last successful fetch is older
|
||||
// than the stale threshold (GitHub outage / revoked key): the card shows
|
||||
// last-known data greyed out rather than blank.
|
||||
SnapshotStale bool `json:"snapshot_stale"`
|
||||
// SnapshotFetchedAt is when the snapshot was last fetched (RFC3339), or null.
|
||||
SnapshotFetchedAt *string `json:"snapshot_fetched_at"`
|
||||
// Diff stats (lines added/removed and file count) sourced from the
|
||||
// `pull_request` webhook payload. Legacy rows that pre-date this
|
||||
// field default to 0; the frontend treats total == 0 as "unknown"
|
||||
@@ -148,51 +175,76 @@ func githubPullRequestToResponse(p db.GithubPullRequest) GitHubPullRequestRespon
|
||||
PRCreatedAt: timestampToString(p.PrCreatedAt),
|
||||
PRUpdatedAt: timestampToString(p.PrUpdatedAt),
|
||||
MergeableState: textToPtr(p.MergeableState),
|
||||
// A bare PR row has no aggregated check counts — webhook
|
||||
// broadcasts of a single PR fall through here and the frontend
|
||||
// re-queries the list for fresh counts.
|
||||
// A bare PR row has no aggregated check counts — webhook broadcasts of a
|
||||
// single PR fall through here and the frontend re-queries the list for
|
||||
// the full snapshot (mergeable / rollup / counts).
|
||||
ChecksConclusion: nil,
|
||||
FailedCheckNames: []string{},
|
||||
Additions: p.Additions,
|
||||
Deletions: p.Deletions,
|
||||
ChangedFiles: p.ChangedFiles,
|
||||
}
|
||||
}
|
||||
|
||||
// prSnapshotStaleThreshold is how old an open PR's last successful fetch may be
|
||||
// before the card greys it out as stale. Healthy pipelines refresh open PRs at
|
||||
// least every sweep interval (~10m), so crossing 30m means refreshes are not
|
||||
// landing (GitHub outage, revoked key) and the shown data is last-known.
|
||||
const prSnapshotStaleThreshold = 30 * time.Minute
|
||||
|
||||
func issuePullRequestRowToResponse(p db.ListPullRequestsByIssueRow) GitHubPullRequestResponse {
|
||||
stale := false
|
||||
if p.SnapshotFetchedAt.Valid && (p.State == "open" || p.State == "draft") {
|
||||
stale = time.Since(p.SnapshotFetchedAt.Time) > prSnapshotStaleThreshold
|
||||
}
|
||||
failedNames := p.FailedCheckNames
|
||||
if failedNames == nil {
|
||||
failedNames = []string{}
|
||||
}
|
||||
return GitHubPullRequestResponse{
|
||||
ID: uuidToString(p.ID),
|
||||
Provider: "github",
|
||||
WorkspaceID: uuidToString(p.WorkspaceID),
|
||||
RepoOwner: p.RepoOwner,
|
||||
RepoName: p.RepoName,
|
||||
Number: p.PrNumber,
|
||||
Title: p.Title,
|
||||
State: p.State,
|
||||
HtmlURL: p.HtmlUrl,
|
||||
Branch: textToPtr(p.Branch),
|
||||
AuthorLogin: textToPtr(p.AuthorLogin),
|
||||
AuthorAvatarURL: textToPtr(p.AuthorAvatarUrl),
|
||||
MergedAt: timestampToPtr(p.MergedAt),
|
||||
ClosedAt: timestampToPtr(p.ClosedAt),
|
||||
PRCreatedAt: timestampToString(p.PrCreatedAt),
|
||||
PRUpdatedAt: timestampToString(p.PrUpdatedAt),
|
||||
MergeableState: textToPtr(p.MergeableState),
|
||||
ChecksConclusion: aggregateChecksConclusion(p.ChecksFailed, p.ChecksPassed, p.ChecksPending, p.ChecksTotal),
|
||||
ChecksPassed: p.ChecksPassed,
|
||||
ChecksFailed: p.ChecksFailed,
|
||||
ChecksPending: p.ChecksPending,
|
||||
Additions: p.Additions,
|
||||
Deletions: p.Deletions,
|
||||
ChangedFiles: p.ChangedFiles,
|
||||
ID: uuidToString(p.ID),
|
||||
Provider: "github",
|
||||
WorkspaceID: uuidToString(p.WorkspaceID),
|
||||
RepoOwner: p.RepoOwner,
|
||||
RepoName: p.RepoName,
|
||||
Number: p.PrNumber,
|
||||
Title: p.Title,
|
||||
State: p.State,
|
||||
HtmlURL: p.HtmlUrl,
|
||||
Branch: textToPtr(p.Branch),
|
||||
AuthorLogin: textToPtr(p.AuthorLogin),
|
||||
AuthorAvatarURL: textToPtr(p.AuthorAvatarUrl),
|
||||
MergedAt: timestampToPtr(p.MergedAt),
|
||||
ClosedAt: timestampToPtr(p.ClosedAt),
|
||||
PRCreatedAt: timestampToString(p.PrCreatedAt),
|
||||
PRUpdatedAt: timestampToString(p.PrUpdatedAt),
|
||||
MergeableState: textToPtr(p.MergeableState),
|
||||
Mergeable: lowerTextPtr(p.ApiMergeable),
|
||||
MergeStateStatus: lowerTextPtr(p.ApiMergeStateStatus),
|
||||
ChecksRollup: lowerTextPtr(p.ChecksRollupState),
|
||||
ChecksConclusion: rollupToConclusion(p.ChecksRollupState, p.ChecksFailed, p.ChecksRunning, p.ChecksPassed),
|
||||
ChecksTotal: p.ChecksTotal,
|
||||
ChecksPassed: p.ChecksPassed,
|
||||
ChecksFailed: p.ChecksFailed,
|
||||
ChecksRunning: p.ChecksRunning,
|
||||
ChecksPending: p.ChecksRunning,
|
||||
FailedCheckNames: failedNames,
|
||||
SnapshotStale: stale,
|
||||
SnapshotFetchedAt: timestampToPtr(p.SnapshotFetchedAt),
|
||||
Additions: p.Additions,
|
||||
Deletions: p.Deletions,
|
||||
ChangedFiles: p.ChangedFiles,
|
||||
}
|
||||
}
|
||||
|
||||
// aggregateChecksConclusion collapses the per-PR check_suite counts into a
|
||||
// single status surfaced to the UI:
|
||||
// - any failed-class suite wins ("failed");
|
||||
// - any not-yet-completed suite makes the PR "pending";
|
||||
// - all completed and in the passed-class is "passed";
|
||||
// - no observed suite at all is nil (rendered as "no checks" / hidden).
|
||||
// aggregateChecksConclusion collapses per-PR commit-status counts into a
|
||||
// single coarse status. Still used by the self-hosted VCS provider path
|
||||
// (Forgejo / Gitea / GitLab), which mirrors commit statuses via webhook rather
|
||||
// than fetching a GitHub-style API snapshot:
|
||||
// - any failed status wins ("failed");
|
||||
// - any not-yet-completed status makes the PR "pending";
|
||||
// - all completed and passed is "passed";
|
||||
// - no observed status at all is nil (rendered as "no checks" / hidden).
|
||||
func aggregateChecksConclusion(failed, passed, pending, total int64) *string {
|
||||
if total == 0 {
|
||||
return nil
|
||||
@@ -211,6 +263,47 @@ func aggregateChecksConclusion(failed, passed, pending, total int64) *string {
|
||||
return &v
|
||||
}
|
||||
|
||||
// lowerTextPtr returns a lowercased *string for a non-empty pgtype.Text, else
|
||||
// nil. Used to expose GraphQL enums (MERGEABLE / CLEAN / SUCCESS …) to the API
|
||||
// in the project's lowercase convention.
|
||||
func lowerTextPtr(t pgtype.Text) *string {
|
||||
if !t.Valid || t.String == "" {
|
||||
return nil
|
||||
}
|
||||
v := strings.ToLower(t.String)
|
||||
return &v
|
||||
}
|
||||
|
||||
// rollupToConclusion derives the coarse compat "checks_conclusion" from the
|
||||
// GraphQL rollup, falling back to the run counts when the rollup enum is
|
||||
// unfamiliar. A null/empty rollup means "no checks yet" → nil (never "passed").
|
||||
func rollupToConclusion(rollup pgtype.Text, failed, running, passed int64) *string {
|
||||
if !rollup.Valid || rollup.String == "" {
|
||||
return nil
|
||||
}
|
||||
var v string
|
||||
switch strings.ToUpper(rollup.String) {
|
||||
case "FAILURE", "ERROR":
|
||||
v = "failed"
|
||||
case "PENDING", "EXPECTED":
|
||||
v = "pending"
|
||||
case "SUCCESS":
|
||||
v = "passed"
|
||||
default:
|
||||
switch {
|
||||
case failed > 0:
|
||||
v = "failed"
|
||||
case running > 0:
|
||||
v = "pending"
|
||||
case passed > 0:
|
||||
v = "passed"
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return &v
|
||||
}
|
||||
|
||||
// ── Connect / state token ───────────────────────────────────────────────────
|
||||
|
||||
// githubAppSlug returns the GitHub App slug used to build the install URL.
|
||||
@@ -577,6 +670,14 @@ func (h *Handler) ListPullRequestsForIssue(w http.ResponseWriter, r *http.Reques
|
||||
out := make([]GitHubPullRequestResponse, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
out = append(out, issuePullRequestRowToResponse(row))
|
||||
// Page-visit trigger (MUL-5265): if this card's snapshot is missing or
|
||||
// older than the view TTL, kick an async refresh. Non-blocking — the
|
||||
// current (possibly stale) response is returned immediately and the
|
||||
// fresh snapshot arrives via the pull_request:updated realtime event.
|
||||
h.PRRefresh.MaybeEnqueueOnView(
|
||||
row.InstallationID, row.RepoOwner, row.RepoName, row.PrNumber,
|
||||
row.SnapshotFetchedAt.Time, row.SnapshotFetchedAt.Valid,
|
||||
)
|
||||
}
|
||||
// PRs from token-based providers (Forgejo / Gitea / GitLab) share the same
|
||||
// card list. They live in their own provider-tagged tables, so they merge
|
||||
@@ -596,6 +697,29 @@ func (h *Handler) ListPullRequestsForIssue(w http.ResponseWriter, r *http.Reques
|
||||
writeJSON(w, http.StatusOK, map[string]any{"pull_requests": out})
|
||||
}
|
||||
|
||||
// broadcastPRSnapshotApplied is the ghsnapshot pipeline's onApplied callback:
|
||||
// once an API snapshot is written to a PR row, re-broadcast the PR so every
|
||||
// open issue detail page re-queries its PR list and picks up the fresh CI /
|
||||
// mergeability state. Runs on a background pipeline goroutine.
|
||||
func (h *Handler) broadcastPRSnapshotApplied(ctx context.Context, prID pgtype.UUID) {
|
||||
pr, err := h.Queries.GetGitHubPullRequestByID(ctx, prID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
issueIDs, err := h.Queries.ListIssueIDsForPullRequest(ctx, prID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
linked := make([]string, 0, len(issueIDs))
|
||||
for _, id := range issueIDs {
|
||||
linked = append(linked, uuidToString(id))
|
||||
}
|
||||
h.publish(protocol.EventPullRequestUpdated, uuidToString(pr.WorkspaceID), "system", "", map[string]any{
|
||||
"pull_request": githubPullRequestToResponse(pr),
|
||||
"linked_issue_ids": linked,
|
||||
})
|
||||
}
|
||||
|
||||
// ── Webhook ─────────────────────────────────────────────────────────────────
|
||||
|
||||
// identifierRe extracts identifiers like "MUL-1510" from text. Case-insensitive
|
||||
@@ -650,8 +774,11 @@ func (h *Handler) HandleGitHubWebhook(w http.ResponseWriter, r *http.Request) {
|
||||
h.handleInstallationEvent(ctx, body)
|
||||
case "pull_request":
|
||||
h.handlePullRequestEvent(ctx, body)
|
||||
case "check_suite":
|
||||
h.handleCheckSuiteEvent(ctx, body)
|
||||
case "check_suite", "check_run", "status":
|
||||
// CI events are pure triggers under Plan C (MUL-5265): their payload is
|
||||
// never read for display. Each just asks the API pipeline to re-fetch
|
||||
// the authoritative snapshot for the PR(s) it concerns.
|
||||
h.triggerPRRefreshFromCIEvent(ctx, body)
|
||||
default:
|
||||
// Acknowledge every event so GitHub doesn't mark the endpoint failing,
|
||||
// but ignore types we don't model.
|
||||
@@ -850,6 +977,102 @@ func (h *Handler) handlePullRequestEvent(ctx context.Context, body []byte) {
|
||||
for _, inst := range insts {
|
||||
h.mirrorPullRequestForWorkspace(ctx, inst.WorkspaceID, inst.InstallationID, &p)
|
||||
}
|
||||
// The PR row(s) now carry the new head; ask the API pipeline for the
|
||||
// authoritative CI + mergeability snapshot for that head. The webhook is
|
||||
// only the doorbell — its own mergeable/checks payload is not used for
|
||||
// display anymore (MUL-5265).
|
||||
h.PRRefresh.Enqueue(p.Installation.ID, p.Repository.Owner.Login, p.Repository.Name, p.PullRequest.Number)
|
||||
}
|
||||
|
||||
// ghCIEventPayload captures the shared shape of the check_suite / check_run /
|
||||
// status webhooks — enough to resolve which PR to refresh. These events are
|
||||
// pure triggers under Plan C: their payload data is never read for display.
|
||||
type ghCIEventPayload struct {
|
||||
Installation struct {
|
||||
ID int64 `json:"id"`
|
||||
} `json:"installation"`
|
||||
Repository struct {
|
||||
Name string `json:"name"`
|
||||
Owner struct {
|
||||
Login string `json:"login"`
|
||||
} `json:"owner"`
|
||||
} `json:"repository"`
|
||||
// status events: top-level commit SHA, no PR number.
|
||||
SHA string `json:"sha"`
|
||||
CheckSuite struct {
|
||||
HeadSHA string `json:"head_sha"`
|
||||
PullRequests []struct {
|
||||
Number int32 `json:"number"`
|
||||
} `json:"pull_requests"`
|
||||
} `json:"check_suite"`
|
||||
CheckRun struct {
|
||||
PullRequests []struct {
|
||||
Number int32 `json:"number"`
|
||||
} `json:"pull_requests"`
|
||||
CheckSuite struct {
|
||||
HeadSHA string `json:"head_sha"`
|
||||
} `json:"check_suite"`
|
||||
} `json:"check_run"`
|
||||
}
|
||||
|
||||
// triggerPRRefreshFromCIEvent enqueues an API refresh for the PR(s) a
|
||||
// check_suite / check_run / status webhook concerns. check_suite/check_run
|
||||
// carry the PR numbers directly; status events carry only a commit SHA, so we
|
||||
// map it back to the mirrored head_sha to find the PR(s).
|
||||
func (h *Handler) triggerPRRefreshFromCIEvent(ctx context.Context, body []byte) {
|
||||
if !h.PRRefresh.Enabled() {
|
||||
return
|
||||
}
|
||||
var p ghCIEventPayload
|
||||
if err := json.Unmarshal(body, &p); err != nil {
|
||||
return
|
||||
}
|
||||
if p.Installation.ID == 0 || p.Repository.Name == "" {
|
||||
return
|
||||
}
|
||||
owner, repo := p.Repository.Owner.Login, p.Repository.Name
|
||||
|
||||
seen := map[int32]struct{}{}
|
||||
enqueue := func(number int32) {
|
||||
if number == 0 {
|
||||
return
|
||||
}
|
||||
if _, ok := seen[number]; ok {
|
||||
return
|
||||
}
|
||||
seen[number] = struct{}{}
|
||||
h.PRRefresh.Enqueue(p.Installation.ID, owner, repo, number)
|
||||
}
|
||||
for _, pr := range p.CheckSuite.PullRequests {
|
||||
enqueue(pr.Number)
|
||||
}
|
||||
for _, pr := range p.CheckRun.PullRequests {
|
||||
enqueue(pr.Number)
|
||||
}
|
||||
if len(seen) > 0 {
|
||||
return
|
||||
}
|
||||
// No PR number in the payload (status event, or a check event whose
|
||||
// pull_requests array was empty) — resolve by head SHA.
|
||||
sha := p.SHA
|
||||
if sha == "" {
|
||||
sha = coalesce(p.CheckSuite.HeadSHA, p.CheckRun.CheckSuite.HeadSHA)
|
||||
}
|
||||
if sha == "" {
|
||||
return
|
||||
}
|
||||
numbers, err := h.Queries.ListGitHubPRNumbersByHeadSHA(ctx, db.ListGitHubPRNumbersByHeadSHAParams{
|
||||
InstallationID: p.Installation.ID,
|
||||
RepoOwner: owner,
|
||||
RepoName: repo,
|
||||
HeadSha: sha,
|
||||
})
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, number := range numbers {
|
||||
enqueue(number)
|
||||
}
|
||||
}
|
||||
|
||||
// mirrorPullRequestForWorkspace mirrors a pull_request webhook into a single
|
||||
@@ -889,13 +1112,6 @@ func (h *Handler) mirrorPullRequestForWorkspace(ctx context.Context, wsID pgtype
|
||||
return
|
||||
}
|
||||
|
||||
// Drain any check_suite events that arrived before this PR row was
|
||||
// mirrored (out-of-order webhook delivery). Each drained row is
|
||||
// replayed through the same upsert path used by live check_suite
|
||||
// events; the DrainPending… query removes them atomically so a
|
||||
// concurrent PR upsert can't double-apply.
|
||||
h.replayPendingCheckSuitesForPR(ctx, pr, wsID)
|
||||
|
||||
workspaceID := uuidToString(wsID)
|
||||
resp := githubPullRequestToResponse(pr)
|
||||
|
||||
@@ -1021,198 +1237,6 @@ func (h *Handler) mirrorPullRequestForWorkspace(ctx context.Context, wsID pgtype
|
||||
})
|
||||
}
|
||||
|
||||
// ── check_suite webhook ────────────────────────────────────────────────────
|
||||
|
||||
type ghCheckSuitePayload struct {
|
||||
Action string `json:"action"`
|
||||
CheckSuite struct {
|
||||
ID int64 `json:"id"`
|
||||
HeadSHA string `json:"head_sha"`
|
||||
Status string `json:"status"`
|
||||
Conclusion string `json:"conclusion"`
|
||||
UpdatedAt string `json:"updated_at"`
|
||||
App struct {
|
||||
ID int64 `json:"id"`
|
||||
} `json:"app"`
|
||||
PullRequests []struct {
|
||||
Number int32 `json:"number"`
|
||||
} `json:"pull_requests"`
|
||||
} `json:"check_suite"`
|
||||
Repository struct {
|
||||
Name string `json:"name"`
|
||||
Owner struct {
|
||||
Login string `json:"login"`
|
||||
} `json:"owner"`
|
||||
} `json:"repository"`
|
||||
Installation struct {
|
||||
ID int64 `json:"id"`
|
||||
} `json:"installation"`
|
||||
}
|
||||
|
||||
// handleCheckSuiteEvent records the CI suite state for each PR the suite
|
||||
// references. We persist all non-terminal actions (`requested`, `rerequested`)
|
||||
// as well as `completed`: a `requested`/`rerequested` event has status
|
||||
// `queued`/`in_progress` and an empty conclusion, which the aggregation query
|
||||
// counts as pending. Without persisting them, the per-PR `checks_pending`
|
||||
// count stays at 0 while CI is mid-run and the PR card falls through to
|
||||
// "checks not reported yet" until the first suite finishes.
|
||||
//
|
||||
// The suite payload may reference multiple PRs (e.g. the same head SHA is
|
||||
// open against several base branches), so we iterate. A reference whose PR
|
||||
// hasn't been mirrored locally is stashed in `github_pending_check_suite`
|
||||
// and replayed when the matching `pull_request` event upserts the PR row.
|
||||
func (h *Handler) handleCheckSuiteEvent(ctx context.Context, body []byte) {
|
||||
var p ghCheckSuitePayload
|
||||
if err := json.Unmarshal(body, &p); err != nil {
|
||||
slog.Warn("github: bad check_suite payload", "err", err)
|
||||
return
|
||||
}
|
||||
if p.Installation.ID == 0 {
|
||||
return
|
||||
}
|
||||
insts, err := h.Queries.ListGitHubInstallationsByInstallationID(ctx, p.Installation.ID)
|
||||
if err != nil {
|
||||
slog.Warn("github: lookup installation failed", "err", err)
|
||||
return
|
||||
}
|
||||
if len(insts) == 0 {
|
||||
return
|
||||
}
|
||||
if len(p.CheckSuite.PullRequests) == 0 {
|
||||
// Forks emit suites whose `pull_requests` array is empty for
|
||||
// the upstream repo. We have no way to attribute the result
|
||||
// without polling, so drop with a hint.
|
||||
slog.Info("github: check_suite has no associated PRs", "suite_id", p.CheckSuite.ID)
|
||||
return
|
||||
}
|
||||
updatedAt := parseGHTimeRequired(p.CheckSuite.UpdatedAt)
|
||||
|
||||
// Fan out to every workspace bound to this installation: each records the
|
||||
// suite against its own mirror of the PR (see handlePullRequestEvent /
|
||||
// MUL-4343).
|
||||
for _, inst := range insts {
|
||||
h.recordCheckSuiteForWorkspace(ctx, inst.WorkspaceID, &p, updatedAt)
|
||||
}
|
||||
}
|
||||
|
||||
// recordCheckSuiteForWorkspace records a check_suite webhook against one
|
||||
// workspace's mirror of each referenced PR. A reference whose PR hasn't been
|
||||
// mirrored in this workspace yet is stashed and replayed when the matching
|
||||
// pull_request event upserts the row. Invoked once per workspace bound to the
|
||||
// delivering installation.
|
||||
func (h *Handler) recordCheckSuiteForWorkspace(ctx context.Context, wsID pgtype.UUID, p *ghCheckSuitePayload, updatedAt pgtype.Timestamptz) {
|
||||
affectedIssues := map[string]struct{}{}
|
||||
recorded := false
|
||||
for _, prRef := range p.CheckSuite.PullRequests {
|
||||
// Scope the lookup to the repo's workspace. The (workspace_id,
|
||||
// repo_owner, repo_name, pr_number) tuple is the real uniqueness key:
|
||||
// a bare (owner, repo, number) lookup could return a row from a
|
||||
// different workspace that also tracks this repo and land the suite
|
||||
// on the wrong PR.
|
||||
pr, err := h.Queries.GetGitHubPullRequest(ctx, db.GetGitHubPullRequestParams{
|
||||
WorkspaceID: wsID,
|
||||
RepoOwner: p.Repository.Owner.Login,
|
||||
RepoName: p.Repository.Name,
|
||||
PrNumber: prRef.Number,
|
||||
})
|
||||
if err != nil {
|
||||
if !errors.Is(err, pgx.ErrNoRows) {
|
||||
slog.Warn("github: lookup pr for check_suite failed", "err", err)
|
||||
continue
|
||||
}
|
||||
// Out-of-order delivery: the suite reached us before the
|
||||
// `pull_request` webhook that mirrors the PR row. Stash the
|
||||
// event keyed by (workspace, repo, pr_number, suite_id); the
|
||||
// PR upsert path will drain and replay it.
|
||||
if err := h.Queries.UpsertPendingCheckSuite(ctx, db.UpsertPendingCheckSuiteParams{
|
||||
WorkspaceID: wsID,
|
||||
InstallationID: p.Installation.ID,
|
||||
RepoOwner: p.Repository.Owner.Login,
|
||||
RepoName: p.Repository.Name,
|
||||
PrNumber: prRef.Number,
|
||||
SuiteID: p.CheckSuite.ID,
|
||||
HeadSha: p.CheckSuite.HeadSHA,
|
||||
AppID: p.CheckSuite.App.ID,
|
||||
Conclusion: strToText(p.CheckSuite.Conclusion),
|
||||
Status: p.CheckSuite.Status,
|
||||
SuiteUpdatedAt: updatedAt,
|
||||
}); err != nil {
|
||||
slog.Warn("github: stash pending check_suite failed",
|
||||
"err", err, "suite_id", p.CheckSuite.ID)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := h.Queries.UpsertPullRequestCheckSuite(ctx, db.UpsertPullRequestCheckSuiteParams{
|
||||
PrID: pr.ID,
|
||||
SuiteID: p.CheckSuite.ID,
|
||||
HeadSha: p.CheckSuite.HeadSHA,
|
||||
AppID: p.CheckSuite.App.ID,
|
||||
Conclusion: strToText(p.CheckSuite.Conclusion),
|
||||
Status: p.CheckSuite.Status,
|
||||
UpdatedAt: updatedAt,
|
||||
}); err != nil {
|
||||
slog.Warn("github: upsert check_suite failed", "err", err, "suite_id", p.CheckSuite.ID)
|
||||
continue
|
||||
}
|
||||
recorded = true
|
||||
issues, err := h.Queries.ListIssueIDsForPullRequest(ctx, pr.ID)
|
||||
if err == nil {
|
||||
for _, id := range issues {
|
||||
affectedIssues[uuidToString(id)] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !recorded {
|
||||
return
|
||||
}
|
||||
// Broadcast on the existing event so the issue page just re-queries the PR
|
||||
// list. We don't pass a single pull_request payload here because a suite can
|
||||
// touch several and the listener already invalidates by issue.
|
||||
linked := make([]string, 0, len(affectedIssues))
|
||||
for id := range affectedIssues {
|
||||
linked = append(linked, id)
|
||||
}
|
||||
h.publish(protocol.EventPullRequestUpdated, uuidToString(wsID), "system", "", map[string]any{
|
||||
"linked_issue_ids": linked,
|
||||
})
|
||||
}
|
||||
|
||||
// replayPendingCheckSuitesForPR drains the stash table for one PR (any
|
||||
// rows left there by a check_suite event that arrived before the PR row
|
||||
// was mirrored) and re-applies each event through the normal upsert
|
||||
// path. Safe to call on every PR upsert: the drain is a single
|
||||
// DELETE … RETURNING, so when there is nothing to replay the helper is
|
||||
// a no-op round-trip.
|
||||
func (h *Handler) replayPendingCheckSuitesForPR(ctx context.Context, pr db.GithubPullRequest, workspaceID pgtype.UUID) {
|
||||
pending, err := h.Queries.DrainPendingCheckSuitesForPR(ctx, db.DrainPendingCheckSuitesForPRParams{
|
||||
WorkspaceID: workspaceID,
|
||||
RepoOwner: pr.RepoOwner,
|
||||
RepoName: pr.RepoName,
|
||||
PrNumber: pr.PrNumber,
|
||||
})
|
||||
if err != nil {
|
||||
slog.Warn("github: drain pending check_suites failed",
|
||||
"err", err, "pr_id", uuidToString(pr.ID))
|
||||
return
|
||||
}
|
||||
for _, row := range pending {
|
||||
if err := h.Queries.UpsertPullRequestCheckSuite(ctx, db.UpsertPullRequestCheckSuiteParams{
|
||||
PrID: pr.ID,
|
||||
SuiteID: row.SuiteID,
|
||||
HeadSha: row.HeadSha,
|
||||
AppID: row.AppID,
|
||||
Conclusion: row.Conclusion,
|
||||
Status: row.Status,
|
||||
UpdatedAt: row.SuiteUpdatedAt,
|
||||
}); err != nil {
|
||||
slog.Warn("github: replay pending check_suite failed",
|
||||
"err", err, "pr_id", uuidToString(pr.ID),
|
||||
"suite_id", row.SuiteID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// derivePRMergeableState resolves the upsert behaviour for the PR row's
|
||||
// mergeable_state column on a `pull_request` webhook. It returns three
|
||||
// states encoded as (value, clear):
|
||||
|
||||
@@ -1463,35 +1463,6 @@ func TestDerivePRMergeableState(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestAggregateChecksConclusion(t *testing.T) {
|
||||
str := func(p *string) string {
|
||||
if p == nil {
|
||||
return "<nil>"
|
||||
}
|
||||
return *p
|
||||
}
|
||||
cases := []struct {
|
||||
name string
|
||||
failed, passed, pending, total int64
|
||||
want string
|
||||
}{
|
||||
{"no_suites_nil", 0, 0, 0, 0, "<nil>"},
|
||||
{"any_failure_wins", 1, 5, 0, 6, "failed"},
|
||||
{"failure_beats_pending", 1, 0, 3, 4, "failed"},
|
||||
{"pending_when_no_failure", 0, 1, 2, 3, "pending"},
|
||||
{"all_passed", 0, 3, 0, 3, "passed"},
|
||||
{"counts_zero_but_total_nonzero_returns_nil", 0, 0, 0, 1, "<nil>"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
got := aggregateChecksConclusion(tc.failed, tc.passed, tc.pending, tc.total)
|
||||
if str(got) != tc.want {
|
||||
t.Errorf("aggregateChecksConclusion = %s, want %s", str(got), tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// firePullRequestWebhookWithHead is like firePullRequestWebhook but lets the
|
||||
// caller control the head SHA and mergeable_state on the payload. The CI
|
||||
// tests need both knobs to exercise head-change semantics.
|
||||
@@ -1537,54 +1508,6 @@ func firePullRequestWebhookWithHead(t *testing.T, secret, identifier string, ins
|
||||
}
|
||||
}
|
||||
|
||||
func fireCheckSuiteWebhook(t *testing.T, secret string, installationID int64, repo string, prNumbers []int32, suiteID, appID int64, headSHA, conclusion, updatedAt string) {
|
||||
t.Helper()
|
||||
fireCheckSuiteWebhookWithStatus(t, secret, installationID, repo, prNumbers,
|
||||
suiteID, appID, headSHA, "completed", "completed", conclusion, updatedAt)
|
||||
}
|
||||
|
||||
// fireCheckSuiteWebhookWithStatus is the parametric form of
|
||||
// fireCheckSuiteWebhook. Tests covering the `requested`/`rerequested` and
|
||||
// `queued`/`in_progress` matrix use it directly; the legacy completed-only
|
||||
// helper above wraps it for existing call sites.
|
||||
func fireCheckSuiteWebhookWithStatus(t *testing.T, secret string, installationID int64, repo string, prNumbers []int32, suiteID, appID int64, headSHA, action, status, conclusion, updatedAt string) {
|
||||
t.Helper()
|
||||
prRefs := make([]map[string]any, 0, len(prNumbers))
|
||||
for _, n := range prNumbers {
|
||||
prRefs = append(prRefs, map[string]any{"number": n})
|
||||
}
|
||||
payload := map[string]any{
|
||||
"action": action,
|
||||
"check_suite": map[string]any{
|
||||
"id": suiteID,
|
||||
"head_sha": headSHA,
|
||||
"status": status,
|
||||
"conclusion": conclusion,
|
||||
"updated_at": updatedAt,
|
||||
"app": map[string]any{"id": appID},
|
||||
"pull_requests": prRefs,
|
||||
},
|
||||
"repository": map[string]any{
|
||||
"name": repo,
|
||||
"owner": map[string]any{"login": "acme"},
|
||||
},
|
||||
"installation": map[string]any{"id": installationID},
|
||||
}
|
||||
raw, _ := json.Marshal(payload)
|
||||
mac := hmac.New(sha256.New, []byte(secret))
|
||||
mac.Write(raw)
|
||||
sig := "sha256=" + hex.EncodeToString(mac.Sum(nil))
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
hookReq := httptest.NewRequest("POST", "/api/webhooks/github", bytes.NewReader(raw))
|
||||
hookReq.Header.Set("X-GitHub-Event", "check_suite")
|
||||
hookReq.Header.Set("X-Hub-Signature-256", sig)
|
||||
testHandler.HandleGitHubWebhook(rec, hookReq)
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Fatalf("check_suite webhook: expected 202, got %d (%s)", rec.Code, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func setupPRTestIssue(t *testing.T, ctx context.Context, secret string) (IssueResponse, int64) {
|
||||
t.Helper()
|
||||
t.Setenv("GITHUB_WEBHOOK_SECRET", secret)
|
||||
@@ -1621,259 +1544,6 @@ func setupPRTestIssue(t *testing.T, ctx context.Context, secret string) (IssueRe
|
||||
return created, installationID
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_AggregatesAcrossApps ensures the list query reports
|
||||
// "failed" when one app's latest suite is a failure and another app's is a
|
||||
// success on the same head. Without per-app aggregation, the last-completed
|
||||
// suite would silently flip the verdict.
|
||||
func TestWebhook_CheckSuite_AggregatesAcrossApps(t *testing.T) {
|
||||
if testHandler == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
const secret = "ci-aggregate-secret"
|
||||
created, installationID := setupPRTestIssue(t, ctx, secret)
|
||||
|
||||
head := "abc1234567890"
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-a", 11, "opened", head, "")
|
||||
// App A → success, App B → failure. The list query must report failed.
|
||||
fireCheckSuiteWebhook(t, secret, installationID, "ci-repo-a", []int32{11}, 1001, 7001, head, "success", "2026-05-01T00:00:00Z")
|
||||
fireCheckSuiteWebhook(t, secret, installationID, "ci-repo-a", []int32{11}, 1002, 7002, head, "failure", "2026-05-01T00:01:00Z")
|
||||
|
||||
rows, err := testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("expected 1 PR row, got %d", len(rows))
|
||||
}
|
||||
got := aggregateChecksConclusion(rows[0].ChecksFailed, rows[0].ChecksPassed, rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
if got == nil || *got != "failed" {
|
||||
t.Errorf("expected aggregate failed, got %v (counts: failed=%d passed=%d pending=%d total=%d)",
|
||||
got, rows[0].ChecksFailed, rows[0].ChecksPassed, rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_OldHeadIgnored asserts that a late-arriving
|
||||
// check_suite for a stale head SHA doesn't contaminate the current head's
|
||||
// pending view. Without the head_sha filter in the aggregation query, the
|
||||
// new head would inherit the old head's "passed" verdict.
|
||||
func TestWebhook_CheckSuite_OldHeadIgnored(t *testing.T) {
|
||||
if testHandler == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
const secret = "ci-oldhead-secret"
|
||||
created, installationID := setupPRTestIssue(t, ctx, secret)
|
||||
|
||||
oldHead := "old1111111111"
|
||||
newHead := "new2222222222"
|
||||
|
||||
// First: open the PR at old head, run a passing suite.
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-b", 22, "opened", oldHead, "")
|
||||
fireCheckSuiteWebhook(t, secret, installationID, "ci-repo-b", []int32{22}, 2001, 8001, oldHead, "success", "2026-05-01T00:00:00Z")
|
||||
|
||||
rows, err := testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
got := aggregateChecksConclusion(rows[0].ChecksFailed, rows[0].ChecksPassed, rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
if got == nil || *got != "passed" {
|
||||
t.Fatalf("setup: expected passed on old head, got %v", got)
|
||||
}
|
||||
|
||||
// Then: synchronize to new head — no new suite yet. Then a late suite
|
||||
// for the OLD head fires (e.g. a delayed delivery). The current aggregate
|
||||
// must be nil (no suite for the new head).
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-b", 22, "synchronize", newHead, "")
|
||||
fireCheckSuiteWebhook(t, secret, installationID, "ci-repo-b", []int32{22}, 2002, 8001, oldHead, "success", "2026-05-01T00:05:00Z")
|
||||
|
||||
rows, err = testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
got = aggregateChecksConclusion(rows[0].ChecksFailed, rows[0].ChecksPassed, rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
if got != nil {
|
||||
t.Errorf("expected no aggregate (nil) after head change, got %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_LateOlderEventIgnored guards the single-row ordering
|
||||
// rule: for the same (pr_id, suite_id) the upsert must not let a later-
|
||||
// delivered older event overwrite the latest one. We send the newer state
|
||||
// (failure) first and then the older (success) and assert the row still
|
||||
// reads failure.
|
||||
func TestWebhook_CheckSuite_LateOlderEventIgnored(t *testing.T) {
|
||||
if testHandler == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
const secret = "ci-ordering-secret"
|
||||
created, installationID := setupPRTestIssue(t, ctx, secret)
|
||||
|
||||
head := "ord1234567890"
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-c", 33, "opened", head, "")
|
||||
// Latest event first.
|
||||
fireCheckSuiteWebhook(t, secret, installationID, "ci-repo-c", []int32{33}, 3001, 9001, head, "failure", "2026-05-01T01:00:00Z")
|
||||
// Late-arriving older event for the same suite.
|
||||
fireCheckSuiteWebhook(t, secret, installationID, "ci-repo-c", []int32{33}, 3001, 9001, head, "success", "2026-05-01T00:00:00Z")
|
||||
|
||||
rows, err := testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
got := aggregateChecksConclusion(rows[0].ChecksFailed, rows[0].ChecksPassed, rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
if got == nil || *got != "failed" {
|
||||
t.Errorf("expected failure to win against later-delivered older success, got %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_QueuedCountsAsPending covers the "CI 跑到一半" path:
|
||||
// GitHub fires `check_suite.requested` with status `queued` and an empty
|
||||
// conclusion while CI is still spinning up. The handler must persist these
|
||||
// non-terminal events so the per-PR `checks_pending` count reflects work in
|
||||
// progress; otherwise the frontend falls through to the "checks not
|
||||
// reported yet" placeholder until the first completed suite arrives.
|
||||
func TestWebhook_CheckSuite_QueuedCountsAsPending(t *testing.T) {
|
||||
if testHandler == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
const secret = "ci-pending-secret"
|
||||
created, installationID := setupPRTestIssue(t, ctx, secret)
|
||||
|
||||
head := "pending1234567"
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-pending", 55, "opened", head, "")
|
||||
// CI just kicked off — `requested` action, status=queued, no conclusion.
|
||||
fireCheckSuiteWebhookWithStatus(t, secret, installationID, "ci-repo-pending", []int32{55}, 4001, 6001, head, "requested", "queued", "", "2026-05-01T00:00:00Z")
|
||||
// A second app's suite starts a moment later with status=in_progress.
|
||||
fireCheckSuiteWebhookWithStatus(t, secret, installationID, "ci-repo-pending", []int32{55}, 4002, 6002, head, "requested", "in_progress", "", "2026-05-01T00:00:30Z")
|
||||
|
||||
rows, err := testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("expected 1 PR row, got %d", len(rows))
|
||||
}
|
||||
if rows[0].ChecksPending != 2 || rows[0].ChecksTotal != 2 ||
|
||||
rows[0].ChecksFailed != 0 || rows[0].ChecksPassed != 0 {
|
||||
t.Fatalf("expected pending=2 total=2 failed=0 passed=0, got pending=%d total=%d failed=%d passed=%d",
|
||||
rows[0].ChecksPending, rows[0].ChecksTotal, rows[0].ChecksFailed, rows[0].ChecksPassed)
|
||||
}
|
||||
got := aggregateChecksConclusion(rows[0].ChecksFailed, rows[0].ChecksPassed, rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
if got == nil || *got != "pending" {
|
||||
t.Errorf("expected aggregate pending while CI is running, got %v", got)
|
||||
}
|
||||
|
||||
// Now one app completes successfully — pending count drops to 1 and the
|
||||
// aggregate stays pending until the second app finishes.
|
||||
fireCheckSuiteWebhookWithStatus(t, secret, installationID, "ci-repo-pending", []int32{55}, 4001, 6001, head, "completed", "completed", "success", "2026-05-01T00:05:00Z")
|
||||
rows, err = testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
if rows[0].ChecksPending != 1 || rows[0].ChecksPassed != 1 || rows[0].ChecksTotal != 2 {
|
||||
t.Fatalf("expected pending=1 passed=1 total=2 after one suite completes, got pending=%d passed=%d total=%d",
|
||||
rows[0].ChecksPending, rows[0].ChecksPassed, rows[0].ChecksTotal)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_OutOfOrderReplaysOnPRUpsert covers the out-of-order
|
||||
// path: a `check_suite` event arrives before the matching `pull_request`
|
||||
// row has been mirrored locally (e.g. webhook reordering, or the PR was
|
||||
// linked to an installation that was suspended/resumed). The handler must
|
||||
// stash the suite and replay it when the PR upsert arrives, otherwise the
|
||||
// PR's first observed suite is silently lost and `checks_pending` stays at
|
||||
// 0 until the next suite ships.
|
||||
func TestWebhook_CheckSuite_OutOfOrderReplaysOnPRUpsert(t *testing.T) {
|
||||
if testHandler == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
const secret = "ci-oooreplay-secret"
|
||||
created, installationID := setupPRTestIssue(t, ctx, secret)
|
||||
|
||||
head := "oo01234567890"
|
||||
// Suite event lands FIRST — the PR row does not exist yet.
|
||||
fireCheckSuiteWebhookWithStatus(t, secret, installationID, "ci-repo-ooo", []int32{66}, 5001, 7501, head, "requested", "in_progress", "", "2026-05-01T00:00:00Z")
|
||||
|
||||
// Verify nothing landed on the PR table yet (no PR row to land on).
|
||||
if rows, err := testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID)); err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
} else if len(rows) != 0 {
|
||||
t.Fatalf("expected 0 PR rows before PR webhook, got %d", len(rows))
|
||||
}
|
||||
|
||||
// Now the pull_request webhook arrives. The handler must drain the
|
||||
// pending stash and replay it onto this PR.
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-ooo", 66, "opened", head, "")
|
||||
|
||||
rows, err := testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("expected 1 PR row after PR webhook, got %d", len(rows))
|
||||
}
|
||||
if rows[0].ChecksPending != 1 || rows[0].ChecksTotal != 1 {
|
||||
t.Fatalf("expected pending=1 total=1 after replay, got pending=%d total=%d",
|
||||
rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
}
|
||||
|
||||
// The next PR upsert (a no-op metadata edit) must NOT re-apply or fail
|
||||
// — the drain is one-shot, so the second pull_request webhook drains
|
||||
// an empty pending list.
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-ooo", 66, "edited", head, "")
|
||||
rows, err = testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
if rows[0].ChecksPending != 1 || rows[0].ChecksTotal != 1 {
|
||||
t.Fatalf("expected pending=1 total=1 after no-op edit, got pending=%d total=%d",
|
||||
rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_OutOfOrderStashKeepsNewer guards the pending
|
||||
// stash against the same out-of-order trap the live table already
|
||||
// handles: while the PR row is still missing, an older event for the
|
||||
// same suite_id must not overwrite a newer payload that was stashed
|
||||
// first. Without the suite_updated_at guard on UpsertPendingCheckSuite,
|
||||
// a late `requested/in_progress` arriving after `completed/success`
|
||||
// would roll the stash back to pending; the subsequent PR upsert would
|
||||
// then replay the stale state and the PR card would stay stuck on
|
||||
// "pending" until the next suite shipped.
|
||||
func TestWebhook_CheckSuite_OutOfOrderStashKeepsNewer(t *testing.T) {
|
||||
if testHandler == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
const secret = "ci-stash-order-secret"
|
||||
created, installationID := setupPRTestIssue(t, ctx, secret)
|
||||
|
||||
head := "stash01234567"
|
||||
// Newer event lands FIRST while the PR row does not exist yet.
|
||||
fireCheckSuiteWebhookWithStatus(t, secret, installationID, "ci-repo-stash", []int32{77}, 6001, 8001, head, "completed", "completed", "success", "2026-05-01T00:05:00Z")
|
||||
// Older event for the SAME suite arrives later (webhook reorder). The
|
||||
// pending stash must keep the newer payload.
|
||||
fireCheckSuiteWebhookWithStatus(t, secret, installationID, "ci-repo-stash", []int32{77}, 6001, 8001, head, "requested", "in_progress", "", "2026-05-01T00:00:00Z")
|
||||
|
||||
// PR webhook arrives — drain replays the (still newer) stash.
|
||||
firePullRequestWebhookWithHead(t, secret, created.Identifier, installationID, "ci-repo-stash", 77, "opened", head, "")
|
||||
|
||||
rows, err := testHandler.Queries.ListPullRequestsByIssue(ctx, parseUUID(created.ID))
|
||||
if err != nil {
|
||||
t.Fatalf("ListPullRequestsByIssue: %v", err)
|
||||
}
|
||||
if len(rows) != 1 {
|
||||
t.Fatalf("expected 1 PR row after PR webhook, got %d", len(rows))
|
||||
}
|
||||
if rows[0].ChecksPassed != 1 || rows[0].ChecksPending != 0 || rows[0].ChecksTotal != 1 {
|
||||
t.Fatalf("expected passed=1 pending=0 total=1 (newer stash preserved), got passed=%d pending=%d total=%d",
|
||||
rows[0].ChecksPassed, rows[0].ChecksPending, rows[0].ChecksTotal)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWebhook_PullRequest_SynchronizeClearsMergeable verifies that
|
||||
// `synchronize` sets mergeable_state to NULL even when the payload still
|
||||
// carries the previous "clean" verdict — the old answer no longer applies
|
||||
@@ -2935,207 +2605,6 @@ func TestWebhook_PullRequest_FansOutToBoundWorkspaces(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_FansOutToBoundWorkspaces mirrors the PR fan-out for CI:
|
||||
// a check_suite event must be recorded against every bound workspace's copy of
|
||||
// the referenced PR, not just one.
|
||||
func TestWebhook_CheckSuite_FansOutToBoundWorkspaces(t *testing.T) {
|
||||
if testHandler == nil || testPool == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
secret := "fanout-cs-secret"
|
||||
t.Setenv("GITHUB_WEBHOOK_SECRET", secret)
|
||||
|
||||
const repo = "fanout-ci-repo"
|
||||
const prNumber int32 = 4344
|
||||
const installationID int64 = 778899102
|
||||
const suiteID int64 = 90019001
|
||||
head := "fanoutsha123456"
|
||||
|
||||
testPool.Exec(ctx, `DELETE FROM github_installation WHERE installation_id = $1`, installationID)
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE slug = $1`, "fanout-ci-ws-a")
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE slug = $1`, "fanout-ci-ws-b")
|
||||
wsA, err := testHandler.Queries.CreateWorkspace(ctx, db.CreateWorkspaceParams{
|
||||
Name: "fanout-ci-ws-a", Slug: "fanout-ci-ws-a", IssuePrefix: "FCA",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateWorkspace A: %v", err)
|
||||
}
|
||||
wsB, err := testHandler.Queries.CreateWorkspace(ctx, db.CreateWorkspaceParams{
|
||||
Name: "fanout-ci-ws-b", Slug: "fanout-ci-ws-b", IssuePrefix: "FCB",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateWorkspace B: %v", err)
|
||||
}
|
||||
|
||||
if _, err := testHandler.Queries.CreateGitHubInstallation(ctx, db.CreateGitHubInstallationParams{
|
||||
WorkspaceID: wsA.ID, InstallationID: installationID, AccountLogin: "acme", AccountType: "User",
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateGitHubInstallation A: %v", err)
|
||||
}
|
||||
if _, err := testHandler.Queries.CreateGitHubInstallation(ctx, db.CreateGitHubInstallationParams{
|
||||
WorkspaceID: wsB.ID, InstallationID: installationID, AccountLogin: "acme", AccountType: "User",
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateGitHubInstallation B: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() {
|
||||
testPool.Exec(ctx, `DELETE FROM github_pull_request_check_suite WHERE pr_id IN (SELECT id FROM github_pull_request WHERE repo_owner = 'acme' AND repo_name = $1)`, repo)
|
||||
testPool.Exec(ctx, `DELETE FROM github_pending_check_suite WHERE repo_owner = 'acme' AND repo_name = $1`, repo)
|
||||
testPool.Exec(ctx, `DELETE FROM github_pull_request WHERE repo_owner = 'acme' AND repo_name = $1`, repo)
|
||||
testPool.Exec(ctx, `DELETE FROM github_installation WHERE installation_id = $1`, installationID)
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, wsA.ID)
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, wsB.ID)
|
||||
})
|
||||
|
||||
// Mirror the PR into both workspaces (no issue needed), then fire CI on the
|
||||
// same head SHA.
|
||||
firePullRequestWebhookWithHead(t, secret, "FCX-1", installationID, repo, prNumber, "opened", head, "")
|
||||
fireCheckSuiteWebhook(t, secret, installationID, repo, []int32{prNumber}, suiteID, 7100, head, "failure", "2026-05-01T00:00:00Z")
|
||||
|
||||
// The suite must be recorded against BOTH workspaces' PR rows.
|
||||
assertRecorded := func(label string, prID any) {
|
||||
var n int
|
||||
if err := testPool.QueryRow(ctx,
|
||||
`SELECT count(*) FROM github_pull_request_check_suite WHERE pr_id = $1 AND suite_id = $2`,
|
||||
prID, suiteID).Scan(&n); err != nil {
|
||||
t.Fatalf("workspace %s: count check suites: %v", label, err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("workspace %s: expected 1 recorded check_suite, got %d", label, n)
|
||||
}
|
||||
}
|
||||
|
||||
prA, err := testHandler.Queries.GetGitHubPullRequest(ctx, db.GetGitHubPullRequestParams{
|
||||
WorkspaceID: wsA.ID, RepoOwner: "acme", RepoName: repo, PrNumber: prNumber,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("workspace A: expected PR mirrored: %v", err)
|
||||
}
|
||||
prB, err := testHandler.Queries.GetGitHubPullRequest(ctx, db.GetGitHubPullRequestParams{
|
||||
WorkspaceID: wsB.ID, RepoOwner: "acme", RepoName: repo, PrNumber: prNumber,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("workspace B: expected PR mirrored: %v", err)
|
||||
}
|
||||
assertRecorded("A", prA.ID)
|
||||
assertRecorded("B", prB.ID)
|
||||
}
|
||||
|
||||
// TestWebhook_CheckSuite_OutOfOrderFansOutToBoundWorkspaces covers the most
|
||||
// error-prone multi-workspace path: a check_suite that arrives BEFORE the PR is
|
||||
// mirrored. Each bound workspace must stash its own pending row, and when the PR
|
||||
// event fans out, each workspace must drain its own pending row and record the
|
||||
// suite — one workspace's stash/drain must not stand in for another's.
|
||||
func TestWebhook_CheckSuite_OutOfOrderFansOutToBoundWorkspaces(t *testing.T) {
|
||||
if testHandler == nil || testPool == nil {
|
||||
t.Skip("handler test fixture not initialized (no DB?)")
|
||||
}
|
||||
ctx := context.Background()
|
||||
secret := "fanout-cs-ooo-secret"
|
||||
t.Setenv("GITHUB_WEBHOOK_SECRET", secret)
|
||||
|
||||
const repo = "fanout-ooo-repo"
|
||||
const prNumber int32 = 4345
|
||||
const installationID int64 = 778899103
|
||||
const suiteID int64 = 90019002
|
||||
head := "ooosha7654321"
|
||||
|
||||
testPool.Exec(ctx, `DELETE FROM github_installation WHERE installation_id = $1`, installationID)
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE slug = $1`, "fanout-ooo-ws-a")
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE slug = $1`, "fanout-ooo-ws-b")
|
||||
wsA, err := testHandler.Queries.CreateWorkspace(ctx, db.CreateWorkspaceParams{
|
||||
Name: "fanout-ooo-ws-a", Slug: "fanout-ooo-ws-a", IssuePrefix: "OOA",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateWorkspace A: %v", err)
|
||||
}
|
||||
wsB, err := testHandler.Queries.CreateWorkspace(ctx, db.CreateWorkspaceParams{
|
||||
Name: "fanout-ooo-ws-b", Slug: "fanout-ooo-ws-b", IssuePrefix: "OOB",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("CreateWorkspace B: %v", err)
|
||||
}
|
||||
|
||||
if _, err := testHandler.Queries.CreateGitHubInstallation(ctx, db.CreateGitHubInstallationParams{
|
||||
WorkspaceID: wsA.ID, InstallationID: installationID, AccountLogin: "acme", AccountType: "User",
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateGitHubInstallation A: %v", err)
|
||||
}
|
||||
if _, err := testHandler.Queries.CreateGitHubInstallation(ctx, db.CreateGitHubInstallationParams{
|
||||
WorkspaceID: wsB.ID, InstallationID: installationID, AccountLogin: "acme", AccountType: "User",
|
||||
}); err != nil {
|
||||
t.Fatalf("CreateGitHubInstallation B: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() {
|
||||
testPool.Exec(ctx, `DELETE FROM github_pull_request_check_suite WHERE pr_id IN (SELECT id FROM github_pull_request WHERE repo_owner = 'acme' AND repo_name = $1)`, repo)
|
||||
testPool.Exec(ctx, `DELETE FROM github_pending_check_suite WHERE repo_owner = 'acme' AND repo_name = $1`, repo)
|
||||
testPool.Exec(ctx, `DELETE FROM github_pull_request WHERE repo_owner = 'acme' AND repo_name = $1`, repo)
|
||||
testPool.Exec(ctx, `DELETE FROM github_installation WHERE installation_id = $1`, installationID)
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, wsA.ID)
|
||||
testPool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, wsB.ID)
|
||||
})
|
||||
|
||||
pendingCount := func(wsID any) int {
|
||||
var n int
|
||||
if err := testPool.QueryRow(ctx,
|
||||
`SELECT count(*) FROM github_pending_check_suite WHERE workspace_id = $1 AND repo_owner = 'acme' AND repo_name = $2 AND pr_number = $3 AND suite_id = $4`,
|
||||
wsID, repo, prNumber, suiteID).Scan(&n); err != nil {
|
||||
t.Fatalf("count pending: %v", err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
suiteCount := func(prID any) int {
|
||||
var n int
|
||||
if err := testPool.QueryRow(ctx,
|
||||
`SELECT count(*) FROM github_pull_request_check_suite WHERE pr_id = $1 AND suite_id = $2`,
|
||||
prID, suiteID).Scan(&n); err != nil {
|
||||
t.Fatalf("count suites: %v", err)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// 1. check_suite arrives BEFORE any PR mirror: each bound workspace stashes
|
||||
// its own pending row.
|
||||
fireCheckSuiteWebhook(t, secret, installationID, repo, []int32{prNumber}, suiteID, 7200, head, "failure", "2026-05-02T00:00:00Z")
|
||||
if got := pendingCount(wsA.ID); got != 1 {
|
||||
t.Fatalf("workspace A: expected 1 pending check_suite, got %d", got)
|
||||
}
|
||||
if got := pendingCount(wsB.ID); got != 1 {
|
||||
t.Fatalf("workspace B: expected 1 pending check_suite, got %d", got)
|
||||
}
|
||||
|
||||
// 2. The PR arrives and fans out: each workspace drains its own pending row
|
||||
// and records the suite against its own PR mirror.
|
||||
firePullRequestWebhookWithHead(t, secret, "OOX-1", installationID, repo, prNumber, "opened", head, "")
|
||||
|
||||
prA, err := testHandler.Queries.GetGitHubPullRequest(ctx, db.GetGitHubPullRequestParams{
|
||||
WorkspaceID: wsA.ID, RepoOwner: "acme", RepoName: repo, PrNumber: prNumber,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("workspace A: expected PR mirrored: %v", err)
|
||||
}
|
||||
prB, err := testHandler.Queries.GetGitHubPullRequest(ctx, db.GetGitHubPullRequestParams{
|
||||
WorkspaceID: wsB.ID, RepoOwner: "acme", RepoName: repo, PrNumber: prNumber,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("workspace B: expected PR mirrored: %v", err)
|
||||
}
|
||||
if got := suiteCount(prA.ID); got != 1 {
|
||||
t.Fatalf("workspace A: expected 1 recorded check_suite after drain, got %d", got)
|
||||
}
|
||||
if got := suiteCount(prB.ID); got != 1 {
|
||||
t.Fatalf("workspace B: expected 1 recorded check_suite after drain, got %d", got)
|
||||
}
|
||||
if got := pendingCount(wsA.ID); got != 0 {
|
||||
t.Fatalf("workspace A: expected pending drained to 0, got %d", got)
|
||||
}
|
||||
if got := pendingCount(wsB.ID); got != 0 {
|
||||
t.Fatalf("workspace B: expected pending drained to 0, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSecondWorkspaceBindDoesNotUnbindFirst is the #4823 regression: binding
|
||||
// the same GitHub App installation in a second workspace must NOT overwrite the
|
||||
// first workspace's binding. Both bindings coexist, and re-binding an existing
|
||||
|
||||
@@ -23,6 +23,7 @@ import (
|
||||
"github.com/multica-ai/multica/server/internal/events"
|
||||
"github.com/multica-ai/multica/server/internal/integrations/channel/engine"
|
||||
composio "github.com/multica-ai/multica/server/internal/integrations/composio"
|
||||
"github.com/multica-ai/multica/server/internal/integrations/ghsnapshot"
|
||||
"github.com/multica-ai/multica/server/internal/integrations/lark"
|
||||
"github.com/multica-ai/multica/server/internal/integrations/slack"
|
||||
obsmetrics "github.com/multica-ai/multica/server/internal/metrics"
|
||||
@@ -246,7 +247,14 @@ type Handler struct {
|
||||
// error rather than silently storing plaintext. Wired in
|
||||
// cmd/server/router.go after New.
|
||||
VCSSecretBox *secretbox.Box
|
||||
cfg Config
|
||||
// PRRefresh drives the GitHub API snapshot pipeline for PR cards (MUL-5265):
|
||||
// webhook / page-visit / TTL triggers → authenticated GraphQL fetch →
|
||||
// head-SHA-guarded atomic snapshot write. Always non-nil, but inert (every
|
||||
// trigger is a no-op) when GITHUB_APP_ID / GITHUB_APP_PRIVATE_KEY are unset,
|
||||
// so the feature degrades cleanly on deployments without a private key.
|
||||
// Wired in cmd/server/router.go after New.
|
||||
PRRefresh *ghsnapshot.Manager
|
||||
cfg Config
|
||||
}
|
||||
|
||||
func New(queries *db.Queries, txStarter txStarter, hub *realtime.Hub, bus *events.Bus, emailService *service.EmailService, store storage.Storage, cfSigner *auth.CloudFrontSigner, analyticsClient analytics.Client, cfg Config, daemonHubs ...*daemonws.Hub) *Handler {
|
||||
@@ -318,6 +326,18 @@ func New(queries *db.Queries, txStarter txStarter, hub *realtime.Hub, bus *event
|
||||
cfg: cfg,
|
||||
}
|
||||
h.WebhookDeliveryWorker = NewWebhookDeliveryWorker(h)
|
||||
|
||||
// GitHub API snapshot pipeline for PR cards (MUL-5265). Built
|
||||
// unconditionally but inert (every trigger no-ops) when the App private key
|
||||
// is unconfigured, so the feature degrades cleanly. main.go calls
|
||||
// h.PRRefresh.Start(ctx) to launch its worker pool + TTL sweeper.
|
||||
ghClient, err := ghsnapshot.NewClientFromEnv()
|
||||
if err != nil {
|
||||
// Malformed key is operator-actionable; the pipeline stays disabled.
|
||||
slog.Warn("github: PR snapshot pipeline disabled (invalid App private key)", "err", err)
|
||||
}
|
||||
h.PRRefresh = ghsnapshot.NewManager(ghClient, queries, txStarter, h.broadcastPRSnapshotApplied)
|
||||
|
||||
return h
|
||||
}
|
||||
|
||||
|
||||
@@ -68,9 +68,12 @@ func vcsPullRequestRowToResponse(p db.ListVCSPullRequestsByIssueRow) GitHubPullR
|
||||
PRUpdatedAt: timestampToString(p.PrUpdatedAt),
|
||||
MergeableState: nil,
|
||||
ChecksConclusion: aggregateChecksConclusion(p.ChecksFailed, p.ChecksPassed, p.ChecksPending, p.ChecksTotal),
|
||||
ChecksTotal: p.ChecksTotal,
|
||||
ChecksPassed: p.ChecksPassed,
|
||||
ChecksFailed: p.ChecksFailed,
|
||||
ChecksPending: p.ChecksPending,
|
||||
ChecksRunning: p.ChecksPending,
|
||||
FailedCheckNames: []string{},
|
||||
Additions: p.Additions,
|
||||
Deletions: p.Deletions,
|
||||
ChangedFiles: p.ChangedFiles,
|
||||
|
||||
Reference in New Issue
Block a user