aboutsummaryrefslogtreecommitdiffstats
path: root/bridge/github/import.go
diff options
context:
space:
mode:
Diffstat (limited to 'bridge/github/import.go')
-rw-r--r--bridge/github/import.go198
1 files changed, 135 insertions, 63 deletions
diff --git a/bridge/github/import.go b/bridge/github/import.go
index dcaf2d05..7c4deb50 100644
--- a/bridge/github/import.go
+++ b/bridge/github/import.go
@@ -28,11 +28,8 @@ type githubImporter struct {
// iterator
iterator *iterator
- // number of imported issues
- importedIssues int
-
- // number of imported identities
- importedIdentities int
+ // send only channel
+ out chan<- core.ImportResult
}
func (gi *githubImporter) Init(conf core.Configuration) error {
@@ -42,40 +39,49 @@ func (gi *githubImporter) Init(conf core.Configuration) error {
// ImportAll iterate over all the configured repository issues and ensure the creation of the
// missing issues / timeline items / edits / label events ...
-func (gi *githubImporter) ImportAll(repo *cache.RepoCache, since time.Time) error {
- gi.iterator = NewIterator(gi.conf[keyOwner], gi.conf[keyProject], gi.conf[keyToken], since)
-
- // Loop over all matching issues
- for gi.iterator.NextIssue() {
- issue := gi.iterator.IssueValue()
- fmt.Printf("importing issue: %v\n", issue.Title)
+func (gi *githubImporter) ImportAll(ctx context.Context, repo *cache.RepoCache, since time.Time) (<-chan core.ImportResult, error) {
+ gi.iterator = NewIterator(ctx, 10, gi.conf[keyOwner], gi.conf[keyProject], gi.conf[keyToken], since)
+ out := make(chan core.ImportResult)
+ gi.out = out
+
+ go func() {
+ defer close(gi.out)
+
+ // Loop over all matching issues
+ for gi.iterator.NextIssue() {
+ issue := gi.iterator.IssueValue()
+ // create issue
+ b, err := gi.ensureIssue(repo, issue)
+ if err != nil {
+ err := fmt.Errorf("issue creation: %v", err)
+ out <- core.NewImportError(err, "")
+ return
+ }
- // create issue
- b, err := gi.ensureIssue(repo, issue)
- if err != nil {
- return fmt.Errorf("issue creation: %v", err)
- }
+ // loop over timeline items
+ for gi.iterator.NextTimelineItem() {
+ item := gi.iterator.TimelineItemValue()
+ if err := gi.ensureTimelineItem(repo, b, item); err != nil {
+ err := fmt.Errorf("timeline item creation: %v", err)
+ out <- core.NewImportError(err, "")
+ return
+ }
+ }
- // loop over timeline items
- for gi.iterator.NextTimelineItem() {
- if err := gi.ensureTimelineItem(repo, b, gi.iterator.TimelineItemValue()); err != nil {
- return fmt.Errorf("timeline item creation: %v", err)
+ // commit bug state
+ if err := b.CommitAsNeeded(); err != nil {
+ err = fmt.Errorf("bug commit: %v", err)
+ out <- core.NewImportError(err, "")
+ return
}
}
- // commit bug state
- if err := b.CommitAsNeeded(); err != nil {
- return fmt.Errorf("bug commit: %v", err)
+ if err := gi.iterator.Error(); err != nil && err != context.Canceled {
+ gi.out <- core.NewImportError(err, "")
}
- }
+ }()
- if err := gi.iterator.Error(); err != nil {
- fmt.Printf("import error: %v\n", err)
- return err
- }
-
- fmt.Printf("Successfully imported %d issues and %d identities from Github\n", gi.importedIssues, gi.importedIdentities)
- return nil
+ return out, nil
}
func (gi *githubImporter) ensureIssue(repo *cache.RepoCache, issue issueTimeline) (*cache.BugCache, error) {
@@ -122,7 +128,9 @@ func (gi *githubImporter) ensureIssue(repo *cache.RepoCache, issue issueTimeline
}
// importing a new bug
- gi.importedIssues++
+ gi.out <- core.NewImportBug(b.Id())
+ } else {
+ gi.out <- core.NewImportNothing("", "bug already imported")
}
} else {
@@ -130,6 +138,7 @@ func (gi *githubImporter) ensureIssue(repo *cache.RepoCache, issue issueTimeline
for i, edit := range issueEdits {
if i == 0 && b != nil {
// The first edit in the github result is the issue creation itself, we already have that
+ gi.out <- core.NewImportNothing("", "bug already imported")
continue
}
@@ -159,13 +168,12 @@ func (gi *githubImporter) ensureIssue(repo *cache.RepoCache, issue issueTimeline
}
// importing a new bug
- gi.importedIssues++
-
+ gi.out <- core.NewImportBug(b.Id())
continue
}
// other edits will be added as CommentEdit operations
- target, err := b.ResolveOperationWithMetadata(keyGithubUrl, issue.Url.String())
+ target, err := b.ResolveOperationWithMetadata(keyGithubId, parseId(issue.Id))
if err != nil {
return nil, err
}
@@ -181,16 +189,16 @@ func (gi *githubImporter) ensureIssue(repo *cache.RepoCache, issue issueTimeline
}
func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.BugCache, item timelineItem) error {
- fmt.Printf("import event item: %s\n", item.Typename)
switch item.Typename {
case "IssueComment":
// collect all comment edits
- commentEdits := []userContentEdit{}
+ var commentEdits []userContentEdit
for gi.iterator.NextCommentEdit() {
commentEdits = append(commentEdits, gi.iterator.CommentEditValue())
}
+ // ensureTimelineComment send import events over out chanel
err := gi.ensureTimelineComment(repo, b, item.IssueComment, commentEdits)
if err != nil {
return fmt.Errorf("timeline comment creation: %v", err)
@@ -199,6 +207,12 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
case "LabeledEvent":
id := parseId(item.LabeledEvent.Id)
_, err := b.ResolveOperationWithMetadata(keyGithubId, id)
+ if err == nil {
+ reason := fmt.Sprintf("operation already imported: %v", item.Typename)
+ gi.out <- core.NewImportNothing("", reason)
+ return nil
+ }
+
if err != cache.ErrNoMatchingOp {
return err
}
@@ -206,7 +220,7 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
if err != nil {
return err
}
- _, err = b.ForceChangeLabelsRaw(
+ op, err := b.ForceChangeLabelsRaw(
author,
item.LabeledEvent.CreatedAt.Unix(),
[]string{
@@ -215,12 +229,21 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
nil,
map[string]string{keyGithubId: id},
)
+ if err != nil {
+ return err
+ }
- return err
+ gi.out <- core.NewImportLabelChange(op.Id())
+ return nil
case "UnlabeledEvent":
id := parseId(item.UnlabeledEvent.Id)
_, err := b.ResolveOperationWithMetadata(keyGithubId, id)
+ if err == nil {
+ reason := fmt.Sprintf("operation already imported: %v", item.Typename)
+ gi.out <- core.NewImportNothing("", reason)
+ return nil
+ }
if err != cache.ErrNoMatchingOp {
return err
}
@@ -229,7 +252,7 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
return err
}
- _, err = b.ForceChangeLabelsRaw(
+ op, err := b.ForceChangeLabelsRaw(
author,
item.UnlabeledEvent.CreatedAt.Unix(),
nil,
@@ -238,7 +261,12 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
},
map[string]string{keyGithubId: id},
)
- return err
+ if err != nil {
+ return err
+ }
+
+ gi.out <- core.NewImportLabelChange(op.Id())
+ return nil
case "ClosedEvent":
id := parseId(item.ClosedEvent.Id)
@@ -246,16 +274,27 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
if err != cache.ErrNoMatchingOp {
return err
}
+ if err == nil {
+ reason := fmt.Sprintf("operation already imported: %v", item.Typename)
+ gi.out <- core.NewImportNothing("", reason)
+ return nil
+ }
author, err := gi.ensurePerson(repo, item.ClosedEvent.Actor)
if err != nil {
return err
}
- _, err = b.CloseRaw(
+ op, err := b.CloseRaw(
author,
item.ClosedEvent.CreatedAt.Unix(),
map[string]string{keyGithubId: id},
)
- return err
+
+ if err != nil {
+ return err
+ }
+
+ gi.out <- core.NewImportStatusChange(op.Id())
+ return nil
case "ReopenedEvent":
id := parseId(item.ReopenedEvent.Id)
@@ -263,16 +302,27 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
if err != cache.ErrNoMatchingOp {
return err
}
+ if err == nil {
+ reason := fmt.Sprintf("operation already imported: %v", item.Typename)
+ gi.out <- core.NewImportNothing("", reason)
+ return nil
+ }
author, err := gi.ensurePerson(repo, item.ReopenedEvent.Actor)
if err != nil {
return err
}
- _, err = b.OpenRaw(
+ op, err := b.OpenRaw(
author,
item.ReopenedEvent.CreatedAt.Unix(),
map[string]string{keyGithubId: id},
)
- return err
+
+ if err != nil {
+ return err
+ }
+
+ gi.out <- core.NewImportStatusChange(op.Id())
+ return nil
case "RenamedTitleEvent":
id := parseId(item.RenamedTitleEvent.Id)
@@ -280,20 +330,31 @@ func (gi *githubImporter) ensureTimelineItem(repo *cache.RepoCache, b *cache.Bug
if err != cache.ErrNoMatchingOp {
return err
}
+ if err == nil {
+ reason := fmt.Sprintf("operation already imported: %v", item.Typename)
+ gi.out <- core.NewImportNothing("", reason)
+ return nil
+ }
author, err := gi.ensurePerson(repo, item.RenamedTitleEvent.Actor)
if err != nil {
return err
}
- _, err = b.SetTitleRaw(
+ op, err := b.SetTitleRaw(
author,
item.RenamedTitleEvent.CreatedAt.Unix(),
string(item.RenamedTitleEvent.CurrentTitle),
map[string]string{keyGithubId: id},
)
- return err
+ if err != nil {
+ return err
+ }
+
+ gi.out <- core.NewImportTitleEdition(op.Id())
+ return nil
default:
- fmt.Printf("ignore event: %v\n", item.Typename)
+ reason := fmt.Sprintf("ignoring timeline type: %v", item.Typename)
+ gi.out <- core.NewImportNothing("", reason)
}
return nil
@@ -307,14 +368,15 @@ func (gi *githubImporter) ensureTimelineComment(repo *cache.RepoCache, b *cache.
}
targetOpID, err := b.ResolveOperationWithMetadata(keyGithubId, parseId(item.Id))
- if err != nil && err != cache.ErrNoMatchingOp {
+ if err == nil {
+ gi.out <- core.NewImportNothing("", "comment already imported")
+ } else if err != cache.ErrNoMatchingOp {
// real error
return err
}
// if no edits are given we create the comment
if len(edits) == 0 {
- // if comment doesn't exist
if err == cache.ErrNoMatchingOp {
cleanText, err := text.Cleanup(string(item.Body))
if err != nil {
@@ -322,7 +384,7 @@ func (gi *githubImporter) ensureTimelineComment(repo *cache.RepoCache, b *cache.
}
// add comment operation
- _, err = b.AddCommentRaw(
+ op, err := b.AddCommentRaw(
author,
item.CreatedAt.Unix(),
cleanText,
@@ -332,12 +394,18 @@ func (gi *githubImporter) ensureTimelineComment(repo *cache.RepoCache, b *cache.
keyGithubUrl: parseId(item.Url.String()),
},
)
- return err
+ if err != nil {
+ return err
+ }
+
+ gi.out <- core.NewImportComment(op.Id())
}
+
} else {
for i, edit := range edits {
if i == 0 && targetOpID != "" {
// The first edit in the github result is the comment creation itself, we already have that
+ gi.out <- core.NewImportNothing("", "comment already imported")
continue
}
@@ -370,7 +438,6 @@ func (gi *githubImporter) ensureTimelineComment(repo *cache.RepoCache, b *cache.
// set target for the nexr edit now that the comment is created
targetOpID = op.Id()
-
continue
}
@@ -386,7 +453,7 @@ func (gi *githubImporter) ensureTimelineComment(repo *cache.RepoCache, b *cache.
func (gi *githubImporter) ensureCommentEdit(repo *cache.RepoCache, b *cache.BugCache, target entity.Id, edit userContentEdit) error {
_, err := b.ResolveOperationWithMetadata(keyGithubId, parseId(edit.Id))
if err == nil {
- // already imported
+ gi.out <- core.NewImportNothing(b.Id(), "edition already imported")
return nil
}
if err != cache.ErrNoMatchingOp {
@@ -394,8 +461,6 @@ func (gi *githubImporter) ensureCommentEdit(repo *cache.RepoCache, b *cache.BugC
return err
}
- fmt.Println("import edition")
-
editor, err := gi.ensurePerson(repo, edit.Editor)
if err != nil {
return err
@@ -404,7 +469,7 @@ func (gi *githubImporter) ensureCommentEdit(repo *cache.RepoCache, b *cache.BugC
switch {
case edit.DeletedAt != nil:
// comment deletion, not supported yet
- fmt.Println("comment deletion is not supported yet")
+ gi.out <- core.NewImportNothing(b.Id(), "comment deletion is not supported yet")
case edit.DeletedAt == nil:
@@ -414,7 +479,7 @@ func (gi *githubImporter) ensureCommentEdit(repo *cache.RepoCache, b *cache.BugC
}
// comment edition
- _, err = b.EditCommentRaw(
+ op, err := b.EditCommentRaw(
editor,
edit.CreatedAt.Unix(),
target,
@@ -427,6 +492,8 @@ func (gi *githubImporter) ensureCommentEdit(repo *cache.RepoCache, b *cache.BugC
if err != nil {
return err
}
+
+ gi.out <- core.NewImportCommentEdition(op.Id())
}
return nil
@@ -450,7 +517,6 @@ func (gi *githubImporter) ensurePerson(repo *cache.RepoCache, actor *actor) (*ca
}
// importing a new identity
- gi.importedIdentities++
var name string
var email string
@@ -471,7 +537,7 @@ func (gi *githubImporter) ensurePerson(repo *cache.RepoCache, actor *actor) (*ca
case "Bot":
}
- return repo.NewIdentityRaw(
+ i, err = repo.NewIdentityRaw(
name,
email,
string(actor.Login),
@@ -480,6 +546,13 @@ func (gi *githubImporter) ensurePerson(repo *cache.RepoCache, actor *actor) (*ca
keyGithubLogin: string(actor.Login),
},
)
+
+ if err != nil {
+ return nil, err
+ }
+
+ gi.out <- core.NewImportIdentity(i.Id())
+ return i, nil
}
func (gi *githubImporter) getGhost(repo *cache.RepoCache) (*cache.IdentityCache, error) {
@@ -500,8 +573,7 @@ func (gi *githubImporter) getGhost(repo *cache.RepoCache) (*cache.IdentityCache,
gc := buildClient(gi.conf[keyToken])
- parentCtx := context.Background()
- ctx, cancel := context.WithTimeout(parentCtx, defaultTimeout)
+ ctx, cancel := context.WithTimeout(gi.iterator.ctx, defaultTimeout)
defer cancel()
err = gc.Query(ctx, &q, variables)
generated by cgit (git 2.34.1) at 2024-11-06 03:16:06 +0000