mirror of
https://github.com/go-gitea/gitea.git
synced 2026-10-26 06:16:56 +09:00
Adds a batched GraphQL path for GitHub migrations that fetches each issue or pull request together with its comments, reviews, reactions and timeline events instead of the REST N+1 pattern, sharply reducing rate-limit pressure on large repositories. - GraphQL is used for authenticated migrations with issues or pull requests and falls back to REST automatically when it fails - timeline events (close/reopen, lock, rename, label, milestone, merge, branch deletion, pin, auto-merge) are imported as typed comments showing their GitHub author - GitBucket issue migrations no longer stop after the first page, GitBucket sends no Link header Fixes: https://github.com/go-gitea/gitea/issues/13243 Fixes: https://github.com/go-gitea/gitea/issues/31856 Signed-off-by: David E. Lovas <david@rexovas.com> Co-authored-by: bircni <bircni@icloud.com> Co-authored-by: silverwind <me@silverwind.io> Co-authored-by: Giteabot <teabot@gitea.io>
717 lines
22 KiB
Go
717 lines
22 KiB
Go
// Copyright 2026 The Gitea Authors. All rights reserved.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package migrations
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/tls"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"maps"
|
|
"net"
|
|
"net/http"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.dev/modules/egress/policy"
|
|
"gitea.dev/modules/json"
|
|
"gitea.dev/modules/log"
|
|
base "gitea.dev/modules/migration"
|
|
|
|
"github.com/google/go-github/v92/github"
|
|
)
|
|
|
|
const (
|
|
githubGraphQLPageSize = 100
|
|
githubGraphQLPRPageSize = 20
|
|
githubGraphQLPointsFloor = 10
|
|
maxGraphQLRateRetries = 5
|
|
maxGraphQLTransientRetries = 5
|
|
graphQLRateResetCap = time.Hour
|
|
graphQLNodeBatchSize = 100
|
|
graphQLLabelPageSize = 30
|
|
graphQLAssigneePageSize = 30
|
|
graphQLReviewRequestPageSize = 50
|
|
)
|
|
|
|
const (
|
|
gqlPageInfoFields = "pageInfo{hasNextPage endCursor}"
|
|
// databaseId is not on the Actor interface, every implementation needs its own fragment
|
|
gqlActorFields = "__typename login ... on User{databaseId} ... on Bot{databaseId} ... on Organization{databaseId} ... on Mannequin{databaseId}"
|
|
// teams in the RequestedReviewer union have no login
|
|
gqlReviewerFields = "__typename ... on User{login databaseId} ... on Bot{login databaseId} ... on Mannequin{login databaseId}"
|
|
gqlReactionFields = "content user{login databaseId}"
|
|
)
|
|
|
|
var graphQLTransientRetryBase = 2 * time.Second
|
|
|
|
var graphQLRateResetFallback = time.Minute // GitHub requires a minute between secondary limit retries.
|
|
|
|
// a deterministic 502 for a too-heavy page never succeeds at the same size, so callers shrink it
|
|
var errGraphQLTransientExhausted = errors.New("github graphql: transient retries exhausted")
|
|
|
|
var _ base.Downloader = &GithubGraphQLDownloader{}
|
|
|
|
// GithubGraphQLDownloader sweeps issues and pull requests over GraphQL, the embedded REST downloader serves the rest and takes over on failure
|
|
type GithubGraphQLDownloader struct {
|
|
*GithubDownloaderV3
|
|
gqlRates []*github.Rate // GraphQL has a separate rate budget
|
|
skipComments bool
|
|
issuesCursor string
|
|
prCursor string
|
|
restFallback bool
|
|
// REST restarts from its first page after a fallback, the migration skips what it already imported
|
|
restIssuePageOffset int
|
|
restPRPageOffset int
|
|
}
|
|
|
|
func newGithubGraphQLDownloader(rest *GithubDownloaderV3, skipComments bool) *GithubGraphQLDownloader {
|
|
return &GithubGraphQLDownloader{GithubDownloaderV3: rest, gqlRates: make([]*github.Rate, len(rest.clients)), skipComments: skipComments}
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) GetIssues(ctx context.Context, page, perPage int) ([]*base.Issue, bool, error) {
|
|
if !g.restFallback {
|
|
issues, isEnd, err := g.getIssuesGraphQL(ctx, page, perPage)
|
|
if !g.fallBackToREST(ctx, err) {
|
|
return issues, isEnd, err
|
|
}
|
|
g.restIssuePageOffset = page - 1
|
|
}
|
|
return g.GithubDownloaderV3.GetIssues(ctx, page-g.restIssuePageOffset, perPage)
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) GetPullRequests(ctx context.Context, page, perPage int) ([]*base.PullRequest, bool, error) {
|
|
if !g.restFallback {
|
|
prs, isEnd, err := g.getPullRequestsGraphQL(ctx, page, perPage)
|
|
if !g.fallBackToREST(ctx, err) {
|
|
return prs, isEnd, err
|
|
}
|
|
g.restPRPageOffset = page - 1
|
|
}
|
|
return g.GithubDownloaderV3.GetPullRequests(ctx, page-g.restPRPageOffset, perPage)
|
|
}
|
|
|
|
// fallBackToREST switches to REST for good when GraphQL fails, e.g. on a server or token without GraphQL support
|
|
func (g *GithubGraphQLDownloader) fallBackToREST(ctx context.Context, err error) bool {
|
|
if err == nil || ctx.Err() != nil {
|
|
return false
|
|
}
|
|
log.Warn("GitHub GraphQL failed, falling back to REST: %v", err)
|
|
g.restFallback = true
|
|
return true
|
|
}
|
|
|
|
// SupportGetRepoComments is false so comments are served per issue from what each sweep page prefetched
|
|
func (g *GithubGraphQLDownloader) SupportGetRepoComments() bool {
|
|
return false
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) GetComments(ctx context.Context, commentable base.Commentable) ([]*base.Comment, bool, error) {
|
|
if prefetched, ok := commentable.GetContext().(*gqlPrefetched); ok {
|
|
return prefetched.comments, false, nil
|
|
}
|
|
return g.GithubDownloaderV3.GetComments(ctx, commentable)
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) GetReviews(ctx context.Context, reviewable base.Reviewable) ([]*base.Review, error) {
|
|
if prefetched, ok := reviewable.GetContext().(*gqlPrefetched); ok {
|
|
return prefetched.reviews, nil
|
|
}
|
|
return g.GithubDownloaderV3.GetReviews(ctx, reviewable)
|
|
}
|
|
|
|
type graphQLRequest struct {
|
|
Query string `json:"query"`
|
|
Variables map[string]any `json:"variables"`
|
|
}
|
|
|
|
type graphQLError struct {
|
|
Type string `json:"type"`
|
|
Message string `json:"message"`
|
|
Path []any `json:"path"`
|
|
}
|
|
|
|
func (e graphQLError) Error() string {
|
|
if e.Type != "" {
|
|
return e.Type + ": " + e.Message
|
|
}
|
|
return e.Message
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) graphQLEndpoint() string {
|
|
restURL := g.clients[0].BaseURL()
|
|
if host, ok := strings.CutSuffix(restURL, "/api/v3/"); ok {
|
|
return host + "/api/graphql"
|
|
}
|
|
return restURL + "graphql"
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) doGraphQL(ctx context.Context, query string, vars map[string]any, out any) error {
|
|
payload, err := json.Marshal(graphQLRequest{Query: query, Variables: vars})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rateWaits, transientRetries := 0, 0
|
|
for {
|
|
if err := g.pickGraphQLClient(ctx); err != nil {
|
|
return err
|
|
}
|
|
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, g.graphQLEndpoint(), bytes.NewReader(payload))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("Accept", "application/json")
|
|
|
|
resp, err := g.getClient().Client().Do(req)
|
|
if err != nil {
|
|
if ctx.Err() != nil || !isTransientTransportError(err) {
|
|
return err
|
|
}
|
|
if err := g.waitGraphQLTransient(ctx, &transientRetries, fmt.Sprintf("request failed: %v", err)); err != nil {
|
|
return err
|
|
}
|
|
continue
|
|
}
|
|
g.recordGraphQLRate(resp.Header)
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
detail := readGraphQLErrorBody(resp.Body)
|
|
resp.Body.Close()
|
|
switch {
|
|
case resp.StatusCode >= http.StatusInternalServerError:
|
|
if err := g.waitGraphQLTransient(ctx, &transientRetries, "status "+resp.Status+detail); err != nil {
|
|
return err
|
|
}
|
|
continue
|
|
case isGraphQLThrottled(resp, detail):
|
|
if rateWaits < maxGraphQLRateRetries*len(g.clients) {
|
|
rateWaits++
|
|
g.restGraphQLClient(resp.Header)
|
|
continue
|
|
}
|
|
return fmt.Errorf("doGraphQL: rate limited, status %s%s", resp.Status, detail)
|
|
default:
|
|
return fmt.Errorf("doGraphQL: unexpected status %s%s", resp.Status, detail)
|
|
}
|
|
}
|
|
|
|
envelope := struct {
|
|
Data any `json:"data"`
|
|
Errors []graphQLError `json:"errors"`
|
|
}{Data: out}
|
|
decodeErr := json.NewDecoder(resp.Body).Decode(&envelope)
|
|
resp.Body.Close()
|
|
if decodeErr != nil {
|
|
if err := g.waitGraphQLTransient(ctx, &transientRetries, fmt.Sprintf("truncated response: %v", decodeErr)); err != nil {
|
|
return err
|
|
}
|
|
continue
|
|
}
|
|
|
|
if !onlyUnresolvedNodes(envelope.Errors) {
|
|
if isGraphQLRateLimited(envelope.Errors) && rateWaits < maxGraphQLRateRetries*len(g.clients) {
|
|
rateWaits++
|
|
g.restGraphQLClient(resp.Header)
|
|
continue
|
|
}
|
|
return fmt.Errorf("doGraphQL: %w", envelope.Errors[0])
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
const maxGraphQLErrorBody = 512
|
|
|
|
func readGraphQLErrorBody(body io.Reader) string {
|
|
buf, err := io.ReadAll(io.LimitReader(body, maxGraphQLErrorBody))
|
|
if err != nil || len(buf) == 0 {
|
|
return ""
|
|
}
|
|
return ": " + strings.Join(strings.Fields(string(buf)), " ")
|
|
}
|
|
|
|
func isGraphQLThrottled(resp *http.Response, body string) bool {
|
|
switch resp.StatusCode {
|
|
case http.StatusTooManyRequests:
|
|
return true
|
|
case http.StatusForbidden:
|
|
return resp.Header.Get("Retry-After") != "" ||
|
|
resp.Header.Get(github.HeaderRateRemaining) == "0" ||
|
|
strings.Contains(strings.ToLower(body), "secondary rate limit")
|
|
}
|
|
return false
|
|
}
|
|
|
|
func isTransientTransportError(err error) bool {
|
|
if errors.Is(err, policy.ErrDenied) {
|
|
return false
|
|
}
|
|
if _, ok := errors.AsType[*tls.CertificateVerificationError](err); ok {
|
|
return false
|
|
}
|
|
if dnsErr, ok := errors.AsType[*net.DNSError](err); ok {
|
|
return !dnsErr.IsNotFound
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) waitGraphQLTransient(ctx context.Context, attempt *int, reason string) error {
|
|
if *attempt >= maxGraphQLTransientRetries {
|
|
return fmt.Errorf("%w: %s", errGraphQLTransientExhausted, reason)
|
|
}
|
|
wait := graphQLTransientRetryBase << *attempt
|
|
*attempt++
|
|
log.Warn("github graphql: transient failure (%s), retrying in %s (attempt %d/%d)",
|
|
reason, wait, *attempt, maxGraphQLTransientRetries)
|
|
timer := time.NewTimer(wait)
|
|
defer timer.Stop()
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-timer.C:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func onlyUnresolvedNodes(errs []graphQLError) bool {
|
|
return !slices.ContainsFunc(errs, func(e graphQLError) bool {
|
|
return e.Type != "NOT_FOUND" || len(e.Path) == 0 || e.Path[0] != "nodes"
|
|
})
|
|
}
|
|
|
|
func isGraphQLRateLimited(errs []graphQLError) bool {
|
|
return slices.ContainsFunc(errs, func(e graphQLError) bool { return e.Type == "RATE_LIMITED" })
|
|
}
|
|
|
|
func isGraphQLTooHeavy(err error) bool {
|
|
var ge graphQLError
|
|
return errors.Is(err, errGraphQLTransientExhausted) || errors.As(err, &ge) && ge.Type == "RESOURCE_LIMITS_EXCEEDED"
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) pickGraphQLClient(ctx context.Context) error {
|
|
for {
|
|
best, soonest := -1, -1
|
|
for i, rate := range g.gqlRates {
|
|
if rate == nil || !time.Now().Before(rate.Reset.Time) {
|
|
g.curClientIdx = i
|
|
return nil
|
|
}
|
|
if rate.Remaining > githubGraphQLPointsFloor && (best < 0 || rate.Remaining > g.gqlRates[best].Remaining) {
|
|
best = i
|
|
}
|
|
if soonest < 0 || rate.Reset.Before(g.gqlRates[soonest].Reset.Time) {
|
|
soonest = i
|
|
}
|
|
}
|
|
if best >= 0 {
|
|
g.curClientIdx = best
|
|
return nil
|
|
}
|
|
wait := min(time.Until(g.gqlRates[soonest].Reset.Time), graphQLRateResetCap)
|
|
log.Info("github graphql: every token's points budget is spent, sleeping %s until the earliest reset", wait.Round(time.Second))
|
|
timer := time.NewTimer(wait)
|
|
select {
|
|
case <-ctx.Done():
|
|
timer.Stop()
|
|
return ctx.Err()
|
|
case <-timer.C:
|
|
g.gqlRates[soonest] = nil // the reset may lie beyond the capped sleep
|
|
}
|
|
}
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) recordGraphQLRate(header http.Header) {
|
|
remaining, err := strconv.Atoi(header.Get(github.HeaderRateRemaining))
|
|
if err != nil {
|
|
return
|
|
}
|
|
rate := &github.Rate{Remaining: remaining}
|
|
if epoch, err := strconv.ParseInt(header.Get(github.HeaderRateReset), 10, 64); err == nil {
|
|
rate.Reset = github.Timestamp{Time: time.Unix(epoch, 0)}
|
|
}
|
|
g.gqlRates[g.curClientIdx] = rate
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) restGraphQLClient(header http.Header) {
|
|
wait := graphQLRateResetWait(header)
|
|
log.Info("github graphql: token %d throttled, resting it for %s", g.curClientIdx, wait.Round(time.Second))
|
|
g.gqlRates[g.curClientIdx] = &github.Rate{Reset: github.Timestamp{Time: time.Now().Add(wait)}}
|
|
}
|
|
|
|
func isPrimaryBudgetSpent(header http.Header) bool {
|
|
remaining := header.Get(github.HeaderRateRemaining)
|
|
return remaining == "" || remaining == "0"
|
|
}
|
|
|
|
func graphQLRateResetWait(header http.Header) time.Duration {
|
|
wait := graphQLRateResetFallback
|
|
if ra := header.Get("Retry-After"); ra != "" {
|
|
if secs, err := strconv.Atoi(strings.TrimSpace(ra)); err == nil && secs > 0 {
|
|
wait = time.Duration(secs) * time.Second
|
|
}
|
|
} else if reset := header.Get(github.HeaderRateReset); reset != "" && isPrimaryBudgetSpent(header) {
|
|
if epoch, err := strconv.ParseInt(reset, 10, 64); err == nil {
|
|
if d := time.Until(time.Unix(epoch, 0)); d > 0 {
|
|
wait = d
|
|
}
|
|
}
|
|
}
|
|
return min(wait, graphQLRateResetCap)
|
|
}
|
|
|
|
// graphQLContentQuery orders by creation since a mutable sort key skips rows, nested comment reactions exceed GitHub's node limit
|
|
func (g *GithubGraphQLDownloader) graphQLContentQuery(connection, prFields, prDiscussion string) string {
|
|
fields := prFields
|
|
if !g.SkipReactions {
|
|
fields += " reactions(first:100){" + gqlPageInfoFields + " nodes{" + gqlReactionFields + "}}"
|
|
}
|
|
if !g.skipComments {
|
|
fields += " comments(first:100){" + gqlPageInfoFields + " nodes{id databaseId body createdAt updatedAt author{" + gqlActorFields + "}}}" + prDiscussion
|
|
}
|
|
return fmt.Sprintf(`
|
|
query($owner:String!,$name:String!,$cursor:String,$first:Int!){
|
|
repository(owner:$owner,name:$name){
|
|
%[1]s(first:$first,after:$cursor,orderBy:{field:CREATED_AT,direction:ASC}){
|
|
%[2]s
|
|
nodes{
|
|
id number title body state createdAt updatedAt closedAt locked
|
|
author{%[3]s}
|
|
milestone{title}
|
|
labels(first:%[4]d){%[2]s nodes{name color description}}
|
|
assignees(first:%[5]d){%[2]s nodes{login}}
|
|
%[6]s
|
|
}
|
|
}
|
|
}
|
|
}`, connection, gqlPageInfoFields, gqlActorFields, graphQLLabelPageSize, graphQLAssigneePageSize, fields)
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) graphQLIssuesQuery() string {
|
|
return g.graphQLContentQuery("issues", "", "")
|
|
}
|
|
|
|
func nodeConnQuery(onType, connection, nodeFields string) string {
|
|
return `query($id:ID!,$cursor:String){node(id:$id){... on ` + onType + `{conn:` + connection +
|
|
`(first:100,after:$cursor){` + gqlPageInfoFields + ` nodes{` + nodeFields + `}}}}}`
|
|
}
|
|
|
|
var (
|
|
graphQLLabelsSweepQuery = nodeConnQuery("Labelable", "labels", "name color description")
|
|
graphQLAssigneesSweepQuery = nodeConnQuery("Assignable", "assignees", "login")
|
|
graphQLReactionsSweepQuery = nodeConnQuery("Reactable", "reactions", gqlReactionFields)
|
|
)
|
|
|
|
const graphQLNodeReactionsQuery = `query($ids:[ID!]!){nodes(ids:$ids){id ... on Reactable{reactions(first:100){` +
|
|
gqlPageInfoFields + ` nodes{` + gqlReactionFields + `}}}}}`
|
|
|
|
type gqlPageInfo struct {
|
|
HasNextPage bool `json:"hasNextPage"`
|
|
EndCursor string `json:"endCursor"`
|
|
}
|
|
|
|
type gqlConn[T any] struct {
|
|
PageInfo gqlPageInfo `json:"pageInfo"`
|
|
Nodes []T `json:"nodes"`
|
|
}
|
|
|
|
type gqlActor struct {
|
|
Typename string `json:"__typename"`
|
|
Login string `json:"login"`
|
|
DatabaseID int64 `json:"databaseId"`
|
|
}
|
|
|
|
func (a *gqlActor) UnmarshalJSON(data []byte) error {
|
|
type plain gqlActor
|
|
if err := json.Unmarshal(data, (*plain)(a)); err != nil {
|
|
return err
|
|
}
|
|
switch {
|
|
case a.Typename == "" && a.Login == "":
|
|
a.Login = "ghost" // Match REST's ghost login with an unset ID for GHES.
|
|
case a.Typename == "Bot":
|
|
a.Login += "[bot]"
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type gqlCommit struct {
|
|
OID string `json:"oid"`
|
|
}
|
|
|
|
type gqlReaction struct {
|
|
Content string `json:"content"`
|
|
User gqlActor `json:"user"`
|
|
}
|
|
|
|
type gqlComment struct {
|
|
ID string `json:"id"`
|
|
DatabaseID int64 `json:"databaseId"`
|
|
Body string `json:"body"`
|
|
CreatedAt time.Time `json:"createdAt"`
|
|
UpdatedAt time.Time `json:"updatedAt"`
|
|
Author gqlActor `json:"author"`
|
|
}
|
|
|
|
type gqlLabel struct {
|
|
Name string `json:"name"`
|
|
Color string `json:"color"`
|
|
Description string `json:"description"`
|
|
}
|
|
|
|
type gqlIssue struct {
|
|
ID string `json:"id"`
|
|
Number int64 `json:"number"`
|
|
Title string `json:"title"`
|
|
Body string `json:"body"`
|
|
State string `json:"state"`
|
|
CreatedAt time.Time `json:"createdAt"`
|
|
UpdatedAt time.Time `json:"updatedAt"`
|
|
ClosedAt *time.Time `json:"closedAt"`
|
|
Locked bool `json:"locked"`
|
|
Author gqlActor `json:"author"`
|
|
Milestone *struct {
|
|
Title string `json:"title"`
|
|
} `json:"milestone"`
|
|
Labels gqlConn[gqlLabel] `json:"labels"`
|
|
Assignees gqlConn[gqlActor] `json:"assignees"`
|
|
Reactions gqlConn[gqlReaction] `json:"reactions"`
|
|
Comments gqlConn[gqlComment] `json:"comments"`
|
|
}
|
|
|
|
type gqlIssuesResponse struct {
|
|
Repository struct {
|
|
Issues gqlConn[gqlIssue] `json:"issues"`
|
|
} `json:"repository"`
|
|
}
|
|
|
|
type gqlPrefetched struct {
|
|
index int64
|
|
comments []*base.Comment
|
|
commentIDs []string
|
|
reviews []*base.Review
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) getIssuesGraphQL(ctx context.Context, page, perPage int) ([]*base.Issue, bool, error) {
|
|
if page <= 1 {
|
|
g.issuesCursor = ""
|
|
}
|
|
var resp gqlIssuesResponse
|
|
if err := g.doGraphQLPage(ctx, g.graphQLIssuesQuery(), g.issuesCursor, graphQLPageSize(perPage, githubGraphQLPageSize), &resp); err != nil {
|
|
return nil, false, err
|
|
}
|
|
conn := resp.Repository.Issues
|
|
prefetched := map[string]*gqlPrefetched{}
|
|
issues := make([]*base.Issue, 0, len(conn.Nodes))
|
|
for i := range conn.Nodes {
|
|
issue, err := g.convertGraphQLIssue(ctx, &conn.Nodes[i], prefetched)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
issues = append(issues, issue)
|
|
}
|
|
if err := g.enrichGraphQLPage(ctx, prefetched); err != nil {
|
|
return nil, false, err
|
|
}
|
|
g.issuesCursor = conn.PageInfo.EndCursor // only now, a failed page is retried from the same cursor
|
|
return issues, !conn.PageInfo.HasNextPage, nil
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) convertGraphQLIssue(ctx context.Context, node *gqlIssue, prefetchedByNodeID map[string]*gqlPrefetched) (*base.Issue, error) {
|
|
labels, err := allNodes(ctx, g, graphQLLabelsSweepQuery, node.ID, node.Labels)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
assignees, err := allNodes(ctx, g, graphQLAssigneesSweepQuery, node.ID, node.Assignees)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
reactions, err := g.allReactions(ctx, node.ID, node.Reactions)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
issue := &base.Issue{
|
|
Number: node.Number,
|
|
Title: node.Title,
|
|
Content: node.Body,
|
|
PosterID: node.Author.DatabaseID,
|
|
PosterName: node.Author.Login,
|
|
State: strings.ToLower(node.State),
|
|
Created: node.CreatedAt,
|
|
Updated: node.UpdatedAt,
|
|
Closed: node.ClosedAt,
|
|
IsLocked: node.Locked,
|
|
Labels: make([]*base.Label, 0, len(labels)),
|
|
Reactions: reactions,
|
|
ForeignIndex: node.Number,
|
|
}
|
|
if node.Milestone != nil {
|
|
issue.Milestone = node.Milestone.Title
|
|
}
|
|
for _, label := range labels {
|
|
issue.Labels = append(issue.Labels, &base.Label{Name: label.Name, Color: label.Color, Description: label.Description})
|
|
}
|
|
for _, assignee := range assignees {
|
|
issue.Assignees = append(issue.Assignees, assignee.Login)
|
|
}
|
|
if g.skipComments {
|
|
return issue, nil
|
|
}
|
|
|
|
prefetched := &gqlPrefetched{index: node.Number}
|
|
issue.Context = prefetched
|
|
prefetchedByNodeID[node.ID] = prefetched
|
|
if node.Comments.PageInfo.HasNextPage {
|
|
prefetched.comments, err = g.getComments(ctx, issue)
|
|
return issue, err
|
|
}
|
|
for _, comment := range node.Comments.Nodes {
|
|
prefetched.commentIDs = append(prefetched.commentIDs, comment.ID)
|
|
prefetched.comments = append(prefetched.comments, &base.Comment{
|
|
IssueIndex: node.Number,
|
|
Index: comment.DatabaseID,
|
|
PosterID: comment.Author.DatabaseID,
|
|
PosterName: comment.Author.Login,
|
|
Content: comment.Body,
|
|
Created: comment.CreatedAt,
|
|
Updated: comment.UpdatedAt,
|
|
})
|
|
}
|
|
return issue, nil
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) enrichGraphQLPage(ctx context.Context, prefetchedByNodeID map[string]*gqlPrefetched) error {
|
|
if !g.SkipReactions {
|
|
comments := map[string]*base.Comment{}
|
|
for _, prefetched := range prefetchedByNodeID {
|
|
for i, id := range prefetched.commentIDs {
|
|
comments[id] = prefetched.comments[i]
|
|
}
|
|
}
|
|
for batch := range slices.Chunk(slices.Sorted(maps.Keys(comments)), graphQLNodeBatchSize) {
|
|
reactions, err := g.fetchNodeReactions(ctx, batch)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for id, reaction := range reactions {
|
|
comments[id].Reactions = reaction
|
|
}
|
|
}
|
|
}
|
|
return g.attachTimelineEvents(ctx, prefetchedByNodeID)
|
|
}
|
|
|
|
func graphQLPageSize(perPage, maxSize int) int {
|
|
if perPage <= 0 {
|
|
return maxSize
|
|
}
|
|
return min(perPage, maxSize) // Keep pages within the caller's database batch size.
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) doGraphQLPage(ctx context.Context, query, cursor string, pageSize int, out any) error {
|
|
vars := map[string]any{"owner": g.repoOwner, "name": g.repoName}
|
|
if cursor != "" {
|
|
vars["cursor"] = cursor
|
|
}
|
|
for {
|
|
vars["first"] = pageSize
|
|
err := g.doGraphQL(ctx, query, vars, out)
|
|
if err == nil || !isGraphQLTooHeavy(err) || pageSize <= 1 {
|
|
return err
|
|
}
|
|
pageSize /= 2
|
|
log.Warn("github graphql: page too heavy (%v), retrying with %d", err, pageSize)
|
|
}
|
|
}
|
|
|
|
func allNodes[T any](ctx context.Context, g *GithubGraphQLDownloader, sweepQuery, nodeID string, conn gqlConn[T]) ([]T, error) {
|
|
nodes := conn.Nodes
|
|
for page := conn.PageInfo; page.HasNextPage; {
|
|
var resp struct {
|
|
Node struct {
|
|
Conn gqlConn[T] `json:"conn"`
|
|
} `json:"node"`
|
|
}
|
|
if err := g.doGraphQL(ctx, sweepQuery, map[string]any{"id": nodeID, "cursor": page.EndCursor}, &resp); err != nil {
|
|
return nil, err
|
|
}
|
|
nodes = append(nodes, resp.Node.Conn.Nodes...)
|
|
page = resp.Node.Conn.PageInfo
|
|
}
|
|
return nodes, nil
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) allReactions(ctx context.Context, nodeID string, conn gqlConn[gqlReaction]) ([]*base.Reaction, error) {
|
|
nodes, err := allNodes(ctx, g, graphQLReactionsSweepQuery, nodeID, conn)
|
|
if err != nil || len(nodes) == 0 {
|
|
return nil, err
|
|
}
|
|
reactions := make([]*base.Reaction, 0, len(nodes))
|
|
for _, reaction := range nodes {
|
|
reactions = append(reactions, &base.Reaction{
|
|
UserID: reaction.User.DatabaseID,
|
|
UserName: reaction.User.Login,
|
|
Content: graphQLReactionContent(reaction.Content),
|
|
})
|
|
}
|
|
return reactions, nil
|
|
}
|
|
|
|
func (g *GithubGraphQLDownloader) fetchNodeReactions(ctx context.Context, ids []string) (map[string][]*base.Reaction, error) {
|
|
var resp struct {
|
|
Nodes []struct {
|
|
ID string `json:"id"`
|
|
Reactions gqlConn[gqlReaction] `json:"reactions"`
|
|
} `json:"nodes"`
|
|
}
|
|
if err := g.doGraphQL(ctx, graphQLNodeReactionsQuery, map[string]any{"ids": ids}, &resp); err != nil {
|
|
if isGraphQLTooHeavy(err) && len(ids) > 1 {
|
|
return fetchHalves(ctx, ids, g.fetchNodeReactions)
|
|
}
|
|
return nil, err
|
|
}
|
|
out := make(map[string][]*base.Reaction, len(resp.Nodes))
|
|
for _, node := range resp.Nodes {
|
|
if node.ID == "" {
|
|
continue
|
|
}
|
|
reactions, err := g.allReactions(ctx, node.ID, node.Reactions)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out[node.ID] = reactions
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func fetchHalves[V any](ctx context.Context, ids []string, fetch func(context.Context, []string) (map[string]V, error)) (map[string]V, error) {
|
|
left, err := fetch(ctx, ids[:len(ids)/2])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
right, err := fetch(ctx, ids[len(ids)/2:])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
maps.Copy(left, right)
|
|
return left, nil
|
|
}
|
|
|
|
func graphQLReactionContent(enum string) string {
|
|
switch enum {
|
|
case "THUMBS_UP":
|
|
return "+1"
|
|
case "THUMBS_DOWN":
|
|
return "-1"
|
|
default:
|
|
return strings.ToLower(enum)
|
|
}
|
|
}
|