1package github
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "fmt"
8 "io/ioutil"
9 "net/http"
10 "os"
11 "strings"
12 "time"
13
14 "github.com/pkg/errors"
15 "github.com/shurcooL/githubv4"
16 "golang.org/x/sync/errgroup"
17
18 "github.com/MichaelMure/git-bug/bridge/core"
19 "github.com/MichaelMure/git-bug/bridge/core/auth"
20 "github.com/MichaelMure/git-bug/cache"
21 "github.com/MichaelMure/git-bug/entities/bug"
22 "github.com/MichaelMure/git-bug/entities/identity"
23 "github.com/MichaelMure/git-bug/entity"
24 "github.com/MichaelMure/git-bug/entity/dag"
25)
26
27var (
28 ErrMissingIdentityToken = errors.New("missing identity token")
29)
30
31// githubExporter implement the Exporter interface
32type githubExporter struct {
33 conf core.Configuration
34
35 // cache identities clients
36 identityClient map[entity.Id]*rateLimitHandlerClient
37
38 // the client to use for non user-specific queries
39 // it's the client associated to the "default-login" config
40 // used for the github V4 API (graphql)
41 defaultClient *rateLimitHandlerClient
42
43 // the token of the default user
44 // it's the token associated to the "default-login" config
45 // used for the github V3 API (REST)
46 defaultToken *auth.Token
47
48 // github repository ID
49 repositoryID string
50
51 // cache identifiers used to speed up exporting operations
52 // cleared for each bug
53 cachedOperationIDs map[entity.Id]string
54
55 // cache labels used to speed up exporting labels events
56 cachedLabels map[string]string
57
58 // channel to send export results
59 out chan<- core.ExportResult
60}
61
62// Init .
63func (ge *githubExporter) Init(_ context.Context, repo *cache.RepoCache, conf core.Configuration) error {
64 ge.conf = conf
65 ge.identityClient = make(map[entity.Id]*rateLimitHandlerClient)
66 ge.cachedOperationIDs = make(map[entity.Id]string)
67 ge.cachedLabels = make(map[string]string)
68
69 // preload all clients
70 err := ge.cacheAllClient(repo)
71 if err != nil {
72 return err
73 }
74
75 return nil
76}
77
78func (ge *githubExporter) cacheAllClient(repo *cache.RepoCache) error {
79 creds, err := auth.List(repo, auth.WithTarget(target), auth.WithKind(auth.KindToken))
80 if err != nil {
81 return err
82 }
83
84 for _, cred := range creds {
85 login, ok := cred.GetMetadata(auth.MetaKeyLogin)
86 if !ok {
87 _, _ = fmt.Fprintf(os.Stderr, "credential %s is not tagged with a Github login\n", cred.ID().Human())
88 continue
89 }
90
91 user, err := repo.ResolveIdentityImmutableMetadata(metaKeyGithubLogin, login)
92 if err == identity.ErrIdentityNotExist {
93 continue
94 }
95 if err != nil {
96 return nil
97 }
98
99 if _, ok := ge.identityClient[user.Id()]; ok {
100 continue
101 }
102
103 client := buildClient(creds[0].(*auth.Token))
104 ge.identityClient[user.Id()] = client
105
106 // assign the default client and token as well
107 if ge.defaultClient == nil && login == ge.conf[confKeyDefaultLogin] {
108 ge.defaultClient = client
109 ge.defaultToken = creds[0].(*auth.Token)
110 }
111 }
112
113 if ge.defaultClient == nil {
114 return fmt.Errorf("no token found for the default login \"%s\"", ge.conf[confKeyDefaultLogin])
115 }
116
117 return nil
118}
119
120// getClientForIdentity return a githubv4 API client configured with the access token of the given identity.
121func (ge *githubExporter) getClientForIdentity(userId entity.Id) (*rateLimitHandlerClient, error) {
122 client, ok := ge.identityClient[userId]
123 if ok {
124 return client, nil
125 }
126
127 return nil, ErrMissingIdentityToken
128}
129
130// ExportAll export all event made by the current user to Github
131func (ge *githubExporter) ExportAll(ctx context.Context, repo *cache.RepoCache, since time.Time) (<-chan core.ExportResult, error) {
132 out := make(chan core.ExportResult)
133 ge.out = out
134
135 var err error
136 // get repository node id
137 ge.repositoryID, err = getRepositoryNodeID(
138 ctx,
139 ge.defaultToken,
140 ge.conf[confKeyOwner],
141 ge.conf[confKeyProject],
142 )
143 if err != nil {
144 return nil, err
145 }
146
147 go func() {
148 defer close(out)
149
150 // query all labels
151 err = ge.cacheGithubLabels(ctx, ge.defaultClient)
152 if err != nil {
153 out <- core.NewExportError(errors.Wrap(err, "can't obtain Github labels"), "")
154 return
155 }
156
157 allIdentitiesIds := make([]entity.Id, 0, len(ge.identityClient))
158 for id := range ge.identityClient {
159 allIdentitiesIds = append(allIdentitiesIds, id)
160 }
161
162 allBugsIds := repo.AllBugsIds()
163
164 for _, id := range allBugsIds {
165 b, err := repo.ResolveBug(id)
166 if err != nil {
167 out <- core.NewExportError(errors.Wrap(err, "can't load bug"), id)
168 return
169 }
170
171 select {
172
173 case <-ctx.Done():
174 // stop iterating if context cancel function is called
175 return
176
177 default:
178 snapshot := b.Snapshot()
179
180 // ignore issues created before since date
181 // TODO: compare the Lamport time instead of using the unix time
182 if snapshot.CreateTime.Before(since) {
183 out <- core.NewExportNothing(b.Id(), "bug created before the since date")
184 continue
185 }
186
187 if snapshot.HasAnyActor(allIdentitiesIds...) {
188 // try to export the bug and it associated events
189 ge.exportBug(ctx, b, out)
190 }
191 }
192 }
193 }()
194
195 return out, nil
196}
197
198// exportBug publish bugs and related events
199func (ge *githubExporter) exportBug(ctx context.Context, b *cache.BugCache, out chan<- core.ExportResult) {
200 snapshot := b.Snapshot()
201 var bugUpdated bool
202
203 var bugGithubID string
204 var bugGithubURL string
205
206 // Special case:
207 // if a user try to export a bug that is not already exported to Github (or imported
208 // from Github) and we do not have the token of the bug author, there is nothing we can do.
209
210 // first operation is always createOp
211 createOp := snapshot.Operations[0].(*bug.CreateOperation)
212 author := snapshot.Author
213
214 // skip bug if origin is not allowed
215 origin, ok := snapshot.GetCreateMetadata(core.MetaKeyOrigin)
216 if ok && origin != target {
217 out <- core.NewExportNothing(b.Id(), fmt.Sprintf("issue tagged with origin: %s", origin))
218 return
219 }
220
221 // get github bug ID
222 githubID, ok := snapshot.GetCreateMetadata(metaKeyGithubId)
223 if ok {
224 githubURL, ok := snapshot.GetCreateMetadata(metaKeyGithubUrl)
225 if !ok {
226 // if we find github ID, github URL must be found too
227 err := fmt.Errorf("incomplete Github metadata: expected to find issue URL")
228 out <- core.NewExportError(err, b.Id())
229 }
230
231 // extract owner and project
232 owner, project, err := splitURL(githubURL)
233 if err != nil {
234 err := fmt.Errorf("bad project url: %v", err)
235 out <- core.NewExportError(err, b.Id())
236 return
237 }
238
239 // ignore issue coming from other repositories
240 if owner != ge.conf[confKeyOwner] && project != ge.conf[confKeyProject] {
241 out <- core.NewExportNothing(b.Id(), fmt.Sprintf("skipping issue from url:%s", githubURL))
242 return
243 }
244
245 // will be used to mark operation related to a bug as exported
246 bugGithubID = githubID
247 bugGithubURL = githubURL
248
249 } else {
250 // check that we have a token for operation author
251 client, err := ge.getClientForIdentity(author.Id())
252 if err != nil {
253 // if bug is still not exported and we do not have the author stop the execution
254 out <- core.NewExportNothing(b.Id(), fmt.Sprintf("missing author token"))
255 return
256 }
257
258 // create bug
259 id, url, err := ge.createGithubIssue(ctx, client, ge.repositoryID, createOp.Title, createOp.Message)
260 if err != nil {
261 err := errors.Wrap(err, "exporting github issue")
262 out <- core.NewExportError(err, b.Id())
263 return
264 }
265
266 out <- core.NewExportBug(b.Id())
267
268 // mark bug creation operation as exported
269 if err := markOperationAsExported(b, createOp.Id(), id, url); err != nil {
270 err := errors.Wrap(err, "marking operation as exported")
271 out <- core.NewExportError(err, b.Id())
272 return
273 }
274
275 // commit operation to avoid creating multiple issues with multiple pushes
276 if err := b.CommitAsNeeded(); err != nil {
277 err := errors.Wrap(err, "bug commit")
278 out <- core.NewExportError(err, b.Id())
279 return
280 }
281
282 // cache bug github ID and URL
283 bugGithubID = id
284 bugGithubURL = url
285 }
286
287 // cache operation github id
288 ge.cachedOperationIDs[createOp.Id()] = bugGithubID
289
290 for _, op := range snapshot.Operations[1:] {
291 // ignore SetMetadata operations
292 if _, ok := op.(dag.OperationDoesntChangeSnapshot); ok {
293 continue
294 }
295
296 // ignore operations already existing in github (due to import or export)
297 // cache the ID of already exported or imported issues and events from Github
298 if id, ok := op.GetMetadata(metaKeyGithubId); ok {
299 ge.cachedOperationIDs[op.Id()] = id
300 continue
301 }
302
303 opAuthor := op.Author()
304 client, err := ge.getClientForIdentity(opAuthor.Id())
305 if err != nil {
306 continue
307 }
308
309 var id, url string
310 switch op := op.(type) {
311 case *bug.AddCommentOperation:
312 // send operation to github
313 id, url, err = ge.addCommentGithubIssue(ctx, client, bugGithubID, op.Message)
314 if err != nil {
315 err := errors.Wrap(err, "adding comment")
316 out <- core.NewExportError(err, b.Id())
317 return
318 }
319
320 out <- core.NewExportComment(op.Id())
321
322 // cache comment id
323 ge.cachedOperationIDs[op.Id()] = id
324
325 case *bug.EditCommentOperation:
326 // Since github doesn't consider the issue body as a comment
327 if op.Target == createOp.Id() {
328
329 // case bug creation operation: we need to edit the Github issue
330 if err := ge.updateGithubIssueBody(ctx, client, bugGithubID, op.Message); err != nil {
331 err := errors.Wrap(err, "editing issue")
332 out <- core.NewExportError(err, b.Id())
333 return
334 }
335
336 out <- core.NewExportCommentEdition(op.Id())
337
338 id = bugGithubID
339 url = bugGithubURL
340
341 } else {
342
343 // case comment edition operation: we need to edit the Github comment
344 commentID, ok := ge.cachedOperationIDs[op.Target]
345 if !ok {
346 panic("unexpected error: comment id not found")
347 }
348
349 eid, eurl, err := ge.editCommentGithubIssue(ctx, client, commentID, op.Message)
350 if err != nil {
351 err := errors.Wrap(err, "editing comment")
352 out <- core.NewExportError(err, b.Id())
353 return
354 }
355
356 out <- core.NewExportCommentEdition(op.Id())
357
358 // use comment id/url instead of issue id/url
359 id = eid
360 url = eurl
361 }
362
363 case *bug.SetStatusOperation:
364 if err := ge.updateGithubIssueStatus(ctx, client, bugGithubID, op.Status); err != nil {
365 err := errors.Wrap(err, "editing status")
366 out <- core.NewExportError(err, b.Id())
367 return
368 }
369
370 out <- core.NewExportStatusChange(op.Id())
371
372 id = bugGithubID
373 url = bugGithubURL
374
375 case *bug.SetTitleOperation:
376 if err := ge.updateGithubIssueTitle(ctx, client, bugGithubID, op.Title); err != nil {
377 err := errors.Wrap(err, "editing title")
378 out <- core.NewExportError(err, b.Id())
379 return
380 }
381
382 out <- core.NewExportTitleEdition(op.Id())
383
384 id = bugGithubID
385 url = bugGithubURL
386
387 case *bug.LabelChangeOperation:
388 if err := ge.updateGithubIssueLabels(ctx, client, bugGithubID, op.Added, op.Removed); err != nil {
389 err := errors.Wrap(err, "updating labels")
390 out <- core.NewExportError(err, b.Id())
391 return
392 }
393
394 out <- core.NewExportLabelChange(op.Id())
395
396 id = bugGithubID
397 url = bugGithubURL
398
399 default:
400 panic("unhandled operation type case")
401 }
402
403 // mark operation as exported
404 if err := markOperationAsExported(b, op.Id(), id, url); err != nil {
405 err := errors.Wrap(err, "marking operation as exported")
406 out <- core.NewExportError(err, b.Id())
407 return
408 }
409
410 // commit at each operation export to avoid exporting same events multiple times
411 if err := b.CommitAsNeeded(); err != nil {
412 err := errors.Wrap(err, "bug commit")
413 out <- core.NewExportError(err, b.Id())
414 return
415 }
416
417 bugUpdated = true
418 }
419
420 if !bugUpdated {
421 out <- core.NewExportNothing(b.Id(), "nothing has been exported")
422 }
423}
424
425// getRepositoryNodeID request github api v3 to get repository node id
426func getRepositoryNodeID(ctx context.Context, token *auth.Token, owner, project string) (string, error) {
427 url := fmt.Sprintf("%s/repos/%s/%s", githubV3Url, owner, project)
428 client := &http.Client{}
429
430 req, err := http.NewRequest("GET", url, nil)
431 if err != nil {
432 return "", err
433 }
434
435 // need the token for private repositories
436 req.Header.Set("Authorization", fmt.Sprintf("token %s", token.Value))
437
438 ctx, cancel := context.WithTimeout(ctx, defaultTimeout)
439 defer cancel()
440 req = req.WithContext(ctx)
441
442 resp, err := client.Do(req)
443 if err != nil {
444 return "", err
445 }
446
447 if resp.StatusCode != http.StatusOK {
448 return "", fmt.Errorf("HTTP error %v retrieving repository node id", resp.StatusCode)
449 }
450
451 aux := struct {
452 NodeID string `json:"node_id"`
453 }{}
454
455 data, _ := ioutil.ReadAll(resp.Body)
456 err = resp.Body.Close()
457 if err != nil {
458 return "", err
459 }
460
461 err = json.Unmarshal(data, &aux)
462 if err != nil {
463 return "", err
464 }
465
466 return aux.NodeID, nil
467}
468
469func markOperationAsExported(b *cache.BugCache, target entity.Id, githubID, githubURL string) error {
470 _, err := b.SetMetadata(
471 target,
472 map[string]string{
473 metaKeyGithubId: githubID,
474 metaKeyGithubUrl: githubURL,
475 },
476 )
477
478 return err
479}
480
481func (ge *githubExporter) cacheGithubLabels(ctx context.Context, gc *rateLimitHandlerClient) error {
482 variables := map[string]interface{}{
483 "owner": githubv4.String(ge.conf[confKeyOwner]),
484 "name": githubv4.String(ge.conf[confKeyProject]),
485 "first": githubv4.Int(10),
486 "after": (*githubv4.String)(nil),
487 }
488
489 q := labelsQuery{}
490
491 hasNextPage := true
492 for hasNextPage {
493 if err := gc.queryExport(ctx, &q, variables, ge.out); err != nil {
494 return err
495 }
496
497 for _, label := range q.Repository.Labels.Nodes {
498 ge.cachedLabels[label.Name] = label.ID
499 }
500
501 hasNextPage = q.Repository.Labels.PageInfo.HasNextPage
502 variables["after"] = q.Repository.Labels.PageInfo.EndCursor
503 }
504
505 return nil
506}
507
508func (ge *githubExporter) getLabelID(label string) (string, error) {
509 label = strings.ToLower(label)
510 for cachedLabel, ID := range ge.cachedLabels {
511 if label == strings.ToLower(cachedLabel) {
512 return ID, nil
513 }
514 }
515
516 return "", fmt.Errorf("didn't find label id in cache")
517}
518
519// create a new label and return it github id
520// NOTE: since createLabel mutation is still in preview mode we use github api v3 to create labels
521// see https://developer.github.com/v4/mutation/createlabel/ and https://developer.github.com/v4/previews/#labels-preview
522func (ge *githubExporter) createGithubLabel(ctx context.Context, label, color string) (string, error) {
523 url := fmt.Sprintf("%s/repos/%s/%s/labels", githubV3Url, ge.conf[confKeyOwner], ge.conf[confKeyProject])
524 client := &http.Client{}
525
526 params := struct {
527 Name string `json:"name"`
528 Color string `json:"color"`
529 Description string `json:"description"`
530 }{
531 Name: label,
532 Color: color,
533 }
534
535 data, err := json.Marshal(params)
536 if err != nil {
537 return "", err
538 }
539
540 req, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
541 if err != nil {
542 return "", err
543 }
544
545 ctx, cancel := context.WithTimeout(ctx, defaultTimeout)
546 defer cancel()
547 req = req.WithContext(ctx)
548
549 // need the token for private repositories
550 req.Header.Set("Authorization", fmt.Sprintf("token %s", ge.defaultToken.Value))
551
552 resp, err := client.Do(req)
553 if err != nil {
554 return "", err
555 }
556
557 if resp.StatusCode != http.StatusCreated {
558 return "", fmt.Errorf("error creating label: response status %v", resp.StatusCode)
559 }
560
561 aux := struct {
562 ID int `json:"id"`
563 NodeID string `json:"node_id"`
564 Color string `json:"color"`
565 }{}
566
567 data, _ = ioutil.ReadAll(resp.Body)
568 defer resp.Body.Close()
569
570 err = json.Unmarshal(data, &aux)
571 if err != nil {
572 return "", err
573 }
574
575 return aux.NodeID, nil
576}
577
578/**
579// create github label using api v4
580func (ge *githubExporter) createGithubLabelV4(gc *githubv4.Client, label, labelColor string) (string, error) {
581 m := createLabelMutation{}
582 input := createLabelInput{
583 RepositoryID: ge.repositoryID,
584 Name: githubv4.String(label),
585 Color: githubv4.String(labelColor),
586 }
587
588 ctx := context.Background()
589
590 if err := gc.mutate(ctx, &m, input, nil); err != nil {
591 return "", err
592 }
593
594 return m.CreateLabel.Label.ID, nil
595}
596*/
597
598func (ge *githubExporter) getOrCreateGithubLabelID(ctx context.Context, gc *rateLimitHandlerClient, repositoryID string, label bug.Label) (string, error) {
599 // try to get label id from cache
600 labelID, err := ge.getLabelID(string(label))
601 if err == nil {
602 return labelID, nil
603 }
604
605 // RGBA to hex color
606 rgba := label.Color().RGBA()
607 hexColor := fmt.Sprintf("%.2x%.2x%.2x", rgba.R, rgba.G, rgba.B)
608
609 ctx, cancel := context.WithTimeout(ctx, defaultTimeout)
610 defer cancel()
611
612 labelID, err = ge.createGithubLabel(ctx, string(label), hexColor)
613 if err != nil {
614 return "", err
615 }
616
617 return labelID, nil
618}
619
620func (ge *githubExporter) getLabelsIDs(ctx context.Context, gc *rateLimitHandlerClient, repositoryID string, labels []bug.Label) ([]githubv4.ID, error) {
621 ids := make([]githubv4.ID, 0, len(labels))
622 var err error
623
624 // check labels ids
625 for _, label := range labels {
626 id, ok := ge.cachedLabels[string(label)]
627 if !ok {
628 // try to query label id
629 id, err = ge.getOrCreateGithubLabelID(ctx, gc, repositoryID, label)
630 if err != nil {
631 return nil, errors.Wrap(err, "get or create github label")
632 }
633
634 // cache label id
635 ge.cachedLabels[string(label)] = id
636 }
637
638 ids = append(ids, githubv4.ID(id))
639 }
640
641 return ids, nil
642}
643
644// create a github issue and return it ID
645func (ge *githubExporter) createGithubIssue(ctx context.Context, gc *rateLimitHandlerClient, repositoryID, title, body string) (string, string, error) {
646 m := &createIssueMutation{}
647 input := githubv4.CreateIssueInput{
648 RepositoryID: repositoryID,
649 Title: githubv4.String(title),
650 Body: (*githubv4.String)(&body),
651 }
652
653 if err := gc.mutate(ctx, m, input, nil, ge.out); err != nil {
654 return "", "", err
655 }
656
657 issue := m.CreateIssue.Issue
658 return issue.ID, issue.URL, nil
659}
660
661// add a comment to an issue and return it ID
662func (ge *githubExporter) addCommentGithubIssue(ctx context.Context, gc *rateLimitHandlerClient, subjectID string, body string) (string, string, error) {
663 m := &addCommentToIssueMutation{}
664 input := githubv4.AddCommentInput{
665 SubjectID: subjectID,
666 Body: githubv4.String(body),
667 }
668
669 if err := gc.mutate(ctx, m, input, nil, ge.out); err != nil {
670 return "", "", err
671 }
672
673 node := m.AddComment.CommentEdge.Node
674 return node.ID, node.URL, nil
675}
676
677func (ge *githubExporter) editCommentGithubIssue(ctx context.Context, gc *rateLimitHandlerClient, commentID, body string) (string, string, error) {
678 m := &updateIssueCommentMutation{}
679 input := githubv4.UpdateIssueCommentInput{
680 ID: commentID,
681 Body: githubv4.String(body),
682 }
683
684 if err := gc.mutate(ctx, m, input, nil, ge.out); err != nil {
685 return "", "", err
686 }
687
688 return commentID, m.UpdateIssueComment.IssueComment.URL, nil
689}
690
691func (ge *githubExporter) updateGithubIssueStatus(ctx context.Context, gc *rateLimitHandlerClient, id string, status bug.Status) error {
692 m := &updateIssueMutation{}
693
694 // set state
695 var state githubv4.IssueState
696
697 switch status {
698 case bug.OpenStatus:
699 state = githubv4.IssueStateOpen
700 case bug.ClosedStatus:
701 state = githubv4.IssueStateClosed
702 default:
703 panic("unknown bug state")
704 }
705
706 input := githubv4.UpdateIssueInput{
707 ID: id,
708 State: &state,
709 }
710
711 if err := gc.mutate(ctx, m, input, nil, ge.out); err != nil {
712 return err
713 }
714
715 return nil
716}
717
718func (ge *githubExporter) updateGithubIssueBody(ctx context.Context, gc *rateLimitHandlerClient, id string, body string) error {
719 m := &updateIssueMutation{}
720 input := githubv4.UpdateIssueInput{
721 ID: id,
722 Body: (*githubv4.String)(&body),
723 }
724
725 if err := gc.mutate(ctx, m, input, nil, ge.out); err != nil {
726 return err
727 }
728
729 return nil
730}
731
732func (ge *githubExporter) updateGithubIssueTitle(ctx context.Context, gc *rateLimitHandlerClient, id, title string) error {
733 m := &updateIssueMutation{}
734 input := githubv4.UpdateIssueInput{
735 ID: id,
736 Title: (*githubv4.String)(&title),
737 }
738
739 if err := gc.mutate(ctx, m, input, nil, ge.out); err != nil {
740 return err
741 }
742
743 return nil
744}
745
746// update github issue labels
747func (ge *githubExporter) updateGithubIssueLabels(ctx context.Context, gc *rateLimitHandlerClient, labelableID string, added, removed []bug.Label) error {
748
749 wg, ctx := errgroup.WithContext(ctx)
750 if len(added) > 0 {
751 wg.Go(func() error {
752 addedIDs, err := ge.getLabelsIDs(ctx, gc, labelableID, added)
753 if err != nil {
754 return err
755 }
756
757 m := &addLabelsToLabelableMutation{}
758 inputAdd := githubv4.AddLabelsToLabelableInput{
759 LabelableID: labelableID,
760 LabelIDs: addedIDs,
761 }
762
763 // add labels
764 if err := gc.mutate(ctx, m, inputAdd, nil, ge.out); err != nil {
765 return err
766 }
767 return nil
768 })
769 }
770
771 if len(removed) > 0 {
772 wg.Go(func() error {
773 removedIDs, err := ge.getLabelsIDs(ctx, gc, labelableID, removed)
774 if err != nil {
775 return err
776 }
777
778 m2 := &removeLabelsFromLabelableMutation{}
779 inputRemove := githubv4.RemoveLabelsFromLabelableInput{
780 LabelableID: labelableID,
781 LabelIDs: removedIDs,
782 }
783
784 // remove label labels
785 if err := gc.mutate(ctx, m2, inputRemove, nil, ge.out); err != nil {
786 return err
787 }
788 return nil
789 })
790 }
791
792 return wg.Wait()
793}