[feature] Receive notification when followed account posts (if desired) (#1680)

* start working on notifs for new posts

* tidy up a bit

* update swagger

* carry over show reblogs + notify from follow req

* test notify on status post

* update column slice

* dedupe update logic + add tests

* fix own boosts not being timelined

* avoid type check, passing unnecessary accounts

* remove unnecessary 'inReplyToID' check

* add a couple todo's for future db functions

docs/api/swagger.yaml

@@ -2774,6 +2774,9 @@ paths:
             description: |-
                 The parameters can also be given in the body of the request, as JSON, if the content-type is set to 'application/json'.
                 The parameters can also be given in the body of the request, as XML, if the content-type is set to 'application/xml'.
+
+                If you already follow (request) the given account, then the follow (request) will be updated instead using the
+                `reblogs` and `notify` parameters.
             operationId: accountFollow
             parameters:
                 - description: ID of the account to follow.

internal/api/client/accounts/follow.go

@@ -35,6 +35,9 @@ import (
 // The parameters can also be given in the body of the request, as JSON, if the content-type is set to 'application/json'.
 // The parameters can also be given in the body of the request, as XML, if the content-type is set to 'application/xml'.
 //
+// If you already follow (request) the given account, then the follow (request) will be updated instead using the
+// `reblogs` and `notify` parameters.
+//
 //	---
 //	tags:
 //	- accounts
@@ -58,11 +61,11 @@ import (
 //		description: Show reblogs from this account.
 //		in: formData
 //	-
+//		name: notify
+//		type: boolean
 //		default: false
 //		description: Notify when this account posts.
 //		in: formData
-//		name: notify
-//		type: boolean
 //
 //	produces:
 //	- application/json

internal/cache/gts.go

@@ -320,6 +320,7 @@ func (c *GTSCaches) initMention() {
 func (c *GTSCaches) initNotification() {
 	c.notification = result.New([]result.Lookup{
 		{Name: "ID"},
+		{Name: "NotificationType.TargetAccountID.OriginAccountID.StatusID"},
 	}, func(n1 *gtsmodel.Notification) *gtsmodel.Notification {
 		n2 := new(gtsmodel.Notification)
 		*n2 = *n1

internal/db/bundb/notification.go

@@ -48,6 +48,31 @@ func (n *notificationDB) GetNotificationByID(ctx context.Context, id string) (*g
 	}, id)
 }
 
+func (n *notificationDB) GetNotification(
+	ctx context.Context,
+	notificationType gtsmodel.NotificationType,
+	targetAccountID string,
+	originAccountID string,
+	statusID string,
+) (*gtsmodel.Notification, db.Error) {
+	return n.state.Caches.GTS.Notification().Load("NotificationType.TargetAccountID.OriginAccountID.StatusID", func() (*gtsmodel.Notification, error) {
+		var notif gtsmodel.Notification
+
+		q := n.conn.NewSelect().
+			Model(&notif).
+			Where("? = ?", bun.Ident("notification_type"), notificationType).
+			Where("? = ?", bun.Ident("target_account_id"), targetAccountID).
+			Where("? = ?", bun.Ident("origin_account_id"), originAccountID).
+			Where("? = ?", bun.Ident("status_id"), statusID)
+
+		if err := q.Scan(ctx); err != nil {
+			return nil, n.conn.ProcessError(err)
+		}
+
+		return &notif, nil
+	}, notificationType, targetAccountID, originAccountID, statusID)
+}
+
 func (n *notificationDB) GetAccountNotifications(ctx context.Context, accountID string, excludeTypes []string, limit int, maxID string, sinceID string) ([]*gtsmodel.Notification, db.Error) {
 	// Ensure reasonable
 	if limit < 0 {

internal/db/bundb/relationship_follow.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
@@ -189,6 +190,26 @@ func (r *relationshipDB) PutFollow(ctx context.Context, follow *gtsmodel.Follow)
 	return nil
 }
 
+func (r *relationshipDB) UpdateFollow(ctx context.Context, follow *gtsmodel.Follow, columns ...string) error {
+	follow.UpdatedAt = time.Now()
+	if len(columns) > 0 {
+		// If we're updating by column, ensure "updated_at" is included.
+		columns = append(columns, "updated_at")
+	}
+
+	return r.state.Caches.GTS.Follow().Store(follow, func() error {
+		if _, err := r.conn.NewUpdate().
+			Model(follow).
+			Where("? = ?", bun.Ident("follow.id"), follow.ID).
+			Column(columns...).
+			Exec(ctx); err != nil {
+			return r.conn.ProcessError(err)
+		}
+
+		return nil
+	})
+}
+
 func (r *relationshipDB) DeleteFollowByID(ctx context.Context, id string) error {
 	if _, err := r.conn.NewDelete().
 		Table("follows").

internal/db/bundb/relationship_follow_req.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
@@ -167,6 +168,26 @@ func (r *relationshipDB) PutFollowRequest(ctx context.Context, follow *gtsmodel.
 	return nil
 }
 
+func (r *relationshipDB) UpdateFollowRequest(ctx context.Context, followRequest *gtsmodel.FollowRequest, columns ...string) error {
+	followRequest.UpdatedAt = time.Now()
+	if len(columns) > 0 {
+		// If we're updating by column, ensure "updated_at" is included.
+		columns = append(columns, "updated_at")
+	}
+
+	return r.state.Caches.GTS.FollowRequest().Store(followRequest, func() error {
+		if _, err := r.conn.NewUpdate().
+			Model(followRequest).
+			Where("? = ?", bun.Ident("follow_request.id"), followRequest.ID).
+			Column(columns...).
+			Exec(ctx); err != nil {
+			return r.conn.ProcessError(err)
+		}
+
+		return nil
+	})
+}
+
 func (r *relationshipDB) AcceptFollowRequest(ctx context.Context, sourceAccountID string, targetAccountID string) (*gtsmodel.Follow, db.Error) {
 	// Get original follow request.
 	followReq, err := r.GetFollowRequest(ctx, sourceAccountID, targetAccountID)
@@ -183,6 +204,8 @@ func (r *relationshipDB) AcceptFollowRequest(ctx context.Context, sourceAccountI
 		TargetAccountID: targetAccountID,
 		TargetAccount:   followReq.TargetAccount,
 		URI:             followReq.URI,
+		ShowReblogs:     followReq.ShowReblogs,
+		Notify:          followReq.Notify,
 	}
 
 	// If the follow already exists, just

internal/db/bundb/relationship_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/id"
+	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
 type RelationshipTestSuite struct {
@@ -861,6 +862,32 @@ func (suite *RelationshipTestSuite) TestUnfollowRequestNotExisting() {
 	suite.Nil(followRequest)
 }
 
+func (suite *RelationshipTestSuite) TestUpdateFollow() {
+	ctx := context.Background()
+
+	follow := &gtsmodel.Follow{}
+	*follow = *suite.testFollows["local_account_1_admin_account"]
+
+	follow.Notify = testrig.TrueBool()
+	if err := suite.db.UpdateFollow(ctx, follow, "notify"); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	dbFollow, err := suite.db.GetFollowByID(ctx, follow.ID)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.True(*dbFollow.Notify)
+
+	relationship, err := suite.db.GetRelationship(ctx, follow.AccountID, follow.TargetAccountID)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.True(relationship.Notifying)
+}
+
 func TestRelationshipTestSuite(t *testing.T) {
 	suite.Run(t, new(RelationshipTestSuite))
 }

internal/db/notification.go

@@ -33,6 +33,10 @@ type Notification interface {
 	// GetNotification returns one notification according to its id.
 	GetNotificationByID(ctx context.Context, id string) (*gtsmodel.Notification, Error)
 
+	// GetNotification gets one notification according to the provided parameters, if it exists.
+	// Since not all notifications are about a status, statusID can be an empty string.
+	GetNotification(ctx context.Context, notificationType gtsmodel.NotificationType, targetAccountID string, originAccountID string, statusID string) (*gtsmodel.Notification, Error)
+
 	// PutNotification will insert the given notification into the database.
 	PutNotification(ctx context.Context, notif *gtsmodel.Notification) error
 

internal/db/relationship.go

@@ -85,9 +85,15 @@ type Relationship interface {
 	// PutFollow attempts to place the given account follow in the database.
 	PutFollow(ctx context.Context, follow *gtsmodel.Follow) error
 
+	// UpdateFollow updates one follow by ID.
+	UpdateFollow(ctx context.Context, follow *gtsmodel.Follow, columns ...string) error
+
 	// PutFollowRequest attempts to place the given account follow request in the database.
 	PutFollowRequest(ctx context.Context, follow *gtsmodel.FollowRequest) error
 
+	// UpdateFollowRequest updates one follow request by ID.
+	UpdateFollowRequest(ctx context.Context, followRequest *gtsmodel.FollowRequest, columns ...string) error
+
 	// DeleteFollowByID deletes a follow from the database with the given ID.
 	DeleteFollowByID(ctx context.Context, id string) error
 

internal/processing/account/account_test.go

@@ -59,6 +59,7 @@ type AccountStandardTestSuite struct {
 	testApplications map[string]*gtsmodel.Application
 	testUsers        map[string]*gtsmodel.User
 	testAccounts     map[string]*gtsmodel.Account
+	testFollows      map[string]*gtsmodel.Follow
 	testAttachments  map[string]*gtsmodel.MediaAttachment
 	testStatuses     map[string]*gtsmodel.Status
 
@@ -72,6 +73,7 @@ func (suite *AccountStandardTestSuite) SetupSuite() {
 	suite.testApplications = testrig.NewTestApplications()
 	suite.testUsers = testrig.NewTestUsers()
 	suite.testAccounts = testrig.NewTestAccounts()
+	suite.testFollows = testrig.NewTestFollows()
 	suite.testAttachments = testrig.NewTestAttachments()
 	suite.testStatuses = testrig.NewTestStatuses()
 }
@@ -80,8 +82,8 @@ func (suite *AccountStandardTestSuite) SetupTest() {
 	suite.state.Caches.Init()
 	testrig.StartWorkers(&suite.state)
 
-	testrig.InitTestLog()
 	testrig.InitTestConfig()
+	testrig.InitTestLog()
 
 	suite.db = testrig.NewTestDB(&suite.state)
 	suite.state.DB = suite.db

internal/processing/account/follow.go

@@ -25,6 +25,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
+	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/id"
@@ -40,24 +41,47 @@ func (p *Processor) FollowCreate(ctx context.Context, requestingAccount *gtsmode
 	}
 
 	// Check if a follow exists already.
-	if follows, err := p.state.DB.IsFollowing(ctx, requestingAccount.ID, targetAccount.ID); err != nil {
+	if follow, err := p.state.DB.GetFollow(
-		err = fmt.Errorf("FollowCreate: db error checking follow: %w", err)
+		gtscontext.SetBarebones(ctx),
+		requestingAccount.ID,
+		targetAccount.ID,
+	); err != nil && !errors.Is(err, db.ErrNoEntries) {
+		err = fmt.Errorf("FollowCreate: db error checking existing follow: %w", err)
 		return nil, gtserror.NewErrorInternalError(err)
-	} else if follows {
+	} else if follow != nil {
-		// Already follows, just return current relationship.
+		// Already follows, update if necessary + return relationship.
-		return p.RelationshipGet(ctx, requestingAccount, form.ID)
+		return p.updateFollow(
+			ctx,
+			requestingAccount,
+			form,
+			follow.ShowReblogs,
+			follow.Notify,
+			func(columns ...string) error { return p.state.DB.UpdateFollow(ctx, follow, columns...) },
+		)
 	}
 
 	// Check if a follow request exists already.
-	if followRequested, err := p.state.DB.IsFollowRequested(ctx, requestingAccount.ID, targetAccount.ID); err != nil {
+	if followRequest, err := p.state.DB.GetFollowRequest(
-		err = fmt.Errorf("FollowCreate: db error checking follow request: %w", err)
+		gtscontext.SetBarebones(ctx),
+		requestingAccount.ID,
+		targetAccount.ID,
+	); err != nil && !errors.Is(err, db.ErrNoEntries) {
+		err = fmt.Errorf("FollowCreate: db error checking existing follow request: %w", err)
 		return nil, gtserror.NewErrorInternalError(err)
-	} else if followRequested {
+	} else if followRequest != nil {
-		// Already follow requested, just return current relationship.
+		// Already requested, update if necessary + return relationship.
-		return p.RelationshipGet(ctx, requestingAccount, form.ID)
+		return p.updateFollow(
+			ctx,
+			requestingAccount,
+			form,
+			followRequest.ShowReblogs,
+			followRequest.Notify,
+			func(columns ...string) error { return p.state.DB.UpdateFollowRequest(ctx, followRequest, columns...) },
+		)
 	}
 
-	// Create and store a new follow request.
+	// Neither follows nor follow requests, so
+	// create and store a new follow request.
 	followID, err := id.NewRandomULID()
 	if err != nil {
 		return nil, gtserror.NewErrorInternalError(err)
@@ -129,6 +153,52 @@ func (p *Processor) FollowRemove(ctx context.Context, requestingAccount *gtsmode
 	Utility functions.
 */
 
+// updateFollow is a utility function for updating an existing
+// follow or followRequest with the parameters provided in the
+// given form. If nothing changes, this function is a no-op and
+// will just return the existing relationship between follow
+// origin and follow target account.
+func (p *Processor) updateFollow(
+	ctx context.Context,
+	requestingAccount *gtsmodel.Account,
+	form *apimodel.AccountFollowRequest,
+	currentShowReblogs *bool,
+	currentNotify *bool,
+	update func(...string) error,
+) (*apimodel.Relationship, gtserror.WithCode) {
+
+	if form.Reblogs == nil && form.Notify == nil {
+		// There's nothing to update.
+		return p.RelationshipGet(ctx, requestingAccount, form.ID)
+	}
+
+	// Including "updated_at", max 3 columns may change.
+	columns := make([]string, 0, 3)
+
+	// Check what we need to update (if anything).
+	if newReblogs := form.Reblogs; newReblogs != nil && *newReblogs != *currentShowReblogs {
+		*currentShowReblogs = *newReblogs
+		columns = append(columns, "show_reblogs")
+	}
+
+	if newNotify := form.Notify; newNotify != nil && *newNotify != *currentNotify {
+		*currentNotify = *newNotify
+		columns = append(columns, "notify")
+	}
+
+	if len(columns) == 0 {
+		// Nothing actually changed.
+		return p.RelationshipGet(ctx, requestingAccount, form.ID)
+	}
+
+	if err := update(columns...); err != nil {
+		err = fmt.Errorf("updateFollow: error updating existing follow (request): %w", err)
+		return nil, gtserror.NewErrorInternalError(err)
+	}
+
+	return p.RelationshipGet(ctx, requestingAccount, form.ID)
+}
+
 // getFollowTarget is a convenience function which:
 //   - Checks if account is trying to follow/unfollow itself.
 //   - Returns not found if there's a block in place between accounts.

internal/processing/account/follow_test.go

@@ -0,0 +1,140 @@
+// GoToSocial
+// Copyright (C) GoToSocial Authors admin@gotosocial.org
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package account_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type FollowTestSuite struct {
+	AccountStandardTestSuite
+}
+
+func (suite *FollowTestSuite) TestUpdateExistingFollowChangeBoth() {
+	ctx := context.Background()
+	requestingAccount := suite.testAccounts["local_account_1"]
+	targetAccount := suite.testAccounts["admin_account"]
+
+	// Change both Reblogs and Notify.
+	// Trace logs should show a query similar to this:
+	//	UPDATE "follows" AS "follow" SET "show_reblogs" = FALSE, "notify" = TRUE, "updated_at" = '2023-04-09 11:42:39.424705+00:00' WHERE ("follow"."id" = '01F8PY8RHWRQZV038T4E8T9YK8')
+	relationship, err := suite.accountProcessor.FollowCreate(ctx, requestingAccount, &apimodel.AccountFollowRequest{
+		ID:      targetAccount.ID,
+		Reblogs: testrig.FalseBool(),
+		Notify:  testrig.TrueBool(),
+	})
+
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.False(relationship.ShowingReblogs)
+	suite.True(relationship.Notifying)
+}
+
+func (suite *FollowTestSuite) TestUpdateExistingFollowChangeNotifyIgnoreReblogs() {
+	ctx := context.Background()
+	requestingAccount := suite.testAccounts["local_account_1"]
+	targetAccount := suite.testAccounts["admin_account"]
+
+	// Change Notify, ignore Reblogs.
+	// Trace logs should show a query similar to this:
+	//	UPDATE "follows" AS "follow" SET "notify" = TRUE, "updated_at" = '2023-04-09 11:40:33.827858+00:00' WHERE ("follow"."id" = '01F8PY8RHWRQZV038T4E8T9YK8')
+	relationship, err := suite.accountProcessor.FollowCreate(ctx, requestingAccount, &apimodel.AccountFollowRequest{
+		ID:     targetAccount.ID,
+		Notify: testrig.TrueBool(),
+	})
+
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.True(relationship.ShowingReblogs)
+	suite.True(relationship.Notifying)
+}
+
+func (suite *FollowTestSuite) TestUpdateExistingFollowChangeNotifySetReblogs() {
+	ctx := context.Background()
+	requestingAccount := suite.testAccounts["local_account_1"]
+	targetAccount := suite.testAccounts["admin_account"]
+
+	// Change Notify, set Reblogs to same value as before.
+	// Trace logs should show a query similar to this:
+	//	UPDATE "follows" AS "follow" SET "notify" = TRUE, "updated_at" = '2023-04-09 11:40:33.827858+00:00' WHERE ("follow"."id" = '01F8PY8RHWRQZV038T4E8T9YK8')
+	relationship, err := suite.accountProcessor.FollowCreate(ctx, requestingAccount, &apimodel.AccountFollowRequest{
+		ID:      targetAccount.ID,
+		Notify:  testrig.TrueBool(),
+		Reblogs: testrig.TrueBool(),
+	})
+
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.True(relationship.ShowingReblogs)
+	suite.True(relationship.Notifying)
+}
+
+func (suite *FollowTestSuite) TestUpdateExistingFollowChangeNothing() {
+	ctx := context.Background()
+	requestingAccount := suite.testAccounts["local_account_1"]
+	targetAccount := suite.testAccounts["admin_account"]
+
+	// Set Notify and Reblogs to same values as before.
+	// Trace logs should show no update query.
+	relationship, err := suite.accountProcessor.FollowCreate(ctx, requestingAccount, &apimodel.AccountFollowRequest{
+		ID:      targetAccount.ID,
+		Notify:  testrig.FalseBool(),
+		Reblogs: testrig.TrueBool(),
+	})
+
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.True(relationship.ShowingReblogs)
+	suite.False(relationship.Notifying)
+}
+
+func (suite *FollowTestSuite) TestUpdateExistingFollowSetNothing() {
+	ctx := context.Background()
+	requestingAccount := suite.testAccounts["local_account_1"]
+	targetAccount := suite.testAccounts["admin_account"]
+
+	// Don't set Notify or Reblogs.
+	// Trace logs should show no update query.
+	relationship, err := suite.accountProcessor.FollowCreate(ctx, requestingAccount, &apimodel.AccountFollowRequest{
+		ID: targetAccount.ID,
+	})
+
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.True(relationship.ShowingReblogs)
+	suite.False(relationship.Notifying)
+}
+
+func TestFollowTestS(t *testing.T) {
+	suite.Run(t, new(FollowTestSuite))
+}

internal/processing/fromclientapi.go

@@ -160,11 +160,7 @@ func (p *Processor) processCreateStatusFromClientAPI(ctx context.Context, client
 		return errors.New("note was not parseable as *gtsmodel.Status")
 	}
 
-	if err := p.timelineStatus(ctx, status); err != nil {
+	if err := p.timelineAndNotifyStatus(ctx, status); err != nil {
-		return err
-	}
-
-	if err := p.notifyStatus(ctx, status); err != nil {
 		return err
 	}
 
@@ -203,7 +199,7 @@ func (p *Processor) processCreateAnnounceFromClientAPI(ctx context.Context, clie
 		return errors.New("boost was not parseable as *gtsmodel.Status")
 	}
 
-	if err := p.timelineStatus(ctx, boostWrapperStatus); err != nil {
+	if err := p.timelineAndNotifyStatus(ctx, boostWrapperStatus); err != nil {
 		return err
 	}
 
@@ -255,7 +251,7 @@ func (p *Processor) processUpdateReportFromClientAPI(ctx context.Context, client
 		return nil
 	}
 
-	return p.notifyReportClosed(ctx, report)
+	return p.emailReportClosed(ctx, report)
 }
 
 func (p *Processor) processAcceptFollowFromClientAPI(ctx context.Context, clientMsg messages.FromClientAPI) error {
@@ -373,7 +369,7 @@ func (p *Processor) processReportAccountFromClientAPI(ctx context.Context, clien
 		}
 	}
 
-	if err := p.notifyReport(ctx, report); err != nil {
+	if err := p.emailReport(ctx, report); err != nil {
 		return fmt.Errorf("processReportAccountFromClientAPI: error notifying report: %w", err)
 	}
 

internal/processing/fromclientapi_test.go

@@ -159,6 +159,79 @@ func (suite *FromClientAPITestSuite) TestProcessStatusDelete() {
 	suite.ErrorIs(err, db.ErrNoEntries)
 }
 
+func (suite *FromClientAPITestSuite) TestProcessNewStatusWithNotification() {
+	ctx := context.Background()
+	postingAccount := suite.testAccounts["admin_account"]
+	receivingAccount := suite.testAccounts["local_account_1"]
+
+	// Update the follow from receiving account -> posting account so
+	// that receiving account wants notifs when posting account posts.
+	follow := &gtsmodel.Follow{}
+	*follow = *suite.testFollows["local_account_1_admin_account"]
+	follow.Notify = testrig.TrueBool()
+	if err := suite.db.UpdateFollow(ctx, follow); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Make a new status from admin account.
+	newStatus := &gtsmodel.Status{
+		ID:                       "01FN4B2F88TF9676DYNXWE1WSS",
+		URI:                      "http://localhost:8080/users/admin/statuses/01FN4B2F88TF9676DYNXWE1WSS",
+		URL:                      "http://localhost:8080/@admin/statuses/01FN4B2F88TF9676DYNXWE1WSS",
+		Content:                  "this status should create a notification",
+		AttachmentIDs:            []string{},
+		TagIDs:                   []string{},
+		MentionIDs:               []string{},
+		EmojiIDs:                 []string{},
+		CreatedAt:                testrig.TimeMustParse("2021-10-20T11:36:45Z"),
+		UpdatedAt:                testrig.TimeMustParse("2021-10-20T11:36:45Z"),
+		Local:                    testrig.TrueBool(),
+		AccountURI:               "http://localhost:8080/users/admin",
+		AccountID:                "01F8MH17FWEB39HZJ76B6VXSKF",
+		InReplyToID:              "",
+		BoostOfID:                "",
+		ContentWarning:           "",
+		Visibility:               gtsmodel.VisibilityFollowersOnly,
+		Sensitive:                testrig.FalseBool(),
+		Language:                 "en",
+		CreatedWithApplicationID: "01F8MGXQRHYF5QPMTMXP78QC2F",
+		Federated:                testrig.FalseBool(),
+		Boostable:                testrig.TrueBool(),
+		Replyable:                testrig.TrueBool(),
+		Likeable:                 testrig.TrueBool(),
+		ActivityStreamsType:      ap.ObjectNote,
+	}
+
+	// Put the status in the db first, to mimic what
+	// would have already happened earlier up the flow.
+	err := suite.db.PutStatus(ctx, newStatus)
+	suite.NoError(err)
+
+	// Process the new status.
+	if err := suite.processor.ProcessFromClientAPI(ctx, messages.FromClientAPI{
+		APObjectType:   ap.ObjectNote,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       newStatus,
+		OriginAccount:  postingAccount,
+	}); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Wait for a notification to appear for the status.
+	if !testrig.WaitFor(func() bool {
+		_, err := suite.db.GetNotification(
+			ctx,
+			gtsmodel.NotificationStatus,
+			receivingAccount.ID,
+			postingAccount.ID,
+			newStatus.ID,
+		)
+		return err == nil
+	}) {
+		suite.FailNow("timed out waiting for new status notification")
+	}
+}
+
 func TestFromClientAPITestSuite(t *testing.T) {
 	suite.Run(t, &FromClientAPITestSuite{})
 }

internal/processing/fromcommon.go

@@ -25,396 +25,27 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/email"
+	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/id"
 	"github.com/superseriousbusiness/gotosocial/internal/stream"
 )
 
-func (p *Processor) notifyStatus(ctx context.Context, status *gtsmodel.Status) error {
+// timelineAndNotifyStatus processes the given new status and inserts it into
-	// if there are no mentions in this status then just bail
+// the HOME timelines of accounts that follow the status author. It will also
-	if len(status.MentionIDs) == 0 {
+// handle notifications for any mentions attached to the account, and also
-		return nil
+// notifications for any local accounts that want a notif when this account posts.
-	}
+func (p *Processor) timelineAndNotifyStatus(ctx context.Context, status *gtsmodel.Status) error {
-
+	// Ensure status fully populated; including account, mentions, etc.
-	if status.Mentions == nil {
-		// there are mentions but they're not fully populated on the status yet so do this
-		mentions, err := p.state.DB.GetMentions(ctx, status.MentionIDs)
-		if err != nil {
-			return fmt.Errorf("notifyStatus: error getting mentions for status %s from the db: %s", status.ID, err)
-		}
-
-		status.Mentions = mentions
-	}
-
-	// now we have mentions as full gtsmodel.Mention structs on the status we can continue
-	for _, m := range status.Mentions {
-		// make sure this is a local account, otherwise we don't need to create a notification for it
-		if m.TargetAccount == nil {
-			a, err := p.state.DB.GetAccountByID(ctx, m.TargetAccountID)
-			if err != nil {
-				// we don't have the account or there's been an error
-				return fmt.Errorf("notifyStatus: error getting account with id %s from the db: %s", m.TargetAccountID, err)
-			}
-			m.TargetAccount = a
-		}
-		if m.TargetAccount.Domain != "" {
-			// not a local account so skip it
-			continue
-		}
-
-		// make sure a notif doesn't already exist for this mention
-		if err := p.state.DB.GetWhere(ctx, []db.Where{
-			{Key: "notification_type", Value: gtsmodel.NotificationMention},
-			{Key: "target_account_id", Value: m.TargetAccountID},
-			{Key: "origin_account_id", Value: m.OriginAccountID},
-			{Key: "status_id", Value: m.StatusID},
-		}, &gtsmodel.Notification{}); err == nil {
-			// notification exists already so just continue
-			continue
-		} else if err != db.ErrNoEntries {
-			// there's a real error in the db
-			return fmt.Errorf("notifyStatus: error checking existence of notification for mention with id %s : %s", m.ID, err)
-		}
-
-		// if we've reached this point we know the mention is for a local account, and the notification doesn't exist, so create it
-		notif := &gtsmodel.Notification{
-			ID:               id.NewULID(),
-			NotificationType: gtsmodel.NotificationMention,
-			TargetAccountID:  m.TargetAccountID,
-			TargetAccount:    m.TargetAccount,
-			OriginAccountID:  status.AccountID,
-			OriginAccount:    status.Account,
-			StatusID:         status.ID,
-			Status:           status,
-		}
-
-		if err := p.state.DB.PutNotification(ctx, notif); err != nil {
-			return fmt.Errorf("notifyStatus: error putting notification in database: %s", err)
-		}
-
-		// now stream the notification to the user
-		apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif)
-		if err != nil {
-			return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err)
-		}
-
-		if err := p.stream.Notify(apiNotif, m.TargetAccount); err != nil {
-			return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err)
-		}
-	}
-
-	return nil
-}
-
-func (p *Processor) notifyFollowRequest(ctx context.Context, followRequest *gtsmodel.FollowRequest) error {
-	// make sure we have the target account pinned on the follow request
-	if followRequest.TargetAccount == nil {
-		a, err := p.state.DB.GetAccountByID(ctx, followRequest.TargetAccountID)
-		if err != nil {
-			return err
-		}
-		followRequest.TargetAccount = a
-	}
-	targetAccount := followRequest.TargetAccount
-
-	// return if this isn't a local account
-	if targetAccount.Domain != "" {
-		// this isn't a local account so we've got nothing to do here
-		return nil
-	}
-
-	notif := &gtsmodel.Notification{
-		ID:               id.NewULID(),
-		NotificationType: gtsmodel.NotificationFollowRequest,
-		TargetAccountID:  followRequest.TargetAccountID,
-		OriginAccountID:  followRequest.AccountID,
-	}
-
-	if err := p.state.DB.PutNotification(ctx, notif); err != nil {
-		return fmt.Errorf("notifyFollowRequest: error putting notification in database: %s", err)
-	}
-
-	// now stream the notification to the user
-	apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif)
-	if err != nil {
-		return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err)
-	}
-
-	if err := p.stream.Notify(apiNotif, targetAccount); err != nil {
-		return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err)
-	}
-
-	return nil
-}
-
-func (p *Processor) notifyFollow(ctx context.Context, follow *gtsmodel.Follow, targetAccount *gtsmodel.Account) error {
-	// return if this isn't a local account
-	if targetAccount.Domain != "" {
-		return nil
-	}
-
-	// first remove the follow request notification
-	if err := p.state.DB.DeleteWhere(ctx, []db.Where{
-		{Key: "notification_type", Value: gtsmodel.NotificationFollowRequest},
-		{Key: "target_account_id", Value: follow.TargetAccountID},
-		{Key: "origin_account_id", Value: follow.AccountID},
-	}, &gtsmodel.Notification{}); err != nil {
-		return fmt.Errorf("notifyFollow: error removing old follow request notification from database: %s", err)
-	}
-
-	// now create the new follow notification
-	notif := &gtsmodel.Notification{
-		ID:               id.NewULID(),
-		NotificationType: gtsmodel.NotificationFollow,
-		TargetAccountID:  follow.TargetAccountID,
-		TargetAccount:    follow.TargetAccount,
-		OriginAccountID:  follow.AccountID,
-		OriginAccount:    follow.Account,
-	}
-	if err := p.state.DB.PutNotification(ctx, notif); err != nil {
-		return fmt.Errorf("notifyFollow: error putting notification in database: %s", err)
-	}
-
-	// now stream the notification to the user
-	apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif)
-	if err != nil {
-		return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err)
-	}
-
-	if err := p.stream.Notify(apiNotif, targetAccount); err != nil {
-		return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err)
-	}
-
-	return nil
-}
-
-func (p *Processor) notifyFave(ctx context.Context, fave *gtsmodel.StatusFave) error {
-	// ignore self-faves
-	if fave.TargetAccountID == fave.AccountID {
-		return nil
-	}
-
-	if fave.TargetAccount == nil {
-		a, err := p.state.DB.GetAccountByID(ctx, fave.TargetAccountID)
-		if err != nil {
-			return err
-		}
-		fave.TargetAccount = a
-	}
-	targetAccount := fave.TargetAccount
-
-	// just return if target isn't a local account
-	if targetAccount.Domain != "" {
-		return nil
-	}
-
-	notif := &gtsmodel.Notification{
-		ID:               id.NewULID(),
-		NotificationType: gtsmodel.NotificationFave,
-		TargetAccountID:  fave.TargetAccountID,
-		TargetAccount:    fave.TargetAccount,
-		OriginAccountID:  fave.AccountID,
-		OriginAccount:    fave.Account,
-		StatusID:         fave.StatusID,
-		Status:           fave.Status,
-	}
-
-	if err := p.state.DB.PutNotification(ctx, notif); err != nil {
-		return fmt.Errorf("notifyFave: error putting notification in database: %s", err)
-	}
-
-	// now stream the notification to the user
-	apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif)
-	if err != nil {
-		return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err)
-	}
-
-	if err := p.stream.Notify(apiNotif, targetAccount); err != nil {
-		return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err)
-	}
-
-	return nil
-}
-
-func (p *Processor) notifyAnnounce(ctx context.Context, status *gtsmodel.Status) error {
-	if status.BoostOfID == "" {
-		// not a boost, nothing to do
-		return nil
-	}
-
-	if status.BoostOf == nil {
-		boostedStatus, err := p.state.DB.GetStatusByID(ctx, status.BoostOfID)
-		if err != nil {
-			return fmt.Errorf("notifyAnnounce: error getting status with id %s: %s", status.BoostOfID, err)
-		}
-		status.BoostOf = boostedStatus
-	}
-
-	if status.BoostOfAccount == nil {
-		boostedAcct, err := p.state.DB.GetAccountByID(ctx, status.BoostOfAccountID)
-		if err != nil {
-			return fmt.Errorf("notifyAnnounce: error getting account with id %s: %s", status.BoostOfAccountID, err)
-		}
-		status.BoostOf.Account = boostedAcct
-		status.BoostOfAccount = boostedAcct
-	}
-
-	if status.BoostOfAccount.Domain != "" {
-		// remote account, nothing to do
-		return nil
-	}
-
-	if status.BoostOfAccountID == status.AccountID {
-		// it's a self boost, nothing to do
-		return nil
-	}
-
-	// make sure a notif doesn't already exist for this announce
-	err := p.state.DB.GetWhere(ctx, []db.Where{
-		{Key: "notification_type", Value: gtsmodel.NotificationReblog},
-		{Key: "target_account_id", Value: status.BoostOfAccountID},
-		{Key: "origin_account_id", Value: status.AccountID},
-		{Key: "status_id", Value: status.ID},
-	}, &gtsmodel.Notification{})
-	if err == nil {
-		// notification exists already so just bail
-		return nil
-	}
-
-	// now create the new reblog notification
-	notif := &gtsmodel.Notification{
-		ID:               id.NewULID(),
-		NotificationType: gtsmodel.NotificationReblog,
-		TargetAccountID:  status.BoostOfAccountID,
-		TargetAccount:    status.BoostOfAccount,
-		OriginAccountID:  status.AccountID,
-		OriginAccount:    status.Account,
-		StatusID:         status.ID,
-		Status:           status,
-	}
-
-	if err := p.state.DB.PutNotification(ctx, notif); err != nil {
-		return fmt.Errorf("notifyAnnounce: error putting notification in database: %s", err)
-	}
-
-	// now stream the notification to the user
-	apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif)
-	if err != nil {
-		return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err)
-	}
-
-	if err := p.stream.Notify(apiNotif, status.BoostOfAccount); err != nil {
-		return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err)
-	}
-
-	return nil
-}
-
-func (p *Processor) notifyReport(ctx context.Context, report *gtsmodel.Report) error {
-	instance, err := p.state.DB.GetInstance(ctx, config.GetHost())
-	if err != nil {
-		return fmt.Errorf("notifyReport: error getting instance: %w", err)
-	}
-
-	toAddresses, err := p.state.DB.GetInstanceModeratorAddresses(ctx)
-	if err != nil {
-		if errors.Is(err, db.ErrNoEntries) {
-			// No registered moderator addresses.
-			return nil
-		}
-		return fmt.Errorf("notifyReport: error getting instance moderator addresses: %w", err)
-	}
-
-	if report.Account == nil {
-		report.Account, err = p.state.DB.GetAccountByID(ctx, report.AccountID)
-		if err != nil {
-			return fmt.Errorf("notifyReport: error getting report account: %w", err)
-		}
-	}
-
-	if report.TargetAccount == nil {
-		report.TargetAccount, err = p.state.DB.GetAccountByID(ctx, report.TargetAccountID)
-		if err != nil {
-			return fmt.Errorf("notifyReport: error getting report target account: %w", err)
-		}
-	}
-
-	reportData := email.NewReportData{
-		InstanceURL:        instance.URI,
-		InstanceName:       instance.Title,
-		ReportURL:          instance.URI + "/settings/admin/reports/" + report.ID,
-		ReportDomain:       report.Account.Domain,
-		ReportTargetDomain: report.TargetAccount.Domain,
-	}
-
-	if err := p.emailSender.SendNewReportEmail(toAddresses, reportData); err != nil {
-		return fmt.Errorf("notifyReport: error emailing instance moderators: %w", err)
-	}
-
-	return nil
-}
-
-func (p *Processor) notifyReportClosed(ctx context.Context, report *gtsmodel.Report) error {
-	user, err := p.state.DB.GetUserByAccountID(ctx, report.Account.ID)
-	if err != nil {
-		return fmt.Errorf("notifyReportClosed: db error getting user: %w", err)
-	}
-
-	if user.ConfirmedAt.IsZero() || !*user.Approved || *user.Disabled || user.Email == "" {
-		// Only email users who:
-		// - are confirmed
-		// - are approved
-		// - are not disabled
-		// - have an email address
-		return nil
-	}
-
-	instance, err := p.state.DB.GetInstance(ctx, config.GetHost())
-	if err != nil {
-		return fmt.Errorf("notifyReportClosed: db error getting instance: %w", err)
-	}
-
-	if report.Account == nil {
-		report.Account, err = p.state.DB.GetAccountByID(ctx, report.AccountID)
-		if err != nil {
-			return fmt.Errorf("notifyReportClosed: error getting report account: %w", err)
-		}
-	}
-
-	if report.TargetAccount == nil {
-		report.TargetAccount, err = p.state.DB.GetAccountByID(ctx, report.TargetAccountID)
-		if err != nil {
-			return fmt.Errorf("notifyReportClosed: error getting report target account: %w", err)
-		}
-	}
-
-	reportClosedData := email.ReportClosedData{
-		Username:             report.Account.Username,
-		InstanceURL:          instance.URI,
-		InstanceName:         instance.Title,
-		ReportTargetUsername: report.TargetAccount.Username,
-		ReportTargetDomain:   report.TargetAccount.Domain,
-		ActionTakenComment:   report.ActionTaken,
-	}
-
-	return p.emailSender.SendReportClosedEmail(user.Email, reportClosedData)
-}
-
-// timelineStatus processes the given new status and inserts it into
-// the HOME timelines of accounts that follow the status author.
-func (p *Processor) timelineStatus(ctx context.Context, status *gtsmodel.Status) error {
-	if status.Account == nil {
-		// ensure status fully populated (including account)
 	if err := p.state.DB.PopulateStatus(ctx, status); err != nil {
-			return fmt.Errorf("timelineStatus: error populating status with id %s: %w", status.ID, err)
+		return fmt.Errorf("timelineAndNotifyStatus: error populating status with id %s: %w", status.ID, err)
-		}
 	}
 
-	// get local followers of the account that posted the status
+	// Get local followers of the account that posted the status.
 	follows, err := p.state.DB.GetAccountLocalFollowers(ctx, status.AccountID)
 	if err != nil {
-		return fmt.Errorf("timelineStatus: error getting followers for account id %s: %w", status.AccountID, err)
+		return fmt.Errorf("timelineAndNotifyStatus: error getting local followers for account id %s: %w", status.AccountID, err)
 	}
 
 	// If the poster is also local, add a fake entry for them
@@ -423,66 +54,276 @@ func (p *Processor) timelineStatus(ctx context.Context, status *gtsmodel.Status)
 		follows = append(follows, &gtsmodel.Follow{
 			AccountID:   status.AccountID,
 			Account:     status.Account,
+			Notify:      func() *bool { b := false; return &b }(), // Account shouldn't notify itself.
+			ShowReblogs: func() *bool { b := true; return &b }(),  // Account should show own reblogs.
 		})
 	}
 
-	var errs gtserror.MultiError
+	// Timeline the status for each local follower of this account.
+	// This will also handle notifying any followers with notify
+	// set to true on their follow.
+	if err := p.timelineAndNotifyStatusForFollowers(ctx, status, follows); err != nil {
+		return fmt.Errorf("timelineAndNotifyStatus: error timelining status %s for followers: %w", status.ID, err)
+	}
+
+	// Notify each local account that's mentioned by this status.
+	if err := p.notifyStatusMentions(ctx, status); err != nil {
+		return fmt.Errorf("timelineAndNotifyStatus: error notifying status mentions for status %s: %w", status.ID, err)
+	}
+
+	return nil
+}
+
+func (p *Processor) timelineAndNotifyStatusForFollowers(ctx context.Context, status *gtsmodel.Status, follows []*gtsmodel.Follow) error {
+	var (
+		errs  = make(gtserror.MultiError, 0, len(follows))
+		boost = status.BoostOfID != ""
+		reply = status.InReplyToURI != ""
+	)
 
 	for _, follow := range follows {
-		// Timeline the status for each local following account.
+		if sr := follow.ShowReblogs; boost && (sr == nil || !*sr) {
-		if err := p.timelineStatusForAccount(ctx, follow.Account, status); err != nil {
+			// This is a boost, but this follower
+			// doesn't want to see those from this
+			// account, so just skip everything.
+			continue
+		}
+
+		// Add status to home timeline for this
+		// follower, and stream it if applicable.
+		if timelined, err := p.timelineStatusForAccount(ctx, follow.Account, status); err != nil {
+			errs.Append(fmt.Errorf("timelineAndNotifyStatusForFollowers: error timelining status: %w", err))
+			continue
+		} else if !timelined {
+			// Status wasn't added to home tomeline,
+			// so we shouldn't notify it either.
+			continue
+		}
+
+		if n := follow.Notify; n == nil || !*n {
+			// This follower doesn't have notifications
+			// set for this account's new posts, so bail.
+			continue
+		}
+
+		if boost || reply {
+			// Don't notify for boosts or replies.
+			continue
+		}
+
+		// If we reach here, we know:
+		//
+		//   - This follower wants to be notified when this account posts.
+		//   - This is a top-level post (not a reply).
+		//   - This is not a boost of another post.
+		//   - The post is visible in this follower's home timeline.
+		//
+		// That means we can officially notify this one.
+		if err := p.notify(
+			ctx,
+			gtsmodel.NotificationStatus,
+			follow.AccountID,
+			status.AccountID,
+			status.ID,
+		); err != nil {
+			errs.Append(fmt.Errorf("timelineAndNotifyStatusForFollowers: error notifying account %s about new status: %w", follow.AccountID, err))
+		}
+	}
+
+	return errs.Combine()
+}
+
+// timelineStatusForAccount puts the given status in the HOME timeline
+// of the account with given accountID, if it's HomeTimelineable.
+//
+// If the status was inserted into the home timeline of the given account,
+// true will be returned + it will also be streamed via websockets to the user.
+func (p *Processor) timelineStatusForAccount(ctx context.Context, account *gtsmodel.Account, status *gtsmodel.Status) (bool, error) {
+	// Make sure the status is timelineable.
+	if timelineable, err := p.filter.StatusHomeTimelineable(ctx, account, status); err != nil {
+		err = fmt.Errorf("timelineStatusForAccount: error getting timelineability for status for timeline with id %s: %w", account.ID, err)
+		return false, err
+	} else if !timelineable {
+		// Nothing to do.
+		return false, nil
+	}
+
+	// Insert status in the home timeline of account.
+	if inserted, err := p.statusTimelines.IngestOne(ctx, account.ID, status); err != nil {
+		err = fmt.Errorf("timelineStatusForAccount: error ingesting status %s: %w", status.ID, err)
+		return false, err
+	} else if !inserted {
+		// Nothing more to do.
+		return false, nil
+	}
+
+	// The status was inserted so stream it to the user.
+	apiStatus, err := p.tc.StatusToAPIStatus(ctx, status, account)
+	if err != nil {
+		err = fmt.Errorf("timelineStatusForAccount: error converting status %s to frontend representation: %w", status.ID, err)
+		return true, err
+	}
+
+	if err := p.stream.Update(apiStatus, account, stream.TimelineHome); err != nil {
+		err = fmt.Errorf("timelineStatusForAccount: error streaming update for status %s: %w", status.ID, err)
+		return true, err
+	}
+
+	return true, nil
+}
+
+func (p *Processor) notifyStatusMentions(ctx context.Context, status *gtsmodel.Status) error {
+	errs := make(gtserror.MultiError, 0, len(status.Mentions))
+
+	for _, m := range status.Mentions {
+		if err := p.notify(
+			ctx,
+			gtsmodel.NotificationMention,
+			m.TargetAccountID,
+			m.OriginAccountID,
+			m.StatusID,
+		); err != nil {
 			errs.Append(err)
 		}
 	}
 
-	if len(errs) != 0 {
+	return errs.Combine()
-		return fmt.Errorf("timelineStatus: one or more errors timelining statuses: %w", errs.Combine())
 }
 
+func (p *Processor) notifyFollowRequest(ctx context.Context, followRequest *gtsmodel.FollowRequest) error {
+	return p.notify(
+		ctx,
+		gtsmodel.NotificationFollowRequest,
+		followRequest.TargetAccountID,
+		followRequest.AccountID,
+		"",
+	)
+}
+
+func (p *Processor) notifyFollow(ctx context.Context, follow *gtsmodel.Follow, targetAccount *gtsmodel.Account) error {
+	// Remove previous follow request notification, if it exists.
+	prevNotif, err := p.state.DB.GetNotification(
+		gtscontext.SetBarebones(ctx),
+		gtsmodel.NotificationFollowRequest,
+		targetAccount.ID,
+		follow.AccountID,
+		"",
+	)
+	if err != nil && !errors.Is(err, db.ErrNoEntries) {
+		// Proper error while checking.
+		return fmt.Errorf("notifyFollow: db error checking for previous follow request notification: %w", err)
+	}
+
+	if prevNotif != nil {
+		// Previous notification existed, delete.
+		if err := p.state.DB.DeleteNotificationByID(ctx, prevNotif.ID); err != nil {
+			return fmt.Errorf("notifyFollow: db error removing previous follow request notification %s: %w", prevNotif.ID, err)
+		}
+	}
+
+	// Now notify the follow itself.
+	return p.notify(
+		ctx,
+		gtsmodel.NotificationFollow,
+		targetAccount.ID,
+		follow.AccountID,
+		"",
+	)
+}
+
+func (p *Processor) notifyFave(ctx context.Context, fave *gtsmodel.StatusFave) error {
+	if fave.TargetAccountID == fave.AccountID {
+		// Self-fave, nothing to do.
 		return nil
 	}
 
-// timelineStatusForAccount puts the given status in the HOME timeline
+	return p.notify(
-// of the account with given accountID, if it's hometimelineable.
+		ctx,
-//
+		gtsmodel.NotificationFave,
-// If the status was inserted into the home timeline of the given account,
+		fave.TargetAccountID,
-// it will also be streamed via websockets to the user.
+		fave.AccountID,
-func (p *Processor) timelineStatusForAccount(ctx context.Context, account *gtsmodel.Account, status *gtsmodel.Status) error {
+		fave.StatusID,
-	// make sure the status is timelineable
+	)
-	if timelineable, err := p.filter.StatusHomeTimelineable(ctx, account, status); err != nil {
+}
-		return fmt.Errorf("timelineStatusForAccount: error getting timelineability for status for timeline with id %s: %w", account.ID, err)
+
-	} else if !timelineable {
+func (p *Processor) notifyAnnounce(ctx context.Context, status *gtsmodel.Status) error {
+	if status.BoostOfID == "" {
+		// Not a boost, nothing to do.
 		return nil
 	}
 
-	// stick the status in the timeline for the account
+	if status.BoostOfAccountID == status.AccountID {
-	if inserted, err := p.statusTimelines.IngestOne(ctx, account.ID, status); err != nil {
+		// Self-boost, nothing to do.
-		return fmt.Errorf("timelineStatusForAccount: error ingesting status %s: %w", status.ID, err)
-	} else if !inserted {
 		return nil
 	}
 
-	// the status was inserted so stream it to the user
+	return p.notify(
-	apiStatus, err := p.tc.StatusToAPIStatus(ctx, status, account)
+		ctx,
+		gtsmodel.NotificationReblog,
+		status.BoostOfAccountID,
+		status.AccountID,
+		status.ID,
+	)
+}
+
+func (p *Processor) notify(
+	ctx context.Context,
+	notificationType gtsmodel.NotificationType,
+	targetAccountID string,
+	originAccountID string,
+	statusID string,
+) error {
+	targetAccount, err := p.state.DB.GetAccountByID(ctx, targetAccountID)
 	if err != nil {
-		return fmt.Errorf("timelineStatusForAccount: error converting status %s to frontend representation: %w", status.ID, err)
+		return fmt.Errorf("notify: error getting target account %s: %w", targetAccountID, err)
-	}
-
-	if err := p.stream.Update(apiStatus, account, stream.TimelineHome); err != nil {
-		return fmt.Errorf("timelineStatusForAccount: error streaming update for status %s: %w", status.ID, err)
 	}
 
+	if !targetAccount.IsLocal() {
+		// Nothing to do.
 		return nil
 	}
 
-// deleteStatusFromTimelines completely removes the given status from all timelines.
+	// Make sure a notification doesn't
-// It will also stream deletion of the status to all open streams.
+	// already exist with these params.
-func (p *Processor) deleteStatusFromTimelines(ctx context.Context, status *gtsmodel.Status) error {
+	if _, err := p.state.DB.GetNotification(
-	if err := p.statusTimelines.WipeItemFromAllTimelines(ctx, status.ID); err != nil {
+		ctx,
-		return err
+		notificationType,
+		targetAccountID,
+		originAccountID,
+		statusID,
+	); err == nil {
+		// Notification exists, nothing to do.
+		return nil
+	} else if !errors.Is(err, db.ErrNoEntries) {
+		// Real error.
+		return fmt.Errorf("notify: error checking existence of notification: %w", err)
 	}
 
-	return p.stream.Delete(status.ID)
+	// Notification doesn't yet exist, so
+	// we need to create + store one.
+	notif := &gtsmodel.Notification{
+		ID:               id.NewULID(),
+		NotificationType: notificationType,
+		TargetAccountID:  targetAccountID,
+		OriginAccountID:  originAccountID,
+		StatusID:         statusID,
+	}
+
+	if err := p.state.DB.PutNotification(ctx, notif); err != nil {
+		return fmt.Errorf("notify: error putting notification in database: %w", err)
+	}
+
+	// Stream notification to the user.
+	apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif)
+	if err != nil {
+		return fmt.Errorf("notify: error converting notification to api representation: %w", err)
+	}
+
+	if err := p.stream.Notify(apiNotif, targetAccount); err != nil {
+		return fmt.Errorf("notify: error streaming notification to account: %w", err)
+	}
+
+	return nil
 }
 
 // wipeStatus contains common logic used to totally delete a status
@@ -494,12 +335,14 @@ func (p *Processor) wipeStatus(ctx context.Context, statusToDelete *gtsmodel.Sta
 	// than delete is that the poster might want to reattach them
 	// to another status immediately (in case of delete + redraft)
 	if deleteAttachments {
+		// todo: p.state.DB.DeleteAttachmentsForStatus
 		for _, a := range statusToDelete.AttachmentIDs {
 			if err := p.media.Delete(ctx, a); err != nil {
 				return err
 			}
 		}
 	} else {
+		// todo: p.state.DB.UnattachAttachmentsForStatus
 		for _, a := range statusToDelete.AttachmentIDs {
 			if _, err := p.media.Unattach(ctx, statusToDelete.Account, a); err != nil {
 				return err
@@ -508,6 +351,7 @@ func (p *Processor) wipeStatus(ctx context.Context, statusToDelete *gtsmodel.Sta
 	}
 
 	// delete all mention entries generated by this status
+	// todo: p.state.DB.DeleteMentionsForStatus
 	for _, id := range statusToDelete.MentionIDs {
 		if err := p.state.DB.DeleteMentionByID(ctx, id); err != nil {
 			return err
@@ -553,3 +397,107 @@ func (p *Processor) wipeStatus(ctx context.Context, statusToDelete *gtsmodel.Sta
 
 	return nil
 }
+
+// deleteStatusFromTimelines completely removes the given status from all timelines.
+// It will also stream deletion of the status to all open streams.
+func (p *Processor) deleteStatusFromTimelines(ctx context.Context, status *gtsmodel.Status) error {
+	if err := p.statusTimelines.WipeItemFromAllTimelines(ctx, status.ID); err != nil {
+		return err
+	}
+
+	return p.stream.Delete(status.ID)
+}
+
+/*
+	EMAIL FUNCTIONS
+*/
+
+func (p *Processor) emailReport(ctx context.Context, report *gtsmodel.Report) error {
+	instance, err := p.state.DB.GetInstance(ctx, config.GetHost())
+	if err != nil {
+		return fmt.Errorf("emailReport: error getting instance: %w", err)
+	}
+
+	toAddresses, err := p.state.DB.GetInstanceModeratorAddresses(ctx)
+	if err != nil {
+		if errors.Is(err, db.ErrNoEntries) {
+			// No registered moderator addresses.
+			return nil
+		}
+		return fmt.Errorf("emailReport: error getting instance moderator addresses: %w", err)
+	}
+
+	if report.Account == nil {
+		report.Account, err = p.state.DB.GetAccountByID(ctx, report.AccountID)
+		if err != nil {
+			return fmt.Errorf("emailReport: error getting report account: %w", err)
+		}
+	}
+
+	if report.TargetAccount == nil {
+		report.TargetAccount, err = p.state.DB.GetAccountByID(ctx, report.TargetAccountID)
+		if err != nil {
+			return fmt.Errorf("emailReport: error getting report target account: %w", err)
+		}
+	}
+
+	reportData := email.NewReportData{
+		InstanceURL:        instance.URI,
+		InstanceName:       instance.Title,
+		ReportURL:          instance.URI + "/settings/admin/reports/" + report.ID,
+		ReportDomain:       report.Account.Domain,
+		ReportTargetDomain: report.TargetAccount.Domain,
+	}
+
+	if err := p.emailSender.SendNewReportEmail(toAddresses, reportData); err != nil {
+		return fmt.Errorf("emailReport: error emailing instance moderators: %w", err)
+	}
+
+	return nil
+}
+
+func (p *Processor) emailReportClosed(ctx context.Context, report *gtsmodel.Report) error {
+	user, err := p.state.DB.GetUserByAccountID(ctx, report.Account.ID)
+	if err != nil {
+		return fmt.Errorf("emailReportClosed: db error getting user: %w", err)
+	}
+
+	if user.ConfirmedAt.IsZero() || !*user.Approved || *user.Disabled || user.Email == "" {
+		// Only email users who:
+		// - are confirmed
+		// - are approved
+		// - are not disabled
+		// - have an email address
+		return nil
+	}
+
+	instance, err := p.state.DB.GetInstance(ctx, config.GetHost())
+	if err != nil {
+		return fmt.Errorf("emailReportClosed: db error getting instance: %w", err)
+	}
+
+	if report.Account == nil {
+		report.Account, err = p.state.DB.GetAccountByID(ctx, report.AccountID)
+		if err != nil {
+			return fmt.Errorf("emailReportClosed: error getting report account: %w", err)
+		}
+	}
+
+	if report.TargetAccount == nil {
+		report.TargetAccount, err = p.state.DB.GetAccountByID(ctx, report.TargetAccountID)
+		if err != nil {
+			return fmt.Errorf("emailReportClosed: error getting report target account: %w", err)
+		}
+	}
+
+	reportClosedData := email.ReportClosedData{
+		Username:             report.Account.Username,
+		InstanceURL:          instance.URI,
+		InstanceName:         instance.Title,
+		ReportTargetUsername: report.TargetAccount.Username,
+		ReportTargetDomain:   report.TargetAccount.Domain,
+		ActionTakenComment:   report.ActionTaken,
+	}
+
+	return p.emailSender.SendReportClosedEmail(user.Email, reportClosedData)
+}

internal/processing/fromfederator.go

@@ -164,11 +164,7 @@ func (p *Processor) processCreateStatusFromFederator(ctx context.Context, federa
 		status.Account = a
 	}
 
-	if err := p.timelineStatus(ctx, status); err != nil {
+	if err := p.timelineAndNotifyStatus(ctx, status); err != nil {
-		return err
-	}
-
-	if err := p.notifyStatus(ctx, status); err != nil {
 		return err
 	}
 
@@ -327,7 +323,7 @@ func (p *Processor) processCreateAnnounceFromFederator(ctx context.Context, fede
 		return fmt.Errorf("error adding dereferenced announce to the db: %s", err)
 	}
 
-	if err := p.timelineStatus(ctx, incomingAnnounce); err != nil {
+	if err := p.timelineAndNotifyStatus(ctx, incomingAnnounce); err != nil {
 		return err
 	}
 
@@ -367,7 +363,7 @@ func (p *Processor) processCreateFlagFromFederator(ctx context.Context, federato
 	// TODO: handle additional side effects of flag creation:
 	// - notify admins by dm / notification
 
-	return p.notifyReport(ctx, incomingReport)
+	return p.emailReport(ctx, incomingReport)
 }
 
 // processUpdateAccountFromFederator handles Activity Update and Object Profile

internal/processing/processor_test.go

@@ -53,6 +53,7 @@ type ProcessingStandardTestSuite struct {
 	testApplications map[string]*gtsmodel.Application
 	testUsers        map[string]*gtsmodel.User
 	testAccounts     map[string]*gtsmodel.Account
+	testFollows      map[string]*gtsmodel.Follow
 	testAttachments  map[string]*gtsmodel.MediaAttachment
 	testStatuses     map[string]*gtsmodel.Status
 	testTags         map[string]*gtsmodel.Tag
@@ -70,6 +71,7 @@ func (suite *ProcessingStandardTestSuite) SetupSuite() {
 	suite.testApplications = testrig.NewTestApplications()
 	suite.testUsers = testrig.NewTestUsers()
 	suite.testAccounts = testrig.NewTestAccounts()
+	suite.testFollows = testrig.NewTestFollows()
 	suite.testAttachments = testrig.NewTestAttachments()
 	suite.testStatuses = testrig.NewTestStatuses()
 	suite.testTags = testrig.NewTestTags()