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