Use transactions for issues, PRs and users

Signed-off-by: Carlos Martín <carlos.martin.sanchez@gmail.com>

Author: carlosms

shallow/issue.go

@@ -14,27 +14,30 @@ import (
 
 type IssueSyncer struct {
 	db     *sql.DB
-	store  *models.IssueStore
 	client *github.Client
 }
 
 func NewIssueSyncer(db *sql.DB, c *github.Client) *IssueSyncer {
 	return &IssueSyncer{
 		db:     db,
-		store:  models.NewIssueStore(db),
 		client: c,
 	}
 }
 
 func (s *IssueSyncer) Sync(owner, repo string, logger log.Logger) error {
+	store := models.NewIssueStore(s.db)
+	return store.Transaction(func(store *models.IssueStore) error {
+		return s.doIssues(store, owner, repo, logger)
+	})
+}
+
+func (s *IssueSyncer) doIssues(store *models.IssueStore, owner, repo string, logger log.Logger) error {
 	opts := &github.IssueListByRepoOptions{}
 	opts.ListOptions.PerPage = listOptionsPerPage
 	opts.State = "all"
 
 	logger.Infof("starting to retrieve issues")
 
-	// TODO transaction for faster times
-
 	// Get the list of all issues
 	for {
 		issues, r, err := s.client.Issues.ListByRepo(context.TODO(), owner, repo, opts)
@@ -49,7 +52,7 @@ func (s *IssueSyncer) Sync(owner, repo string, logger log.Logger) error {
 
 			logger := logger.With(log.Fields{"issue": i.GetNumber()})
 
-			_, err := s.store.FindOne(models.NewIssueQuery().
+			_, err := store.FindOne(models.NewIssueQuery().
 				Where(kallax.And(
 					kallax.Eq(models.Schema.Issue.RepositoryOwner, owner),
 					kallax.Eq(models.Schema.Issue.RepositoryName, repo),
@@ -70,7 +73,7 @@ func (s *IssueSyncer) Sync(owner, repo string, logger log.Logger) error {
 			record := models.NewIssue()
 			record.Issue = *i
 
-			err = s.store.Insert(record)
+			err = store.Insert(record)
 			if err != nil {
 				logger.Errorf(err, "failed to write the resource into the DB")
 				return fmt.Errorf("failed to write the resource into the DB: %v", err)

shallow/pull_request.go

@@ -14,27 +14,30 @@ import (
 
 type PullRequestSyncer struct {
 	db     *sql.DB
-	store  *models.PullRequestStore
 	client *github.Client
 }
 
 func NewPullRequestSyncer(db *sql.DB, c *github.Client) *PullRequestSyncer {
 	return &PullRequestSyncer{
 		db:     db,
-		store:  models.NewPullRequestStore(db),
 		client: c,
 	}
 }
 
 func (s *PullRequestSyncer) Sync(owner, repo string, logger log.Logger) error {
+	store := models.NewPullRequestStore(s.db)
+	return store.Transaction(func(store *models.PullRequestStore) error {
+		return s.doPRs(store, owner, repo, logger)
+	})
+}
+
+func (s *PullRequestSyncer) doPRs(store *models.PullRequestStore, owner, repo string, logger log.Logger) error {
 	opts := &github.PullRequestListOptions{}
 	opts.ListOptions.PerPage = listOptionsPerPage
 	opts.State = "all"
 
 	logger.Infof("starting to retrieve PRs")
 
-	// TODO transaction for faster times
-
 	// Get the list of all PRs
 	for {
 		prs, r, err := s.client.PullRequests.List(context.TODO(), owner, repo, opts)
@@ -45,7 +48,7 @@ func (s *PullRequestSyncer) Sync(owner, repo string, logger log.Logger) error {
 		for _, pr := range prs {
 			logger := logger.With(log.Fields{"pr": pr.GetNumber()})
 
-			_, err := s.store.FindOne(models.NewPullRequestQuery().
+			_, err := store.FindOne(models.NewPullRequestQuery().
 				Where(kallax.And(
 					kallax.Eq(models.Schema.Issue.RepositoryOwner, owner),
 					kallax.Eq(models.Schema.Issue.RepositoryName, repo),
@@ -66,7 +69,7 @@ func (s *PullRequestSyncer) Sync(owner, repo string, logger log.Logger) error {
 			record := models.NewPullRequest()
 			record.PullRequest = *pr
 
-			err = s.store.Insert(record)
+			err = store.Insert(record)
 			if err != nil {
 				logger.Errorf(err, "failed to write the resource into the DB")
 				return fmt.Errorf("failed to write the resource into the DB: %v", err)

shallow/user.go

@@ -14,26 +14,29 @@ import (
 
 type UserSyncer struct {
 	db     *sql.DB
-	store  *models.UserStore
 	client *github.Client
 }
 
 func NewUserSyncer(db *sql.DB, c *github.Client) *UserSyncer {
 	return &UserSyncer{
 		db:     db,
-		store:  models.NewUserStore(db),
 		client: c,
 	}
 }
 
 func (s *UserSyncer) Sync(org string, logger log.Logger) error {
+	store := models.NewUserStore(s.db)
+	return store.Transaction(func(store *models.UserStore) error {
+		return s.doUsers(store, org, logger)
+	})
+}
+
+func (s *UserSyncer) doUsers(store *models.UserStore, org string, logger log.Logger) error {
 	opts := &github.ListMembersOptions{}
 	opts.ListOptions.PerPage = listOptionsPerPage
 
 	logger.Infof("starting to retrieve users")
 
-	// TODO transaction for faster times
-
 	// Get the list of all users
 	for {
 		users, r, err := s.client.Organizations.ListMembers(context.TODO(), org, opts)
@@ -44,7 +47,7 @@ func (s *UserSyncer) Sync(org string, logger log.Logger) error {
 		for _, user := range users {
 			logger := logger.With(log.Fields{"user": user.GetLogin()})
 
-			_, err := s.store.FindOne(models.NewUserQuery().
+			_, err := store.FindOne(models.NewUserQuery().
 				Where(kallax.And(
 					kallax.Eq(models.Schema.User.ID, user.GetID()),
 				)),
@@ -62,7 +65,7 @@ func (s *UserSyncer) Sync(org string, logger log.Logger) error {
 			record := models.NewUser()
 			record.User = *user
 
-			err = s.store.Insert(record)
+			err = store.Insert(record)
 			if err != nil {
 				logger.Errorf(err, "failed to write the resource into the DB")
 				return fmt.Errorf("failed to write the resource into the DB: %v", err)