[feature] Add List functionality (#1802)

* start working on lists

* further list work

* test list db functions nicely

* more work on lists

* peepoopeepoo

* poke

* start list timeline func

* we're getting there lads

* couldn't be me working on stuff... could it?

* hook up handlers

* fiddling

* weeee

* woah

* screaming, pissing

* fix streaming being a whiny baby

* lint, small test fix, swagger

* tidying up, testing

* fucked! by the linter

* move timelines to state like a boss

* add timeline start to tests using state

* invalidate lists

internal/processing/stream/open.go

@@ -31,60 +31,65 @@ import (
 )
 
 // Open returns a new Stream for the given account, which will contain a channel for passing messages back to the caller.
-func (p *Processor) Open(ctx context.Context, account *gtsmodel.Account, streamTimeline string) (*stream.Stream, gtserror.WithCode) {
+func (p *Processor) Open(ctx context.Context, account *gtsmodel.Account, streamType string) (*stream.Stream, gtserror.WithCode) {
 	l := log.WithContext(ctx).WithFields(kv.Fields{
 		{"account", account.ID},
-		{"streamType", streamTimeline},
+		{"streamType", streamType},
 	}...)
 	l.Debug("received open stream request")
 
-	// each stream needs a unique ID so we know to close it
-	streamID, err := id.NewRandomULID()
+	var (
+		streamID string
+		err      error
+	)
+
+	// Each stream needs a unique ID so we know to close it.
+	streamID, err = id.NewRandomULID()
 	if err != nil {
-		return nil, gtserror.NewErrorInternalError(fmt.Errorf("error generating stream id: %s", err))
+		return nil, gtserror.NewErrorInternalError(fmt.Errorf("error generating stream id: %w", err))
 	}
 
-	// Each stream can be subscibed to multiple timelines.
+	// Each stream can be subscibed to multiple types.
 	// Record them in a set, and include the initial one
-	// if it was given to us
-	timelines := map[string]bool{}
-	if streamTimeline != "" {
-		timelines[streamTimeline] = true
+	// if it was given to us.
+	streamTypes := map[string]any{}
+	if streamType != "" {
+		streamTypes[streamType] = true
 	}
 
-	thisStream := &stream.Stream{
-		ID:        streamID,
-		Timelines: timelines,
-		Messages:  make(chan *stream.Message, 100),
-		Hangup:    make(chan interface{}, 1),
-		Connected: true,
+	newStream := &stream.Stream{
+		ID:          streamID,
+		StreamTypes: streamTypes,
+		Messages:    make(chan *stream.Message, 100),
+		Hangup:      make(chan interface{}, 1),
+		Connected:   true,
 	}
-	go p.waitToCloseStream(account, thisStream)
+	go p.waitToCloseStream(account, newStream)
 
 	v, ok := p.streamMap.Load(account.ID)
-	if !ok || v == nil {
-		// there is no entry in the streamMap for this account yet, so make one and store it
-		streamsForAccount := &stream.StreamsForAccount{
-			Streams: []*stream.Stream{
-				thisStream,
-			},
-		}
-		p.streamMap.Store(account.ID, streamsForAccount)
-	} else {
-		// there is an entry in the streamMap for this account
-		// parse the interface as a streamsForAccount
+	if ok {
+		// There is an entry in the streamMap
+		// for this account. Parse it out.
 		streamsForAccount, ok := v.(*stream.StreamsForAccount)
 		if !ok {
 			return nil, gtserror.NewErrorInternalError(errors.New("stream map error"))
 		}
 
-		// append this stream to it
+		// Append new stream to existing entry.
 		streamsForAccount.Lock()
-		streamsForAccount.Streams = append(streamsForAccount.Streams, thisStream)
+		streamsForAccount.Streams = append(streamsForAccount.Streams, newStream)
 		streamsForAccount.Unlock()
+	} else {
+		// There is no entry in the streamMap for
+		// this account yet. Create one and store it.
+		p.streamMap.Store(account.ID, &stream.StreamsForAccount{
+			Streams: []*stream.Stream{
+				newStream,
+			},
+		})
 	}
 
-	return thisStream, nil
+	return newStream, nil
 }
 
 // waitToCloseStream waits until the hangup channel is closed for the given stream.

internal/processing/stream/stream.go

@@ -18,7 +18,6 @@
 package stream
 
 import (
-	"errors"
 	"sync"
 
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
@@ -40,37 +39,38 @@ func New(state *state.State, oauthServer oauth.Server) Processor {
 }
 
 // toAccount streams the given payload with the given event type to any streams currently open for the given account ID.
-func (p *Processor) toAccount(payload string, event string, timelines []string, accountID string) error {
+func (p *Processor) toAccount(payload string, event string, streamTypes []string, accountID string) error {
+	// Load all streams open for this account.
 	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")
+		return nil // No entry = nothing to stream.
 	}
+	streamsForAccount := v.(*stream.StreamsForAccount) //nolint:forcetypeassert
 
 	streamsForAccount.Lock()
 	defer streamsForAccount.Unlock()
+
 	for _, s := range streamsForAccount.Streams {
 		s.Lock()
 		defer s.Unlock()
+
 		if !s.Connected {
 			continue
 		}
 
-		for _, t := range timelines {
-			if _, found := s.Timelines[t]; found {
+	typeLoop:
+		for _, streamType := range streamTypes {
+			if _, found := s.StreamTypes[streamType]; found {
 				s.Messages <- &stream.Message{
-					Stream:  []string{string(t)},
+					Stream:  []string{streamType},
 					Event:   string(event),
 					Payload: payload,
 				}
-				// break out to the outer loop, to avoid sending duplicates
-				// of the same event to the same stream
-				break
+
+				// Break out to the outer loop,
+				// to avoid sending duplicates of
+				// the same event to the same stream.
+				break typeLoop
 			}
 		}
 	}

internal/processing/stream/update.go

@@ -27,11 +27,11 @@ import (
 )
 
 // Update streams the given update to any open, appropriate streams belonging to the given account.
-func (p *Processor) Update(s *apimodel.Status, account *gtsmodel.Account, timeline string) error {
+func (p *Processor) Update(s *apimodel.Status, account *gtsmodel.Account, streamTypes []string) error {
 	bytes, err := json.Marshal(s)
 	if err != nil {
 		return fmt.Errorf("error marshalling status to json: %s", err)
 	}
 
-	return p.toAccount(string(bytes), stream.EventTypeUpdate, []string{timeline}, account.ID)
+	return p.toAccount(string(bytes), stream.EventTypeUpdate, streamTypes, account.ID)
 }