[feature] List replies policy, refactor async workers (#2087)

* Add/update some DB functions.

* move async workers into subprocessor

* rename FromFederator -> FromFediAPI

* update home timeline check to include check for current status first before moving to parent status

* change streamMap to pointer to mollify linter

* update followtoas func signature

* fix merge

* remove errant debug log

* don't use separate errs.Combine() check to wrap errs

* wrap parts of workers functionality in sub-structs

* populate report using new db funcs

* embed federator (tiny bit tidier)

* flesh out error msg, add continue(!)

* fix other error messages to be more specific

* better, nicer

* give parseURI util function a bit more util

* missing headers

* use pointers for subprocessors

1 files changed, 589 insertions, 0 deletions

diff --git a/internal/processing/workers/fromclientapi_test.go b/internal/processing/workers/fromclientapi_test.go
new file mode 100644
index 000000000..6690a43db
--- /dev/null
+++ b/internal/processing/workers/fromclientapi_test.go
@@ -0,0 +1,589 @@
+// 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 workers_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/suite"
+	"github.com/superseriousbusiness/gotosocial/internal/ap"
+	"github.com/superseriousbusiness/gotosocial/internal/config"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/id"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
+	"github.com/superseriousbusiness/gotosocial/internal/util"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type FromClientAPITestSuite struct {
+	WorkersTestSuite
+}
+
+func (suite *FromClientAPITestSuite) newStatus(
+	ctx context.Context,
+	account *gtsmodel.Account,
+	visibility gtsmodel.Visibility,
+	replyToStatus *gtsmodel.Status,
+	boostOfStatus *gtsmodel.Status,
+) *gtsmodel.Status {
+	var (
+		protocol = config.GetProtocol()
+		host     = config.GetHost()
+		statusID = id.NewULID()
+	)
+
+	// Make a new status from given account.
+	newStatus := &gtsmodel.Status{
+		ID:                  statusID,
+		URI:                 protocol + "://" + host + "/users/" + account.Username + "/statuses/" + statusID,
+		URL:                 protocol + "://" + host + "/@" + account.Username + "/statuses/" + statusID,
+		Content:             "pee pee poo poo",
+		Local:               util.Ptr(true),
+		AccountURI:          account.URI,
+		AccountID:           account.ID,
+		Visibility:          visibility,
+		ActivityStreamsType: ap.ObjectNote,
+		Federated:           util.Ptr(true),
+		Boostable:           util.Ptr(true),
+		Replyable:           util.Ptr(true),
+		Likeable:            util.Ptr(true),
+	}
+
+	if replyToStatus != nil {
+		// Status is a reply.
+		newStatus.InReplyToAccountID = replyToStatus.AccountID
+		newStatus.InReplyToID = replyToStatus.ID
+		newStatus.InReplyToURI = replyToStatus.URI
+
+		// Mention the replied-to account.
+		mention := &gtsmodel.Mention{
+			ID:               id.NewULID(),
+			StatusID:         statusID,
+			OriginAccountID:  account.ID,
+			OriginAccountURI: account.URI,
+			TargetAccountID:  replyToStatus.AccountID,
+		}
+
+		if err := suite.db.PutMention(ctx, mention); err != nil {
+			suite.FailNow(err.Error())
+		}
+		newStatus.Mentions = []*gtsmodel.Mention{mention}
+		newStatus.MentionIDs = []string{mention.ID}
+	}
+
+	if boostOfStatus != nil {
+		// Status is a boost.
+
+	}
+
+	// Put the status in the db, to mimic what would
+	// have already happened earlier up the flow.
+	if err := suite.db.PutStatus(ctx, newStatus); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	return newStatus
+}
+
+func (suite *FromClientAPITestSuite) checkStreamed(
+	str *stream.Stream,
+	expectMessage bool,
+	expectPayload string,
+	expectEventType string,
+) {
+	var msg *stream.Message
+streamLoop:
+	for {
+		select {
+		case msg = <-str.Messages:
+			break streamLoop // Got it.
+		case <-time.After(5 * time.Second):
+			break streamLoop // Didn't get it.
+		}
+	}
+
+	if expectMessage && msg == nil {
+		suite.FailNow("expected a message but message was nil")
+	}
+
+	if !expectMessage && msg != nil {
+		suite.FailNow("expected no message but message was not nil")
+	}
+
+	if expectPayload != "" && msg.Payload != expectPayload {
+		suite.FailNow("", "expected payload %s but payload was: %s", expectPayload, msg.Payload)
+	}
+
+	if expectEventType != "" && msg.Event != expectEventType {
+		suite.FailNow("", "expected event type %s but event type was: %s", expectEventType, msg.Event)
+	}
+}
+
+func (suite *FromClientAPITestSuite) statusJSON(
+	ctx context.Context,
+	status *gtsmodel.Status,
+	requestingAccount *gtsmodel.Account,
+) string {
+	apiStatus, err := suite.typeconverter.StatusToAPIStatus(
+		ctx,
+		status,
+		requestingAccount,
+	)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	statusJSON, err := json.Marshal(apiStatus)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	return string(statusJSON)
+}
+
+func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithNotification() {
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["admin_account"]
+		receivingAccount = suite.testAccounts["local_account_1"]
+		testList         = suite.testLists["local_account_1_list_1"]
+		streams          = suite.openStreams(ctx, receivingAccount, []string{testList.ID})
+		homeStream       = streams[stream.TimelineHome]
+		listStream       = streams[stream.TimelineList+":"+testList.ID]
+		notifStream      = streams[stream.TimelineNotifications]
+
+		// Admin account posts a new top-level status.
+		status = suite.newStatus(
+			ctx,
+			postingAccount,
+			gtsmodel.VisibilityPublic,
+			nil,
+			nil,
+		)
+		statusJSON = suite.statusJSON(
+			ctx,
+			status,
+			receivingAccount,
+		)
+	)
+
+	// Update the follow from receiving account -> posting account so
+	// that receiving account wants notifs when posting account posts.
+	follow := new(gtsmodel.Follow)
+	*follow = *suite.testFollows["local_account_1_admin_account"]
+
+	follow.Notify = util.Ptr(true)
+	if err := suite.db.UpdateFollow(ctx, follow); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Process the new status.
+	if err := suite.processor.Workers().ProcessFromClientAPI(
+		ctx,
+		messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			GTSModel:       status,
+			OriginAccount:  postingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Check message in home stream.
+	suite.checkStreamed(
+		homeStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+
+	// Check message in list stream.
+	suite.checkStreamed(
+		listStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+
+	// Wait for a notification to appear for the status.
+	var notif *gtsmodel.Notification
+	if !testrig.WaitFor(func() bool {
+		var err error
+		notif, err = suite.db.GetNotification(
+			ctx,
+			gtsmodel.NotificationStatus,
+			receivingAccount.ID,
+			postingAccount.ID,
+			status.ID,
+		)
+		return err == nil
+	}) {
+		suite.FailNow("timed out waiting for new status notification")
+	}
+
+	apiNotif, err := suite.typeconverter.NotificationToAPINotification(ctx, notif)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	notifJSON, err := json.Marshal(apiNotif)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Check message in notification stream.
+	suite.checkStreamed(
+		notifStream,
+		true,
+		string(notifJSON),
+		stream.EventTypeNotification,
+	)
+}
+
+func (suite *FromClientAPITestSuite) TestProcessCreateStatusReply() {
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["admin_account"]
+		receivingAccount = suite.testAccounts["local_account_1"]
+		testList         = suite.testLists["local_account_1_list_1"]
+		streams          = suite.openStreams(ctx, receivingAccount, []string{testList.ID})
+		homeStream       = streams[stream.TimelineHome]
+		listStream       = streams[stream.TimelineList+":"+testList.ID]
+
+		// Admin account posts a reply to turtle.
+		// Since turtle is followed by zork, and
+		// the default replies policy for this list
+		// is to show replies to followed accounts,
+		// post should also show in the list stream.
+		status = suite.newStatus(
+			ctx,
+			postingAccount,
+			gtsmodel.VisibilityPublic,
+			suite.testStatuses["local_account_2_status_1"],
+			nil,
+		)
+		statusJSON = suite.statusJSON(
+			ctx,
+			status,
+			receivingAccount,
+		)
+	)
+
+	// Process the new status.
+	if err := suite.processor.Workers().ProcessFromClientAPI(
+		ctx,
+		messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			GTSModel:       status,
+			OriginAccount:  postingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Check message in home stream.
+	suite.checkStreamed(
+		homeStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+
+	// Check message in list stream.
+	suite.checkStreamed(
+		listStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+}
+
+func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyListOnlyOK() {
+	// We're modifying the test list so take a copy.
+	testList := new(gtsmodel.List)
+	*testList = *suite.testLists["local_account_1_list_1"]
+
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["admin_account"]
+		receivingAccount = suite.testAccounts["local_account_1"]
+		streams          = suite.openStreams(ctx, receivingAccount, []string{testList.ID})
+		homeStream       = streams[stream.TimelineHome]
+		listStream       = streams[stream.TimelineList+":"+testList.ID]
+
+		// Admin account posts a reply to turtle.
+		status = suite.newStatus(
+			ctx,
+			postingAccount,
+			gtsmodel.VisibilityPublic,
+			suite.testStatuses["local_account_2_status_1"],
+			nil,
+		)
+		statusJSON = suite.statusJSON(
+			ctx,
+			status,
+			receivingAccount,
+		)
+	)
+
+	// Modify replies policy of test list to show replies
+	// only to other accounts in the same list. Since turtle
+	// and admin are in the same list, this means the reply
+	// should be shown in the list.
+	testList.RepliesPolicy = gtsmodel.RepliesPolicyList
+	if err := suite.db.UpdateList(ctx, testList, "replies_policy"); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Process the new status.
+	if err := suite.processor.Workers().ProcessFromClientAPI(
+		ctx,
+		messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			GTSModel:       status,
+			OriginAccount:  postingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Check message in home stream.
+	suite.checkStreamed(
+		homeStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+
+	// Check message in list stream.
+	suite.checkStreamed(
+		listStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+}
+
+func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyListOnlyNo() {
+	// We're modifying the test list so take a copy.
+	testList := new(gtsmodel.List)
+	*testList = *suite.testLists["local_account_1_list_1"]
+
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["admin_account"]
+		receivingAccount = suite.testAccounts["local_account_1"]
+		streams          = suite.openStreams(ctx, receivingAccount, []string{testList.ID})
+		homeStream       = streams[stream.TimelineHome]
+		listStream       = streams[stream.TimelineList+":"+testList.ID]
+
+		// Admin account posts a reply to turtle.
+		status = suite.newStatus(
+			ctx,
+			postingAccount,
+			gtsmodel.VisibilityPublic,
+			suite.testStatuses["local_account_2_status_1"],
+			nil,
+		)
+		statusJSON = suite.statusJSON(
+			ctx,
+			status,
+			receivingAccount,
+		)
+	)
+
+	// Modify replies policy of test list to show replies
+	// only to other accounts in the same list. We're
+	// about to remove turtle from the same list as admin,
+	// so the new post should not be streamed to the list.
+	testList.RepliesPolicy = gtsmodel.RepliesPolicyList
+	if err := suite.db.UpdateList(ctx, testList, "replies_policy"); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Remove turtle from the list.
+	if err := suite.db.DeleteListEntry(ctx, suite.testListEntries["local_account_1_list_1_entry_1"].ID); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Process the new status.
+	if err := suite.processor.Workers().ProcessFromClientAPI(
+		ctx,
+		messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			GTSModel:       status,
+			OriginAccount:  postingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Check message in home stream.
+	suite.checkStreamed(
+		homeStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+
+	// Check message NOT in list stream.
+	suite.checkStreamed(
+		listStream,
+		false,
+		"",
+		"",
+	)
+}
+
+func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyListRepliesPolicyNone() {
+	// We're modifying the test list so take a copy.
+	testList := new(gtsmodel.List)
+	*testList = *suite.testLists["local_account_1_list_1"]
+
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["admin_account"]
+		receivingAccount = suite.testAccounts["local_account_1"]
+		streams          = suite.openStreams(ctx, receivingAccount, []string{testList.ID})
+		homeStream       = streams[stream.TimelineHome]
+		listStream       = streams[stream.TimelineList+":"+testList.ID]
+
+		// Admin account posts a reply to turtle.
+		status = suite.newStatus(
+			ctx,
+			postingAccount,
+			gtsmodel.VisibilityPublic,
+			suite.testStatuses["local_account_2_status_1"],
+			nil,
+		)
+		statusJSON = suite.statusJSON(
+			ctx,
+			status,
+			receivingAccount,
+		)
+	)
+
+	// Modify replies policy of test list.
+	// Since we're modifying the list to not
+	// show any replies, the post should not
+	// be streamed to the list.
+	testList.RepliesPolicy = gtsmodel.RepliesPolicyNone
+	if err := suite.db.UpdateList(ctx, testList, "replies_policy"); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Process the new status.
+	if err := suite.processor.Workers().ProcessFromClientAPI(
+		ctx,
+		messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			GTSModel:       status,
+			OriginAccount:  postingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Check message in home stream.
+	suite.checkStreamed(
+		homeStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+
+	// Check message NOT in list stream.
+	suite.checkStreamed(
+		listStream,
+		false,
+		"",
+		"",
+	)
+}
+
+func (suite *FromClientAPITestSuite) TestProcessStatusDelete() {
+	var (
+		ctx                  = context.Background()
+		deletingAccount      = suite.testAccounts["local_account_1"]
+		receivingAccount     = suite.testAccounts["local_account_2"]
+		deletedStatus        = suite.testStatuses["local_account_1_status_1"]
+		boostOfDeletedStatus = suite.testStatuses["admin_account_status_4"]
+		streams              = suite.openStreams(ctx, receivingAccount, nil)
+		homeStream           = streams[stream.TimelineHome]
+	)
+
+	// Delete the status from the db first, to mimic what
+	// would have already happened earlier up the flow
+	if err := suite.db.DeleteStatusByID(ctx, deletedStatus.ID); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Process the status delete.
+	if err := suite.processor.Workers().ProcessFromClientAPI(
+		ctx,
+		messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityDelete,
+			GTSModel:       deletedStatus,
+			OriginAccount:  deletingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Stream should have the delete
+	// of admin's boost in it now.
+	suite.checkStreamed(
+		homeStream,
+		true,
+		boostOfDeletedStatus.ID,
+		stream.EventTypeDelete,
+	)
+
+	// Stream should also have the delete
+	// of the message itself in it.
+	suite.checkStreamed(
+		homeStream,
+		true,
+		deletedStatus.ID,
+		stream.EventTypeDelete,
+	)
+
+	// Boost should no longer be in the database.
+	if !testrig.WaitFor(func() bool {
+		_, err := suite.db.GetStatusByID(ctx, boostOfDeletedStatus.ID)
+		return errors.Is(err, db.ErrNoEntries)
+	}) {
+		suite.FailNow("timed out waiting for status delete")
+	}
+}
+
+func TestFromClientAPITestSuite(t *testing.T) {
+	suite.Run(t, &FromClientAPITestSuite{})
+}