Move notifywatch to service layer (#33825)

No logic change.

3 files changed, 183 insertions, 19 deletions

diff --git a/services/feed/feed.go b/services/feed/feed.go
index 93bf875fd0..a1c327fb51 100644
--- a/services/feed/feed.go
+++ b/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
+	})
+}
diff --git a/services/feed/feed_test.go b/services/feed/feed_test.go
index 1e4d029e18..243bc046b0 100644
--- a/services/feed/feed_test.go
+++ b/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,
+	})
+}
diff --git a/services/feed/notifier.go b/services/feed/notifier.go
index 3aaf885c9a..64aeccdfd2 100644
--- a/services/feed/notifier.go
+++ b/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,