[feature] Implement filters_changed stream event (superseriousbusines…

…s#2972)

docs/api/swagger.yaml

@@ -8651,7 +8651,7 @@ paths:
                                     `update`: a new status has been received.
                                     `notification`: a new notification has been received.
                                     `delete`: a status has been deleted.
-                                    `filters_changed`: not implemented.
+                                    `filters_changed`: filters (including keywords and statuses) have changed.
                                 enum:
                                     - update
                                     - notification
@@ -8668,6 +8668,7 @@ paths:
                                     If `event` = `update`, then the payload will be a JSON string of a status.
                                     If `event` = `notification`, then the payload will be a JSON string of a notification.
                                     If `event` = `delete`, then the payload will be a status ID.
+                                    If `event` = `filters_changed`, then there is no payload.
                                 example: '{"id":"01FC3TZ5CFG6H65GCKCJRKA669","created_at":"2021-08-02T16:25:52Z","sensitive":false,"spoiler_text":"","visibility":"public","language":"en","uri":"https://gts.superseriousbusiness.org/users/dumpsterqueer/statuses/01FC3TZ5CFG6H65GCKCJRKA669","url":"https://gts.superseriousbusiness.org/@dumpsterqueer/statuses/01FC3TZ5CFG6H65GCKCJRKA669","replies_count":0,"reblogs_count":0,"favourites_count":0,"favourited":false,"reblogged":false,"muted":false,"bookmarked":fals…//gts.superseriousbusiness.org/fileserver/01JNN207W98SGG3CBJ76R5MVDN/header/original/019036W043D8FXPJKSKCX7G965.png","header_static":"https://gts.superseriousbusiness.org/fileserver/01JNN207W98SGG3CBJ76R5MVDN/header/small/019036W043D8FXPJKSKCX7G965.png","followers_count":33,"following_count":28,"statuses_count":126,"last_status_at":"2021-08-02T16:25:52Z","emojis":[],"fields":[]},"media_attachments":[],"mentions":[],"tags":[],"emojis":[],"card":null,"poll":null,"text":"a"}'
                                 type: string
                             stream:

internal/api/client/filters/v1/filter_test.go

@@ -18,6 +18,10 @@
 package v1_test
 
 import (
+	"context"
+	"testing"
+	"time"
+
 	"github.com/stretchr/testify/suite"
 	filtersV1 "github.com/superseriousbusiness/gotosocial/internal/api/client/filters/v1"
 	"github.com/superseriousbusiness/gotosocial/internal/config"
@@ -30,9 +34,9 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/processing"
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/storage"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/testrig"
-	"testing"
 )
 
 type FiltersTestSuite struct {
@@ -112,6 +116,44 @@ func (suite *FiltersTestSuite) TearDownTest() {
 	testrig.StopWorkers(&suite.state)
 }
 
+func (suite *FiltersTestSuite) openHomeStream(account *gtsmodel.Account) *stream.Stream {
+	stream, err := suite.processor.Stream().Open(context.Background(), account, stream.TimelineHome)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+	return stream
+}
+
+func (suite *FiltersTestSuite) checkStreamed(
+	str *stream.Stream,
+	expectMessage bool,
+	expectPayload string,
+	expectEventType string,
+) {
+	// Set a 5s timeout on context.
+	ctx := context.Background()
+	ctx, cncl := context.WithTimeout(ctx, time.Second*5)
+	defer cncl()
+
+	msg, ok := str.Recv(ctx)
+
+	if expectMessage && !ok {
+		suite.FailNow("expected a message but message was not received")
+	}
+
+	if !expectMessage && ok {
+		suite.FailNow("expected no message but message was received")
+	}
+
+	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 TestFiltersTestSuite(t *testing.T) {
 	suite.Run(t, new(FiltersTestSuite))
 }

internal/api/client/filters/v1/filterdelete_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -88,12 +89,16 @@ func (suite *FiltersTestSuite) deleteFilter(
 }
 
 func (suite *FiltersTestSuite) TestDeleteFilter() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 
 	err := suite.deleteFilter(id, http.StatusOK, "")
 	if err != nil {
 		suite.FailNow(err.Error())
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestDeleteAnotherAccountsFilter() {

internal/api/client/filters/v1/filterpost_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -116,6 +117,8 @@ func (suite *FiltersTestSuite) postFilter(
 }
 
 func (suite *FiltersTestSuite) TestPostFilterFull() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	phrase := "GNU/Linux"
 	context := []string{"home", "public"}
 	irreversible := false
@@ -137,9 +140,13 @@ func (suite *FiltersTestSuite) TestPostFilterFull() {
 	if suite.NotNil(filter.ExpiresAt) {
 		suite.NotEmpty(*filter.ExpiresAt)
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterFullJSON() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	// Use a numeric literal with a fractional part to test the JSON-specific handling for non-integer "expires_in".
 	requestJson := `{
 		"phrase":"GNU/Linux",
@@ -166,9 +173,13 @@ func (suite *FiltersTestSuite) TestPostFilterFullJSON() {
 	if suite.NotNil(filter.ExpiresAt) {
 		suite.NotEmpty(*filter.ExpiresAt)
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterMinimal() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	phrase := "GNU/Linux"
 	context := []string{"home"}
 	filter, err := suite.postFilter(&phrase, &context, nil, nil, nil, nil, http.StatusOK, "")
@@ -185,6 +196,8 @@ func (suite *FiltersTestSuite) TestPostFilterMinimal() {
 	suite.False(filter.Irreversible)
 	suite.False(filter.WholeWord)
 	suite.Nil(filter.ExpiresAt)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterEmptyPhrase() {

internal/api/client/filters/v1/filterput_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -119,6 +120,8 @@ func (suite *FiltersTestSuite) putFilter(
 }
 
 func (suite *FiltersTestSuite) TestPutFilterFull() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 	phrase := "GNU/Linux"
 	context := []string{"home", "public"}
@@ -141,9 +144,13 @@ func (suite *FiltersTestSuite) TestPutFilterFull() {
 	if suite.NotNil(filter.ExpiresAt) {
 		suite.NotEmpty(*filter.ExpiresAt)
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterFullJSON() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 	// Use a numeric literal with a fractional part to test the JSON-specific handling for non-integer "expires_in".
 	requestJson := `{
@@ -171,9 +178,13 @@ func (suite *FiltersTestSuite) TestPutFilterFullJSON() {
 	if suite.NotNil(filter.ExpiresAt) {
 		suite.NotEmpty(*filter.ExpiresAt)
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterMinimal() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 	phrase := "GNU/Linux"
 	context := []string{"home"}
@@ -191,6 +202,8 @@ func (suite *FiltersTestSuite) TestPutFilterMinimal() {
 	suite.False(filter.Irreversible)
 	suite.False(filter.WholeWord)
 	suite.Nil(filter.ExpiresAt)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterEmptyPhrase() {

internal/api/client/filters/v2/filter_test.go

@@ -18,7 +18,9 @@
 package v2_test
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 	filtersV2 "github.com/superseriousbusiness/gotosocial/internal/api/client/filters/v2"
@@ -32,6 +34,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/processing"
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/storage"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
@@ -113,6 +116,44 @@ func (suite *FiltersTestSuite) TearDownTest() {
 	testrig.StopWorkers(&suite.state)
 }
 
+func (suite *FiltersTestSuite) openHomeStream(account *gtsmodel.Account) *stream.Stream {
+	stream, err := suite.processor.Stream().Open(context.Background(), account, stream.TimelineHome)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+	return stream
+}
+
+func (suite *FiltersTestSuite) checkStreamed(
+	str *stream.Stream,
+	expectMessage bool,
+	expectPayload string,
+	expectEventType string,
+) {
+	// Set a 5s timeout on context.
+	ctx := context.Background()
+	ctx, cncl := context.WithTimeout(ctx, time.Second*5)
+	defer cncl()
+
+	msg, ok := str.Recv(ctx)
+
+	if expectMessage && !ok {
+		suite.FailNow("expected a message but message was not received")
+	}
+
+	if !expectMessage && ok {
+		suite.FailNow("expected no message but message was received")
+	}
+
+	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 TestFiltersTestSuite(t *testing.T) {
 	suite.Run(t, new(FiltersTestSuite))
 }

internal/api/client/filters/v2/filterdelete_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -88,12 +89,16 @@ func (suite *FiltersTestSuite) deleteFilter(
 }
 
 func (suite *FiltersTestSuite) TestDeleteFilter() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilters["local_account_1_filter_1"].ID
 
 	err := suite.deleteFilter(id, http.StatusOK, "")
 	if err != nil {
 		suite.FailNow(err.Error())
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestDeleteAnotherAccountsFilter() {

internal/api/client/filters/v2/filterkeyworddelete_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -88,12 +89,16 @@ func (suite *FiltersTestSuite) deleteFilterKeyword(
 }
 
 func (suite *FiltersTestSuite) TestDeleteFilterKeyword() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 
 	err := suite.deleteFilterKeyword(id, http.StatusOK, "")
 	if err != nil {
 		suite.FailNow(err.Error())
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestDeleteAnotherAccountsFilterKeyword() {

internal/api/client/filters/v2/filterkeywordpost_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -107,6 +108,8 @@ func (suite *FiltersTestSuite) postFilterKeyword(
 }
 
 func (suite *FiltersTestSuite) TestPostFilterKeywordFull() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterID := suite.testFilters["local_account_1_filter_1"].ID
 	keyword := "fnords"
 	wholeWord := true
@@ -117,9 +120,13 @@ func (suite *FiltersTestSuite) TestPostFilterKeywordFull() {
 
 	suite.Equal(keyword, filterKeyword.Keyword)
 	suite.Equal(wholeWord, filterKeyword.WholeWord)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterKeywordFullJSON() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterID := suite.testFilters["local_account_1_filter_1"].ID
 	requestJson := `{
 		"keyword": "fnords",
@@ -132,9 +139,13 @@ func (suite *FiltersTestSuite) TestPostFilterKeywordFullJSON() {
 
 	suite.Equal("fnords", filterKeyword.Keyword)
 	suite.True(filterKeyword.WholeWord)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterKeywordMinimal() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterID := suite.testFilters["local_account_1_filter_1"].ID
 	keyword := "fnords"
 	filterKeyword, err := suite.postFilterKeyword(filterID, &keyword, nil, nil, http.StatusOK, "")
@@ -144,6 +155,8 @@ func (suite *FiltersTestSuite) TestPostFilterKeywordMinimal() {
 
 	suite.Equal(keyword, filterKeyword.Keyword)
 	suite.False(filterKeyword.WholeWord)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterKeywordEmptyKeyword() {