[feature] Implement backfilling statuses thru scheduled_at (#3685)

* Implement backfilling statuses thru scheduled_at

* Forbid mentioning others in backfills

* Update error messages & codes

* Add new tests for backfilled statuses

* Test that backfilling doesn't timeline or notify

* Fix check for absence of notification

* Test that backfills do not cause federation

* Fix type of apimodel.StatusCreateRequest.ScheduledAt in tests

* Add config file switch and min date check

2 files changed, 173 insertions, 8 deletions

diff --git a/internal/processing/workers/fromclientapi.go b/internal/processing/workers/fromclientapi.go
index c5dfc157d..a208d97b0 100644
--- a/internal/processing/workers/fromclientapi.go
+++ b/internal/processing/workers/fromclientapi.go
@@ -260,9 +260,16 @@ func (p *clientAPI) CreateUser(ctx context.Context, cMsg *messages.FromClientAPI
 }
 
 func (p *clientAPI) CreateStatus(ctx context.Context, cMsg *messages.FromClientAPI) error {
-	status, ok := cMsg.GTSModel.(*gtsmodel.Status)
-	if !ok {
-		return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel)
+	var status *gtsmodel.Status
+	backfill := false
+	if backfillStatus, ok := cMsg.GTSModel.(*gtsmodel.BackfillStatus); ok {
+		status = backfillStatus.Status
+		backfill = true
+	} else {
+		status, ok = cMsg.GTSModel.(*gtsmodel.Status)
+		if !ok {
+			return gtserror.Newf("%T not parseable as *gtsmodel.Status or *gtsmodel.BackfillStatus", cMsg.GTSModel)
+		}
 	}
 
 	// If pending approval is true then status must
@@ -344,12 +351,14 @@ func (p *clientAPI) CreateStatus(ctx context.Context, cMsg *messages.FromClientA
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
-	if err := p.surface.timelineAndNotifyStatus(ctx, status); err != nil {
-		log.Errorf(ctx, "error timelining and notifying status: %v", err)
-	}
+	if !backfill {
+		if err := p.surface.timelineAndNotifyStatus(ctx, status); err != nil {
+			log.Errorf(ctx, "error timelining and notifying status: %v", err)
+		}
 
-	if err := p.federate.CreateStatus(ctx, status); err != nil {
-		log.Errorf(ctx, "error federating status: %v", err)
+		if err := p.federate.CreateStatus(ctx, status); err != nil {
+			log.Errorf(ctx, "error federating status: %v", err)
+		}
 	}
 
 	if status.InReplyToID != "" {
diff --git a/internal/processing/workers/fromclientapi_test.go b/internal/processing/workers/fromclientapi_test.go
index acb25673d..1d70eb96c 100644
--- a/internal/processing/workers/fromclientapi_test.go
+++ b/internal/processing/workers/fromclientapi_test.go
@@ -368,6 +368,162 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithNotification() {
 	suite.checkWebPushed(testStructs.WebPushSender, receivingAccount.ID, gtsmodel.NotificationStatus)
 }
 
+// Even with notifications on for a user, backfilling a status should not notify or timeline it.
+func (suite *FromClientAPITestSuite) TestProcessCreateBackfilledStatusWithNotification() {
+	testStructs := testrig.SetupTestStructs(rMediaPath, rTemplatePath)
+	defer testrig.TearDownTestStructs(testStructs)
+
+	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,
+			testStructs.Processor,
+			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,
+			testStructs.State,
+			postingAccount,
+			gtsmodel.VisibilityPublic,
+			nil,
+			nil,
+			nil,
+			false,
+			nil,
+		)
+	)
+
+	// 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 := testStructs.State.DB.UpdateFollow(ctx, follow); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Process the new status as a backfill.
+	if err := testStructs.Processor.Workers().ProcessFromClientAPI(
+		ctx,
+		&messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			GTSModel:       &gtsmodel.BackfillStatus{Status: status},
+			Origin:         postingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// There should be no message in the home stream.
+	suite.checkStreamed(
+		homeStream,
+		false,
+		"",
+		"",
+	)
+
+	// There should be no message in the list stream.
+	suite.checkStreamed(
+		listStream,
+		false,
+		"",
+		"",
+	)
+
+	// No notification should appear for the status.
+	if testrig.WaitFor(func() bool {
+		var err error
+		_, err = testStructs.State.DB.GetNotification(
+			ctx,
+			gtsmodel.NotificationStatus,
+			receivingAccount.ID,
+			postingAccount.ID,
+			status.ID,
+		)
+		return err == nil
+	}) {
+		suite.FailNow("a status notification was created, but should not have been")
+	}
+
+	// There should be no message in the notification stream.
+	suite.checkStreamed(
+		notifStream,
+		false,
+		"",
+		"",
+	)
+
+	// There should be no Web Push status notification.
+	suite.checkNotWebPushed(testStructs.WebPushSender, receivingAccount.ID)
+}
+
+// Backfilled statuses should not federate when created.
+func (suite *FromClientAPITestSuite) TestProcessCreateBackfilledStatusWithRemoteFollower() {
+	testStructs := testrig.SetupTestStructs(rMediaPath, rTemplatePath)
+	defer testrig.TearDownTestStructs(testStructs)
+
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["local_account_1"]
+		receivingAccount = suite.testAccounts["remote_account_1"]
+
+		// Local account posts a new top-level status.
+		status = suite.newStatus(
+			ctx,
+			testStructs.State,
+			postingAccount,
+			gtsmodel.VisibilityPublic,
+			nil,
+			nil,
+			nil,
+			false,
+			nil,
+		)
+	)
+
+	// Follow the local account from the remote account.
+	follow := &gtsmodel.Follow{
+		ID:              "01JJHW9RW28SC1NEPZ0WBJQ4ZK",
+		CreatedAt:       testrig.TimeMustParse("2022-05-14T13:21:09+02:00"),
+		UpdatedAt:       testrig.TimeMustParse("2022-05-14T13:21:09+02:00"),
+		AccountID:       receivingAccount.ID,
+		TargetAccountID: postingAccount.ID,
+		ShowReblogs:     util.Ptr(true),
+		URI:             "http://fossbros-anonymous.io/users/foss_satan/follow/01JJHWEVC7F8W2JDW1136K431K",
+		Notify:          util.Ptr(false),
+	}
+
+	if err := testStructs.State.DB.PutFollow(ctx, follow); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// Process the new status as a backfill.
+	if err := testStructs.Processor.Workers().ProcessFromClientAPI(
+		ctx,
+		&messages.FromClientAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			GTSModel:       &gtsmodel.BackfillStatus{Status: status},
+			Origin:         postingAccount,
+		},
+	); err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	// No deliveries should be queued.
+	suite.Zero(testStructs.State.Workers.Delivery.Queue.Len())
+}
+
 func (suite *FromClientAPITestSuite) TestProcessCreateStatusReply() {
 	testStructs := testrig.SetupTestStructs(rMediaPath, rTemplatePath)
 	defer testrig.TearDownTestStructs(testStructs)