[feature] add streaming of statuses and status updates to LOCAL / PUBLIC timelines (#4353)

This adds streaming of statuses and edits to LOCAL and PUBLIC timeline types. Note that in the future we should probably rearrange some of the surface code so we don't perform so many repeated mute and visibility checks on the same status in sequence.

closes https://codeberg.org/superseriousbusiness/gotosocial/issues/4342

Reviewed-on: https://codeberg.org/superseriousbusiness/gotosocial/pulls/4353
Co-authored-by: kim <grufwub@gmail.com>
Co-committed-by: kim <grufwub@gmail.com>

1 files changed, 110 insertions, 11 deletions

diff --git a/internal/processing/workers/fromclientapi_test.go b/internal/processing/workers/fromclientapi_test.go
index 7da34ff42..5967d4d34 100644
--- a/internal/processing/workers/fromclientapi_test.go
+++ b/internal/processing/workers/fromclientapi_test.go
@@ -262,9 +262,10 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithNotification() {
 			receivingAccount,
 			[]string{testList.ID},
 		)
-		homeStream  = streams[stream.TimelineHome]
-		listStream  = streams[stream.TimelineList+":"+testList.ID]
-		notifStream = streams[stream.TimelineNotifications]
+		publicStream = streams[stream.TimelinePublic]
+		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(
@@ -310,6 +311,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithNotification() {
 		receivingAccount,
 	)
 
+	// Check message in public stream.
+	suite.checkStreamed(
+		publicStream,
+		true,
+		statusJSON,
+		stream.EventTypeUpdate,
+	)
+
 	// Check message in home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -379,9 +388,10 @@ func (suite *FromClientAPITestSuite) TestProcessCreateBackfilledStatusWithNotifi
 			receivingAccount,
 			[]string{testList.ID},
 		)
-		homeStream  = streams[stream.TimelineHome]
-		listStream  = streams[stream.TimelineList+":"+testList.ID]
-		notifStream = streams[stream.TimelineNotifications]
+		publicStream = streams[stream.TimelinePublic]
+		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(
@@ -420,6 +430,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateBackfilledStatusWithNotifi
 		suite.FailNow(err.Error())
 	}
 
+	// There should be no message in public stream.
+	suite.checkStreamed(
+		publicStream,
+		false,
+		"",
+		"",
+	)
+
 	// There should be no message in the home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -530,6 +548,7 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReply() {
 		receivingAccount = suite.testAccounts["local_account_1"]
 		testList         = suite.testLists["local_account_1_list_1"]
 		streams          = suite.openStreams(ctx, testStructs.Processor, receivingAccount, []string{testList.ID})
+		publicStream     = streams[stream.TimelinePublic]
 		homeStream       = streams[stream.TimelineHome]
 		listStream       = streams[stream.TimelineList+":"+testList.ID]
 
@@ -571,6 +590,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReply() {
 		receivingAccount,
 	)
 
+	// Check message *not* in public stream.
+	suite.checkStreamed(
+		publicStream,
+		false,
+		"",
+		"",
+	)
+
 	// Check message in home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -732,6 +759,7 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 		postingAccount   = suite.testAccounts["admin_account"]
 		receivingAccount = suite.testAccounts["local_account_1"]
 		streams          = suite.openStreams(ctx, testStructs.Processor, receivingAccount, []string{testList.ID})
+		publicStream     = streams[stream.TimelinePublic]
 		homeStream       = streams[stream.TimelineHome]
 		listStream       = streams[stream.TimelineList+":"+testList.ID]
 
@@ -778,6 +806,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 		receivingAccount,
 	)
 
+	// Check message *not* in public stream.
+	suite.checkStreamed(
+		publicStream,
+		false,
+		"",
+		"",
+	)
+
 	// Check message in home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -811,6 +847,7 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 		postingAccount   = suite.testAccounts["admin_account"]
 		receivingAccount = suite.testAccounts["local_account_1"]
 		streams          = suite.openStreams(ctx, testStructs.Processor, receivingAccount, []string{testList.ID})
+		publicStream     = streams[stream.TimelinePublic]
 		homeStream       = streams[stream.TimelineHome]
 		listStream       = streams[stream.TimelineList+":"+testList.ID]
 
@@ -863,6 +900,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 		receivingAccount,
 	)
 
+	// Check message *not* in public stream.
+	suite.checkStreamed(
+		publicStream,
+		false,
+		"",
+		"",
+	)
+
 	// Check message in home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -896,6 +941,7 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyListRepliesPoli
 		postingAccount   = suite.testAccounts["admin_account"]
 		receivingAccount = suite.testAccounts["local_account_1"]
 		streams          = suite.openStreams(ctx, testStructs.Processor, receivingAccount, []string{testList.ID})
