diff --git a/knotserver/ingester.go b/knotserver/ingester.go --- a/knotserver/ingester.go +++ b/knotserver/ingester.go @@ -13,8 +13,9 @@ comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/xrpc" - "github.com/bluesky-social/jetstream/pkg/models" + jmodels "github.com/bluesky-social/jetstream/pkg/models" "tangled.org/core/api/tangled" + "tangled.org/core/appview/models" "tangled.org/core/knotserver/db" "tangled.org/core/knotserver/git" "tangled.org/core/log" @@ -22,7 +23,7 @@ "tangled.org/core/workflow" ) -func (h *Knot) processPublicKey(ctx context.Context, event *models.Event) error { +func (h *Knot) processPublicKey(ctx context.Context, event *jmodels.Event) error { l := log.FromContext(ctx) raw := json.RawMessage(event.Commit.Record) did := event.Did @@ -44,7 +45,7 @@ return nil } -func (h *Knot) processKnotMember(ctx context.Context, event *models.Event) error { +func (h *Knot) processKnotMember(ctx context.Context, event *jmodels.Event) error { l := log.FromContext(ctx) raw := json.RawMessage(event.Commit.Record) did := event.Did @@ -84,32 +85,25 @@ return nil } -func (h *Knot) processPull(ctx context.Context, event *models.Event) error { - raw := json.RawMessage(event.Commit.Record) - did := event.Did +// returns a repo path on disk if present, and error if not +type targetRepo struct { + RepoPath string + OwnerDid string + RepoName string + RepoDid string +} - var record tangled.RepoPull - if err := json.Unmarshal(raw, &record); err != nil { - return fmt.Errorf("failed to unmarshal record: %w", err) - } - - l := log.FromContext(ctx) - l = l.With("handler", "processPull") - l = l.With("did", did) - +func (h *Knot) validatePullRecord(ctx context.Context, record *tangled.RepoPull) (*targetRepo, error) { if record.Target == nil { - return fmt.Errorf("ignoring pull record: target repo is nil") + return nil, fmt.Errorf("ignoring pull record: target repo is nil") } - - l = l.With("target_repo", record.Target.Repo, "target_repo_did", record.Target.RepoDid) - l = l.With("target_branch", record.Target.Branch) if record.Source == nil { - return fmt.Errorf("ignoring pull record: not a branch-based pull request") + return nil, fmt.Errorf("ignoring pull record: not a branch-based pull request") } if record.Source.Repo != nil || record.Source.RepoDid != nil { - return fmt.Errorf("ignoring pull record: fork based pull") + return nil, fmt.Errorf("ignoring pull record: fork based pull") } var repoPath, ownerDid, repoName, repoDid string @@ -119,19 +113,19 @@ var lookupErr error repoPath, ownerDid, repoName, lookupErr = h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid) if lookupErr != nil { - return fmt.Errorf("unknown target repo DID %s: %w", repoDid, lookupErr) + return nil, fmt.Errorf("unknown target repo DID %s: %w", repoDid, lookupErr) } case record.Target.Repo != nil: // TODO: get rid of this PDS fetch once all repos have DIDs repoAt, parseErr := syntax.ParseATURI(*record.Target.Repo) if parseErr != nil { - return fmt.Errorf("failed to parse ATURI: %w", parseErr) + return nil, fmt.Errorf("failed to parse ATURI: %w", parseErr) } ident, resolveErr := h.resolver.ResolveIdent(ctx, repoAt.Authority().String()) if resolveErr != nil || ident.Handle.IsInvalidHandle() { - return fmt.Errorf("failed to resolve handle: %w", resolveErr) + return nil, fmt.Errorf("failed to resolve handle: %w", resolveErr) } xrpcc := xrpc.Client{ @@ -140,13 +134,13 @@ resp, getErr := comatproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String()) if getErr != nil { - return fmt.Errorf("failed to resolve repo: %w", getErr) + return nil, fmt.Errorf("failed to resolve repo: %w", getErr) } repo := resp.Value.Val.(*tangled.Repo) if repo.Knot != h.c.Server.Hostname { - return fmt.Errorf("rejected pull record: not this knot, %s != %s", repo.Knot, h.c.Server.Hostname) + return nil, fmt.Errorf("rejected pull record: not this knot, %s != %s", repo.Knot, h.c.Server.Hostname) } ownerDid = ident.DID.String() @@ -154,27 +148,83 @@ repoDid, didErr := h.db.GetRepoDid(ownerDid, repoName) if didErr != nil { - return fmt.Errorf("failed to resolve repo DID for %s/%s: %w", ownerDid, repoName, didErr) + return nil, fmt.Errorf("failed to resolve repo DID for %s/%s: %w", ownerDid, repoName, didErr) } var lookupErr error repoPath, _, _, lookupErr = h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid) if lookupErr != nil { - return fmt.Errorf("failed to resolve repo on disk: %w", lookupErr) + return nil, fmt.Errorf("failed to resolve repo on disk: %w", lookupErr) } default: - return fmt.Errorf("ignoring pull record: target has neither repo nor repoDid") + return nil, fmt.Errorf("ignoring pull record: target has neither repo nor repoDid") } - gr, err := git.Open(repoPath, record.Source.Sha) + _, err := git.Open(repoPath, record.Source.Branch) if err != nil { - return fmt.Errorf("failed to open git repository: %w", err) + return nil, fmt.Errorf("failed to open git repository: %w", err) + } + + return &targetRepo{ + RepoPath: repoPath, + OwnerDid: ownerDid, + RepoName: repoName, + RepoDid: repoDid, + }, nil +} + +func (h *Knot) fetchLatestSubmission(ctx context.Context, did, rkey string, record *tangled.RepoPull) (*models.PullSubmission, error) { + // resolve the PR owner's identity to fetch the blob from their PDS + prOwnerIdent, err := h.resolver.ResolveIdent(ctx, did) + if err != nil || prOwnerIdent.Handle.IsInvalidHandle() { + return nil, fmt.Errorf("failed to resolve PR owner handle: %w", err) + } + + roundNumber := len(record.Rounds) - 1 + round := record.Rounds[roundNumber] + + // fetch the blob from the PR owner's PDS + prOwnerPds := prOwnerIdent.PDSEndpoint() + blobUrl, err := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", prOwnerPds)) + if err != nil { + return nil, fmt.Errorf("failed to construct blob URL: %w", err) + } + q := blobUrl.Query() + q.Set("cid", round.PatchBlob.Ref.String()) + q.Set("did", did) + blobUrl.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobUrl.String(), nil) + if err != nil { + return nil, fmt.Errorf("failed to create blob request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + blobResp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to fetch blob: %w", err) + } + defer blobResp.Body.Close() + + blob := io.ReadCloser(blobResp.Body) + latestSubmission, err := models.PullSubmissionFromRecord(did, rkey, roundNumber, round, &blob) + if err != nil { + return nil, fmt.Errorf("failed to parse submission: %w", err) + } + + return latestSubmission, nil +} + +func (h *Knot) discoverWorkflows(ctx context.Context, repoPath, sha string) (workflow.RawPipeline, error) { + gr, err := git.Open(repoPath, sha) + if err != nil { + return nil, fmt.Errorf("failed to open git repository: %w", err) } workflowDir, err := gr.FileTree(ctx, workflow.WorkflowDir) if err != nil { - return fmt.Errorf("failed to open workflow directory: %w", err) + return nil, fmt.Errorf("failed to open workflow directory: %w", err) } var pipeline workflow.RawPipeline @@ -195,11 +245,17 @@ }) } + return pipeline, nil +} + +func (h *Knot) compilePipeline(ctx context.Context, targetRepo *targetRepo, sourceBranch, sourceSha, targetBranch string, rawPipeline workflow.RawPipeline) tangled.Pipeline { + l := log.FromContext(ctx) + trigger := tangled.Pipeline_PullRequestTriggerData{ Action: "create", - SourceBranch: record.Source.Branch, - SourceSha: record.Source.Sha, - TargetBranch: record.Target.Branch, + SourceBranch: sourceBranch, + SourceSha: sourceSha, + TargetBranch: targetBranch, } compiler := workflow.Compiler{ @@ -207,23 +263,79 @@ Kind: string(workflow.TriggerKindPullRequest), PullRequest: &trigger, Repo: &tangled.Pipeline_TriggerRepo{ - Did: ownerDid, Knot: h.c.Server.Hostname, - Repo: &repoName, - RepoDid: &repoDid, + RepoDid: &targetRepo.RepoDid, + Did: targetRepo.OwnerDid, + Repo: &targetRepo.RepoName, }, }, } - cp := compiler.Compile(compiler.Parse(pipeline)) - eventJson, err := json.Marshal(cp) - if err != nil { - return fmt.Errorf("failed to marshal pipeline event: %w", err) + l.Info("raw", "raw", rawPipeline) + parsed := compiler.Parse(rawPipeline) + l.Info("parsed", "parsed", parsed) + compiled := compiler.Compile(parsed) + + l.Info("compiler diagnostics", "diagnostics", compiler.Diagnostics) + + return compiled +} + +func (h *Knot) processPull(ctx context.Context, event *jmodels.Event) error { + raw := json.RawMessage(event.Commit.Record) + rkey := event.Commit.RKey + did := event.Did + + var record tangled.RepoPull + if err := json.Unmarshal(raw, &record); err != nil { + return fmt.Errorf("failed to unmarshal record: %w", err) } + + l := log.FromContext(ctx) + l = l.With("handler", "processPull") + l = l.With("did", did) + + l.Info("validating pull record") + targetRepo, err := h.validatePullRecord(ctx, &record) + if err != nil { + l.Warn("pull record did not validate, skipping...") + return err + } + + l = l.With("target_repo", record.Target.Repo) + l = l.With("target_branch", record.Target.Branch) + + l.Info("fetching latest submission") + latestSubmission, err := h.fetchLatestSubmission(ctx, did, rkey, &record) + if err != nil { + return err + } + + sha := latestSubmission.SourceRev + if sha == "" { + return fmt.Errorf("failed to extract source SHA from pull submission") + } + l = l.With("sha", sha) + + l.Info("discovering workflows", "repo_path", targetRepo.RepoPath) + pipeline, err := h.discoverWorkflows(ctx, targetRepo.RepoPath, sha) + if err != nil { + return err + } + + l.Info("compiling pipeline", "workflow_count", len(pipeline)) + cp := h.compilePipeline(ctx, targetRepo, record.Source.Branch, sha, record.Target.Branch, pipeline) // do not run empty pipelines if cp.Workflows == nil { + l.Info("skipping empty pipeline") return nil + } + + l.Info("marshaling pipeline event") + eventJson, err := json.Marshal(cp) + if err != nil { + return fmt.Errorf("failed to marshal pipeline event: %w", err) } ev := db.Event{ @@ -232,11 +344,12 @@ EventJson: string(eventJson), } + l.Info("inserting pipeline event") return h.db.InsertEvent(ev, h.n) } // duplicated from add collaborator -func (h *Knot) processCollaborator(ctx context.Context, event *models.Event) error { +func (h *Knot) processCollaborator(ctx context.Context, event *jmodels.Event) error { raw := json.RawMessage(event.Commit.Record) did := event.Did @@ -357,8 +470,8 @@ return nil } -func (h *Knot) processMessages(ctx context.Context, event *models.Event) error { - if event.Kind != models.EventKindCommit { +func (h *Knot) processMessages(ctx context.Context, event *jmodels.Event) error { + if event.Kind != jmodels.EventKindCommit { return nil } diff --git a/patchutil/patchutil.go b/patchutil/patchutil.go --- a/patchutil/patchutil.go +++ b/patchutil/patchutil.go @@ -17,9 +17,8 @@ func ExtractPatches(formatPatch string) ([]types.FormatPatch, error) { patches := splitFormatPatch(formatPatch) - result := []types.FormatPatch{} - - for _, patch := range patches { + result := make([]types.FormatPatch, len(patches)) + for i, patch := range patches { files, headerStr, err := gitdiff.Parse(strings.NewReader(patch)) if err != nil { return nil, fmt.Errorf("failed to parse patch: %w", err) @@ -30,11 +29,11 @@ return nil, fmt.Errorf("failed to parse patch header: %w", err) } - result = append(result, types.FormatPatch{ + result[i] = types.FormatPatch{ Files: files, PatchHeader: header, Raw: patch, - }) + } } return result, nil diff --git a/appview/middleware/middleware.go b/appview/middleware/middleware.go --- a/appview/middleware/middleware.go +++ b/appview/middleware/middleware.go @@ -266,7 +266,7 @@ return } - pr, err := db.GetPull(mw.db, f.RepoAt(), prIdInt) + pr, err := db.GetPull(mw.db, orm.FilterEq("repo_at", f.RepoAt()), orm.FilterEq("pull_id", prIdInt)) if err != nil { l.Error("failed to get pull and comments", "err", err) mw.pages.Error404(w) @@ -275,21 +275,14 @@ ctx := context.WithValue(r.Context(), "pull", pr) - if pr.IsStacked() { - stack, err := db.GetStack(mw.db, pr.StackId) - if err != nil { - l.Error("failed to get stack", "err", err) - return - } - abandonedPulls, err := db.GetAbandonedPulls(mw.db, pr.StackId) - if err != nil { - l.Error("failed to get abandoned pulls", "err", err) - return - } - - ctx = context.WithValue(ctx, "stack", stack) - ctx = context.WithValue(ctx, "abandonedPulls", abandonedPulls) + stack, err := db.GetStack(mw.db, pr.AtUri()) + if err != nil { + l.Error("failed to get stack", "err", err) + mw.pages.Error404(w) + return } + + ctx = context.WithValue(ctx, "stack", stack) next.ServeHTTP(w, r.WithContext(ctx)) }) diff --git a/appview/models/pull.go b/appview/models/pull.go --- a/appview/models/pull.go +++ b/appview/models/pull.go @@ -162,8 +162,10 @@ var targetRepoAt syntax.ATURI var targetBranch string if record.Target != nil { - if uri, err := syntax.ParseATURI(record.Target.Repo); err == nil { - targetRepoAt = uri + if record.Target.Repo != nil { + if uri, err := syntax.ParseATURI(*record.Target.Repo); err == nil { + targetRepoAt = uri + } } targetBranch = record.Target.Branch } diff --git a/appview/pulls/pulls.go b/appview/pulls/pulls.go --- a/appview/pulls/pulls.go +++ b/appview/pulls/pulls.go @@ -47,7 +47,6 @@ lexutil "github.com/bluesky-social/indigo/lex/util" indigoxrpc "github.com/bluesky-social/indigo/xrpc" "github.com/go-chi/chi/v5" - "github.com/google/uuid" ) const ApplicationGzip = "application/gzip" @@ -101,21 +100,28 @@ // htmx fragment func (s *Pulls) PullActions(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "PullActions") + switch r.Method { case http.MethodGet: user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } + f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid) // can be nil if this pull is not stacked stack, _ := r.Context().Value("stack").(models.Stack) @@ -127,7 +133,7 @@ } if roundNumber >= len(pull.Submissions) { http.Error(w, "bad round id", http.StatusBadRequest) - s.logger.Error("failed to parse round id", "err", err) + l.Error("failed to parse round id", "err", err, "round_number", roundNumber) return } @@ -153,23 +159,30 @@ } func (s *Pulls) repoPullHelper(w http.ResponseWriter, r *http.Request, interdiff bool) { + l := s.logger.With("handler", "repoPullHelper", "interdiff", interdiff) + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } + f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid) backlinks, err := db.GetBacklinks(s.db, pull.AtUri()) if err != nil { - s.logger.Error("failed to get pull backlinks", "err", err) + l.Error("failed to get pull backlinks", "err", err) s.pages.Notice(w, "pull-error", "Failed to get pull. Try again later.") return } @@ -181,7 +194,7 @@ } if roundIdInt >= len(pull.Submissions) { http.Error(w, "bad round id", http.StatusBadRequest) - s.logger.Error("failed to parse round id", "err", err) + l.Error("failed to parse round id", "err", err, "round_number", roundIdInt) return } @@ -192,7 +205,6 @@ // can be nil if this pull is not stacked stack, _ := r.Context().Value("stack").(models.Stack) - abandonedPulls, _ := r.Context().Value("abandonedPulls").([]*models.Pull) mergeCheckResponse := s.mergeCheck(r, f, pull, stack) branchDeleteStatus := s.branchDeleteStatus(r, f, pull) @@ -210,9 +222,6 @@ for _, p := range stack { shas = append(shas, p.LatestSha()) } - for _, p := range abandonedPulls { - shas = append(shas, p.LatestSha()) - } ps, err := db.GetPipelineStatuses( s.db, @@ -223,7 +232,7 @@ orm.FilterIn("p.sha", shas), ) if err != nil { - s.logger.Error("failed to fetch pipeline statuses", "err", err) + l.Error("failed to fetch pipeline statuses", "err", err) // non-fatal } @@ -233,7 +242,7 @@ reactionMap, err := db.GetReactionMap(s.db, 20, pull.AtUri()) if err != nil { - s.logger.Error("failed to get pull reactions", "err", err) + l.Error("failed to get pull reactions", "err", err) } userReactions := map[models.ReactionKind]bool{} @@ -247,7 +256,7 @@ orm.FilterContains("scope", tangled.RepoPullNSID), ) if err != nil { - s.logger.Error("failed to fetch labels", "err", err) + l.Error("failed to fetch labels", "err", err) s.pages.Error503(w) return } @@ -264,14 +273,14 @@ if interdiff { currentPatch, err := patchutil.AsDiff(pull.Submissions[roundIdInt].CombinedPatch()) if err != nil { - s.logger.Error("failed to interdiff; current patch malformed", "err", err) + l.Error("failed to interdiff; current patch malformed", "err", err, "round_number", roundIdInt) s.pages.Notice(w, fmt.Sprintf("interdiff-error-%d", roundIdInt), "Failed to calculate interdiff; current patch is invalid.") return } previousPatch, err := patchutil.AsDiff(pull.Submissions[roundIdInt-1].CombinedPatch()) if err != nil { - s.logger.Error("failed to interdiff; previous patch malformed", "err", err) + l.Error("failed to interdiff; previous patch malformed", "err", err, "round_number", roundIdInt) s.pages.Notice(w, fmt.Sprintf("interdiff-error-%d", roundIdInt), "Failed to calculate interdiff; previous patch is invalid.") return } @@ -284,7 +293,6 @@ RepoInfo: s.repoResolver.GetRepoInfo(r, user), Pull: pull, Stack: stack, - AbandonedPulls: abandonedPulls, Backlinks: backlinks, BranchDeleteStatus: branchDeleteStatus, MergeCheck: mergeCheckResponse, @@ -303,9 +311,11 @@ } func (s *Pulls) RepoSinglePull(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "RepoSinglePull") + pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } @@ -328,15 +338,12 @@ Host: host, } - patch := pull.LatestPatch() - if pull.IsStacked() { - // combine patches of substack - subStack := stack.Below(pull) - // collect the portion of the stack that is mergeable - mergeable := subStack.Mergeable() - // combine each patch - patch = mergeable.CombinedPatch() - } + // combine patches of substack + subStack := stack.Below(pull) + // collect the portion of the stack that is mergeable + mergeable := subStack.Mergeable() + // combine each patch + patch := mergeable.CombinedPatch() resp, xe := tangled.RepoMergeCheck( r.Context(), @@ -349,7 +356,7 @@ }, ) if err := xrpcclient.HandleXrpcErr(xe); err != nil { - s.logger.Error("failed to check for mergeability", "err", err) + s.logger.Error("failed to check for mergeability", "err", err, "pull_id", pull.PullId, "target_branch", pull.TargetBranch) return types.MergeCheckResponse{ Error: fmt.Sprintf("failed to check merge status: %s", err.Error()), } @@ -426,7 +433,7 @@ } func (s *Pulls) resubmitCheck(r *http.Request, repo *models.Repo, pull *models.Pull, stack models.Stack) pages.ResubmitResult { - if pull.State == models.PullMerged || pull.State == models.PullDeleted || pull.PullSource == nil { + if pull.State == models.PullMerged || pull.State == models.PullAbandoned || pull.PullSource == nil { return pages.Unknown } @@ -441,21 +448,17 @@ branchResp, err := tangled.GitTempGetBranch(r.Context(), xrpcc, pull.PullSource.Branch, sourceRepo.String()) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.branches", "err", xrpcerr) + s.logger.Error("failed to call XRPC repo.branches", "err", xrpcerr, "pull_id", pull.PullId, "branch", pull.PullSource.Branch) return pages.Unknown } - s.logger.Error("failed to reach knotserver", "err", err) + s.logger.Error("failed to reach knotserver", "err", err, "pull_id", pull.PullId) return pages.Unknown } targetBranch := branchResp - latestSourceRev := pull.LatestSha() - - if pull.IsStacked() && stack != nil { - top := stack[0] - latestSourceRev = top.LatestSha() - } + top := stack[0] + latestSourceRev := top.LatestSha() if latestSourceRev != targetBranch.Hash { return pages.ShouldResubmit @@ -473,18 +476,21 @@ } func (s *Pulls) RepoPullPatchRaw(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "RepoPullPatchRaw") + pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId) roundId := chi.URLParam(r, "round") roundIdInt, err := strconv.Atoi(roundId) if err != nil || roundIdInt >= len(pull.Submissions) { http.Error(w, "bad round id", http.StatusBadRequest) - s.logger.Error("failed to parse round id", "err", err) + l.Error("failed to parse round id", "err", err, "round_id_str", roundId) return } @@ -496,14 +502,19 @@ l := s.logger.With("handler", "RepoPulls") user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } + params := r.URL.Query() page := pagination.FromContext(r.Context()) f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } + l = l.With("repo_at", f.RepoAt().String()) query := searchquery.Parse(params.Get("q")) @@ -618,7 +629,6 @@ countOpts := searchOpts countOpts.Page = pagination.Page{Limit: 1} for _, ps := range []models.PullState{models.PullOpen, models.PullMerged, models.PullClosed} { - ps := ps countOpts.State = &ps countRes, err := s.indexer.Search(r.Context(), countOpts) if err != nil { @@ -670,7 +680,7 @@ if p.PullSource.RepoAt != nil { pullSourceRepo, err = db.GetRepoByAtUri(s.db, p.PullSource.RepoAt.String()) if err != nil { - s.logger.Error("failed to get repo by at uri", "err", err) + l.Error("failed to get repo by at uri", "err", err, "repo_at", p.PullSource.RepoAt.String()) continue } else { p.PullSource.Repo = pullSourceRepo @@ -679,30 +689,59 @@ } } - // we want to group all stacked PRs into just one list - stacks := make(map[string]models.Stack) + var stacks []models.Stack var shas []string - n := 0 + + pullMap := make(map[string]*models.Pull) for _, p := range pulls { - // store the sha for later shas = append(shas, p.LatestSha()) - // this PR is stacked - if p.StackId != "" { - // we have already seen this PR stack - if _, seen := stacks[p.StackId]; seen { - stacks[p.StackId] = append(stacks[p.StackId], p) - // skip this PR - } else { - stacks[p.StackId] = nil - pulls[n] = p - n++ - } - } else { - pulls[n] = p - n++ - } + pullMap[p.AtUri().String()] = p } - pulls = pulls[:n] + + // track which PRs have been added to stacks + visited := make(map[string]bool) + + // group stacked PRs together using dependent_on relationships + for _, p := range pulls { + if visited[p.AtUri().String()] { + continue + } + + root := p + for root.DependentOn != nil { + if parent, ok := pullMap[root.DependentOn.String()]; ok { + root = parent + } else { + break // parent not in current page + } + } + + var stack models.Stack + current := root + for { + if visited[current.AtUri().String()] { + break + } + stack = append(stack, current) + visited[current.AtUri().String()] = true + + found := false + for _, candidate := range pulls { + if candidate.DependentOn != nil && + candidate.DependentOn.String() == current.AtUri().String() { + current = candidate + found = true + break + } + } + if !found { + break + } + } + + slices.Reverse(stack) + stacks = append(stacks, stack) + } ps, err := db.GetPipelineStatuses( s.db, @@ -713,7 +752,7 @@ orm.FilterIn("p.sha", shas), ) if err != nil { - s.logger.Error("failed to fetch pipeline statuses", "err", err) + l.Warn("failed to fetch pipeline statuses", "err", err) // non-fatal } m := make(map[string]models.Pipeline) @@ -757,25 +796,32 @@ } func (s *Pulls) PullComment(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "PullComment") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } + f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid) roundNumberStr := chi.URLParam(r, "round") roundNumber, err := strconv.Atoi(roundNumberStr) if err != nil || roundNumber >= len(pull.Submissions) { http.Error(w, "bad round id", http.StatusBadRequest) - s.logger.Error("failed to parse round id", "err", err) + l.Error("failed to parse round id", "err", err, "round_number_str", roundNumberStr) return } @@ -800,7 +846,7 @@ // Start a transaction tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - s.logger.Error("failed to start transaction", "err", err) + l.Error("failed to start transaction", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } @@ -810,7 +856,7 @@ client, err := s.oauth.AuthorizedClient(r) if err != nil { - s.logger.Error("failed to get authorized client", "err", err) + l.Error("failed to get authorized client", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } @@ -827,7 +873,7 @@ }, }) if err != nil { - s.logger.Error("failed to create pull comment", "err", err) + l.Error("failed to create pull comment", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } @@ -846,14 +892,14 @@ // Create the pull comment in the database with the commentAt field commentId, err := db.NewPullComment(tx, comment) if err != nil { - s.logger.Error("failed to create pull comment", "err", err) + l.Error("failed to create pull comment in database", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } // Commit the transaction if err = tx.Commit(); err != nil { - s.logger.Error("failed to commit transaction", "err", err) + l.Error("failed to commit transaction", "err", err) s.pages.Notice(w, "pull-comment", "Failed to create comment.") return } @@ -867,12 +913,19 @@ } func (s *Pulls) NewPull(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "NewPull") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } + f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } + l = l.With("repo_at", f.RepoAt().String()) switch r.Method { case http.MethodGet: @@ -881,17 +934,17 @@ xrpcBytes, err := tangled.GitTempListBranches(r.Context(), xrpcc, "", 0, f.RepoAt().String()) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.branches", "err", xrpcerr) + l.Error("failed to call XRPC repo.branches", "err", xrpcerr) s.pages.Error503(w) return } - s.logger.Error("failed to fetch branches", "err", err) + l.Error("failed to fetch branches", "err", err) return } var result types.RepoBranchesResponse if err := json.Unmarshal(xrpcBytes, &result); err != nil { - s.logger.Error("failed to decode XRPC response", "err", err) + l.Error("failed to decode XRPC response", "err", err) s.pages.Error503(w) return } @@ -1034,6 +1087,8 @@ sourceBranch string, isStacked bool, ) { + l := s.logger.With("handler", "handleBranchBasedPull", "user", user.Active.Did, "target_branch", targetBranch, "source_branch", sourceBranch, "is_stacked", isStacked) + scheme := "http" if !s.config.Core.Dev { scheme = "https" @@ -1046,18 +1101,18 @@ xrpcBytes, err := tangled.RepoCompare(r.Context(), xrpcc, repo.RepoIdentifier(), targetBranch, sourceBranch) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.compare", "err", xrpcerr) + l.Error("failed to call XRPC repo.compare", "err", xrpcerr) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } - s.logger.Error("failed to compare", "err", err) + l.Error("failed to compare", "err", err) s.pages.Notice(w, "pull", err.Error()) return } var comparison types.RepoFormatPatchResponse if err := json.Unmarshal(xrpcBytes, &comparison); err != nil { - s.logger.Error("failed to decode XRPC compare response", "err", err) + l.Error("failed to decode XRPC compare response", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -1077,7 +1132,6 @@ } recordPullSource := &tangled.RepoPull_Source{ Branch: sourceBranch, - Sha: comparison.Rev2, } s.createPullRequest(w, r, repo, user, title, body, targetBranch, patch, combined, sourceRev, pullSource, recordPullSource, isStacked) @@ -1094,6 +1148,8 @@ } func (s *Pulls) handleForkBasedPull(w http.ResponseWriter, r *http.Request, repo *models.Repo, user *oauth.MultiAccountUser, forkRepo string, title, body, targetBranch, sourceBranch string, isStacked bool) { + l := s.logger.With("handler", "handleForkBasedPull", "user", user.Active.Did, "fork_repo", forkRepo, "target_branch", targetBranch, "source_branch", sourceBranch, "is_stacked", isStacked) + repoString := strings.SplitN(forkRepo, "/", 2) forkOwnerDid := repoString[0] repoName := repoString[1] @@ -1102,7 +1158,7 @@ s.pages.Notice(w, "pull", "No such fork.") return } else if err != nil { - s.logger.Error("failed to fetch fork:", "err", err) + l.Error("failed to fetch fork", "err", err, "fork_owner_did", forkOwnerDid, "repo_name", repoName) s.pages.Notice(w, "pull", "Failed to fetch fork.") return } @@ -1155,18 +1211,18 @@ forkXrpcBytes, err := tangled.RepoCompare(r.Context(), forkXrpcc, fork.RepoIdentifier(), hiddenRef, sourceBranch) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.compare for fork", "err", xrpcerr) + l.Error("failed to call XRPC repo.compare for fork", "err", xrpcerr, "hidden_ref", hiddenRef) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } - s.logger.Error("failed to compare across branches", "err", err) + l.Error("failed to compare across branches", "err", err, "hidden_ref", hiddenRef) s.pages.Notice(w, "pull", err.Error()) return } var comparison types.RepoFormatPatchResponse if err := json.Unmarshal(forkXrpcBytes, &comparison); err != nil { - s.logger.Error("failed to decode XRPC compare response for fork", "err", err) + l.Error("failed to decode XRPC compare response for fork", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -1191,7 +1247,6 @@ recordPullSource := &tangled.RepoPull_Source{ Branch: sourceBranch, Repo: &forkAtUriStr, - Sha: sourceRev, } if fork.RepoDid != "" { recordPullSource.RepoDid = &fork.RepoDid @@ -1213,6 +1268,8 @@ recordPullSource *tangled.RepoPull_Source, isStacked bool, ) { + l := s.logger.With("handler", "createPullRequest", "user", user.Active.Did, "target_branch", targetBranch, "is_stacked", isStacked) + if isStacked { // creates a series of PRs, each linking to the previous, identified by jj's change-id s.createStackedPullRequest( @@ -1230,14 +1287,14 @@ client, err := s.oauth.AuthorizedClient(r) if err != nil { - s.logger.Error("failed to get authorized client", "err", err) + l.Error("failed to get authorized client", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - s.logger.Error("failed to start tx", "err", err) + l.Error("failed to start tx", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -1267,10 +1324,32 @@ mentions, references := s.mentionsResolver.Resolve(r.Context(), body) rkey := tid.TID() + + blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(patch), ApplicationGzip) + if err != nil { + l.Error("failed to upload patch", "err", err) + s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") + return + } + + record := tangled.RepoPull{ + Title: title, + Body: &body, + Target: repoPullTarget(repo, targetBranch), + Source: recordPullSource, + CreatedAt: time.Now().Format(time.RFC3339), + Rounds: []*tangled.RepoPull_Round{ + { + CreatedAt: time.Now().Format(time.RFC3339), + PatchBlob: blob.Blob, + }, + }, + } initialSubmission := models.PullSubmission{ Patch: patch, Combined: combined, SourceRev: sourceRev, + Blob: *blob.Blob, } pull := &models.Pull{ Title: title, @@ -1285,10 +1364,26 @@ &initialSubmission, }, PullSource: pullSource, + State: models.PullOpen, } - err = db.NewPull(tx, pull) + + _, err = comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ + Collection: tangled.RepoPullNSID, + Repo: user.Active.Did, + Rkey: rkey, + Record: &lexutil.LexiconTypeDecoder{ + Val: &record, + }, + }) if err != nil { - s.logger.Error("failed to create pull request", "err", err) + l.Error("failed to create pull request", "err", err) + s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") + return + } + + err = db.PutPull(tx, pull) + if err != nil { + l.Error("failed to create pull request in database", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -1299,35 +1394,8 @@ return } - blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(patch), ApplicationGzip) - if err != nil { - s.logger.Error("failed to upload patch", "err", err) - s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") - return - } - - _, err = comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ - Collection: tangled.RepoPullNSID, - Repo: user.Active.Did, - Rkey: rkey, - Record: &lexutil.LexiconTypeDecoder{ - Val: &tangled.RepoPull{ - Title: title, - Target: repoPullTarget(repo, targetBranch), - PatchBlob: blob.Blob, - Source: recordPullSource, - CreatedAt: time.Now().Format(time.RFC3339), - }, - }, - }) - if err != nil { - s.logger.Error("failed to create pull request", "err", err) - s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") - return - } - if err = tx.Commit(); err != nil { - s.logger.Error("failed to create pull request", "err", err) + l.Error("failed to commit transaction for pull request", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -1348,57 +1416,63 @@ sourceRev string, pullSource *models.PullSource, ) { + l := s.logger.With("handler", "createStackedPullRequest", "user", user.Active.Did, "target_branch", targetBranch, "source_rev", sourceRev) + // run some necessary checks for stacked-prs first // must be branch or fork based if sourceRev == "" { - s.logger.Warn("stacked PR from patch-based pull") + l.Error("stacked PR from patch-based pull") s.pages.Notice(w, "pull", "Stacking is only supported on branch and fork based pull-requests.") return } formatPatches, err := patchutil.ExtractPatches(patch) if err != nil { - s.logger.Error("failed to extract patches", "err", err) + l.Error("failed to extract patches", "err", err) s.pages.Notice(w, "pull", fmt.Sprintf("Failed to extract patches: %v", err)) return } // must have atleast 1 patch to begin with if len(formatPatches) == 0 { - s.logger.Error("empty patches") + l.Error("empty patches") s.pages.Notice(w, "pull", "No patches found in the generated format-patch.") - return - } - - // build a stack out of this patch - stackId := uuid.New() - stack, err := s.newStack(r.Context(), repo, user, targetBranch, patch, pullSource, stackId.String()) - if err != nil { - s.logger.Error("failed to create stack", "err", err) - s.pages.Notice(w, "pull", fmt.Sprintf("Failed to create stack: %v", err)) return } client, err := s.oauth.AuthorizedClient(r) if err != nil { - s.logger.Error("failed to get authorized client", "err", err) + l.Error("failed to get authorized client", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") + return + } + + // first upload all blobs + blobs := make([]*lexutil.LexBlob, len(formatPatches)) + for i, p := range formatPatches { + blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(p.Raw), ApplicationGzip) + if err != nil { + l.Error("failed to upload patch blob", "err", err, "patch_index", i) + s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") + return + } + l.Info("uploaded blob", "idx", i+1, "total", len(formatPatches)) + blobs[i] = blob.Blob + } + + // build a stack out of this patch + stack, err := s.newStack(r.Context(), repo, user, targetBranch, pullSource, formatPatches, blobs) + if err != nil { + l.Error("failed to create stack", "err", err) + s.pages.Notice(w, "pull", fmt.Sprintf("Failed to create stack: %v", err)) return } // apply all record creations at once var writes []*comatproto.RepoApplyWrites_Input_Writes_Elem for _, p := range stack { - blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(p.LatestPatch()), ApplicationGzip) - if err != nil { - s.logger.Error("failed to upload patch blob", "err", err) - s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") - return - } - record := p.AsRecord() - record.PatchBlob = blob.Blob writes = append(writes, &comatproto.RepoApplyWrites_Input_Writes_Elem{ RepoApplyWrites_Create: &comatproto.RepoApplyWrites_Create{ Collection: tangled.RepoPullNSID, @@ -1414,7 +1488,7 @@ Writes: writes, }) if err != nil { - s.logger.Error("failed to create stacked pull request", "err", err) + l.Error("failed to create stacked pull request", "err", err) s.pages.Notice(w, "pull", "Failed to create stacked pull request. Try again later.") return } @@ -1422,16 +1496,16 @@ // create all pulls at once tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - s.logger.Error("failed to start tx", "err", err) + l.Error("failed to start tx", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } defer tx.Rollback() for _, p := range stack { - err = db.NewPull(tx, p) + err = db.PutPull(tx, p) if err != nil { - s.logger.Error("failed to create pull request", "err", err) + l.Error("failed to create pull request in database", "err", err, "pull_rkey", p.Rkey) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -1439,7 +1513,7 @@ } if err = tx.Commit(); err != nil { - s.logger.Error("failed to create pull request", "err", err) + l.Error("failed to commit transaction for pull requests", "err", err) s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -1456,9 +1530,11 @@ } func (s *Pulls) ValidatePatch(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "ValidatePatch") + _, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } @@ -1469,7 +1545,7 @@ } if err := s.validator.ValidatePatch(&patch); err != nil { - s.logger.Error("failed to validate patch", "err", err) + l.Error("failed to validate patch", "err", err) s.pages.Notice(w, "patch-error", "Invalid patch format. Please provide a valid git diff or format-patch.") return } @@ -1490,10 +1566,12 @@ } func (s *Pulls) CompareBranchesFragment(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "CompareBranchesFragment") + user := s.oauth.GetMultiAccountUser(r) f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } @@ -1501,14 +1579,14 @@ xrpcBytes, err := tangled.GitTempListBranches(r.Context(), xrpcc, "", 0, f.RepoAt().String()) if err != nil { - s.logger.Error("failed to fetch branches", "err", err) + l.Error("failed to fetch branches", "err", err) s.pages.Error503(w) return } var result types.RepoBranchesResponse if err := json.Unmarshal(xrpcBytes, &result); err != nil { - s.logger.Error("failed to decode XRPC response", "err", err) + l.Error("failed to decode XRPC response", "err", err) s.pages.Error503(w) return } @@ -1533,11 +1611,16 @@ } func (s *Pulls) CompareForksFragment(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "CompareForksFragment") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } forks, err := db.GetForksByDid(s.db, user.Active.Did) if err != nil { - s.logger.Error("failed to get forks", "err", err) + l.Error("failed to get forks", "err", err) return } @@ -1549,11 +1632,16 @@ } func (s *Pulls) CompareForksBranchesFragment(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "CompareForksBranchesFragment") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } @@ -1570,25 +1658,25 @@ orm.FilterEq("name", forkName), ) if err != nil { - s.logger.Error("failed to get repo", "did", forkOwnerDid, "name", forkName, "err", err) + l.Error("failed to get repo", "fork_owner_did", forkOwnerDid, "fork_name", forkName, "err", err) return } sourceXrpcBytes, err := tangled.GitTempListBranches(r.Context(), xrpcc, "", 0, repo.RepoAt().String()) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.branches for source", "err", xrpcerr) + l.Error("failed to call XRPC repo.branches for source", "err", xrpcerr) s.pages.Error503(w) return } - s.logger.Error("failed to fetch source branches", "err", err) + l.Error("failed to fetch source branches", "err", err) return } // Decode source branches var sourceBranches types.RepoBranchesResponse if err := json.Unmarshal(sourceXrpcBytes, &sourceBranches); err != nil { - s.logger.Error("failed to decode source branches XRPC response", "err", err) + l.Error("failed to decode source branches XRPC response", "err", err) s.pages.Error503(w) return } @@ -1596,18 +1684,18 @@ targetXrpcBytes, err := tangled.GitTempListBranches(r.Context(), xrpcc, "", 0, f.RepoAt().String()) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.branches for target", "err", xrpcerr) + l.Error("failed to call XRPC repo.branches for target", "err", xrpcerr) s.pages.Error503(w) return } - s.logger.Error("failed to fetch target branches", "err", err) + l.Error("failed to fetch target branches", "err", err) return } // Decode target branches var targetBranches types.RepoBranchesResponse if err := json.Unmarshal(targetXrpcBytes, &targetBranches); err != nil { - s.logger.Error("failed to decode target branches XRPC response", "err", err) + l.Error("failed to decode target branches XRPC response", "err", err) s.pages.Error503(w) return } @@ -1624,14 +1712,20 @@ } func (s *Pulls) ResubmitPull(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "ResubmitPull") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid) switch r.Method { case http.MethodGet: @@ -1655,23 +1749,29 @@ } func (s *Pulls) resubmitPatch(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "resubmitPatch") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid) f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } if user.Active.Did != pull.OwnerDid { - s.logger.Warn("unauthorized user") + l.Error("unauthorized user", "actual_user", user.Active.Did, "expected_owner", pull.OwnerDid) w.WriteHeader(http.StatusUnauthorized) return } @@ -1682,30 +1782,36 @@ } func (s *Pulls) resubmitBranch(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "resubmitBranch") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "resubmit-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid, "target_branch", pull.TargetBranch) f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } if user.Active.Did != pull.OwnerDid { - s.logger.Warn("unauthorized user") + l.Error("unauthorized user", "actual_user", user.Active.Did, "expected_owner", pull.OwnerDid) w.WriteHeader(http.StatusUnauthorized) return } roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} if !roles.IsPushAllowed() { - s.logger.Warn("unauthorized user") + l.Error("unauthorized user - no push permission") w.WriteHeader(http.StatusUnauthorized) return } @@ -1722,18 +1828,18 @@ xrpcBytes, err := tangled.RepoCompare(r.Context(), xrpcc, f.RepoIdentifier(), pull.TargetBranch, pull.PullSource.Branch) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.compare", "err", xrpcerr) + l.Error("failed to call XRPC repo.compare", "err", xrpcerr, "source_branch", pull.PullSource.Branch) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } - s.logger.Error("compare request failed", "err", err) + l.Error("compare request failed", "err", err, "source_branch", pull.PullSource.Branch) s.pages.Notice(w, "resubmit-error", err.Error()) return } var comparison types.RepoFormatPatchResponse if err := json.Unmarshal(xrpcBytes, &comparison); err != nil { - s.logger.Error("failed to decode XRPC compare response", "err", err) + l.Error("failed to decode XRPC compare response", "err", err) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } @@ -1746,30 +1852,36 @@ } func (s *Pulls) resubmitFork(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "resubmitFork") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "resubmit-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid, "target_branch", pull.TargetBranch) f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to get repo and knot", "err", err) + l.Error("failed to get repo and knot", "err", err) return } if user.Active.Did != pull.OwnerDid { - s.logger.Warn("unauthorized user") + l.Error("unauthorized user", "actual_user", user.Active.Did, "expected_owner", pull.OwnerDid) w.WriteHeader(http.StatusUnauthorized) return } forkRepo, err := db.GetRepoByAtUri(s.db, pull.PullSource.RepoAt.String()) if err != nil { - s.logger.Error("failed to get source repo", "err", err) + l.Error("failed to get source repo", "err", err, "repo_at", pull.PullSource.RepoAt.String()) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } @@ -1782,7 +1894,7 @@ oauth.WithDev(s.config.Core.Dev), ) if err != nil { - s.logger.Error("failed to connect to knot server", "err", err) + l.Error("failed to connect to knot server", "err", err, "fork_knot", forkRepo.Knot) return } @@ -1800,7 +1912,7 @@ return } if !resp.Success { - s.logger.Warn("failed to update tracking ref", "err", resp.Error) + l.Error("failed to update tracking ref", "err", resp.Error, "fork_ref", pull.PullSource.Branch, "remote_ref", pull.TargetBranch) s.pages.Notice(w, "resubmit-error", "Failed to update tracking ref.") return } @@ -1815,18 +1927,18 @@ forkXrpcBytes, err := tangled.RepoCompare(r.Context(), &indigoxrpc.Client{Host: forkHost}, forkRepo.RepoIdentifier(), hiddenRef, pull.PullSource.Branch) if err != nil { if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil { - s.logger.Error("failed to call XRPC repo.compare for fork", "err", xrpcerr) + l.Error("failed to call XRPC repo.compare for fork", "err", xrpcerr, "hidden_ref", hiddenRef, "source_branch", pull.PullSource.Branch) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } - s.logger.Error("failed to compare branches", "err", err) + l.Error("failed to compare branches", "err", err, "hidden_ref", hiddenRef, "source_branch", pull.PullSource.Branch) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } var forkComparison types.RepoFormatPatchResponse if err := json.Unmarshal(forkXrpcBytes, &forkComparison); err != nil { - s.logger.Error("failed to decode XRPC compare response for fork", "err", err) + l.Error("failed to decode XRPC compare response for fork", "err", err) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } @@ -1851,9 +1963,12 @@ combined string, sourceRev string, ) { - if pull.IsStacked() { - s.logger.Info("resubmitting stacked PR") - s.resubmitStackedPullHelper(w, r, repo, user, pull, patch, pull.StackId) + l := s.logger.With("handler", "resubmitPullHelper", "user", user.Active.Did, "pull_id", pull.PullId, "target_branch", pull.TargetBranch) + + stack := r.Context().Value("stack").(models.Stack) + if stack != nil && len(stack) != 1 { + l.Info("resubmitting stacked PR", "stack_size", len(stack)) + s.resubmitStackedPullHelper(w, r, repo, user, pull, patch) return } @@ -1875,28 +1990,15 @@ } } - tx, err := s.db.BeginTx(r.Context(), nil) - if err != nil { - s.logger.Error("failed to start tx", "err", err) - s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") - return - } - defer tx.Rollback() - pullAt := pull.AtUri() newRoundNumber := len(pull.Submissions) newPatch := patch newSourceRev := sourceRev combinedPatch := combined - err = db.ResubmitPull(tx, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev) - if err != nil { - s.logger.Error("failed to create pull request", "err", err) - s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") - return - } + client, err := s.oauth.AuthorizedClient(r) if err != nil { - s.logger.Error("failed to authorize client", "err", err) + l.Error("failed to authorize client", "err", err) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } @@ -1904,23 +2006,23 @@ ex, err := comatproto.RepoGetRecord(r.Context(), client, "", tangled.RepoPullNSID, user.Active.Did, pull.Rkey) if err != nil { // failed to get record + l.Error("failed to get record from PDS", "err", err, "rkey", pull.Rkey) s.pages.Notice(w, "resubmit-error", "Failed to update pull, no record found on PDS.") return } blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(patch), ApplicationGzip) if err != nil { - s.logger.Error("failed to upload patch blob", "err", err) + l.Error("failed to upload patch blob", "err", err) s.pages.Notice(w, "resubmit-error", "Failed to update pull request on the PDS. Try again later.") return } record := pull.AsRecord() - record.PatchBlob = blob.Blob + record.Rounds = append(record.Rounds, &tangled.RepoPull_Round{ + CreatedAt: time.Now().Format(time.RFC3339), + PatchBlob: blob.Blob, + }) record.CreatedAt = time.Now().Format(time.RFC3339) - - if record.Source != nil { - record.Source.Sha = newSourceRev - } _, err = comatproto.RepoPutRecord(r.Context(), client, &comatproto.RepoPutRecord_Input{ Collection: tangled.RepoPullNSID, @@ -1932,14 +2034,15 @@ }, }) if err != nil { - s.logger.Error("failed to update record", "err", err) + l.Error("failed to update record on PDS", "err", err, "rkey", pull.Rkey) s.pages.Notice(w, "resubmit-error", "Failed to update pull request on the PDS. Try again later.") return } - if err = tx.Commit(); err != nil { - s.logger.Error("failed to commit transaction", "err", err) - s.pages.Notice(w, "resubmit-error", "Failed to resubmit pull.") + err = db.ResubmitPull(s.db, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, blob.Blob) + if err != nil { + l.Error("failed to resubmit pull request in database", "err", err, "round_number", newRoundNumber) + s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") return } @@ -1954,14 +2057,50 @@ user *oauth.MultiAccountUser, pull *models.Pull, patch string, - stackId string, ) { + l := s.logger.With("handler", "resubmitStackedPullHelper", "user", user.Active.Did, "pull_id", pull.PullId, "target_branch", pull.TargetBranch) + targetBranch := pull.TargetBranch origStack, _ := r.Context().Value("stack").(models.Stack) - newStack, err := s.newStack(r.Context(), repo, user, targetBranch, patch, pull.PullSource, stackId) + + formatPatches, err := patchutil.ExtractPatches(patch) if err != nil { - s.logger.Error("failed to create resubmitted stack", "err", err) + l.Error("failed to extract patches", "err", err) + s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Failed to parse patches.") + return + } + + // must have atleast 1 patch to begin with + if len(formatPatches) == 0 { + l.Error("no patches found in the generated format-patch") + s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request: No patches found in the generated patch.") + return + } + + client, err := s.oauth.AuthorizedClient(r) + if err != nil { + l.Error("failed to get authorized client", "err", err) + s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") + return + } + + // first upload all blobs + blobs := make([]*lexutil.LexBlob, len(formatPatches)) + for i, p := range formatPatches { + blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(p.Raw), ApplicationGzip) + if err != nil { + l.Error("failed to upload patch blob", "err", err, "patch_index", i) + s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") + return + } + l.Info("uploaded blob", "idx", i+1, "total", len(formatPatches)) + blobs[i] = blob.Blob + } + + newStack, err := s.newStack(r.Context(), repo, user, targetBranch, pull.PullSource, formatPatches, blobs) + if err != nil { + l.Error("failed to create resubmitted stack", "err", err) s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") return } @@ -1970,10 +2109,10 @@ origById := make(map[string]*models.Pull) newById := make(map[string]*models.Pull) for _, p := range origStack { - origById[p.ChangeId] = p + origById[p.LatestSubmission().ChangeId()] = p } for _, p := range newStack { - newById[p.ChangeId] = p + newById[p.LatestSubmission().ChangeId()] = p } // commits that got deleted: corresponding pull is closed @@ -1985,41 +2124,50 @@ // pulls in original stack but not in new one for _, op := range origStack { - if _, ok := newById[op.ChangeId]; !ok { - deletions[op.ChangeId] = op + if _, ok := newById[op.LatestSubmission().ChangeId()]; !ok { + deletions[op.LatestSubmission().ChangeId()] = op } } // pulls in new stack but not in original one for _, np := range newStack { - if _, ok := origById[np.ChangeId]; !ok { - additions[np.ChangeId] = np + if _, ok := origById[np.LatestSubmission().ChangeId()]; !ok { + additions[np.LatestSubmission().ChangeId()] = np } } // NOTE: this loop can be written in any of above blocks, // but is written separately in the interest of simpler code for _, np := range newStack { - if op, ok := origById[np.ChangeId]; ok { + if op, ok := origById[np.LatestSubmission().ChangeId()]; ok { // pull exists in both stacks - updated[op.ChangeId] = struct{}{} + updated[op.LatestSubmission().ChangeId()] = struct{}{} } } + // NOTE: we can go through the newStack and update dependent relations and + // rkeys now that we know which ones have been updated + // update dependentOn relations for the entire stack + var parentAt *syntax.ATURI + for _, np := range newStack { + if op, ok := origById[np.LatestSubmission().ChangeId()]; ok { + // pull exists in both stacks + np.Rkey = op.Rkey + } + np.DependentOn = parentAt + x := np.AtUri() + parentAt = &x + } + + l = l.With("additions", len(additions), "deletions", len(deletions), "updates", len(updated)) + tx, err := s.db.Begin() if err != nil { - s.logger.Error("failed to start transaction", "err", err) + l.Error("failed to start transaction", "err", err) s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") return } defer tx.Rollback() - - client, err := s.oauth.AuthorizedClient(r) - if err != nil { - s.logger.Error("failed to authorize client", "err", err) - s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") - return - } // pds updates to make var writes []*comatproto.RepoApplyWrites_Input_Writes_Elem @@ -2031,9 +2179,9 @@ continue } - err := db.DeletePull(tx, p.RepoAt, p.PullId) + err := db.AbandonPulls(tx, orm.FilterEq("repo_at", p.RepoAt), orm.FilterEq("at_uri", p.AtUri())) if err != nil { - s.logger.Error("failed to delete pull", "err", err, "pull_id", p.PullId) + l.Error("failed to delete pull", "err", err, "pull_id", p.PullId) s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") return } @@ -2047,21 +2195,27 @@ // new pulls are created for _, p := range additions { - err := db.NewPull(tx, p) + blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(p.LatestPatch()), ApplicationGzip) if err != nil { - s.logger.Error("failed to create pull", "err", err, "pull_id", p.PullId) + l.Error("failed to upload patch blob for new pull", "err", err, "change_id", p.LatestSubmission().ChangeId()) + s.pages.Notice(w, "resubmit-error", "Failed to update pull request on the PDS. Try again later.") + return + } + p.Submissions[0].Blob = *blob.Blob + + if err = db.PutPull(tx, p); err != nil { + l.Error("failed to create pull", "err", err, "pull_id", p.PullId, "change_id", p.LatestSubmission().ChangeId()) s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") return } - blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(patch), ApplicationGzip) - if err != nil { - s.logger.Error("failed to upload patch blob", "err", err) - s.pages.Notice(w, "resubmit-error", "Failed to update pull request on the PDS. Try again later.") - return - } record := p.AsRecord() - record.PatchBlob = blob.Blob + record.Rounds = []*tangled.RepoPull_Round{ + { + CreatedAt: time.Now().Format(time.RFC3339), + PatchBlob: blob.Blob, + }, + } writes = append(writes, &comatproto.RepoApplyWrites_Input_Writes_Elem{ RepoApplyWrites_Create: &comatproto.RepoApplyWrites_Create{ Collection: tangled.RepoPullNSID, @@ -2084,26 +2238,44 @@ } // resubmit the new pull + np.Rkey = op.Rkey pullAt := op.AtUri() newRoundNumber := len(op.Submissions) newPatch := np.LatestPatch() combinedPatch := np.LatestSubmission().Combined newSourceRev := np.LatestSha() - err := db.ResubmitPull(tx, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev) + + blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(newPatch), ApplicationGzip) if err != nil { - s.logger.Error("failed to update pull", "err", err, "pull_id", op.PullId) + l.Error("failed to upload patch blob for update", "err", err, "change_id", id, "pull_id", op.PullId) + s.pages.Notice(w, "resubmit-error", "Failed to update pull request on the PDS. Try again later.") + return + } + + // create new round + err = db.ResubmitPull(tx, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, blob.Blob) + if err != nil { + l.Error("failed to update pull in database", "err", err, "pull_id", op.PullId, "round_number", newRoundNumber) s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") return } - blob, err := xrpc.RepoUploadBlob(r.Context(), client, gz(patch), ApplicationGzip) - if err != nil { - s.logger.Error("failed to upload patch blob", "err", err) - s.pages.Notice(w, "resubmit-error", "Failed to update pull request on the PDS. Try again later.") - return + // update dependent-on relation + if np.DependentOn != nil { + err := db.SetDependentOn(tx, *np.DependentOn, orm.FilterEq("at_uri", np.AtUri())) + if err != nil { + l.Error("failed to update pull in database", "err", err, "pull_id", op.PullId, "round_number", newRoundNumber) + s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") + return + } } + record := np.AsRecord() - record.PatchBlob = blob.Blob + record.Rounds = op.AsRecord().Rounds + record.Rounds = append(record.Rounds, &tangled.RepoPull_Round{ + CreatedAt: time.Now().Format(time.RFC3339), + PatchBlob: blob.Blob, + }) writes = append(writes, &comatproto.RepoApplyWrites_Input_Writes_Elem{ RepoApplyWrites_Update: &comatproto.RepoApplyWrites_Update{ Collection: tangled.RepoPullNSID, @@ -2115,38 +2287,20 @@ }) } - // update parent-change-id relations for the entire stack - for _, p := range newStack { - err := db.SetPullParentChangeId( - tx, - p.ParentChangeId, - // these should be enough filters to be unique per-stack - orm.FilterEq("repo_at", p.RepoAt.String()), - orm.FilterEq("owner_did", p.OwnerDid), - orm.FilterEq("change_id", p.ChangeId), - ) - - if err != nil { - s.logger.Error("failed to update pull", "err", err, "pull_id", p.PullId) - s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") - return - } - } - - err = tx.Commit() - if err != nil { - s.logger.Error("failed to resubmit pull", "err", err) - s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") - return - } - _, err = comatproto.RepoApplyWrites(r.Context(), client, &comatproto.RepoApplyWrites_Input{ Repo: user.Active.Did, Writes: writes, }) if err != nil { - s.logger.Error("failed to create stacked pull request", "err", err) + l.Error("failed to apply writes for stacked pull request", "err", err, "writes_count", len(writes)) s.pages.Notice(w, "pull", "Failed to create stacked pull request. Try again later.") + return + } + + err = tx.Commit() + if err != nil { + l.Error("failed to commit resubmit transaction", "err", err) + s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") return } @@ -2155,51 +2309,54 @@ } func (s *Pulls) MergePull(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "MergePull") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } + f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to resolve repo:", "err", err) + l.Error("failed to resolve repo", "err", err) s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") return } + l = l.With("repo_at", f.RepoAt().String()) pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") + s.pages.Notice(w, "pull-merge-error", "Failed to merge patch. Try again later.") + return + } + l = l.With("pull_id", pull.PullId, "target_branch", pull.TargetBranch) + + stack, ok := r.Context().Value("stack").(models.Stack) + if !ok { + l.Error("failed to get stack") s.pages.Notice(w, "pull-merge-error", "Failed to merge patch. Try again later.") return } - var pullsToMerge models.Stack - pullsToMerge = append(pullsToMerge, pull) - if pull.IsStacked() { - stack, ok := r.Context().Value("stack").(models.Stack) - if !ok { - s.logger.Error("failed to get stack") - s.pages.Notice(w, "pull-merge-error", "Failed to merge patch. Try again later.") - return - } - - // combine patches of substack - subStack := stack.StrictlyBelow(pull) - // collect the portion of the stack that is mergeable - mergeable := subStack.Mergeable() - // add to total patch - pullsToMerge = append(pullsToMerge, mergeable...) - } + // combine patches of substack + subStack := stack.Below(pull) + // collect the portion of the stack that is mergeable + pullsToMerge := subStack.Mergeable() + l = l.With("pulls_to_merge", len(pullsToMerge)) patch := pullsToMerge.CombinedPatch() ident, err := s.idResolver.ResolveIdent(r.Context(), pull.OwnerDid) if err != nil { - s.logger.Error("resolving identity", "err", err) + l.Error("failed to resolve identity", "err", err, "owner_did", pull.OwnerDid) w.WriteHeader(http.StatusNotFound) return } email, err := db.GetPrimaryEmail(s.db, pull.OwnerDid) if err != nil { - s.logger.Error("failed to get primary email", "err", err) + l.Warn("failed to get primary email", "err", err, "owner_did", pull.OwnerDid) } authorName := ident.Handle.String() @@ -2227,7 +2384,7 @@ oauth.WithDev(s.config.Core.Dev), ) if err != nil { - s.logger.Error("failed to connect to knot server", "err", err) + l.Error("failed to connect to knot server", "err", err, "knot", f.Knot) s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") return } @@ -2240,26 +2397,28 @@ tx, err := s.db.Begin() if err != nil { - s.logger.Error("failed to start transaction", "err", err) + l.Error("failed to start transaction", "err", err) s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") return } defer tx.Rollback() + var atUris []syntax.ATURI for _, p := range pullsToMerge { - err := db.MergePull(tx, f.RepoAt(), p.PullId) - if err != nil { - s.logger.Error("failed to update pull request status in database", "err", err) - s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") - return - } + atUris = append(atUris, p.AtUri()) p.State = models.PullMerged + } + err = db.MergePulls(tx, orm.FilterEq("repo_at", f.RepoAt()), orm.FilterIn("at_uri", atUris)) + if err != nil { + l.Error("failed to update pull request status in database", "err", err) + s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") + return } err = tx.Commit() if err != nil { // TODO: this is unsound, we should also revert the merge from the knotserver here - s.logger.Error("failed to update pull request status in database", "err", err) + l.Error("failed to commit merge transaction", "err", err) s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") return } @@ -2274,20 +2433,26 @@ } func (s *Pulls) ClosePull(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "ClosePull") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("malformed middleware", "err", err) + l.Error("failed to resolve repo", "err", err) return } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid) // auth filter: only owner or collaborators can close roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} @@ -2296,7 +2461,7 @@ isPullAuthor := user.Active.Did == pull.OwnerDid isCloseAllowed := isOwner || isCollaborator || isPullAuthor if !isCloseAllowed { - s.logger.Warn("failed to close pull: unauthorized") + l.Error("unauthorized to close pull", "is_owner", isOwner, "is_collaborator", isCollaborator, "is_pull_author", isPullAuthor) s.pages.Notice(w, "pull-close", "You are unauthorized to close this pull.") return } @@ -2304,36 +2469,33 @@ // Start a transaction tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - s.logger.Error("failed to start transaction", "err", err) + l.Error("failed to start transaction", "err", err) s.pages.Notice(w, "pull-close", "Failed to close pull.") return } defer tx.Rollback() - var pullsToClose []*models.Pull - pullsToClose = append(pullsToClose, pull) - - // if this PR is stacked, then we want to close all PRs below this one on the stack - if pull.IsStacked() { - stack := r.Context().Value("stack").(models.Stack) - subStack := stack.StrictlyBelow(pull) - pullsToClose = append(pullsToClose, subStack...) - } - + // if this PR is stacked, then we want to close all PRs above this one on the stack + stack := r.Context().Value("stack").(models.Stack) + pullsToClose := stack.Above(pull) + var atUris []syntax.ATURI for _, p := range pullsToClose { - // Close the pull in the database - err = db.ClosePull(tx, f.RepoAt(), p.PullId) - if err != nil { - s.logger.Error("failed to close pull", "err", err) - s.pages.Notice(w, "pull-close", "Failed to close pull.") - return - } + atUris = append(atUris, p.AtUri()) p.State = models.PullClosed + } + err = db.ClosePulls( + tx, + orm.FilterEq("repo_at", f.RepoAt()), + orm.FilterIn("at_uri", atUris), + ) + if err != nil { + l.Error("failed to close pulls in database", "err", err, "pulls_to_close", len(pullsToClose)) + s.pages.Notice(w, "pull-close", "Failed to close pull.") } // Commit the transaction if err = tx.Commit(); err != nil { - s.logger.Error("failed to commit transaction", "err", err) + l.Error("failed to commit transaction", "err", err) s.pages.Notice(w, "pull-close", "Failed to close pull.") return } @@ -2347,21 +2509,27 @@ } func (s *Pulls) ReopenPull(w http.ResponseWriter, r *http.Request) { + l := s.logger.With("handler", "ReopenPull") + user := s.oauth.GetMultiAccountUser(r) + if user != nil && user.Active != nil { + l = l.With("user", user.Active.Did) + } f, err := s.repoResolver.Resolve(r) if err != nil { - s.logger.Error("failed to resolve repo", "err", err) + l.Error("failed to resolve repo", "err", err) s.pages.Notice(w, "pull-reopen", "Failed to reopen pull.") return } pull, ok := r.Context().Value("pull").(*models.Pull) if !ok { - s.logger.Error("failed to get pull") + l.Error("failed to get pull") s.pages.Notice(w, "pull-error", "Failed to edit patch. Try again later.") return } + l = l.With("pull_id", pull.PullId, "pull_owner", pull.OwnerDid, "state", pull.State) // auth filter: only owner or collaborators can close roles := repoinfo.RolesInRepo{Roles: s.enforcer.GetPermissionsInRepo(user.Active.Did, f.Knot, f.RepoIdentifier())} @@ -2370,7 +2538,7 @@ isPullAuthor := user.Active.Did == pull.OwnerDid isCloseAllowed := isOwner || isCollaborator || isPullAuthor if !isCloseAllowed { - s.logger.Warn("failed to close pull: unauthorized") + l.Error("unauthorized to reopen pull", "is_owner", isOwner, "is_collaborator", isCollaborator, "is_pull_author", isPullAuthor) s.pages.Notice(w, "pull-close", "You are unauthorized to close this pull.") return } @@ -2378,36 +2546,33 @@ // Start a transaction tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - s.logger.Error("failed to start transaction", "err", err) + l.Error("failed to start transaction", "err", err) s.pages.Notice(w, "pull-reopen", "Failed to reopen pull.") return } defer tx.Rollback() - var pullsToReopen []*models.Pull - pullsToReopen = append(pullsToReopen, pull) - // if this PR is stacked, then we want to reopen all PRs above this one on the stack - if pull.IsStacked() { - stack := r.Context().Value("stack").(models.Stack) - subStack := stack.StrictlyAbove(pull) - pullsToReopen = append(pullsToReopen, subStack...) - } - + stack := r.Context().Value("stack").(models.Stack) + pullsToReopen := stack.Below(pull) + var atUris []syntax.ATURI for _, p := range pullsToReopen { - // Close the pull in the database - err = db.ReopenPull(tx, f.RepoAt(), p.PullId) - if err != nil { - s.logger.Error("failed to close pull", "err", err) - s.pages.Notice(w, "pull-close", "Failed to close pull.") - return - } + atUris = append(atUris, p.AtUri()) p.State = models.PullOpen + } + err = db.ReopenPulls( + tx, + orm.FilterEq("repo_at", f.RepoAt()), + orm.FilterIn("at_uri", atUris), + ) + if err != nil { + l.Error("failed to reopen pulls in database", "err", err, "pulls_to_reopen", len(pullsToReopen)) + s.pages.Notice(w, "pull-close", "Failed to reopen pull.") } // Commit the transaction if err = tx.Commit(); err != nil { - s.logger.Error("failed to commit transaction", "err", err) + l.Error("failed to commit transaction", "err", err) s.pages.Notice(w, "pull-reopen", "Failed to reopen pull.") return } @@ -2420,23 +2585,20 @@ s.pages.HxLocation(w, fmt.Sprintf("/%s/pulls/%d", ownerSlashRepo, pull.PullId)) } -func (s *Pulls) newStack(ctx context.Context, repo *models.Repo, user *oauth.MultiAccountUser, targetBranch, patch string, pullSource *models.PullSource, stackId string) (models.Stack, error) { - formatPatches, err := patchutil.ExtractPatches(patch) - if err != nil { - return nil, fmt.Errorf("Failed to extract patches: %v", err) - } - - // must have atleast 1 patch to begin with - if len(formatPatches) == 0 { - return nil, fmt.Errorf("No patches found in the generated format-patch.") - } - - // the stack is identified by a UUID +func (s *Pulls) newStack( + ctx context.Context, + repo *models.Repo, + user *oauth.MultiAccountUser, + targetBranch string, + pullSource *models.PullSource, + formatPatches []types.FormatPatch, + blobs []*lexutil.LexBlob, +) (models.Stack, error) { var stack models.Stack - parentChangeId := "" - for _, fp := range formatPatches { + var parentAtUri *syntax.ATURI + for i, fp := range formatPatches { // all patches must have a jj change-id - changeId, err := fp.ChangeId() + _, err := fp.ChangeId() if err != nil { return nil, fmt.Errorf("Stacking is only supported if all patches contain a change-id commit header.") } @@ -2451,6 +2613,7 @@ Patch: fp.Raw, SourceRev: fp.SHA, Combined: fp.Raw, + Blob: *blobs[i], } pull := models.Pull{ Title: title, @@ -2466,15 +2629,16 @@ }, PullSource: pullSource, Created: time.Now(), + State: models.PullOpen, - StackId: stackId, - ChangeId: changeId, - ParentChangeId: parentChangeId, + DependentOn: parentAtUri, + Repo: repo, } stack = append(stack, &pull) - parentChangeId = changeId + parent := pull.AtUri() + parentAtUri = &parent } return stack, nil diff --git a/appview/notify/db/db.go b/appview/notify/db/db.go --- a/appview/notify/db/db.go +++ b/appview/notify/db/db.go @@ -285,8 +285,8 @@ l := log.FromContext(ctx) pull, err := db.GetPull(n.db, - syntax.ATURI(comment.RepoAt), - comment.PullId, + orm.FilterEq("repo_at", syntax.ATURI(comment.RepoAt)), + orm.FilterEq("pull_id", comment.PullId), ) if err != nil { l.Error("failed to get pulls", "err", err)