replace async client API / federator msg processing with worker pools (#497)

* replace async client API / federator msg processing with worker pools
* appease our lord-and-saviour, the linter

5 files changed, 37 insertions, 15 deletions

diff --git a/testrig/federatingdb.go b/testrig/federatingdb.go
index f01a4e9fe..3f71274ca 100644
--- a/testrig/federatingdb.go
+++ b/testrig/federatingdb.go
@@ -3,9 +3,11 @@ package testrig
 import (
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation/federatingdb"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // NewTestFederatingDB returns a federating DB with the underlying db
-func NewTestFederatingDB(db db.DB) federatingdb.DB {
-	return federatingdb.New(db)
+func NewTestFederatingDB(db db.DB, fedWorker *worker.Worker[messages.FromFederator]) federatingdb.DB {
+	return federatingdb.New(db, fedWorker)
 }
diff --git a/testrig/federator.go b/testrig/federator.go
index 1b5e0fdc5..475ed3346 100644
--- a/testrig/federator.go
+++ b/testrig/federator.go
@@ -23,10 +23,12 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation"
 	"github.com/superseriousbusiness/gotosocial/internal/media"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/transport"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // NewTestFederator returns a federator with the given database and (mock!!) transport controller.
-func NewTestFederator(db db.DB, tc transport.Controller, storage *kv.KVStore, mediaManager media.Manager) federation.Federator {
-	return federation.NewFederator(db, NewTestFederatingDB(db), tc, NewTestTypeConverter(db), mediaManager)
+func NewTestFederator(db db.DB, tc transport.Controller, storage *kv.KVStore, mediaManager media.Manager, fedWorker *worker.Worker[messages.FromFederator]) federation.Federator {
+	return federation.NewFederator(db, NewTestFederatingDB(db, fedWorker), tc, NewTestTypeConverter(db), mediaManager)
 }
diff --git a/testrig/processor.go b/testrig/processor.go
index 4f3f8ae8a..c0fbd8a74 100644
--- a/testrig/processor.go
+++ b/testrig/processor.go
@@ -24,10 +24,12 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/email"
 	"github.com/superseriousbusiness/gotosocial/internal/federation"
 	"github.com/superseriousbusiness/gotosocial/internal/media"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/processing"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // NewTestProcessor returns a Processor suitable for testing purposes
-func NewTestProcessor(db db.DB, storage *kv.KVStore, federator federation.Federator, emailSender email.Sender, mediaManager media.Manager) processing.Processor {
-	return processing.NewProcessor(NewTestTypeConverter(db), federator, NewTestOauthServer(db), mediaManager, storage, db, emailSender)
+func NewTestProcessor(db db.DB, storage *kv.KVStore, federator federation.Federator, emailSender email.Sender, mediaManager media.Manager, clientWorker *worker.Worker[messages.FromClientAPI], fedWorker *worker.Worker[messages.FromFederator]) processing.Processor {
+	return processing.NewProcessor(NewTestTypeConverter(db), federator, NewTestOauthServer(db), mediaManager, storage, db, emailSender, clientWorker, fedWorker)
 }
diff --git a/testrig/testmodels.go b/testrig/testmodels.go
index 73dabaed6..5b06245a9 100644
--- a/testrig/testmodels.go
+++ b/testrig/testmodels.go
@@ -42,6 +42,8 @@ import (
 	"github.com/superseriousbusiness/activity/streams/vocab"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // NewTestTokens returns a map of tokens keyed according to which account the token belongs to.
@@ -505,7 +507,7 @@ func NewTestAccounts() map[string]*gtsmodel.Account {
 	}
 
 	if diff := len(accounts) - len(preserializedKeys); diff > 0 {
-		var keyStrings = make([]string, diff)
+		keyStrings := make([]string, diff)
 		for i := 0; i < diff; i++ {
 			priv, _ := rsa.GenerateKey(rand.Reader, 2048)
 			key, _ := x509.MarshalPKCS8PrivateKey(priv)
@@ -1823,8 +1825,13 @@ func GetSignatureForActivity(activity pub.Activity, pubKeyID string, privkey cry
 		},
 	}
 
+	// Create temporary federator worker for transport controller
+	fedWorker := worker.New[messages.FromFederator](-1, -1)
+	_ = fedWorker.Start()
+	defer func() { _ = fedWorker.Stop() }()
+
 	// use the client to create a new transport
-	c := NewTestTransportController(client, NewTestDB())
+	c := NewTestTransportController(client, NewTestDB(), fedWorker)
 	tp, err := c.NewTransport(pubKeyID, privkey)
 	if err != nil {
 		panic(err)
@@ -1865,8 +1872,13 @@ func GetSignatureForDereference(pubKeyID string, privkey crypto.PrivateKey, dest
 		},
 	}
 
+	// Create temporary federator worker for transport controller
+	fedWorker := worker.New[messages.FromFederator](-1, -1)
+	_ = fedWorker.Start()
+	defer func() { _ = fedWorker.Stop() }()
+
 	// use the client to create a new transport
-	c := NewTestTransportController(client, NewTestDB())
+	c := NewTestTransportController(client, NewTestDB(), fedWorker)
 	tp, err := c.NewTransport(pubKeyID, privkey)
 	if err != nil {
 		panic(err)
@@ -1899,7 +1911,8 @@ func newAPPerson(
 	avatarContentType string,
 	headerURL *url.URL,
 	headerContentType string,
-	manuallyApprovesFollowers bool) vocab.ActivityStreamsPerson {
+	manuallyApprovesFollowers bool,
+) vocab.ActivityStreamsPerson {
 	person := streams.NewActivityStreamsPerson()
 
 	// id should be the activitypub URI of this user
@@ -2082,7 +2095,8 @@ func newAPGroup(
 	avatarContentType string,
 	headerURL *url.URL,
 	headerContentType string,
-	manuallyApprovesFollowers bool) vocab.ActivityStreamsGroup {
+	manuallyApprovesFollowers bool,
+) vocab.ActivityStreamsGroup {
 	group := streams.NewActivityStreamsGroup()
 
 	// id should be the activitypub URI of this group
@@ -2303,8 +2317,8 @@ func newAPNote(
 	noteCC []*url.URL,
 	noteSensitive bool,
 	noteMentions []vocab.ActivityStreamsMention,
-	noteAttachments []vocab.ActivityStreamsImage) vocab.ActivityStreamsNote {
-
+	noteAttachments []vocab.ActivityStreamsImage,
+) vocab.ActivityStreamsNote {
 	// create the note itself
 	note := streams.NewActivityStreamsNote()
 
diff --git a/testrig/transportcontroller.go b/testrig/transportcontroller.go
index 90eab5ab3..943be7a61 100644
--- a/testrig/transportcontroller.go
+++ b/testrig/transportcontroller.go
@@ -26,7 +26,9 @@ import (
 	"github.com/superseriousbusiness/activity/pub"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/transport"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // NewTestTransportController returns a test transport controller with the given http client.
@@ -38,8 +40,8 @@ import (
 // Unlike the other test interfaces provided in this package, you'll probably want to call this function
 // PER TEST rather than per suite, so that the do function can be set on a test by test (or even more granular)
 // basis.
-func NewTestTransportController(client pub.HttpClient, db db.DB) transport.Controller {
-	return transport.NewController(db, NewTestFederatingDB(db), &federation.Clock{}, client)
+func NewTestTransportController(client pub.HttpClient, db db.DB, fedWorker *worker.Worker[messages.FromFederator]) transport.Controller {
+	return transport.NewController(db, NewTestFederatingDB(db, fedWorker), &federation.Clock{}, client)
 }
 
 // NewMockHTTPClient returns a client that conforms to the pub.HttpClient interface,