+		publicStream     = streams[stream.TimelinePublic]
 		homeStream       = streams[stream.TimelineHome]
 		listStream       = streams[stream.TimelineList+":"+testList.ID]
 
@@ -942,6 +988,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyListRepliesPoli
 		receivingAccount,
 	)
 
+	// Check message *not* in public stream.
+	suite.checkStreamed(
+		publicStream,
+		false,
+		"",
+		"",
+	)
+
 	// Check message in home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -972,6 +1026,7 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoost() {
 		receivingAccount = suite.testAccounts["local_account_1"]
 		testList         = suite.testLists["local_account_1_list_1"]
 		streams          = suite.openStreams(ctx, testStructs.Processor, receivingAccount, []string{testList.ID})
+		publicStream     = streams[stream.TimelinePublic]
 		homeStream       = streams[stream.TimelineHome]
 		listStream       = streams[stream.TimelineList+":"+testList.ID]
 
@@ -1009,6 +1064,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoost() {
 		receivingAccount,
 	)
 
+	// Check message *not* in public stream.
+	suite.checkStreamed(
+		publicStream,
+		false,
+		"",
+		"",
+	)
+
 	// Check message in home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -1039,6 +1102,7 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostNoReblogs() {
 		receivingAccount = suite.testAccounts["local_account_1"]
 		testList         = suite.testLists["local_account_1_list_1"]
 		streams          = suite.openStreams(ctx, testStructs.Processor, receivingAccount, []string{testList.ID})
+		publicStream     = streams[stream.TimelinePublic]
 		homeStream       = streams[stream.TimelineHome]
 		listStream       = streams[stream.TimelineList+":"+testList.ID]
 
@@ -1078,6 +1142,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostNoReblogs() {
 		suite.FailNow(err.Error())
 	}
 
+	// Check message *not* in public stream.
+	suite.checkStreamed(
+		publicStream,
+		false,
+		"",
+		"",
+	)
+
 	// Check message NOT in home stream.
 	suite.checkStreamed(
 		homeStream,
@@ -1763,8 +1835,9 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithAuthorOnExclusiv
 			receivingAccount,
 			[]string{testList.ID},
 		)
-		homeStream = streams[stream.TimelineHome]
-		listStream = streams[stream.TimelineList+":"+testList.ID]
+		publicStream = streams[stream.TimelinePublic]
+		homeStream   = streams[stream.TimelineHome]
+		listStream   = streams[stream.TimelineList+":"+testList.ID]
 
 		// postingAccount posts a new public status not mentioning anyone.
 		status = suite.newStatus(
@@ -1802,6 +1875,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithAuthorOnExclusiv
 		suite.FailNow(err.Error())
 	}
 
+	// Check status in public stream.
+	suite.checkStreamed(
+		publicStream,
+		true,
+		"",
+		stream.EventTypeUpdate,
+	)
+
 	// Check status in list stream.
 	suite.checkStreamed(
 		listStream,
@@ -1857,6 +1938,7 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithAuthorOnExclusiv
 				testExclusiveList.ID,
 			},
 		)
+		publicStream        = streams[stream.TimelinePublic]
 		homeStream          = streams[stream.TimelineHome]
 		inclusiveListStream = streams[stream.TimelineList+":"+testInclusiveList.ID]
 		exclusiveListStream = streams[stream.TimelineList+":"+testExclusiveList.ID]
@@ -1911,6 +1993,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithAuthorOnExclusiv
 		suite.FailNow(err.Error())
 	}
 
+	// Check status in public stream.
+	suite.checkStreamed(
+		publicStream,
+		true,
+		"",
+		stream.EventTypeUpdate,
+	)
+
 	// Check status in inclusive list stream.
 	suite.checkStreamed(
 		inclusiveListStream,
@@ -1957,9 +2047,10 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithAuthorOnExclusiv
 			receivingAccount,
 			[]string{testList.ID},
 		)
-		homeStream  = streams[stream.TimelineHome]
-		listStream  = streams[stream.TimelineList+":"+testList.ID]
-		notifStream = streams[stream.TimelineNotifications]
+		publicStream = streams[stream.TimelinePublic]
+		homeStream   = streams[stream.TimelineHome]
+		listStream   = streams[stream.TimelineList+":"+testList.ID]
+		notifStream  = streams[stream.TimelineNotifications]
 
 		// postingAccount posts a new public status not mentioning anyone.
 		status = suite.newStatus(
@@ -2005,6 +2096,14 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithAuthorOnExclusiv
 		suite.FailNow(err.Error())
 	}
 
+	// Check status in public stream.
+	suite.checkStreamed(
+		publicStream,
+		true,
+		"",
+		stream.EventTypeUpdate,
+	)
+
 	// Check status in list stream.
 	suite.checkStreamed(
 		listStream,