[feature] Implement filters_changed stream event (#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() {

internal/api/client/filters/v2/filterkeywordput_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) putFilterKeyword(
 }
 
 func (suite *FiltersTestSuite) TestPutFilterKeywordFull() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterKeywordID := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 	keyword := "fnords"
 	wholeWord := true
@@ -117,9 +120,13 @@ func (suite *FiltersTestSuite) TestPutFilterKeywordFull() {
 
 	suite.Equal(keyword, filterKeyword.Keyword)
 	suite.Equal(wholeWord, filterKeyword.WholeWord)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterKeywordFullJSON() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterKeywordID := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 	requestJson := `{
 		"keyword": "fnords",
@@ -132,9 +139,13 @@ func (suite *FiltersTestSuite) TestPutFilterKeywordFullJSON() {
 
 	suite.Equal("fnords", filterKeyword.Keyword)
 	suite.True(filterKeyword.WholeWord)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterKeywordMinimal() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterKeywordID := suite.testFilterKeywords["local_account_1_filter_1_keyword_1"].ID
 	keyword := "fnords"
 	filterKeyword, err := suite.putFilterKeyword(filterKeywordID, &keyword, nil, nil, http.StatusOK, "")
@@ -144,6 +155,8 @@ func (suite *FiltersTestSuite) TestPutFilterKeywordMinimal() {
 
 	suite.Equal(keyword, filterKeyword.Keyword)
 	suite.False(filterKeyword.WholeWord)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterKeywordEmptyKeyword() {

internal/api/client/filters/v2/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"
 )
 
@@ -104,6 +105,8 @@ func (suite *FiltersTestSuite) postFilter(title *string, context *[]string, acti
 }
 
 func (suite *FiltersTestSuite) TestPostFilterFull() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	title := "GNU/Linux"
 	context := []string{"home", "public"}
 	action := "warn"
@@ -125,9 +128,13 @@ func (suite *FiltersTestSuite) TestPostFilterFull() {
 	}
 	suite.Empty(filter.Keywords)
 	suite.Empty(filter.Statuses)
+
+	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 := `{
 		"title": "GNU/Linux",
@@ -155,9 +162,13 @@ func (suite *FiltersTestSuite) TestPostFilterFullJSON() {
 	}
 	suite.Empty(filter.Keywords)
 	suite.Empty(filter.Statuses)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterMinimal() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	title := "GNU/Linux"
 	context := []string{"home"}
 	filter, err := suite.postFilter(&title, &context, nil, nil, nil, http.StatusOK, "")
@@ -175,6 +186,8 @@ func (suite *FiltersTestSuite) TestPostFilterMinimal() {
 	suite.Nil(filter.ExpiresAt)
 	suite.Empty(filter.Keywords)
 	suite.Empty(filter.Statuses)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterEmptyTitle() {

internal/api/client/filters/v2/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"
 )
 
@@ -106,6 +107,8 @@ func (suite *FiltersTestSuite) putFilter(filterID string, title *string, context
 }
 
 func (suite *FiltersTestSuite) TestPutFilterFull() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilters["local_account_1_filter_2"].ID
 	title := "messy synoptic varblabbles"
 	context := []string{"home", "public"}
@@ -128,9 +131,13 @@ func (suite *FiltersTestSuite) TestPutFilterFull() {
 	}
 	suite.Len(filter.Keywords, 3)
 	suite.Len(filter.Statuses, 0)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterFullJSON() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilters["local_account_1_filter_2"].ID
 	// Use a numeric literal with a fractional part to test the JSON-specific handling for non-integer "expires_in".
 	requestJson := `{
@@ -158,9 +165,13 @@ func (suite *FiltersTestSuite) TestPutFilterFullJSON() {
 	}
 	suite.Len(filter.Keywords, 3)
 	suite.Len(filter.Statuses, 0)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterMinimal() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilters["local_account_1_filter_1"].ID
 	title := "GNU/Linux"
 	context := []string{"home"}
@@ -177,6 +188,8 @@ func (suite *FiltersTestSuite) TestPutFilterMinimal() {
 	suite.ElementsMatch(context, filterContext)
 	suite.Equal(apimodel.FilterActionWarn, filter.FilterAction)
 	suite.Nil(filter.ExpiresAt)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPutFilterEmptyTitle() {

internal/api/client/filters/v2/filterstatusdelete_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"
 )
 
@@ -50,7 +51,7 @@ func (suite *FiltersTestSuite) deleteFilterStatus(
 	ctx.AddParam("id", filterStatusID)
 
 	// trigger the handler
-	suite.filtersModule.FilterDELETEHandler(ctx)
+	suite.filtersModule.FilterStatusDELETEHandler(ctx)
 
 	// read the response
 	result := recorder.Result()
@@ -85,12 +86,16 @@ func (suite *FiltersTestSuite) deleteFilterStatus(
 }
 
 func (suite *FiltersTestSuite) TestDeleteFilterStatus() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	id := suite.testFilterStatuses["local_account_1_filter_3_status_1"].ID
 
 	err := suite.deleteFilterStatus(id, http.StatusOK, "")
 	if err != nil {
 		suite.FailNow(err.Error())
 	}
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestDeleteAnotherAccountsFilterStatus() {

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

@@ -30,6 +30,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"
 )
 
@@ -102,6 +103,8 @@ func (suite *FiltersTestSuite) postFilterStatus(
 }
 
 func (suite *FiltersTestSuite) TestPostFilterStatus() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterID := suite.testFilters["local_account_1_filter_1"].ID
 	statusID := suite.testStatuses["admin_account_status_1"].ID
 	filterStatus, err := suite.postFilterStatus(filterID, &statusID, nil, http.StatusOK, "")
@@ -110,9 +113,13 @@ func (suite *FiltersTestSuite) TestPostFilterStatus() {
 	}
 
 	suite.Equal(statusID, filterStatus.StatusID)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterStatusJSON() {
+	homeStream := suite.openHomeStream(suite.testAccounts["local_account_1"])
+
 	filterID := suite.testFilters["local_account_1_filter_1"].ID
 	requestJson := `{
 		"status_id": "01F8MH75CBF9JFX4ZAD54N0W0R"
@@ -123,6 +130,8 @@ func (suite *FiltersTestSuite) TestPostFilterStatusJSON() {
 	}
 
 	suite.Equal(suite.testStatuses["admin_account_status_1"].ID, filterStatus.StatusID)
+
+	suite.checkStreamed(homeStream, true, "", stream.EventTypeFiltersChanged)
 }
 
 func (suite *FiltersTestSuite) TestPostFilterStatusEmptyStatusID() {

internal/api/client/streaming/stream.go

@@ -125,7 +125,7 @@ import (
 //							`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.
 //						type: string
 //						enum:
 //						- update
@@ -142,6 +142,7 @@ import (
 //							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.
 //						type: string
 //						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\"}"
 //		'401':

internal/processing/filters/v1/create.go

@@ -83,5 +83,13 @@ func (p *Processor) Create(ctx context.Context, account *gtsmodel.Account, form
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
-	return p.apiFilter(ctx, filterKeyword)
+	apiFilter, errWithCode := p.apiFilter(ctx, filterKeyword)
+	if errWithCode != nil {
+		return nil, errWithCode
+	}
+
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
+	return apiFilter, nil
 }

internal/processing/filters/v1/delete.go

@@ -63,5 +63,8 @@ func (p *Processor) Delete(
 		}
 	}
 
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
 	return nil
 }

internal/processing/filters/v1/filters.go

@@ -18,6 +18,7 @@
 package v1
 
 import (
+	"github.com/superseriousbusiness/gotosocial/internal/processing/stream"
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 )
@@ -25,11 +26,13 @@ import (
 type Processor struct {
 	state     *state.State
 	converter *typeutils.Converter
+	stream    *stream.Processor
 }
 
-func New(state *state.State, converter *typeutils.Converter) Processor {
+func New(state *state.State, converter *typeutils.Converter, stream *stream.Processor) Processor {
 	return Processor{
 		state:     state,
 		converter: converter,
+		stream:    stream,
 	}
 }

internal/processing/filters/v1/update.go

@@ -163,5 +163,13 @@ func (p *Processor) Update(
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
-	return p.apiFilter(ctx, filterKeyword)
+	apiFilter, errWithCode := p.apiFilter(ctx, filterKeyword)
+	if errWithCode != nil {
+		return nil, errWithCode
+	}
+
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
+	return apiFilter, nil
 }

internal/processing/filters/v2/create.go

@@ -71,5 +71,13 @@ func (p *Processor) Create(ctx context.Context, account *gtsmodel.Account, form
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
-	return p.apiFilter(ctx, filter)
+	apiFilter, errWithCode := p.apiFilter(ctx, filter)
+	if errWithCode != nil {
+		return nil, errWithCode
+	}
+
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
+	return apiFilter, nil
 }

internal/processing/filters/v2/delete.go

@@ -49,5 +49,8 @@ func (p *Processor) Delete(
 		return gtserror.NewErrorInternalError(err)
 	}
 
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
 	return nil
 }

internal/processing/filters/v2/filters.go

@@ -18,6 +18,7 @@
 package v2
 
 import (
+	"github.com/superseriousbusiness/gotosocial/internal/processing/stream"
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 )
@@ -25,11 +26,13 @@ import (
 type Processor struct {
 	state     *state.State
 	converter *typeutils.Converter
+	stream    *stream.Processor
 }
 
-func New(state *state.State, converter *typeutils.Converter) Processor {
+func New(state *state.State, converter *typeutils.Converter, stream *stream.Processor) Processor {
 	return Processor{
 		state:     state,
 		converter: converter,
+		stream:    stream,
 	}
 }

internal/processing/filters/v2/keywordcreate.go

@@ -63,5 +63,8 @@ func (p *Processor) KeywordCreate(ctx context.Context, account *gtsmodel.Account
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
 	return p.converter.FilterKeywordToAPIFilterKeyword(ctx, filterKeyword), nil
 }

internal/processing/filters/v2/keyworddelete.go

@@ -49,5 +49,8 @@ func (p *Processor) KeywordDelete(
 		return gtserror.NewErrorInternalError(err)
 	}
 
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
 	return nil
 }

internal/processing/filters/v2/keywordupdate.go

@@ -62,5 +62,8 @@ func (p *Processor) KeywordUpdate(
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
 	return p.converter.FilterKeywordToAPIFilterKeyword(ctx, filterKeyword), nil
 }

internal/processing/filters/v2/statuscreate.go

@@ -62,5 +62,8 @@ func (p *Processor) StatusCreate(ctx context.Context, account *gtsmodel.Account,
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
 	return p.converter.FilterStatusToAPIFilterStatus(ctx, filterStatus), nil
 }

internal/processing/filters/v2/statusdelete.go

@@ -49,5 +49,8 @@ func (p *Processor) StatusDelete(
 		return gtserror.NewErrorInternalError(err)
 	}
 
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
 	return nil
 }

internal/processing/filters/v2/update.go

@@ -121,5 +121,13 @@ func (p *Processor) Update(
 	filter.Keywords = filterKeywords
 	filter.Statuses = filterStatuses
 
-	return p.apiFilter(ctx, filter)
+	apiFilter, errWithCode := p.apiFilter(ctx, filter)
+	if errWithCode != nil {
+		return nil, errWithCode
+	}
+
+	// Send a filters changed event.
+	p.stream.FiltersChanged(ctx, account)
+
+	return apiFilter, nil
 }

internal/processing/processor.go

@@ -189,8 +189,8 @@ func NewProcessor(
 	processor.account = account.New(&common, state, converter, mediaManager, federator, filter, parseMentionFunc)
 	processor.admin = admin.New(state, cleaner, converter, mediaManager, federator.TransportController(), emailSender)
 	processor.fedi = fedi.New(state, &common, converter, federator, filter)
-	processor.filtersv1 = filtersv1.New(state, converter)
-	processor.filtersv2 = filtersv2.New(state, converter)
+	processor.filtersv1 = filtersv1.New(state, converter, &processor.stream)
+	processor.filtersv2 = filtersv2.New(state, converter, &processor.stream)
 	processor.list = list.New(state, converter)
 	processor.markers = markers.New(state, converter)
 	processor.polls = polls.New(&common, state, converter)

internal/processing/stream/filterschanged.go

@@ -0,0 +1,36 @@
+// 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 stream
+
+import (
+	"context"
+
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
+)
+
+// FiltersChanged streams a filters changed event to any open, appropriate streams belonging to the given account.
+// Filter changes have no payload.
+func (p *Processor) FiltersChanged(ctx context.Context, account *gtsmodel.Account) {
+	p.streams.Post(ctx, account.ID, stream.Message{
+		Event: stream.EventTypeFiltersChanged,
+		Stream: []string{
+			stream.TimelineHome,
+		},
+	})
+}

internal/stream/stream.go

@@ -42,6 +42,10 @@ const (
 	// user's timeline has been edited (yes this
 	// is a confusing name, blame Mastodon ...).
 	EventTypeStatusUpdate = "status.update"
+
+	// EventTypeFiltersChanged -- the user's filters
+	// (including keywords and statuses) have changed.
+	EventTypeFiltersChanged = "filters_changed"
 )
 
 const (