package handler import ( "context" "errors" "io" "log/slog" "net/http" "github.com/go-chi/chi/v5" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/multica-ai/multica/server/internal/integrations/vcs" "github.com/multica-ai/multica/server/internal/issuestatus" db "github.com/multica-ai/multica/server/pkg/db/generated" "github.com/multica-ai/multica/server/pkg/protocol" ) // ── Response mappers ──────────────────────────────────────────────────────── // vcsPullRequestToResponse maps a stored VCS PR onto the shared PR response // shape for single-PR webhook broadcasts (no aggregated check counts; the // frontend re-queries the issue's PR list for fresh counts). func vcsPullRequestToResponse(p db.VcsPullRequest) GitHubPullRequestResponse { return GitHubPullRequestResponse{ ID: uuidToString(p.ID), Provider: p.Provider, 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: nil, ChecksConclusion: nil, Additions: p.Additions, Deletions: p.Deletions, ChangedFiles: p.ChangedFiles, } } // vcsPullRequestRowToResponse maps an issue's PR-list row, which carries the // aggregated commit-status counts, onto the shared response shape. func vcsPullRequestRowToResponse(p db.ListVCSPullRequestsByIssueRow) GitHubPullRequestResponse { return GitHubPullRequestResponse{ ID: uuidToString(p.ID), Provider: p.Provider, 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: 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, } } // ── Webhook ───────────────────────────────────────────────────────────────── // HandleVCSWebhook (POST /api/webhooks/vcs/{connectionId}) authenticates and // mirrors webhooks from any token-based Git provider. The connection id in the path // selects the workspace, the provider, and the decryption secret; the provider // adapter handles the provider-specific signature scheme, event header, and // payload shape, returning normalized events to the shared mirror logic below. func (h *Handler) HandleVCSWebhook(w http.ResponseWriter, r *http.Request) { // Where the integration is off (the managed cloud) the endpoint behaves as // if it does not exist — a bare 404 that reveals nothing about config, the // same response a genuinely unknown connection id gets below. if !h.isVCSAvailable() { writeError(w, http.StatusNotFound, "unknown connection") return } if !h.isVCSConfigured() { writeError(w, http.StatusNotFound, "unknown connection") return } connUUID, ok := parseUUIDOrBadRequest(w, chi.URLParam(r, "connectionId"), "connection id") if !ok { return } body, err := io.ReadAll(io.LimitReader(r.Body, 10<<20)) // 10 MiB cap if err != nil { writeError(w, http.StatusBadRequest, "read body failed") return } conn, err := h.Queries.GetVCSConnectionByID(r.Context(), connUUID) if err != nil { if !errors.Is(err, pgx.ErrNoRows) { slog.Warn("vcs: lookup connection failed", "err", err) } writeError(w, http.StatusNotFound, "unknown connection") return } provider, ok := vcs.For(conn.Provider) if !ok { slog.Error("vcs: connection has unknown provider", "provider", conn.Provider) writeError(w, http.StatusInternalServerError, "unknown provider") return } secret, err := h.openVCSSecret(conn.WebhookSecretEncrypted) if err != nil { slog.Error("vcs: decrypt webhook secret failed", "err", err) writeError(w, http.StatusInternalServerError, "secret error") return } if !provider.VerifySignature(secret, r.Header, body) { writeError(w, http.StatusUnauthorized, "invalid signature") return } switch provider.EventKind(r.Header) { case vcs.EventPullRequest: if pr, err := provider.ParsePullRequest(body); err != nil { slog.Warn("vcs: bad pull_request payload", "provider", conn.Provider, "err", err) } else { h.mirrorVCSPullRequest(r.Context(), conn, pr) } case vcs.EventCIStatus: if st, err := provider.ParseCIStatus(body); err != nil { slog.Warn("vcs: bad status payload", "provider", conn.Provider, "err", err) } else { h.mirrorVCSCIStatus(r.Context(), conn, st) } default: // Acknowledge unmodelled events so the provider doesn't flag the hook. } w.WriteHeader(http.StatusAccepted) } func (h *Handler) mirrorVCSPullRequest(ctx context.Context, conn db.VcsConnection, ev vcs.PullRequestEvent) { if ev.RepoOwner == "" || ev.RepoName == "" || ev.Number == 0 { slog.Warn("vcs: pull_request missing repo identity", "provider", conn.Provider) return } // The stored state before this event, so a merge completes issues once — // when it happens — and not again on a later event of a merged PR. prevState := "" if prev, err := h.Queries.GetVCSPullRequestByKey(ctx, db.GetVCSPullRequestByKeyParams{ ConnectionID: conn.ID, RepoOwner: ev.RepoOwner, RepoName: ev.RepoName, PrNumber: ev.Number, }); err == nil { prevState = prev.State } pr, err := h.Queries.UpsertVCSPullRequest(ctx, db.UpsertVCSPullRequestParams{ WorkspaceID: conn.WorkspaceID, ConnectionID: conn.ID, Provider: conn.Provider, RepoOwner: ev.RepoOwner, RepoName: ev.RepoName, PrNumber: ev.Number, Title: ev.Title, State: ev.State, HtmlUrl: ev.HTMLURL, Branch: ptrToText(strPtrOrNil(ev.Branch)), AuthorLogin: ptrToText(strPtrOrNil(ev.AuthorLogin)), AuthorAvatarUrl: ptrToText(strPtrOrNil(ev.AuthorAvatarURL)), MergedAt: parseGHTime(ev.MergedAt), ClosedAt: parseGHTime(ev.ClosedAt), PrCreatedAt: parseGHTimeRequired(ev.CreatedAt), PrUpdatedAt: parseGHTimeRequired(ev.UpdatedAt), Additions: ev.Additions, Deletions: ev.Deletions, ChangedFiles: ev.ChangedFiles, HeadSha: ev.HeadSHA, }) if err != nil { slog.Warn("vcs: upsert pr failed", "err", err) return } // Out-of-order guard for the link write. UpsertVCSPullRequest keeps the // newer persisted row on a stale redelivery, so `pr` may reflect a newer // event than this `ev`. Everything the link pass decides below comes from // `ev`, so acting on a stale event would undo what the newer one already // recorded (e.g. a redelivered older event dropping a link the newer title // carries). If the persisted row is strictly newer than this event, the // newer event already linked and published — stop here. (An event with no // usable timestamp falls back to now(), which is never strictly after the // stored value, so it proceeds.) evUpdatedAt := parseGHTimeRequired(ev.UpdatedAt) if pr.PrUpdatedAt.Valid && evUpdatedAt.Valid && pr.PrUpdatedAt.Time.After(evUpdatedAt.Time) { return } workspaceID := uuidToString(conn.WorkspaceID) resp := vcsPullRequestToResponse(pr) // Auto-link to issues by identifiers in the title, branch, and closing // keywords. Connecting a provider is the opt-in, so there is no separate // per-workspace flag. The issue-side machinery is shared with GitHub // (reconcileAutoLinks, maybeAutoCompleteIssue). A connection belongs to // exactly one workspace, so there is no cross-workspace ambiguity to settle. linkedIssueIDs := make([]string, 0) ws, err := h.Queries.GetWorkspace(ctx, conn.WorkspaceID) if err == nil { var touched map[pgtype.UUID]struct{} idents := prClaimedIdentifiers(ev.Title, ev.Body, ev.Branch) linkedIssueIDs, touched = h.reconcileAutoLinks(ctx, ws, pr.ID, ev.State, prAutoLinkInput{ idents: idents, permits: func(string) bool { return true }, ambiguous: func(string) bool { return false }, link: func(issueID pgtype.UUID) (int64, error) { return h.Queries.LinkIssueToVCSPullRequest(ctx, db.LinkIssueToVCSPullRequestParams{IssueID: issueID, PullRequestID: pr.ID}) }, unlink: func(issueID pgtype.UUID) (int64, error) { return h.Queries.UnlinkIssueFromVCSPullRequest(ctx, db.UnlinkIssueFromVCSPullRequestParams{IssueID: issueID, PullRequestID: pr.ID}) }, listAuto: func() ([]pgtype.UUID, error) { return h.Queries.ListAutoLinkedIssueIDsForVCSPullRequest(ctx, pr.ID) }, }) if ev.State == "merged" && prevState != "merged" { issueIDs, err := h.Queries.ListIssueIDsForVCSPullRequest(ctx, pr.ID) if err != nil { slog.Warn("vcs: list linked issues failed", "err", err) } for _, id := range issueIDs { touched[id] = struct{}{} } } resolver := issuestatus.NewResolver(conn.WorkspaceID) for issueID := range touched { h.maybeAutoCompleteIssue(ctx, conn.WorkspaceID, issueID, resolver) } } else { slog.Warn("vcs: load workspace failed", "err", err) } h.publish(protocol.EventPullRequestUpdated, workspaceID, "system", "", map[string]any{ "pull_request": resp, "linked_issue_ids": linkedIssueIDs, }) } func (h *Handler) mirrorVCSCIStatus(ctx context.Context, conn db.VcsConnection, ev vcs.CIStatusEvent) { if ev.SHA == "" || ev.State == "" { return } // Use the provider's own event timestamp so UpsertVCSCommitStatus's // monotonic guard has something real to compare — writing time.Now() here // made the guard always true, so an out-of-order redelivery could regress a // status. Falls back to now() only when the payload carried no timestamp. if err := h.Queries.UpsertVCSCommitStatus(ctx, db.UpsertVCSCommitStatusParams{ ConnectionID: conn.ID, Sha: ev.SHA, Context: ev.Context, State: ev.State, TargetUrl: ptrToText(strPtrOrNil(ev.TargetURL)), Description: ptrToText(strPtrOrNil(ev.Description)), UpdatedAt: parseGHTimeRequired(ev.UpdatedAt), }); err != nil { slog.Warn("vcs: upsert commit status failed", "err", err) return } issueIDs, err := h.Queries.ListIssueIDsForVCSPRHead(ctx, db.ListIssueIDsForVCSPRHeadParams{ ConnectionID: conn.ID, HeadSha: ev.SHA, }) if err != nil { slog.Warn("vcs: lookup issues for status failed", "err", err) return } workspaceID := uuidToString(conn.WorkspaceID) for _, issueID := range issueIDs { h.publish(protocol.EventPullRequestUpdated, workspaceID, "system", "", map[string]any{ "issue_id": uuidToString(issueID), }) } }