Move notifywatch to service layer (#33825)

No logic change.

models/activities/action.go

@@ -16,9 +16,7 @@ import (
 	"code.gitea.io/gitea/models/db"
 	issues_model "code.gitea.io/gitea/models/issues"
 	"code.gitea.io/gitea/models/organization"
-	access_model "code.gitea.io/gitea/models/perm/access"
 	repo_model "code.gitea.io/gitea/models/repo"
-	"code.gitea.io/gitea/models/unit"
 	user_model "code.gitea.io/gitea/models/user"
 	"code.gitea.io/gitea/modules/git"
 	"code.gitea.io/gitea/modules/log"
@@ -200,15 +198,13 @@ func (a *Action) LoadActUser(ctx context.Context) {
 	}
 }
 
-func (a *Action) LoadRepo(ctx context.Context) {
+func (a *Action) LoadRepo(ctx context.Context) error {
 	if a.Repo != nil {
-		return
+		return nil
 	}
 	var err error
 	a.Repo, err = repo_model.GetRepositoryByID(ctx, a.RepoID)
-	if err != nil {
+	return err
-		log.Error("repo_model.GetRepositoryByID(%d): %v", a.RepoID, err)
-	}
 }
 
 // GetActFullName gets the action's user full name.
@@ -250,7 +246,7 @@ func (a *Action) GetActDisplayNameTitle(ctx context.Context) string {
 
 // GetRepoUserName returns the name of the action repository owner.
 func (a *Action) GetRepoUserName(ctx context.Context) string {
-	a.LoadRepo(ctx)
+	_ = a.LoadRepo(ctx)
 	if a.Repo == nil {
 		return "(non-existing-repo)"
 	}
@@ -265,7 +261,7 @@ func (a *Action) ShortRepoUserName(ctx context.Context) string {
 
 // GetRepoName returns the name of the action repository.
 func (a *Action) GetRepoName(ctx context.Context) string {
-	a.LoadRepo(ctx)
+	_ = a.LoadRepo(ctx)
 	if a.Repo == nil {
 		return "(non-existing-repo)"
 	}
@@ -567,130 +563,6 @@ func DeleteOldActions(ctx context.Context, olderThan time.Duration) (err error)
 	return err
 }
 
-// NotifyWatchers creates batch of actions for every watcher.
-// It could insert duplicate actions for a repository action, like this:
-// * Original action: UserID=1 (the real actor), ActUserID=1
-// * Organization action: UserID=100 (the repo's org), ActUserID=1
-// * Watcher action: UserID=20 (a user who is watching a repo), ActUserID=1
-func NotifyWatchers(ctx context.Context, actions ...*Action) error {
-	var watchers []*repo_model.Watch
-	var repo *repo_model.Repository
-	var err error
-	var permCode []bool
-	var permIssue []bool
-	var permPR []bool
-
-	e := db.GetEngine(ctx)
-
-	for _, act := range actions {
-		repoChanged := repo == nil || repo.ID != act.RepoID
-
-		if repoChanged {
-			// Add feeds for user self and all watchers.
-			watchers, err = repo_model.GetWatchers(ctx, act.RepoID)
-			if err != nil {
-				return fmt.Errorf("get watchers: %w", err)
-			}
-		}
-
-		// Add feed for actioner.
-		act.UserID = act.ActUserID
-		if _, err = e.Insert(act); err != nil {
-			return fmt.Errorf("insert new actioner: %w", err)
-		}
-
-		if repoChanged {
-			act.LoadRepo(ctx)
-			repo = act.Repo
-
-			// check repo owner exist.
-			if err := act.Repo.LoadOwner(ctx); err != nil {
-				return fmt.Errorf("can't get repo owner: %w", err)
-			}
-		} else if act.Repo == nil {
-			act.Repo = repo
-		}
-
-		// Add feed for organization
-		if act.Repo.Owner.IsOrganization() && act.ActUserID != act.Repo.Owner.ID {
-			act.ID = 0
-			act.UserID = act.Repo.Owner.ID
-			if err = db.Insert(ctx, act); err != nil {
-				return fmt.Errorf("insert new actioner: %w", err)
-			}
-		}
-
-		if repoChanged {
-			permCode = make([]bool, len(watchers))
-			permIssue = make([]bool, len(watchers))
-			permPR = make([]bool, len(watchers))
-			for i, watcher := range watchers {
-				user, err := user_model.GetUserByID(ctx, watcher.UserID)
-				if err != nil {
-					permCode[i] = false
-					permIssue[i] = false
-					permPR[i] = false
-					continue
-				}
-				perm, err := access_model.GetUserRepoPermission(ctx, repo, user)
-				if err != nil {
-					permCode[i] = false
-					permIssue[i] = false
-					permPR[i] = false
-					continue
-				}
-				permCode[i] = perm.CanRead(unit.TypeCode)
-				permIssue[i] = perm.CanRead(unit.TypeIssues)
-				permPR[i] = perm.CanRead(unit.TypePullRequests)
-			}
-		}
-
-		for i, watcher := range watchers {
-			if act.ActUserID == watcher.UserID {
-				continue
-			}
-			act.ID = 0
-			act.UserID = watcher.UserID
-			act.Repo.Units = nil
-
-			switch act.OpType {
-			case ActionCommitRepo, ActionPushTag, ActionDeleteTag, ActionPublishRelease, ActionDeleteBranch:
-				if !permCode[i] {
-					continue
-				}
-			case ActionCreateIssue, ActionCommentIssue, ActionCloseIssue, ActionReopenIssue:
-				if !permIssue[i] {
-					continue
-				}
-			case ActionCreatePullRequest, ActionCommentPull, ActionMergePullRequest, ActionClosePullRequest, ActionReopenPullRequest, ActionAutoMergePullRequest:
-				if !permPR[i] {
-					continue
-				}
-			}
-
-			if err = db.Insert(ctx, act); err != nil {
-				return fmt.Errorf("insert new action: %w", err)
-			}
-		}
-	}
-	return nil
-}
-
-// NotifyWatchersActions creates batch of actions for every watcher.
-func NotifyWatchersActions(ctx context.Context, acts []*Action) error {
-	ctx, committer, err := db.TxContext(ctx)
-	if err != nil {
-		return err
-	}
-	defer committer.Close()
-	for _, act := range acts {
-		if err := NotifyWatchers(ctx, act); err != nil {
-			return err
-		}
-	}
-	return committer.Commit()
-}
-
 // DeleteIssueActions delete all actions related with issueID
 func DeleteIssueActions(ctx context.Context, repoID, issueID, issueIndex int64) error {
 	// delete actions assigned to this issue

models/activities/action_test.go

@@ -82,43 +82,6 @@ func TestActivityReadable(t *testing.T) {
 	}
 }
 
-func TestNotifyWatchers(t *testing.T) {
-	assert.NoError(t, unittest.PrepareTestDatabase())
-
-	action := &activities_model.Action{
-		ActUserID: 8,
-		RepoID:    1,
-		OpType:    activities_model.ActionStarRepo,
-	}
-	assert.NoError(t, activities_model.NotifyWatchers(db.DefaultContext, action))
-
-	// One watchers are inactive, thus action is only created for user 8, 1, 4, 11
-	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
-		ActUserID: action.ActUserID,
-		UserID:    8,
-		RepoID:    action.RepoID,
-		OpType:    action.OpType,
-	})
-	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
-		ActUserID: action.ActUserID,
-		UserID:    1,
-		RepoID:    action.RepoID,
-		OpType:    action.OpType,
-	})
-	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
-		ActUserID: action.ActUserID,
-		UserID:    4,
-		RepoID:    action.RepoID,
-		OpType:    action.OpType,
-	})
-	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
-		ActUserID: action.ActUserID,
-		UserID:    11,
-		RepoID:    action.RepoID,
-		OpType:    action.OpType,
-	})
-}
-
 func TestConsistencyUpdateAction(t *testing.T) {
 	if !setting.Database.Type.IsSQLite3() {
 		t.Skip("Test is only for SQLite database.")

routers/web/feed/convert.go

@@ -50,7 +50,7 @@ func toReleaseLink(ctx *context.Context, act *activities_model.Action) string {
 
 // renderCommentMarkdown renders the comment markdown to html
 func renderCommentMarkdown(ctx *context.Context, act *activities_model.Action, content string) template.HTML {
-	act.LoadRepo(ctx)
+	_ = act.LoadRepo(ctx)
 	if act.Repo == nil {
 		return ""
 	}

services/feed/feed.go

@@ -5,11 +5,138 @@ package feed
 
 import (
 	"context"
+	"fmt"
 
 	activities_model "code.gitea.io/gitea/models/activities"
+	"code.gitea.io/gitea/models/db"
+	access_model "code.gitea.io/gitea/models/perm/access"
+	repo_model "code.gitea.io/gitea/models/repo"
+	"code.gitea.io/gitea/models/unit"
+	user_model "code.gitea.io/gitea/models/user"
+	"code.gitea.io/gitea/modules/setting"
 )
 
 // GetFeeds returns actions according to the provided options
 func GetFeeds(ctx context.Context, opts activities_model.GetFeedsOptions) (activities_model.ActionList, int64, error) {
 	return activities_model.GetFeeds(ctx, opts)
 }
+
+// notifyWatchers creates batch of actions for every watcher.
+// It could insert duplicate actions for a repository action, like this:
+// * Original action: UserID=1 (the real actor), ActUserID=1
+// * Organization action: UserID=100 (the repo's org), ActUserID=1
+// * Watcher action: UserID=20 (a user who is watching a repo), ActUserID=1
+func notifyWatchers(ctx context.Context, act *activities_model.Action, watchers []*repo_model.Watch, permCode, permIssue, permPR []bool) error {
+	// Add feed for actioner.
+	act.UserID = act.ActUserID
+	if err := db.Insert(ctx, act); err != nil {
+		return fmt.Errorf("insert new actioner: %w", err)
+	}
+
+	// Add feed for organization
+	if act.Repo.Owner.IsOrganization() && act.ActUserID != act.Repo.Owner.ID {
+		act.ID = 0
+		act.UserID = act.Repo.Owner.ID
+		if err := db.Insert(ctx, act); err != nil {
+			return fmt.Errorf("insert new actioner: %w", err)
+		}
+	}
+
+	for i, watcher := range watchers {
+		if act.ActUserID == watcher.UserID {
+			continue
+		}
+		act.ID = 0
+		act.UserID = watcher.UserID
+		act.Repo.Units = nil
+
+		switch act.OpType {
+		case activities_model.ActionCommitRepo, activities_model.ActionPushTag, activities_model.ActionDeleteTag, activities_model.ActionPublishRelease, activities_model.ActionDeleteBranch:
+			if !permCode[i] {
+				continue
+			}
+		case activities_model.ActionCreateIssue, activities_model.ActionCommentIssue, activities_model.ActionCloseIssue, activities_model.ActionReopenIssue:
+			if !permIssue[i] {
+				continue
+			}
+		case activities_model.ActionCreatePullRequest, activities_model.ActionCommentPull, activities_model.ActionMergePullRequest, activities_model.ActionClosePullRequest, activities_model.ActionReopenPullRequest, activities_model.ActionAutoMergePullRequest:
+			if !permPR[i] {
+				continue
+			}
+		}
+
+		if err := db.Insert(ctx, act); err != nil {
+			return fmt.Errorf("insert new action: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// NotifyWatchersActions creates batch of actions for every watcher.
+func NotifyWatchers(ctx context.Context, acts ...*activities_model.Action) error {
+	return db.WithTx(ctx, func(ctx context.Context) error {
+		if len(acts) == 0 {
+			return nil
+		}
+
+		repoID := acts[0].RepoID
+		if repoID == 0 {
+			setting.PanicInDevOrTesting("action should belong to a repo")
+			return nil
+		}
+		if err := acts[0].LoadRepo(ctx); err != nil {
+			return err
+		}
+		repo := acts[0].Repo
+		if err := repo.LoadOwner(ctx); err != nil {
+			return err
+		}
+
+		actUserID := acts[0].ActUserID
+
+		// Add feeds for user self and all watchers.
+		watchers, err := repo_model.GetWatchers(ctx, repoID)
+		if err != nil {
+			return fmt.Errorf("get watchers: %w", err)
+		}
+
+		permCode := make([]bool, len(watchers))
+		permIssue := make([]bool, len(watchers))
+		permPR := make([]bool, len(watchers))
+		for i, watcher := range watchers {
+			user, err := user_model.GetUserByID(ctx, watcher.UserID)
+			if err != nil {
+				permCode[i] = false
+				permIssue[i] = false
+				permPR[i] = false
+				continue
+			}
+			perm, err := access_model.GetUserRepoPermission(ctx, repo, user)
+			if err != nil {
+				permCode[i] = false
+				permIssue[i] = false
+				permPR[i] = false
+				continue
+			}
+			permCode[i] = perm.CanRead(unit.TypeCode)
+			permIssue[i] = perm.CanRead(unit.TypeIssues)
+			permPR[i] = perm.CanRead(unit.TypePullRequests)
+		}
+
+		for _, act := range acts {
+			if act.RepoID != repoID {
+				setting.PanicInDevOrTesting("action should belong to the same repo, expected[%d], got[%d] ", repoID, act.RepoID)
+			}
+			if act.ActUserID != actUserID {
+				setting.PanicInDevOrTesting("action should have the same actor, expected[%d], got[%d] ", actUserID, act.ActUserID)
+			}
+
+			act.Repo = repo
+			if err := notifyWatchers(ctx, act, watchers, permCode, permIssue, permPR); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}

services/feed/feed_test.go

@@ -163,3 +163,40 @@ func TestRepoActions(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Len(t, actions, 1)
 }
+
+func TestNotifyWatchers(t *testing.T) {
+	assert.NoError(t, unittest.PrepareTestDatabase())
+
+	action := &activities_model.Action{
+		ActUserID: 8,
+		RepoID:    1,
+		OpType:    activities_model.ActionStarRepo,
+	}
+	assert.NoError(t, NotifyWatchers(db.DefaultContext, action))
+
+	// One watchers are inactive, thus action is only created for user 8, 1, 4, 11
+	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
+		ActUserID: action.ActUserID,
+		UserID:    8,
+		RepoID:    action.RepoID,
+		OpType:    action.OpType,
+	})
+	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
+		ActUserID: action.ActUserID,
+		UserID:    1,
+		RepoID:    action.RepoID,
+		OpType:    action.OpType,
+	})
+	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
+		ActUserID: action.ActUserID,
+		UserID:    4,
+		RepoID:    action.RepoID,
+		OpType:    action.OpType,
+	})
+	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{
+		ActUserID: action.ActUserID,
+		UserID:    11,
+		RepoID:    action.RepoID,
+		OpType:    action.OpType,
+	})
+}

services/feed/notifier.go

@@ -49,7 +49,7 @@ func (a *actionNotifier) NewIssue(ctx context.Context, issue *issues_model.Issue
 	}
 	repo := issue.Repo
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: issue.Poster.ID,
 		ActUser:   issue.Poster,
 		OpType:    activities_model.ActionCreateIssue,
@@ -90,7 +90,7 @@ func (a *actionNotifier) IssueChangeStatus(ctx context.Context, doer *user_model
 	}
 
 	// Notify watchers for whatever action comes in, ignore if no action type.
-	if err := activities_model.NotifyWatchers(ctx, act); err != nil {
+	if err := NotifyWatchers(ctx, act); err != nil {
 		log.Error("NotifyWatchers: %v", err)
 	}
 }
@@ -126,7 +126,7 @@ func (a *actionNotifier) CreateIssueComment(ctx context.Context, doer *user_mode
 	}
 
 	// Notify watchers for whatever action comes in, ignore if no action type.
-	if err := activities_model.NotifyWatchers(ctx, act); err != nil {
+	if err := NotifyWatchers(ctx, act); err != nil {
 		log.Error("NotifyWatchers: %v", err)
 	}
 }
@@ -145,7 +145,7 @@ func (a *actionNotifier) NewPullRequest(ctx context.Context, pull *issues_model.
 		return
 	}
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: pull.Issue.Poster.ID,
 		ActUser:   pull.Issue.Poster,
 		OpType:    activities_model.ActionCreatePullRequest,
@@ -159,7 +159,7 @@ func (a *actionNotifier) NewPullRequest(ctx context.Context, pull *issues_model.
 }
 
 func (a *actionNotifier) RenameRepository(ctx context.Context, doer *user_model.User, repo *repo_model.Repository, oldRepoName string) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionRenameRepo,
@@ -173,7 +173,7 @@ func (a *actionNotifier) RenameRepository(ctx context.Context, doer *user_model.
 }
 
 func (a *actionNotifier) TransferRepository(ctx context.Context, doer *user_model.User, repo *repo_model.Repository, oldOwnerName string) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionTransferRepo,
@@ -187,7 +187,7 @@ func (a *actionNotifier) TransferRepository(ctx context.Context, doer *user_mode
 }
 
 func (a *actionNotifier) CreateRepository(ctx context.Context, doer, u *user_model.User, repo *repo_model.Repository) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionCreateRepo,
@@ -200,7 +200,7 @@ func (a *actionNotifier) CreateRepository(ctx context.Context, doer, u *user_mod
 }
 
 func (a *actionNotifier) ForkRepository(ctx context.Context, doer *user_model.User, oldRepo, repo *repo_model.Repository) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionCreateRepo,
@@ -265,13 +265,13 @@ func (a *actionNotifier) PullRequestReview(ctx context.Context, pr *issues_model
 		actions = append(actions, action)
 	}
 
-	if err := activities_model.NotifyWatchersActions(ctx, actions); err != nil {
+	if err := NotifyWatchers(ctx, actions...); err != nil {
 		log.Error("notify watchers '%d/%d': %v", review.Reviewer.ID, review.Issue.RepoID, err)
 	}
 }
 
 func (*actionNotifier) MergePullRequest(ctx context.Context, doer *user_model.User, pr *issues_model.PullRequest) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionMergePullRequest,
@@ -285,7 +285,7 @@ func (*actionNotifier) MergePullRequest(ctx context.Context, doer *user_model.Us
 }
 
 func (*actionNotifier) AutoMergePullRequest(ctx context.Context, doer *user_model.User, pr *issues_model.PullRequest) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionAutoMergePullRequest,
@@ -303,7 +303,7 @@ func (*actionNotifier) NotifyPullRevieweDismiss(ctx context.Context, doer *user_
 	if len(review.OriginalAuthor) > 0 {
 		reviewerName = review.OriginalAuthor
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionPullReviewDismissed,
@@ -337,7 +337,7 @@ func (a *actionNotifier) PushCommits(ctx context.Context, pusher *user_model.Use
 		opType = activities_model.ActionDeleteBranch
 	}
 
-	if err = activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err = NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: pusher.ID,
 		ActUser:   pusher,
 		OpType:    opType,
@@ -357,7 +357,7 @@ func (a *actionNotifier) CreateRef(ctx context.Context, doer *user_model.User, r
 		// has sent same action in `PushCommits`, so skip it.
 		return
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    opType,
@@ -376,7 +376,7 @@ func (a *actionNotifier) DeleteRef(ctx context.Context, doer *user_model.User, r
 		// has sent same action in `PushCommits`, so skip it.
 		return
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    opType,
@@ -402,7 +402,7 @@ func (a *actionNotifier) SyncPushCommits(ctx context.Context, pusher *user_model
 		return
 	}
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: repo.OwnerID,
 		ActUser:   repo.MustOwner(ctx),
 		OpType:    activities_model.ActionMirrorSyncPush,
@@ -423,7 +423,7 @@ func (a *actionNotifier) SyncCreateRef(ctx context.Context, doer *user_model.Use
 		return
 	}
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: repo.OwnerID,
 		ActUser:   repo.MustOwner(ctx),
 		OpType:    activities_model.ActionMirrorSyncCreate,
@@ -443,7 +443,7 @@ func (a *actionNotifier) SyncDeleteRef(ctx context.Context, doer *user_model.Use
 		return
 	}
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: repo.OwnerID,
 		ActUser:   repo.MustOwner(ctx),
 		OpType:    activities_model.ActionMirrorSyncDelete,
@@ -461,7 +461,7 @@ func (a *actionNotifier) NewRelease(ctx context.Context, rel *repo_model.Release
 		log.Error("LoadAttributes: %v", err)
 		return
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := NotifyWatchers(ctx, &activities_model.Action{
 		ActUserID: rel.PublisherID,
 		ActUser:   rel.Publisher,
 		OpType:    activities_model.ActionPublishRelease,