[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

1 files changed, 104 insertions, 69 deletions

diff --git a/internal/processing/fromclientapi_test.go b/internal/processing/fromclientapi_test.go
index 0b641c091..808f02cd6 100644
--- a/internal/processing/fromclientapi_test.go
+++ b/internal/processing/fromclientapi_test.go
@@ -20,6 +20,7 @@ package processing_test
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/suite"
@@ -36,24 +37,21 @@ type FromClientAPITestSuite struct {
 	ProcessingStandardTestSuite
 }
 
+// This test ensures that when admin_account posts a new
+// status, it ends up in the correct streaming timelines
+// of local_account_1, which follows it.
 func (suite *FromClientAPITestSuite) TestProcessStreamNewStatus() {
-	ctx := context.Background()
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["admin_account"]
+		receivingAccount = suite.testAccounts["local_account_1"]
+		testList         = suite.testLists["local_account_1_list_1"]
+		streams          = suite.openStreams(ctx, receivingAccount, []string{testList.ID})
+		homeStream       = streams[stream.TimelineHome]
+		listStream       = streams[stream.TimelineList+":"+testList.ID]
+	)
 
-	// let's say that the admin account posts a new status: it should end up in the
-	// timeline of any account that follows it and has a stream open
-	postingAccount := suite.testAccounts["admin_account"]
-	receivingAccount := suite.testAccounts["local_account_1"]
-
-	// open a home timeline stream for zork
-	wssStream, errWithCode := suite.processor.Stream().Open(ctx, receivingAccount, stream.TimelineHome)
-	suite.NoError(errWithCode)
-
-	// open another stream for zork, but for a different timeline;
-	// this shouldn't get stuff streamed into it, since it's for the public timeline
-	irrelevantStream, errWithCode := suite.processor.Stream().Open(ctx, receivingAccount, stream.TimelinePublic)
-	suite.NoError(errWithCode)
-
-	// make a new status from admin account
+	// Make a new status from admin account.
 	newStatus := &gtsmodel.Status{
 		ID:                       "01FN4B2F88TF9676DYNXWE1WSS",
 		URI:                      "http://localhost:8080/users/admin/statuses/01FN4B2F88TF9676DYNXWE1WSS",
@@ -82,87 +80,110 @@ func (suite *FromClientAPITestSuite) TestProcessStreamNewStatus() {
 		ActivityStreamsType:      ap.ObjectNote,
 	}
 
-	// put the status in the db first, to mimic what would have already happened earlier up the flow
-	err := suite.db.PutStatus(ctx, newStatus)
-	suite.NoError(err)
+	// Put the status in the db first, to mimic what
+	// would have already happened earlier up the flow.
+	if err := suite.db.PutStatus(ctx, newStatus); err != nil {
+		suite.FailNow(err.Error())
+	}
 
-	// process the new status
-	err = suite.processor.ProcessFromClientAPI(ctx, messages.FromClientAPI{
+	// Process the new status.
+	if err := suite.processor.ProcessFromClientAPI(ctx, messages.FromClientAPI{
 		APObjectType:   ap.ObjectNote,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       newStatus,
 		OriginAccount:  postingAccount,
-	})
-	suite.NoError(err)
+	}); err != nil {
+		suite.FailNow(err.Error())
+	}
 
-	// zork's stream should have the newly created status in it now
-	msg := <-wssStream.Messages
-	suite.Equal(stream.EventTypeUpdate, msg.Event)
-	suite.NotEmpty(msg.Payload)
-	suite.EqualValues([]string{stream.TimelineHome}, msg.Stream)
-	statusStreamed := &apimodel.Status{}
-	err = json.Unmarshal([]byte(msg.Payload), statusStreamed)
-	suite.NoError(err)
-	suite.Equal("01FN4B2F88TF9676DYNXWE1WSS", statusStreamed.ID)
-	suite.Equal("this status should stream :)", statusStreamed.Content)
+	// Check message in home stream.
+	homeMsg := <-homeStream.Messages
+	suite.Equal(stream.EventTypeUpdate, homeMsg.Event)
+	suite.EqualValues([]string{stream.TimelineHome}, homeMsg.Stream)
+	suite.Empty(homeStream.Messages) // Stream should now be empty.
 
-	// and stream should now be empty
-	suite.Empty(wssStream.Messages)
+	// Check status from home stream.
+	homeStreamStatus := &apimodel.Status{}
+	if err := json.Unmarshal([]byte(homeMsg.Payload), homeStreamStatus); err != nil {
+		suite.FailNow(err.Error())
+	}
+	suite.Equal(newStatus.ID, homeStreamStatus.ID)
+	suite.Equal(newStatus.Content, homeStreamStatus.Content)
+
+	// Check message in list stream.
+	listMsg := <-listStream.Messages
+	suite.Equal(stream.EventTypeUpdate, listMsg.Event)
+	suite.EqualValues([]string{stream.TimelineList + ":" + testList.ID}, listMsg.Stream)
+	suite.Empty(listStream.Messages) // Stream should now be empty.
 
-	// the irrelevant messages stream should also be empty
-	suite.Empty(irrelevantStream.Messages)
+	// Check status from list stream.
+	listStreamStatus := &apimodel.Status{}
+	if err := json.Unmarshal([]byte(listMsg.Payload), listStreamStatus); err != nil {
+		suite.FailNow(err.Error())
+	}
+	suite.Equal(newStatus.ID, listStreamStatus.ID)
+	suite.Equal(newStatus.Content, listStreamStatus.Content)
 }
 
 func (suite *FromClientAPITestSuite) TestProcessStatusDelete() {
-	ctx := context.Background()
+	var (
+		ctx                  = context.Background()
+		deletingAccount      = suite.testAccounts["local_account_1"]
+		receivingAccount     = suite.testAccounts["local_account_2"]
+		deletedStatus        = suite.testStatuses["local_account_1_status_1"]
+		boostOfDeletedStatus = suite.testStatuses["admin_account_status_4"]
+		streams              = suite.openStreams(ctx, receivingAccount, nil)
+		homeStream           = streams[stream.TimelineHome]
+	)
 
-	deletingAccount := suite.testAccounts["local_account_1"]
-	receivingAccount := suite.testAccounts["local_account_2"]
-
-	deletedStatus := suite.testStatuses["local_account_1_status_1"]
-	boostOfDeletedStatus := suite.testStatuses["admin_account_status_4"]
-
-	// open a home timeline stream for turtle, who follows zork
-	wssStream, errWithCode := suite.processor.Stream().Open(ctx, receivingAccount, stream.TimelineHome)
-	suite.NoError(errWithCode)
-
-	// delete the status from the db first, to mimic what would have already happened earlier up the flow
-	err := suite.db.DeleteStatusByID(ctx, deletedStatus.ID)
-	suite.NoError(err)
+	// Delete the status from the db first, to mimic what
+	// would have already happened earlier up the flow
+	if err := suite.db.DeleteStatusByID(ctx, deletedStatus.ID); err != nil {
+		suite.FailNow(err.Error())
+	}
 
-	// process the status delete
-	err = suite.processor.ProcessFromClientAPI(ctx, messages.FromClientAPI{
+	// Process the status delete.
+	if err := suite.processor.ProcessFromClientAPI(ctx, messages.FromClientAPI{
 		APObjectType:   ap.ObjectNote,
 		APActivityType: ap.ActivityDelete,
 		GTSModel:       deletedStatus,
 		OriginAccount:  deletingAccount,
-	})
-	suite.NoError(err)
+	}); err != nil {
+		suite.FailNow(err.Error())
+	}
 
-	// turtle's stream should have the delete of admin's boost in it now
-	msg := <-wssStream.Messages
+	// Stream should have the delete of admin's boost in it now.
+	msg := <-homeStream.Messages
 	suite.Equal(stream.EventTypeDelete, msg.Event)
 	suite.Equal(boostOfDeletedStatus.ID, msg.Payload)
 	suite.EqualValues([]string{stream.TimelineHome}, msg.Stream)
 
-	// turtle's stream should also have the delete of the message itself in it
-	msg = <-wssStream.Messages
+	// Stream should also have the delete of the message itself in it.
+	msg = <-homeStream.Messages
 	suite.Equal(stream.EventTypeDelete, msg.Event)
 	suite.Equal(deletedStatus.ID, msg.Payload)
 	suite.EqualValues([]string{stream.TimelineHome}, msg.Stream)
 
-	// stream should now be empty
-	suite.Empty(wssStream.Messages)
+	// Stream should now be empty.
+	suite.Empty(homeStream.Messages)
 
-	// the boost should no longer be in the database
-	_, err = suite.db.GetStatusByID(ctx, boostOfDeletedStatus.ID)
-	suite.ErrorIs(err, db.ErrNoEntries)
+	// Boost should no longer be in the database.
+	if !testrig.WaitFor(func() bool {
+		_, err := suite.db.GetStatusByID(ctx, boostOfDeletedStatus.ID)
+		return errors.Is(err, db.ErrNoEntries)
+	}) {
+		suite.FailNow("timed out waiting for status delete")
+	}
 }
 
 func (suite *FromClientAPITestSuite) TestProcessNewStatusWithNotification() {
-	ctx := context.Background()
-	postingAccount := suite.testAccounts["admin_account"]
-	receivingAccount := suite.testAccounts["local_account_1"]
+	var (
+		ctx              = context.Background()
+		postingAccount   = suite.testAccounts["admin_account"]
+		receivingAccount = suite.testAccounts["local_account_1"]
+		streams          = suite.openStreams(ctx, receivingAccount, nil)
+		notifStream      = streams[stream.TimelineNotifications]
+	)
 
 	// Update the follow from receiving account -> posting account so
 	// that receiving account wants notifs when posting account posts.
@@ -204,8 +225,9 @@ func (suite *FromClientAPITestSuite) TestProcessNewStatusWithNotification() {
 
 	// Put the status in the db first, to mimic what
 	// would have already happened earlier up the flow.
-	err := suite.db.PutStatus(ctx, newStatus)
-	suite.NoError(err)
+	if err := suite.db.PutStatus(ctx, newStatus); err != nil {
+		suite.FailNow(err.Error())
+	}
 
 	// Process the new status.
 	if err := suite.processor.ProcessFromClientAPI(ctx, messages.FromClientAPI{
@@ -230,6 +252,19 @@ func (suite *FromClientAPITestSuite) TestProcessNewStatusWithNotification() {
 	}) {
 		suite.FailNow("timed out waiting for new status notification")
 	}
+
+	// Check message in notification stream.
+	notifMsg := <-notifStream.Messages
+	suite.Equal(stream.EventTypeNotification, notifMsg.Event)
+	suite.EqualValues([]string{stream.TimelineNotifications}, notifMsg.Stream)
+	suite.Empty(notifStream.Messages) // Stream should now be empty.
+
+	// Check notif.
+	notif := &apimodel.Notification{}
+	if err := json.Unmarshal([]byte(notifMsg.Payload), notif); err != nil {
+		suite.FailNow(err.Error())
+	}
+	suite.Equal(newStatus.ID, notif.Status.ID)
 }
 
 func TestFromClientAPITestSuite(t *testing.T) {