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

1 files changed, 29 insertions, 30 deletions

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, ";"))
 	}