Refactor/tidy (#261)

* tidy up streaming

* cut down code duplication

* test get followers/following

* test streaming processor

* fix some test models

* add TimeMustParse

* fix uri / url typo

* make trace logging less verbose

* make logging more consistent

* disable quote on logging

* remove context.Background

* remove many extraneous mastodon references

* regenerate swagger

* don't log query on no rows result

* log latency first for easier reading

13 files changed, 420 insertions, 145 deletions

diff --git a/internal/processing/streaming/authorize.go b/internal/processing/streaming/authorize.go
index f938a0c0c..1a5724f51 100644
--- a/internal/processing/streaming/authorize.go
+++ b/internal/processing/streaming/authorize.go
@@ -1,3 +1,21 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming
 
 import (
@@ -8,7 +26,7 @@ import (
 )
 
 func (p *processor) AuthorizeStreamingRequest(ctx context.Context, accessToken string) (*gtsmodel.Account, error) {
-	ti, err := p.oauthServer.LoadAccessToken(context.Background(), accessToken)
+	ti, err := p.oauthServer.LoadAccessToken(ctx, accessToken)
 	if err != nil {
 		return nil, fmt.Errorf("AuthorizeStreamingRequest: error loading access token: %s", err)
 	}
diff --git a/internal/processing/streaming/authorize_test.go b/internal/processing/streaming/authorize_test.go
new file mode 100644
index 000000000..f52396250
--- /dev/null
+++ b/internal/processing/streaming/authorize_test.go
@@ -0,0 +1,48 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type AuthorizeTestSuite struct {
+	StreamingTestSuite
+}
+
+func (suite *AuthorizeTestSuite) TestAuthorize() {
+	account1, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), suite.testTokens["local_account_1"].Access)
+	suite.NoError(err)
+	suite.Equal(suite.testAccounts["local_account_1"].ID, account1.ID)
+
+	account2, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), suite.testTokens["local_account_2"].Access)
+	suite.NoError(err)
+	suite.Equal(suite.testAccounts["local_account_2"].ID, account2.ID)
+
+	noAccount, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), "aaaaaaaaaaaaaaaaaaaaa!!")
+	suite.EqualError(err, "AuthorizeStreamingRequest: error loading access token: no entries")
+	suite.Nil(noAccount)
+}
+
+func TestAuthorizeTestSuite(t *testing.T) {
+	suite.Run(t, &AuthorizeTestSuite{})
+}
diff --git a/internal/processing/streaming/notification.go b/internal/processing/streaming/notification.go
new file mode 100644
index 000000000..870490be4
--- /dev/null
+++ b/internal/processing/streaming/notification.go
@@ -0,0 +1,37 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming
+
+import (
+	"encoding/json"
+	"fmt"
+
+	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
+)
+
+func (p *processor) StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error {
+	bytes, err := json.Marshal(n)
+	if err != nil {
+		return fmt.Errorf("error marshalling notification to json: %s", err)
+	}
+
+	return p.streamToAccount(string(bytes), stream.EventTypeNotification, account.ID)
+}
diff --git a/internal/processing/streaming/notification_test.go b/internal/processing/streaming/notification_test.go
new file mode 100644
index 000000000..fa77a8f92
--- /dev/null
+++ b/internal/processing/streaming/notification_test.go
@@ -0,0 +1,60 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type NotificationTestSuite struct {
+	StreamingTestSuite
+}
+
+func (suite *NotificationTestSuite) TestStreamNotification() {
+	account := suite.testAccounts["local_account_1"]
+
+	openStream, errWithCode := suite.streamingProcessor.OpenStreamForAccount(context.Background(), account, "user")
+	suite.NoError(errWithCode)
+
+	followAccount := suite.testAccounts["remote_account_1"]
+	followAccountAPIModel, err := testrig.NewTestTypeConverter(suite.db).AccountToAPIAccountPublic(context.Background(), followAccount)
+	suite.NoError(err)
+
+	notification := &apimodel.Notification{
+		ID:        "01FH57SJCMDWQGEAJ0X08CE3WV",
+		Type:      "follow",
+		CreatedAt: "2021-10-04T10:52:36+02:00",
+		Account:   followAccountAPIModel,
+	}
+
+	err = suite.streamingProcessor.StreamNotificationToAccount(notification, account)
+	suite.NoError(err)
+
+	msg := <-openStream.Messages
+	suite.Equal(`{"id":"01FH57SJCMDWQGEAJ0X08CE3WV","type":"follow","created_at":"2021-10-04T10:52:36+02:00","account":{"id":"01F8MH5ZK5VRH73AKHQM6Y9VNX","username":"foss_satan","acct":"foss_satan@fossbros-anonymous.io","display_name":"big gerald","locked":false,"bot":false,"created_at":"2021-09-26T12:52:36+02:00","note":"i post about like, i dunno, stuff, or whatever!!!!","url":"http://fossbros-anonymous.io/@foss_satan","avatar":"","avatar_static":"","header":"","header_static":"","followers_count":0,"following_count":0,"statuses_count":0,"last_status_at":"","emojis":[],"fields":[]}}`, msg.Payload)
+}
+
+func TestNotificationTestSuite(t *testing.T) {
+	suite.Run(t, &NotificationTestSuite{})
+}
diff --git a/internal/processing/streaming/openstream.go b/internal/processing/streaming/openstream.go
index d4e4eef9f..74b6486f5 100644
--- a/internal/processing/streaming/openstream.go
+++ b/internal/processing/streaming/openstream.go
@@ -1,3 +1,21 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming
 
 import (
diff --git a/internal/processing/streaming/openstream_test.go b/internal/processing/streaming/openstream_test.go
new file mode 100644
index 000000000..6c4134997
--- /dev/null
+++ b/internal/processing/streaming/openstream_test.go
@@ -0,0 +1,41 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type OpenStreamTestSuite struct {
+	StreamingTestSuite
+}
+
+func (suite *OpenStreamTestSuite) TestOpenStream() {
+	account := suite.testAccounts["local_account_1"]
+
+	_, errWithCode := suite.streamingProcessor.OpenStreamForAccount(context.Background(), account, "user")
+	suite.NoError(errWithCode)
+}
+
+func TestOpenStreamTestSuite(t *testing.T) {
+	suite.Run(t, &OpenStreamTestSuite{})
+}
diff --git a/internal/processing/streaming/streamdelete.go b/internal/processing/streaming/streamdelete.go
index cd541bc57..8332c37dc 100644
--- a/internal/processing/streaming/streamdelete.go
+++ b/internal/processing/streaming/streamdelete.go
@@ -1,3 +1,21 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming
 
 import (
@@ -10,39 +28,20 @@ import (
 func (p *processor) StreamDelete(statusID string) error {
 	errs := []string{}
 
-	// we want to range through ALL streams for ALL accounts here to make sure it's very clear to everyone that the status has been deleted
-	p.streamMap.Range(func(k interface{}, v interface{}) bool {
-		// the key of this map should be an accountID (string)
-		accountID, ok := k.(string)
-		if !ok {
-			errs = append(errs, "key in streamMap was not a string!")
-			return false
-		}
-
-		// the value of the map should be a buncha streams
-		streamsForAccount, ok := v.(*stream.StreamsForAccount)
-		if !ok {
-			errs = append(errs, fmt.Sprintf("stream map error for account stream %s", accountID))
-		}
-
-		// lock the streams while we work on them
-		streamsForAccount.Lock()
-		defer streamsForAccount.Unlock()
-		for _, s := range streamsForAccount.Streams {
-			// lock each individual stream as we work on it
-			s.Lock()
-			defer s.Unlock()
-			if s.Connected {
-				s.Messages <- &stream.Message{
-					Stream:  []string{s.Type},
-					Event:   "delete",
-					Payload: statusID,
-				}
-			}
-		}
+	// get all account IDs with open streams
+	accountIDs := []string{}
+	p.streamMap.Range(func(k interface{}, _ interface{}) bool {
+		accountIDs = append(accountIDs, k.(string))
 		return true
 	})
 
+	// stream the delete to every account
+	for _, accountID := range accountIDs {
+		if err := p.streamToAccount(statusID, stream.EventTypeDelete, accountID); err != nil {
+			errs = append(errs, err.Error())
+		}
+	}
+
 	if len(errs) != 0 {
 		return fmt.Errorf("one or more errors streaming status delete: %s", strings.Join(errs, ";"))
 	}
diff --git a/internal/processing/streaming/streaming.go b/internal/processing/streaming/streaming.go
index 610d4a9d2..abce30cd1 100644
--- a/internal/processing/streaming/streaming.go
+++ b/internal/processing/streaming/streaming.go
@@ -1,3 +1,21 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming
 
 import (
@@ -6,14 +24,11 @@ import (
 
 	"github.com/sirupsen/logrus"
 	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
-	"github.com/superseriousbusiness/gotosocial/internal/config"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
 	"github.com/superseriousbusiness/gotosocial/internal/stream"
-	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
-	"github.com/superseriousbusiness/gotosocial/internal/visibility"
 )
 
 // Processor wraps a bunch of functions for processing streaming.
@@ -22,8 +37,8 @@ type Processor interface {
 	AuthorizeStreamingRequest(ctx context.Context, accessToken string) (*gtsmodel.Account, error)
 	// OpenStreamForAccount returns a new Stream for the given account, which will contain a channel for passing messages back to the caller.
 	OpenStreamForAccount(ctx context.Context, account *gtsmodel.Account, streamType string) (*stream.Stream, gtserror.WithCode)
-	// StreamStatusToAccount streams the given status to any open, appropriate streams belonging to the given account.
-	StreamStatusToAccount(s *apimodel.Status, account *gtsmodel.Account) error
+	// StreamUpdateToAccount streams the given update to any open, appropriate streams belonging to the given account.
+	StreamUpdateToAccount(s *apimodel.Status, account *gtsmodel.Account) error
 	// StreamNotificationToAccount streams the given notification to any open, appropriate streams belonging to the given account.
 	StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error
 	// StreamDelete streams the delete of the given statusID to *ALL* open streams.
@@ -31,22 +46,16 @@ type Processor interface {
 }
 
 type processor struct {
-	tc          typeutils.TypeConverter
-	config      *config.Config
 	db          db.DB
-	filter      visibility.Filter
 	log         *logrus.Logger
 	oauthServer oauth.Server
 	streamMap   *sync.Map
 }
 
 // New returns a new status processor.
-func New(db db.DB, tc typeutils.TypeConverter, oauthServer oauth.Server, config *config.Config, log *logrus.Logger) Processor {
+func New(db db.DB, oauthServer oauth.Server, log *logrus.Logger) Processor {
 	return &processor{
-		tc:          tc,
-		config:      config,
 		db:          db,
-		filter:      visibility.NewFilter(db, log),
 		log:         log,
 		oauthServer: oauthServer,
 		streamMap:   &sync.Map{},
diff --git a/internal/processing/streaming/streaming_test.go b/internal/processing/streaming/streaming_test.go
new file mode 100644
index 000000000..acc090b06
--- /dev/null
+++ b/internal/processing/streaming/streaming_test.go
@@ -0,0 +1,55 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming_test
+
+import (
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/suite"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/processing/streaming"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type StreamingTestSuite struct {
+	suite.Suite
+	testAccounts map[string]*gtsmodel.Account
+	testTokens   map[string]*gtsmodel.Token
+	db           db.DB
+	oauthServer  oauth.Server
+	log          *logrus.Logger
+
+	streamingProcessor streaming.Processor
+}
+
+func (suite *StreamingTestSuite) SetupTest() {
+	suite.testAccounts = testrig.NewTestAccounts()
+	suite.testTokens = testrig.NewTestTokens()
+	suite.db = testrig.NewTestDB()
+	suite.oauthServer = testrig.NewTestOauthServer(suite.db)
+	suite.log = testrig.NewTestLog()
+	suite.streamingProcessor = streaming.New(suite.db, suite.oauthServer, suite.log)
+
+	testrig.StandardDBSetup(suite.db, suite.testAccounts)
+}
+
+func (suite *StreamingTestSuite) TearDownTest() {
+	testrig.StandardDBTeardown(suite.db)
+}
diff --git a/internal/processing/streaming/streamnotification.go b/internal/processing/streaming/streamnotification.go
deleted file mode 100644
index d8460874f..000000000
--- a/internal/processing/streaming/streamnotification.go
+++ /dev/null
@@ -1,51 +0,0 @@
-package streaming
-
-import (
-	"encoding/json"
-	"errors"
-	"fmt"
-
-	"github.com/sirupsen/logrus"
-	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
-	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
-	"github.com/superseriousbusiness/gotosocial/internal/stream"
-)
-
-func (p *processor) StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error {
-	l := p.log.WithFields(logrus.Fields{
-		"func":    "StreamNotificationToAccount",
-		"account": account.ID,
-	})
-	v, ok := p.streamMap.Load(account.ID)
-	if !ok {
-		// no open connections so nothing to stream
-		return nil
-	}
-
-	streamsForAccount, ok := v.(*stream.StreamsForAccount)
-	if !ok {
-		return errors.New("stream map error")
-	}
-
-	notificationBytes, err := json.Marshal(n)
-	if err != nil {
-		return fmt.Errorf("error marshalling notification to json: %s", err)
-	}
-
-	streamsForAccount.Lock()
-	defer streamsForAccount.Unlock()
-	for _, s := range streamsForAccount.Streams {
-		s.Lock()
-		defer s.Unlock()
-		if s.Connected {
-			l.Debugf("streaming notification to stream id %s", s.ID)
-			s.Messages <- &stream.Message{
-				Stream:  []string{s.Type},
-				Event:   "notification",
-				Payload: string(notificationBytes),
-			}
-		}
-	}
-
-	return nil
-}
diff --git a/internal/processing/streaming/streamstatus.go b/internal/processing/streaming/streamstatus.go
deleted file mode 100644
index f4d6b2629..000000000
--- a/internal/processing/streaming/streamstatus.go
+++ /dev/null
@@ -1,51 +0,0 @@
-package streaming
-
-import (
-	"encoding/json"
-	"errors"
-	"fmt"
-
-	"github.com/sirupsen/logrus"
-	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
-	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
-	"github.com/superseriousbusiness/gotosocial/internal/stream"
-)
-
-func (p *processor) StreamStatusToAccount(s *apimodel.Status, account *gtsmodel.Account) error {
-	l := p.log.WithFields(logrus.Fields{
-		"func":    "StreamStatusForAccount",
-		"account": account.ID,
-	})
-	v, ok := p.streamMap.Load(account.ID)
-	if !ok {
-		// no open connections so nothing to stream
-		return nil
-	}
-
-	streamsForAccount, ok := v.(*stream.StreamsForAccount)
-	if !ok {
-		return errors.New("stream map error")
-	}
-
-	statusBytes, err := json.Marshal(s)
-	if err != nil {
-		return fmt.Errorf("error marshalling status to json: %s", err)
-	}
-
-	streamsForAccount.Lock()
-	defer streamsForAccount.Unlock()
-	for _, s := range streamsForAccount.Streams {
-		s.Lock()
-		defer s.Unlock()
-		if s.Connected {
-			l.Debugf("streaming status to stream id %s", s.ID)
-			s.Messages <- &stream.Message{
-				Stream:  []string{s.Type},
-				Event:   "update",
-				Payload: string(statusBytes),
-			}
-		}
-	}
-
-	return nil
-}
diff --git a/internal/processing/streaming/streamtoaccount.go b/internal/processing/streaming/streamtoaccount.go
new file mode 100644
index 000000000..140910ab7
--- /dev/null
+++ b/internal/processing/streaming/streamtoaccount.go
@@ -0,0 +1,55 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming
+
+import (
+	"errors"
+
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
+)
+
+// streamToAccount streams the given payload with the given event type to any streams currently open for the given account ID.
+func (p *processor) streamToAccount(payload string, event stream.EventType, accountID string) error {
+	v, ok := p.streamMap.Load(accountID)
+	if !ok {
+		// no open connections so nothing to stream
+		return nil
+	}
+
+	streamsForAccount, ok := v.(*stream.StreamsForAccount)
+	if !ok {
+		return errors.New("stream map error")
+	}
+
+	streamsForAccount.Lock()
+	defer streamsForAccount.Unlock()
+	for _, s := range streamsForAccount.Streams {
+		s.Lock()
+		defer s.Unlock()
+		if s.Connected {
+			s.Messages <- &stream.Message{
+				Stream:  []string{s.Type},
+				Event:   string(event),
+				Payload: payload,
+			}
+		}
+	}
+
+	return nil
+}
diff --git a/internal/processing/streaming/update.go b/internal/processing/streaming/update.go
new file mode 100644
index 000000000..da7dcb6ce
--- /dev/null
+++ b/internal/processing/streaming/update.go
@@ -0,0 +1,37 @@
+/*
+   GoToSocial
+   Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org
+
+   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 streaming
+
+import (
+	"encoding/json"
+	"fmt"
+
+	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/stream"
+)
+
+func (p *processor) StreamUpdateToAccount(s *apimodel.Status, account *gtsmodel.Account) error {
+	bytes, err := json.Marshal(s)
+	if err != nil {
+		return fmt.Errorf("error marshalling status to json: %s", err)
+	}
+
+	return p.streamToAccount(string(bytes), stream.EventTypeUpdate, account.ID)
+}