2 files changed, 56 insertions, 0 deletions
diff --git a/internal/processing/synchronous/streaming/streamdelete.go b/internal/processing/synchronous/streaming/streamdelete.go
new file mode 100644
index 000000000..2282c29ae
--- /dev/null
+++ b/internal/processing/synchronous/streaming/streamdelete.go
@@ -0,0 +1,51 @@
+package streaming
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+)
+
+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.(*gtsmodel.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 _, stream := range streamsForAccount.Streams {
+			// lock each individual stream as we work on it
+			stream.Lock()
+			defer stream.Unlock()
+			if stream.Connected {
+				stream.Messages <- &gtsmodel.Message{
+					Stream:  []string{stream.Type},
+					Event:   "delete",
+					Payload: statusID,
+				}
+			}
+		}
+		return true
+	})
+
+	if len(errs) != 0 {
+		return fmt.Errorf("one or more errors streaming status delete: %s", strings.Join(errs, ";"))
+	}
+
+	return nil
+}
diff --git a/internal/processing/synchronous/streaming/streaming.go b/internal/processing/synchronous/streaming/streaming.go
index 9fd628547..de75b8f27 100644
--- a/internal/processing/synchronous/streaming/streaming.go
+++ b/internal/processing/synchronous/streaming/streaming.go
@@ -18,9 +18,14 @@ import (
 type Processor interface {
 	// AuthorizeStreamingRequest returns an oauth2 token info in response to an access token query from the streaming API
 	AuthorizeStreamingRequest(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(account *gtsmodel.Account, streamType string) (*gtsmodel.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
+	// 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.
+	StreamDelete(statusID string) error
 }
 
 type processor struct {