[performance] update remaining worker pools to use queues (#2865)

* start replacing client + federator + media workers with new worker + queue types

* refactor federatingDB.Delete(), drop queued messages when deleting account / status

* move all queue purging to the processor workers

* undo toolchain updates

* code comments, ensure dereferencer worker pool gets started

* update gruf libraries in readme

* start the job scheduler separately to the worker pools

* reshuffle ordering or server.go + remove duplicate worker start / stop

* update go-list version

* fix vendoring

* move queue invalidation to before wipeing / deletion, to ensure queued work not dropped

* add logging to worker processing functions in testrig, don't start workers in unexpected places

* update go-structr to add (+then rely on) QueueCtx{} type

* ensure more worker pools get started properly in tests

* fix remaining broken tests relying on worker queue logic

* fix account test suite queue popping logic, ensure noop workers do not pull from queue

* move back accidentally shuffled account deletion order

* ensure error (non nil!!) gets passed in refactored federatingDB{}.Delete()

* silently drop deletes from accounts not permitted to

* don't warn log on forwarded deletes

* make if else clauses easier to parse

* use getFederatorMsg()

* improved code comment

* improved code comment re: requesting account delete checks

* remove boolean result from worker start / stop since false = already running or already stopped

* remove optional passed-in http.client

* remove worker starting from the admin CLI commands (we don't need to handle side-effects)

* update prune cli to start scheduler but not all of the workers

* fix rebase issues

* remove redundant return statements

* i'm sorry sir linter

34 files changed, 449 insertions, 407 deletions

diff --git a/internal/processing/account/account_test.go b/internal/processing/account/account_test.go
index 244fc89d6..10d5f91e1 100644
--- a/internal/processing/account/account_test.go
+++ b/internal/processing/account/account_test.go
@@ -19,6 +19,7 @@ package account_test
 
 import (
 	"context"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
@@ -48,7 +49,6 @@ type AccountStandardTestSuite struct {
 	state               state.State
 	mediaManager        *media.Manager
 	oauthServer         oauth.Server
-	fromClientAPIChan   chan messages.FromClientAPI
 	transportController transport.Controller
 	federator           *federation.Federator
 	emailSender         email.Sender
@@ -68,6 +68,13 @@ type AccountStandardTestSuite struct {
 	accountProcessor account.Processor
 }
 
+func (suite *AccountStandardTestSuite) getClientMsg(timeout time.Duration) (*messages.FromClientAPI, bool) {
+	ctx := context.Background()
+	ctx, cncl := context.WithTimeout(ctx, timeout)
+	defer cncl()
+	return suite.state.Workers.Client.Queue.PopCtx(ctx)
+}
+
 func (suite *AccountStandardTestSuite) SetupSuite() {
 	suite.testTokens = testrig.NewTestTokens()
 	suite.testClients = testrig.NewTestClients()
@@ -101,13 +108,6 @@ func (suite *AccountStandardTestSuite) SetupTest() {
 	suite.mediaManager = testrig.NewTestMediaManager(&suite.state)
 	suite.oauthServer = testrig.NewTestOauthServer(suite.db)
 
-	suite.fromClientAPIChan = make(chan messages.FromClientAPI, 100)
-	suite.state.Workers.EnqueueClientAPI = func(ctx context.Context, msgs ...messages.FromClientAPI) {
-		for _, msg := range msgs {
-			suite.fromClientAPIChan <- msg
-		}
-	}
-
 	suite.transportController = testrig.NewTestTransportController(&suite.state, testrig.NewMockHTTPClient(nil, "../../../testrig/media"))
 	suite.federator = testrig.NewTestFederator(&suite.state, suite.transportController, suite.mediaManager)
 	suite.sentEmails = make(map[string]string)
diff --git a/internal/processing/account/block.go b/internal/processing/account/block.go
index 49f81557d..d3904bffa 100644
--- a/internal/processing/account/block.go
+++ b/internal/processing/account/block.go
@@ -83,16 +83,16 @@ func (p *Processor) BlockCreate(ctx context.Context, requestingAccount *gtsmodel
 	}
 
 	// Process block side effects (federation etc).
-	msgs = append(msgs, messages.FromClientAPI{
+	msgs = append(msgs, &messages.FromClientAPI{
 		APObjectType:   ap.ActivityBlock,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       block,
-		OriginAccount:  requestingAccount,
-		TargetAccount:  targetAccount,
+		Origin:         requestingAccount,
+		Target:         targetAccount,
 	})
 
 	// Batch queue accreted client api messages.
-	p.state.Workers.EnqueueClientAPI(ctx, msgs...)
+	p.state.Workers.Client.Queue.Push(msgs...)
 
 	return p.RelationshipGet(ctx, requestingAccount, targetAccountID)
 }
@@ -120,12 +120,12 @@ func (p *Processor) BlockRemove(ctx context.Context, requestingAccount *gtsmodel
 	existingBlock.TargetAccount = targetAccount
 
 	// Process block removal side effects (federation etc).
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActivityBlock,
 		APActivityType: ap.ActivityUndo,
 		GTSModel:       existingBlock,
-		OriginAccount:  requestingAccount,
-		TargetAccount:  targetAccount,
+		Origin:         requestingAccount,
+		Target:         targetAccount,
 	})
 
 	return p.RelationshipGet(ctx, requestingAccount, targetAccountID)
diff --git a/internal/processing/account/create.go b/internal/processing/account/create.go
index 12b2d5e57..761165356 100644
--- a/internal/processing/account/create.go
+++ b/internal/processing/account/create.go
@@ -126,11 +126,11 @@ func (p *Processor) Create(
 
 	// There are side effects for creating a new account
 	// (confirmation emails etc), perform these async.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ObjectProfile,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       user,
-		OriginAccount:  user.Account,
+		Origin:         user.Account,
 	})
 
 	return user, nil
diff --git a/internal/processing/account/delete.go b/internal/processing/account/delete.go
index a900c566d..3f051edf0 100644
--- a/internal/processing/account/delete.go
+++ b/internal/processing/account/delete.go
@@ -102,16 +102,13 @@ func (p *Processor) Delete(
 // and the above Delete function will be called afterwards from the processor, to clear
 // out the account's bits and bobs, and stubbify it.
 func (p *Processor) DeleteSelf(ctx context.Context, account *gtsmodel.Account) gtserror.WithCode {
-	fromClientAPIMessage := messages.FromClientAPI{
+	// Process the delete side effects asynchronously.
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActorPerson,
 		APActivityType: ap.ActivityDelete,
-		OriginAccount:  account,
-		TargetAccount:  account,
-	}
-
-	// Process the delete side effects asynchronously.
-	p.state.Workers.EnqueueClientAPI(ctx, fromClientAPIMessage)
-
+		Origin:         account,
+		Target:         account,
+	})
 	return nil
 }
 
@@ -193,7 +190,8 @@ func (p *Processor) deleteAccountFollows(ctx context.Context, account *gtsmodel.
 
 	var (
 		// Use this slice to batch unfollow messages.
-		msgs = []messages.FromClientAPI{}
+		msgs = []*messages.FromClientAPI{}
+
 		// To avoid checking if account is local over + over
 		// inside the subsequent loops, just generate static
 		// side effects function once now.
@@ -214,7 +212,7 @@ func (p *Processor) deleteAccountFollows(ctx context.Context, account *gtsmodel.
 		}
 		if msg := unfollowSideEffects(ctx, account, follow); msg != nil {
 			// There was a side effect to process.
-			msgs = append(msgs, *msg)
+			msgs = append(msgs, msg)
 		}
 	}
 
@@ -244,13 +242,13 @@ func (p *Processor) deleteAccountFollows(ctx context.Context, account *gtsmodel.
 
 		if msg := unfollowSideEffects(ctx, account, follow); msg != nil {
 			// There was a side effect to process.
-			msgs = append(msgs, *msg)
+			msgs = append(msgs, msg)
 		}
 	}
 
 	// Process accreted messages in serial.
 	for _, msg := range msgs {
-		if err := p.state.Workers.ProcessFromClientAPI(ctx, msg); err != nil {
+		if err := p.state.Workers.Client.Process(ctx, msg); err != nil {
 			log.Errorf(
 				ctx,
 				"error processing %s of %s during Delete of account %s: %v",
@@ -306,8 +304,8 @@ func (p *Processor) unfollowSideEffectsFunc(local bool) func(
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel:       follow,
-			OriginAccount:  account,
-			TargetAccount:  follow.TargetAccount,
+			Origin:         account,
+			Target:         follow.TargetAccount,
 		}
 	}
 }
@@ -337,7 +335,7 @@ func (p *Processor) deleteAccountStatuses(
 		statuses []*gtsmodel.Status
 		err      error
 		maxID    string
-		msgs     = []messages.FromClientAPI{}
+		msgs     = []*messages.FromClientAPI{}
 	)
 
 statusLoop:
@@ -404,29 +402,29 @@ statusLoop:
 					continue
 				}
 
-				msgs = append(msgs, messages.FromClientAPI{
+				msgs = append(msgs, &messages.FromClientAPI{
 					APObjectType:   ap.ActivityAnnounce,
 					APActivityType: ap.ActivityUndo,
 					GTSModel:       status,
-					OriginAccount:  boost.Account,
-					TargetAccount:  account,
+					Origin:         boost.Account,
+					Target:         account,
 				})
 			}
 
 			// Now prepare to Delete status.
-			msgs = append(msgs, messages.FromClientAPI{
+			msgs = append(msgs, &messages.FromClientAPI{
 				APObjectType:   ap.ObjectNote,
 				APActivityType: ap.ActivityDelete,
 				GTSModel:       status,
-				OriginAccount:  account,
-				TargetAccount:  account,
+				Origin:         account,
+				Target:         account,
 			})
 		}
 	}
 
 	// Process accreted messages in serial.
 	for _, msg := range msgs {
-		if err := p.state.Workers.ProcessFromClientAPI(ctx, msg); err != nil {
+		if err := p.state.Workers.Client.Process(ctx, msg); err != nil {
 			log.Errorf(
 				ctx,
 				"error processing %s of %s during Delete of account %s: %v",
diff --git a/internal/processing/account/follow.go b/internal/processing/account/follow.go
index 7f28bb2c0..402c764e8 100644
--- a/internal/processing/account/follow.go
+++ b/internal/processing/account/follow.go
@@ -117,12 +117,12 @@ func (p *Processor) FollowCreate(ctx context.Context, requestingAccount *gtsmode
 	} else {
 		// Otherwise we leave the follow request as it is,
 		// and we handle the rest of the process async.
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+		p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       fr,
-			OriginAccount:  requestingAccount,
-			TargetAccount:  targetAccount,
+			Origin:         requestingAccount,
+			Target:         targetAccount,
 		})
 	}
 
@@ -143,7 +143,7 @@ func (p *Processor) FollowRemove(ctx context.Context, requestingAccount *gtsmode
 	}
 
 	// Batch queue accreted client api messages.
-	p.state.Workers.EnqueueClientAPI(ctx, msgs...)
+	p.state.Workers.Client.Queue.Push(msgs...)
 
 	return p.RelationshipGet(ctx, requestingAccount, targetAccountID)
 }
@@ -225,8 +225,8 @@ func (p *Processor) getFollowTarget(ctx context.Context, requester *gtsmodel.Acc
 // If a follow and/or follow request was removed this way, one or two
 // messages will be returned which should then be processed by a client
 // api worker.
-func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Account, targetAccount *gtsmodel.Account) ([]messages.FromClientAPI, error) {
-	var msgs []messages.FromClientAPI
+func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Account, targetAccount *gtsmodel.Account) ([]*messages.FromClientAPI, error) {
+	var msgs []*messages.FromClientAPI
 
 	// Get follow from requesting account to target account.
 	follow, err := p.state.DB.GetFollow(ctx, requestingAccount.ID, targetAccount.ID)
@@ -251,7 +251,7 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 		}
 
 		// Follow status changed, process side effects.
-		msgs = append(msgs, messages.FromClientAPI{
+		msgs = append(msgs, &messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -259,8 +259,8 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 				TargetAccountID: targetAccount.ID,
 				URI:             follow.URI,
 			},
-			OriginAccount: requestingAccount,
-			TargetAccount: targetAccount,
+			Origin: requestingAccount,
+			Target: targetAccount,
 		})
 	}
 
@@ -287,7 +287,7 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 		}
 
 		// Follow status changed, process side effects.
-		msgs = append(msgs, messages.FromClientAPI{
+		msgs = append(msgs, &messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -295,8 +295,8 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 				TargetAccountID: targetAccount.ID,
 				URI:             followReq.URI,
 			},
-			OriginAccount: requestingAccount,
-			TargetAccount: targetAccount,
+			Origin: requestingAccount,
+			Target: targetAccount,
 		})
 	}
 
diff --git a/internal/processing/account/follow_request.go b/internal/processing/account/follow_request.go
index c054637c8..6f6c7ba2d 100644
--- a/internal/processing/account/follow_request.go
+++ b/internal/processing/account/follow_request.go
@@ -40,12 +40,12 @@ func (p *Processor) FollowRequestAccept(ctx context.Context, requestingAccount *
 	if follow.Account != nil {
 		// Only enqueue work in the case we have a request creating account stored.
 		// NOTE: due to how AcceptFollowRequest works, the inverse shouldn't be possible.
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+		p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityAccept,
 			GTSModel:       follow,
-			OriginAccount:  follow.Account,
-			TargetAccount:  follow.TargetAccount,
+			Origin:         follow.Account,
+			Target:         follow.TargetAccount,
 		})
 	}
 
@@ -67,12 +67,12 @@ func (p *Processor) FollowRequestReject(ctx context.Context, requestingAccount *
 	if followRequest.Account != nil {
 		// Only enqueue work in the case we have a request creating account stored.
 		// NOTE: due to how GetFollowRequest works, the inverse shouldn't be possible.
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+		p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityReject,
 			GTSModel:       followRequest,
-			OriginAccount:  followRequest.Account,
-			TargetAccount:  followRequest.TargetAccount,
+			Origin:         followRequest.Account,
+			Target:         followRequest.TargetAccount,
 		})
 	}
 
diff --git a/internal/processing/account/follow_test.go b/internal/processing/account/follow_test.go
index c269dc710..9ea8ce1b8 100644
--- a/internal/processing/account/follow_test.go
+++ b/internal/processing/account/follow_test.go
@@ -25,7 +25,6 @@ import (
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
-	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/util"
 )
 
@@ -152,18 +151,11 @@ func (suite *FollowTestSuite) TestFollowRequestLocal() {
 	}
 
 	// There should be a message going to the worker.
-	var cMsg messages.FromClientAPI
-	select {
-	case cMsg = <-suite.fromClientAPIChan:
-		// No problem.
-	case <-time.After(5 * time.Second):
-		suite.FailNow("timed out waiting for message")
-	}
-
+	cMsg, _ := suite.getClientMsg(5 * time.Second)
 	suite.Equal(ap.ActivityCreate, cMsg.APActivityType)
 	suite.Equal(ap.ActivityFollow, cMsg.APObjectType)
-	suite.Equal(requestingAccount.ID, cMsg.OriginAccount.ID)
-	suite.Equal(targetAccount.ID, cMsg.TargetAccount.ID)
+	suite.Equal(requestingAccount.ID, cMsg.Origin.ID)
+	suite.Equal(targetAccount.ID, cMsg.Target.ID)
 }
 
 func TestFollowTestS(t *testing.T) {
diff --git a/internal/processing/account/move.go b/internal/processing/account/move.go
index 602e8c021..63187dfd1 100644
--- a/internal/processing/account/move.go
+++ b/internal/processing/account/move.go
@@ -317,12 +317,12 @@ func (p *Processor) MoveSelf(
 	}
 
 	// Everything seems OK, process Move side effects async.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActorPerson,
 		APActivityType: ap.ActivityMove,
 		GTSModel:       move,
-		OriginAccount:  originAcct,
-		TargetAccount:  targetAcct,
+		Origin:         originAcct,
+		Target:         targetAcct,
 	})
 
 	return nil
diff --git a/internal/processing/account/move_test.go b/internal/processing/account/move_test.go
index dfa0ea4e4..c1a931252 100644
--- a/internal/processing/account/move_test.go
+++ b/internal/processing/account/move_test.go
@@ -70,29 +70,23 @@ func (suite *MoveTestSuite) TestMoveAccountOK() {
 		suite.FailNow(err.Error())
 	}
 
-	// There should be a msg heading back to fromClientAPI.
-	select {
-	case msg := <-suite.fromClientAPIChan:
-		move, ok := msg.GTSModel.(*gtsmodel.Move)
-		if !ok {
-			suite.FailNow("", "could not cast %T to *gtsmodel.Move", move)
-		}
-
-		now := time.Now()
-		suite.WithinDuration(now, move.CreatedAt, 5*time.Second)
-		suite.WithinDuration(now, move.UpdatedAt, 5*time.Second)
-		suite.WithinDuration(now, move.AttemptedAt, 5*time.Second)
-		suite.Zero(move.SucceededAt)
-		suite.NotZero(move.ID)
-		suite.Equal(requestingAcct.URI, move.OriginURI)
-		suite.NotNil(move.Origin)
-		suite.Equal(targetAcct.URI, move.TargetURI)
-		suite.NotNil(move.Target)
-		suite.NotZero(move.URI)
-
-	case <-time.After(5 * time.Second):
-		suite.FailNow("time out waiting for message")
+	// There should be a message going to the worker.
+	cMsg, _ := suite.getClientMsg(5 * time.Second)
+	move, ok := cMsg.GTSModel.(*gtsmodel.Move)
+	if !ok {
+		suite.FailNow("", "could not cast %T to *gtsmodel.Move", move)
 	}
+	now := time.Now()
+	suite.WithinDuration(now, move.CreatedAt, 5*time.Second)
+	suite.WithinDuration(now, move.UpdatedAt, 5*time.Second)
+	suite.WithinDuration(now, move.AttemptedAt, 5*time.Second)
+	suite.Zero(move.SucceededAt)
+	suite.NotZero(move.ID)
+	suite.Equal(requestingAcct.URI, move.OriginURI)
+	suite.NotNil(move.Origin)
+	suite.Equal(targetAcct.URI, move.TargetURI)
+	suite.NotNil(move.Target)
+	suite.NotZero(move.URI)
 
 	// Move should be in the database now.
 	move, err := suite.state.DB.GetMoveByOriginTarget(
diff --git a/internal/processing/account/update.go b/internal/processing/account/update.go
index 670620e19..7f2749503 100644
--- a/internal/processing/account/update.go
+++ b/internal/processing/account/update.go
@@ -296,11 +296,11 @@ func (p *Processor) Update(ctx context.Context, account *gtsmodel.Account, form
 		return nil, gtserror.NewErrorInternalError(fmt.Errorf("could not update account settings %s: %s", account.ID, err))
 	}
 
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ObjectProfile,
 		APActivityType: ap.ActivityUpdate,
 		GTSModel:       account,
-		OriginAccount:  account,
+		Origin:         account,
 	})
 
 	acctSensitive, err := p.converter.AccountToAPIAccountSensitive(ctx, account)
diff --git a/internal/processing/account/update_test.go b/internal/processing/account/update_test.go
index 76ad3abe8..ad09ff25c 100644
--- a/internal/processing/account/update_test.go
+++ b/internal/processing/account/update_test.go
@@ -20,6 +20,7 @@ package account_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
@@ -31,20 +32,6 @@ type AccountUpdateTestSuite struct {
 	AccountStandardTestSuite
 }
 
-func (suite *AccountStandardTestSuite) checkClientAPIChan(accountID string) {
-	msg := <-suite.fromClientAPIChan
-
-	// Profile update.
-	suite.Equal(ap.ActivityUpdate, msg.APActivityType)
-	suite.Equal(ap.ObjectProfile, msg.APObjectType)
-
-	// Correct account updated.
-	if msg.OriginAccount == nil {
-		suite.FailNow("expected msg.OriginAccount not to be nil")
-	}
-	suite.Equal(accountID, msg.OriginAccount.ID)
-}
-
 func (suite *AccountUpdateTestSuite) TestAccountUpdateSimple() {
 	testAccount := &gtsmodel.Account{}
 	*testAccount = *suite.testAccounts["local_account_1"]
@@ -73,7 +60,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateSimple() {
 	suite.Equal(noteExpected, apiAccount.Note)
 
 	// We should have an update in the client api channel.
-	suite.checkClientAPIChan(testAccount.ID)
+	msg, _ := suite.getClientMsg(5 * time.Second)
+
+	// Profile update.
+	suite.Equal(ap.ActivityUpdate, msg.APActivityType)
+	suite.Equal(ap.ObjectProfile, msg.APObjectType)
+
+	// Correct account updated.
+	if msg.Origin == nil {
+		suite.FailNow("expected msg.OriginAccount not to be nil")
+	}
+	suite.Equal(testAccount.ID, msg.Origin.ID)
 
 	// Check database model of account as well.
 	dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID)
@@ -113,7 +110,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateWithMention() {
 	suite.Equal(noteExpected, apiAccount.Note)
 
 	// We should have an update in the client api channel.
-	suite.checkClientAPIChan(testAccount.ID)
+	msg, _ := suite.getClientMsg(5 * time.Second)
+
+	// Profile update.
+	suite.Equal(ap.ActivityUpdate, msg.APActivityType)
+	suite.Equal(ap.ObjectProfile, msg.APObjectType)
+
+	// Correct account updated.
+	if msg.Origin == nil {
+		suite.FailNow("expected msg.OriginAccount not to be nil")
+	}
+	suite.Equal(testAccount.ID, msg.Origin.ID)
 
 	// Check database model of account as well.
 	dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID)
@@ -159,7 +166,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateWithMarkdownNote() {
 	suite.Equal(noteExpected, apiAccount.Note)
 
 	// We should have an update in the client api channel.
-	suite.checkClientAPIChan(testAccount.ID)
+	msg, _ := suite.getClientMsg(5 * time.Second)
+
+	// Profile update.
+	suite.Equal(ap.ActivityUpdate, msg.APActivityType)
+	suite.Equal(ap.ObjectProfile, msg.APObjectType)
+
+	// Correct account updated.
+	if msg.Origin == nil {
+		suite.FailNow("expected msg.OriginAccount not to be nil")
+	}
+	suite.Equal(testAccount.ID, msg.Origin.ID)
 
 	// Check database model of account as well.
 	dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID)
@@ -234,7 +251,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateWithFields() {
 	suite.EqualValues(emojisExpected, apiAccount.Emojis)
 
 	// We should have an update in the client api channel.
-	suite.checkClientAPIChan(testAccount.ID)
+	msg, _ := suite.getClientMsg(5 * time.Second)
+
+	// Profile update.
+	suite.Equal(ap.ActivityUpdate, msg.APActivityType)
+	suite.Equal(ap.ObjectProfile, msg.APObjectType)
+
+	// Correct account updated.
+	if msg.Origin == nil {
+		suite.FailNow("expected msg.OriginAccount not to be nil")
+	}
+	suite.Equal(testAccount.ID, msg.Origin.ID)
 
 	// Check database model of account as well.
 	dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID)
@@ -281,7 +308,17 @@ func (suite *AccountUpdateTestSuite) TestAccountUpdateNoteNotFields() {
 	suite.Equal(fieldsBefore, len(apiAccount.Fields))
 
 	// We should have an update in the client api channel.
-	suite.checkClientAPIChan(testAccount.ID)
+	msg, _ := suite.getClientMsg(5 * time.Second)
+
+	// Profile update.
+	suite.Equal(ap.ActivityUpdate, msg.APActivityType)
+	suite.Equal(ap.ObjectProfile, msg.APObjectType)
+
+	// Correct account updated.
+	if msg.Origin == nil {
+		suite.FailNow("expected msg.OriginAccount not to be nil")
+	}
+	suite.Equal(testAccount.ID, msg.Origin.ID)
 
 	// Check database model of account as well.
 	dbAccount, err := suite.db.GetAccountByID(ctx, testAccount.ID)
diff --git a/internal/processing/admin/accountaction.go b/internal/processing/admin/accountaction.go
index 155d8c0b4..7fd1047c4 100644
--- a/internal/processing/admin/accountaction.go
+++ b/internal/processing/admin/accountaction.go
@@ -80,13 +80,13 @@ func (p *Processor) accountActionSuspend(
 			Text:           text,
 		},
 		func(ctx context.Context) gtserror.MultiError {
-			if err := p.state.Workers.ProcessFromClientAPI(
+			if err := p.state.Workers.Client.Process(
 				ctx,
-				messages.FromClientAPI{
+				&messages.FromClientAPI{
 					APObjectType:   ap.ActorPerson,
 					APActivityType: ap.ActivityDelete,
-					OriginAccount:  adminAcct,
-					TargetAccount:  targetAcct,
+					Origin:         adminAcct,
+					Target:         targetAcct,
 				},
 			); err != nil {
 				errs := gtserror.NewMultiError(1)
diff --git a/internal/processing/admin/accountapprove.go b/internal/processing/admin/accountapprove.go
index c8a49e089..ebc91ee0c 100644
--- a/internal/processing/admin/accountapprove.go
+++ b/internal/processing/admin/accountapprove.go
@@ -54,12 +54,12 @@ func (p *Processor) AccountApprove(
 
 	if !*user.Approved {
 		// Process approval side effects asynschronously.
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+		p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 			APObjectType:   ap.ActorPerson,
 			APActivityType: ap.ActivityAccept,
 			GTSModel:       user,
-			OriginAccount:  adminAcct,
-			TargetAccount:  user.Account,
+			Origin:         adminAcct,
+			Target:         user.Account,
 		})
 	}
 
diff --git a/internal/processing/admin/accountreject.go b/internal/processing/admin/accountreject.go
index eee2b2ff5..e7d54be41 100644
--- a/internal/processing/admin/accountreject.go
+++ b/internal/processing/admin/accountreject.go
@@ -101,12 +101,12 @@ func (p *Processor) AccountReject(
 	}
 
 	// Process rejection side effects asynschronously.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActorPerson,
 		APActivityType: ap.ActivityReject,
 		GTSModel:       deniedUser,
-		OriginAccount:  adminAcct,
-		TargetAccount:  user.Account,
+		Origin:         adminAcct,
+		Target:         user.Account,
 	})
 
 	return apiAccount, nil
diff --git a/internal/processing/admin/actions.go b/internal/processing/admin/actions.go
index 0c959b1c5..968e45baa 100644
--- a/internal/processing/admin/actions.go
+++ b/internal/processing/admin/actions.go
@@ -23,6 +23,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/log"
@@ -97,8 +98,10 @@ func (a *Actions) Run(
 	// we're done modifying it for now.
 	a.m.Unlock()
 
-	// Do the rest of the work asynchronously.
-	a.state.Workers.ClientAPI.Enqueue(func(ctx context.Context) {
+	go func() {
+		// Use a background context with existing values.
+		ctx = gtscontext.WithValues(context.Background(), ctx)
+
 		// Run the thing and collect errors.
 		if errs := f(ctx); errs != nil {
 			action.Errors = make([]string, 0, len(errs))
@@ -119,7 +122,7 @@ func (a *Actions) Run(
 		if err := a.state.DB.UpdateAdminAction(ctx, action, "completed_at", "errors"); err != nil {
 			log.Errorf(ctx, "db error marking action %s as completed: %q", actionKey, err)
 		}
-	})
+	}()
 
 	return nil
 }
diff --git a/internal/processing/admin/domainblock.go b/internal/processing/admin/domainblock.go
index 9fe5dc847..2fe10c97b 100644
--- a/internal/processing/admin/domainblock.go
+++ b/internal/processing/admin/domainblock.go
@@ -201,15 +201,13 @@ func (p *Processor) domainBlockSideEffects(
 	// process an account delete message to remove
 	// that account's posts, media, etc.
 	if err := p.rangeDomainAccounts(ctx, block.Domain, func(account *gtsmodel.Account) {
-		cMsg := messages.FromClientAPI{
+		if err := p.state.Workers.Client.Process(ctx, &messages.FromClientAPI{
 			APObjectType:   ap.ActorPerson,
 			APActivityType: ap.ActivityDelete,
 			GTSModel:       block,
-			OriginAccount:  account,
-			TargetAccount:  account,
-		}
-
-		if err := p.state.Workers.ProcessFromClientAPI(ctx, cMsg); err != nil {
+			Origin:         account,
+			Target:         account,
+		}); err != nil {
 			errs.Append(err)
 		}
 	}); err != nil {
diff --git a/internal/processing/admin/report.go b/internal/processing/admin/report.go
index 32f05719b..d2186cfa2 100644
--- a/internal/processing/admin/report.go
+++ b/internal/processing/admin/report.go
@@ -140,12 +140,12 @@ func (p *Processor) ReportResolve(ctx context.Context, account *gtsmodel.Account
 	}
 
 	// Process side effects of closing the report.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActivityFlag,
 		APActivityType: ap.ActivityUpdate,
 		GTSModel:       report,
-		OriginAccount:  account,
-		TargetAccount:  report.Account,
+		Origin:         account,
+		Target:         report.Account,
 	})
 
 	apimodelReport, err := p.converter.ReportToAdminAPIReport(ctx, updatedReport, account)
diff --git a/internal/processing/polls/expiry.go b/internal/processing/polls/expiry.go
index 59d0f17fe..d02a05f0d 100644
--- a/internal/processing/polls/expiry.go
+++ b/internal/processing/polls/expiry.go
@@ -116,11 +116,11 @@ func (p *Processor) onExpiry(pollID string) func(context.Context, time.Time) {
 
 		// Enqueue a status update operation to the client API worker,
 		// this will asynchronously send an update with the Poll close time.
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+		p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 			APActivityType: ap.ActivityUpdate,
 			APObjectType:   ap.ObjectNote,
 			GTSModel:       status,
-			OriginAccount:  status.Account,
+			Origin:         status.Account,
 		})
 	}
 }
diff --git a/internal/processing/polls/vote.go b/internal/processing/polls/vote.go
index 5bc523978..c970fe106 100644
--- a/internal/processing/polls/vote.go
+++ b/internal/processing/polls/vote.go
@@ -96,11 +96,11 @@ func (p *Processor) PollVote(ctx context.Context, requester *gtsmodel.Account, p
 	poll.IncrementVotes(choices)
 
 	// Enqueue worker task to handle side-effects of user poll vote(s).
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APActivityType: ap.ActivityCreate,
 		APObjectType:   ap.ActivityQuestion,
 		GTSModel:       vote, // the vote choices
-		OriginAccount:  requester,
+		Origin:         requester,
 	})
 
 	// Return converted API model poll.
diff --git a/internal/processing/processor_test.go b/internal/processing/processor_test.go
index 2a1d47e46..767e8b5ef 100644
--- a/internal/processing/processor_test.go
+++ b/internal/processing/processor_test.go
@@ -95,7 +95,6 @@ func (suite *ProcessingStandardTestSuite) SetupSuite() {
 
 func (suite *ProcessingStandardTestSuite) SetupTest() {
 	suite.state.Caches.Init()
-	testrig.StartNoopWorkers(&suite.state)
 
 	testrig.InitTestConfig()
 	testrig.InitTestLog()
@@ -124,8 +123,7 @@ func (suite *ProcessingStandardTestSuite) SetupTest() {
 	suite.emailSender = testrig.NewEmailSender("../../web/template/", nil)
 
 	suite.processor = processing.NewProcessor(cleaner.New(&suite.state), suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, &suite.state, suite.emailSender)
-	suite.state.Workers.EnqueueClientAPI = suite.processor.Workers().EnqueueClientAPI
-	suite.state.Workers.EnqueueFediAPI = suite.processor.Workers().EnqueueFediAPI
+	testrig.StartWorkers(&suite.state, suite.processor.Workers())
 
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 	testrig.StandardStorageSetup(suite.storage, "../../testrig/media")
diff --git a/internal/processing/report/create.go b/internal/processing/report/create.go
index c65ae0d52..cac600006 100644
--- a/internal/processing/report/create.go
+++ b/internal/processing/report/create.go
@@ -91,12 +91,12 @@ func (p *Processor) Create(ctx context.Context, account *gtsmodel.Account, form
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ObjectProfile,
 		APActivityType: ap.ActivityFlag,
 		GTSModel:       report,
-		OriginAccount:  account,
-		TargetAccount:  targetAccount,
+		Origin:         account,
+		Target:         targetAccount,
 	})
 
 	apiReport, err := p.converter.ReportToAPIReport(ctx, report)
diff --git a/internal/processing/status/boost.go b/internal/processing/status/boost.go
index 5fae695fd..1c1da4ca7 100644
--- a/internal/processing/status/boost.go
+++ b/internal/processing/status/boost.go
@@ -89,12 +89,12 @@ func (p *Processor) BoostCreate(
 	}
 
 	// Process side effects asynchronously.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActivityAnnounce,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       boost,
-		OriginAccount:  requester,
-		TargetAccount:  target.Account,
+		Origin:         requester,
+		Target:         target.Account,
 	})
 
 	return p.c.GetAPIStatus(ctx, requester, boost)
@@ -141,12 +141,12 @@ func (p *Processor) BoostRemove(
 
 	if boost != nil {
 		// Status was boosted. Process unboost side effects asynchronously.
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+		p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityUndo,
 			GTSModel:       boost,
-			OriginAccount:  requester,
-			TargetAccount:  target.Account,
+			Origin:         requester,
+			Target:         target.Account,
 		})
 	}
 
diff --git a/internal/processing/status/create.go b/internal/processing/status/create.go
index d758fc0fb..80cc65c7f 100644
--- a/internal/processing/status/create.go
+++ b/internal/processing/status/create.go
@@ -143,11 +143,11 @@ func (p *Processor) Create(
 	}
 
 	// send it back to the client API worker for async side-effects.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ObjectNote,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       status,
-		OriginAccount:  requester,
+		Origin:         requester,
 	})
 
 	if status.Poll != nil {
diff --git a/internal/processing/status/delete.go b/internal/processing/status/delete.go
index 261086bdb..700909f44 100644
--- a/internal/processing/status/delete.go
+++ b/internal/processing/status/delete.go
@@ -51,12 +51,12 @@ func (p *Processor) Delete(ctx context.Context, requestingAccount *gtsmodel.Acco
 	}
 
 	// Process delete side effects.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ObjectNote,
 		APActivityType: ap.ActivityDelete,
 		GTSModel:       targetStatus,
-		OriginAccount:  requestingAccount,
-		TargetAccount:  requestingAccount,
+		Origin:         requestingAccount,
+		Target:         requestingAccount,
 	})
 
 	return apiStatus, nil
diff --git a/internal/processing/status/fave.go b/internal/processing/status/fave.go
index 7b71725ab..dd961c082 100644
--- a/internal/processing/status/fave.go
+++ b/internal/processing/status/fave.go
@@ -107,12 +107,12 @@ func (p *Processor) FaveCreate(ctx context.Context, requestingAccount *gtsmodel.
 	}
 
 	// Process new status fave side effects.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActivityLike,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       gtsFave,
-		OriginAccount:  requestingAccount,
-		TargetAccount:  targetStatus.Account,
+		Origin:         requestingAccount,
+		Target:         targetStatus.Account,
 	})
 
 	return p.c.GetAPIStatus(ctx, requestingAccount, targetStatus)
@@ -137,12 +137,12 @@ func (p *Processor) FaveRemove(ctx context.Context, requestingAccount *gtsmodel.
 	}
 
 	// Process remove status fave side effects.
-	p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 		APObjectType:   ap.ActivityLike,
 		APActivityType: ap.ActivityUndo,
 		GTSModel:       existingFave,
-		OriginAccount:  requestingAccount,
-		TargetAccount:  targetStatus.Account,
+		Origin:         requestingAccount,
+		Target:         targetStatus.Account,
 	})
 
 	return p.c.GetAPIStatus(ctx, requestingAccount, targetStatus)
diff --git a/internal/processing/workers/federate.go b/internal/processing/workers/federate.go
index e737513f5..3538c9958 100644
--- a/internal/processing/workers/federate.go
+++ b/internal/processing/workers/federate.go
@@ -75,12 +75,6 @@ func (f *federate) DeleteAccount(ctx context.Context, account *gtsmodel.Account)
 		return nil
 	}
 
-	// Drop any queued outgoing AP requests to / from account,
-	// (this stops any queued likes, boosts, creates etc).
-	f.state.Workers.Delivery.Queue.Delete("ActorID", account.URI)
-	f.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI)
-	f.state.Workers.Delivery.Queue.Delete("TargetID", account.URI)
-
 	// Parse relevant URI(s).
 	outboxIRI, err := parseURI(account.OutboxURI)
 	if err != nil {
@@ -228,16 +222,6 @@ func (f *federate) DeleteStatus(ctx context.Context, status *gtsmodel.Status) er
 		return nil
 	}
 
-	// Drop any queued outgoing http requests for status,
-	// (this stops any queued likes, boosts, creates etc).
-	f.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI)
-	f.state.Workers.Delivery.Queue.Delete("TargetID", status.URI)
-
-	// Ensure the status model is fully populated.
-	if err := f.state.DB.PopulateStatus(ctx, status); err != nil {
-		return gtserror.Newf("error populating status: %w", err)
-	}
-
 	// Parse the outbox URI of the status author.
 	outboxIRI, err := parseURI(status.Account.OutboxURI)
 	if err != nil {
diff --git a/internal/processing/workers/fromclientapi.go b/internal/processing/workers/fromclientapi.go
index 1412ea003..4564afbb9 100644
--- a/internal/processing/workers/fromclientapi.go
+++ b/internal/processing/workers/fromclientapi.go
@@ -25,7 +25,6 @@ import (
 	"codeberg.org/gruf/go-logger/v2/level"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
-	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/log"
@@ -45,29 +44,15 @@ type clientAPI struct {
 	surface   *surface
 	federate  *federate
 	account   *account.Processor
-	utilF     *utilF
+	utils     *utils
 }
 
-func (p *Processor) EnqueueClientAPI(cctx context.Context, msgs ...messages.FromClientAPI) {
-	_ = p.workers.ClientAPI.MustEnqueueCtx(cctx, func(wctx context.Context) {
-		// Copy caller ctx values to worker's.
-		wctx = gtscontext.WithValues(wctx, cctx)
-
-		// Process worker messages.
-		for _, msg := range msgs {
-			if err := p.ProcessFromClientAPI(wctx, msg); err != nil {
-				log.Errorf(wctx, "error processing client API message: %v", err)
-			}
-		}
-	})
-}
-
-func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Allocate new log fields slice
 	fields := make([]kv.Field, 3, 4)
 	fields[0] = kv.Field{"activityType", cMsg.APActivityType}
 	fields[1] = kv.Field{"objectType", cMsg.APObjectType}
-	fields[2] = kv.Field{"fromAccount", cMsg.OriginAccount.Username}
+	fields[2] = kv.Field{"fromAccount", cMsg.Origin.Username}
 
 	// Include GTSModel in logs if appropriate.
 	if cMsg.GTSModel != nil &&
@@ -217,7 +202,7 @@ func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg messages.From
 	return gtserror.Newf("unhandled: %s %s", cMsg.APActivityType, cMsg.APObjectType)
 }
 
-func (p *clientAPI) CreateAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	newUser, ok := cMsg.GTSModel.(*gtsmodel.User)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.User", cMsg.GTSModel)
@@ -241,14 +226,14 @@ func (p *clientAPI) CreateAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) CreateStatus(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateStatus(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	status, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel)
 	}
 
 	// Update stats for the actor account.
-	if err := p.utilF.incrementStatusesCount(ctx, cMsg.OriginAccount, status); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, cMsg.Origin, status); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -269,7 +254,7 @@ func (p *clientAPI) CreateStatus(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Cast the create poll vote attached to message.
 	vote, ok := cMsg.GTSModel.(*gtsmodel.PollVote)
 	if !ok {
@@ -310,14 +295,14 @@ func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg messages.FromClient
 	return nil
 }
 
-func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	followRequest, ok := cMsg.GTSModel.(*gtsmodel.FollowRequest)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", cMsg.GTSModel)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.incrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.incrementFollowRequestsCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -338,7 +323,7 @@ func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg messages.FromClien
 	return nil
 }
 
-func (p *clientAPI) CreateLike(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateLike(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	fave, ok := cMsg.GTSModel.(*gtsmodel.StatusFave)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", cMsg.GTSModel)
@@ -364,14 +349,14 @@ func (p *clientAPI) CreateLike(ctx context.Context, cMsg messages.FromClientAPI)
 	return nil
 }
 
-func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	boost, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel)
 	}
 
 	// Update stats for the actor account.
-	if err := p.utilF.incrementStatusesCount(ctx, cMsg.OriginAccount, boost); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, cMsg.Origin, boost); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -396,7 +381,7 @@ func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg messages.FromClient
 	return nil
 }
 
-func (p *clientAPI) CreateBlock(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateBlock(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	block, ok := cMsg.GTSModel.(*gtsmodel.Block)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Block", cMsg.GTSModel)
@@ -430,7 +415,7 @@ func (p *clientAPI) CreateBlock(ctx context.Context, cMsg messages.FromClientAPI
 	return nil
 }
 
-func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Cast the updated Status model attached to msg.
 	status, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
@@ -462,7 +447,7 @@ func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	account, ok := cMsg.GTSModel.(*gtsmodel.Account)
 	if !ok {
 		return gtserror.Newf("cannot cast %T -> *gtsmodel.Account", cMsg.GTSModel)
@@ -475,7 +460,7 @@ func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) UpdateReport(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UpdateReport(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	report, ok := cMsg.GTSModel.(*gtsmodel.Report)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Report", cMsg.GTSModel)
@@ -494,23 +479,23 @@ func (p *clientAPI) UpdateReport(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	follow, ok := cMsg.GTSModel.(*gtsmodel.Follow)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Follow", cMsg.GTSModel)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.decrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.decrementFollowRequestsCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
-	if err := p.utilF.incrementFollowersCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.incrementFollowersCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.incrementFollowingCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.incrementFollowingCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -525,14 +510,14 @@ func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	followReq, ok := cMsg.GTSModel.(*gtsmodel.FollowRequest)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", cMsg.GTSModel)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.decrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.decrementFollowRequestsCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -546,19 +531,19 @@ func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg messages.FromC
 	return nil
 }
 
-func (p *clientAPI) UndoFollow(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoFollow(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	follow, ok := cMsg.GTSModel.(*gtsmodel.Follow)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Follow", cMsg.GTSModel)
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.decrementFollowingCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.decrementFollowingCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.decrementFollowersCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.decrementFollowersCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -569,7 +554,7 @@ func (p *clientAPI) UndoFollow(ctx context.Context, cMsg messages.FromClientAPI)
 	return nil
 }
 
-func (p *clientAPI) UndoBlock(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoBlock(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	block, ok := cMsg.GTSModel.(*gtsmodel.Block)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Block", cMsg.GTSModel)
@@ -582,7 +567,7 @@ func (p *clientAPI) UndoBlock(ctx context.Context, cMsg messages.FromClientAPI)
 	return nil
 }
 
-func (p *clientAPI) UndoFave(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoFave(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	statusFave, ok := cMsg.GTSModel.(*gtsmodel.StatusFave)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", cMsg.GTSModel)
@@ -599,7 +584,7 @@ func (p *clientAPI) UndoFave(ctx context.Context, cMsg messages.FromClientAPI) e
 	return nil
 }
 
-func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	status, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel)
@@ -610,7 +595,7 @@ func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAP
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.decrementStatusesCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.decrementStatusesCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -629,7 +614,7 @@ func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Don't delete attachments, just unattach them:
 	// this request comes from the client API and the
 	// poster may want to use attachments again later.
@@ -648,12 +633,26 @@ func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAP
 		return gtserror.Newf("db error populating status: %w", err)
 	}
 
-	if err := p.utilF.wipeStatus(ctx, status, deleteAttachments); err != nil {
+	// Drop any outgoing queued AP requests about / targeting
+	// this status, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", status.URI)
+
+	// Drop any incoming queued client messages about / targeting
+	// status, (stops processing of local origin data for status).
+	p.state.Workers.Client.Queue.Delete("TargetURI", status.URI)
+
+	// Drop any incoming queued federator messages targeting status,
+	// (stops processing of remote origin data targeting this status).
+	p.state.Workers.Federator.Queue.Delete("TargetURI", status.URI)
+
+	// First perform the actual status deletion.
+	if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil {
 		log.Errorf(ctx, "error wiping status: %v", err)
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.decrementStatusesCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.decrementStatusesCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -670,7 +669,7 @@ func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// The originID of the delete, one of:
 	//   - ID of a domain block, for which
 	//     this account delete is a side effect.
@@ -684,21 +683,41 @@ func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg messages.FromClientA
 	} else {
 		// Origin is whichever account
 		// originated this message.
-		originID = cMsg.OriginAccount.ID
+		originID = cMsg.Origin.ID
 	}
 
-	if err := p.federate.DeleteAccount(ctx, cMsg.TargetAccount); err != nil {
+	// Extract target account.
+	account := cMsg.Target
+
+	// Drop any outgoing queued AP requests to / from / targeting
+	// this account, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ActorID", account.URI)
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", account.URI)
+
+	// Drop any incoming queued client messages to / from this
+	// account, (stops processing of local origin data for acccount).
+	p.state.Workers.Client.Queue.Delete("Origin.ID", account.ID)
+	p.state.Workers.Client.Queue.Delete("Target.ID", account.ID)
+	p.state.Workers.Client.Queue.Delete("TargetURI", account.URI)
+
+	// Drop any incoming queued federator messages to this account,
+	// (stops processing of remote origin data targeting this account).
+	p.state.Workers.Federator.Queue.Delete("Receiving.ID", account.ID)
+	p.state.Workers.Federator.Queue.Delete("TargetURI", account.URI)
+
+	if err := p.federate.DeleteAccount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error federating account delete: %v", err)
 	}
 
-	if err := p.account.Delete(ctx, cMsg.TargetAccount, originID); err != nil {
+	if err := p.account.Delete(ctx, cMsg.Target, originID); err != nil {
 		log.Errorf(ctx, "error deleting account: %v", err)
 	}
 
 	return nil
 }
 
-func (p *clientAPI) ReportAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) ReportAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	report, ok := cMsg.GTSModel.(*gtsmodel.Report)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Report", cMsg.GTSModel)
@@ -719,28 +738,28 @@ func (p *clientAPI) ReportAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) MoveAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) MoveAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Redirect each local follower of
 	// OriginAccount to follow move target.
-	p.utilF.redirectFollowers(ctx, cMsg.OriginAccount, cMsg.TargetAccount)
+	p.utils.redirectFollowers(ctx, cMsg.Origin, cMsg.Target)
 
 	// At this point, we know OriginAccount has the
 	// Move set on it. Just make sure it's populated.
-	if err := p.state.DB.PopulateMove(ctx, cMsg.OriginAccount.Move); err != nil {
+	if err := p.state.DB.PopulateMove(ctx, cMsg.Origin.Move); err != nil {
 		return gtserror.Newf("error populating Move: %w", err)
 	}
 
 	// Now send the Move message out to
 	// OriginAccount's (remote) followers.
-	if err := p.federate.MoveAccount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.federate.MoveAccount(ctx, cMsg.Origin); err != nil {
 		return gtserror.Newf("error federating account move: %w", err)
 	}
 
 	// Mark the move attempt as successful.
-	cMsg.OriginAccount.Move.SucceededAt = cMsg.OriginAccount.Move.AttemptedAt
+	cMsg.Origin.Move.SucceededAt = cMsg.Origin.Move.AttemptedAt
 	if err := p.state.DB.UpdateMove(
 		ctx,
-		cMsg.OriginAccount.Move,
+		cMsg.Origin.Move,
 		"succeeded_at",
 	); err != nil {
 		return gtserror.Newf("error marking move as successful: %w", err)
@@ -749,7 +768,7 @@ func (p *clientAPI) MoveAccount(ctx context.Context, cMsg messages.FromClientAPI
 	return nil
 }
 
-func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	newUser, ok := cMsg.GTSModel.(*gtsmodel.User)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.User", cMsg.GTSModel)
@@ -772,17 +791,17 @@ func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) RejectAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) RejectAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	deniedUser, ok := cMsg.GTSModel.(*gtsmodel.DeniedUser)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.DeniedUser", cMsg.GTSModel)
 	}
 
 	// Remove the account.
-	if err := p.state.DB.DeleteAccount(ctx, cMsg.TargetAccount.ID); err != nil {
+	if err := p.state.DB.DeleteAccount(ctx, cMsg.Target.ID); err != nil {
 		log.Errorf(ctx,
 			"db error deleting account %s: %v",
-			cMsg.TargetAccount.ID, err,
+			cMsg.Target.ID, err,
 		)
 	}
 
diff --git a/internal/processing/workers/fromclientapi_test.go b/internal/processing/workers/fromclientapi_test.go
index 5e294597d..3aa0fbdad 100644
--- a/internal/processing/workers/fromclientapi_test.go
+++ b/internal/processing/workers/fromclientapi_test.go
@@ -197,11 +197,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithNotification() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -291,11 +291,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReply() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -355,11 +355,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyMuted() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -409,11 +409,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostMuted() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -467,11 +467,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -540,11 +540,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -608,11 +608,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyListRepliesPoli
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -664,11 +664,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoost() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -729,11 +729,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostNoReblogs() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -776,11 +776,11 @@ func (suite *FromClientAPITestSuite) TestProcessStatusDelete() {
 	// Process the status delete.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityDelete,
 			GTSModel:       deletedStatus,
-			OriginAccount:  deletingAccount,
+			Origin:         deletingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
diff --git a/internal/processing/workers/fromfediapi.go b/internal/processing/workers/fromfediapi.go
index 0b1106a9e..fcd5b38f2 100644
--- a/internal/processing/workers/fromfediapi.go
+++ b/internal/processing/workers/fromfediapi.go
@@ -19,13 +19,14 @@ package workers
 
 import (
 	"context"
+	"errors"
 
 	"codeberg.org/gruf/go-kv"
 	"codeberg.org/gruf/go-logger/v2/level"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation/dereferencing"
 
-	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/log"
@@ -43,34 +44,20 @@ type fediAPI struct {
 	surface  *surface
 	federate *federate
 	account  *account.Processor
-	utilF    *utilF
+	utils    *utils
 }
 
-func (p *Processor) EnqueueFediAPI(cctx context.Context, msgs ...messages.FromFediAPI) {
-	_ = p.workers.Federator.MustEnqueueCtx(cctx, func(wctx context.Context) {
-		// Copy caller ctx values to worker's.
-		wctx = gtscontext.WithValues(wctx, cctx)
-
-		// Process worker messages.
-		for _, msg := range msgs {
-			if err := p.ProcessFromFediAPI(wctx, msg); err != nil {
-				log.Errorf(wctx, "error processing fedi API message: %v", err)
-			}
-		}
-	})
-}
-
-func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Allocate new log fields slice
 	fields := make([]kv.Field, 3, 5)
 	fields[0] = kv.Field{"activityType", fMsg.APActivityType}
 	fields[1] = kv.Field{"objectType", fMsg.APObjectType}
-	fields[2] = kv.Field{"toAccount", fMsg.ReceivingAccount.Username}
+	fields[2] = kv.Field{"toAccount", fMsg.Receiving.Username}
 
-	if fMsg.APIri != nil {
+	if fMsg.APIRI != nil {
 		// An IRI was supplied, append to log
 		fields = append(fields, kv.Field{
-			"iri", fMsg.APIri,
+			"iri", fMsg.APIRI,
 		})
 	}
 
@@ -168,7 +155,7 @@ func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg messages.FromFe
 	return gtserror.Newf("unhandled: %s %s", fMsg.APActivityType, fMsg.APObjectType)
 }
 
-func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateStatus(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	var (
 		status     *gtsmodel.Status
 		statusable ap.Statusable
@@ -178,11 +165,11 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	var ok bool
 
 	switch {
-	case fMsg.APObjectModel != nil:
+	case fMsg.APObject != nil:
 		// A model was provided, extract this from message.
-		statusable, ok = fMsg.APObjectModel.(ap.Statusable)
+		statusable, ok = fMsg.APObject.(ap.Statusable)
 		if !ok {
-			return gtserror.Newf("cannot cast %T -> ap.Statusable", fMsg.APObjectModel)
+			return gtserror.Newf("cannot cast %T -> ap.Statusable", fMsg.APObject)
 		}
 
 		// Create bare-bones model to pass
@@ -196,7 +183,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 		// statusable model, which it will use to further flesh out
 		// the bare bones model and insert it into the database.
 		status, statusable, err = p.federate.RefreshStatus(ctx,
-			fMsg.ReceivingAccount.Username,
+			fMsg.Receiving.Username,
 			bareStatus,
 			statusable,
 			// Force refresh within 5min window.
@@ -206,15 +193,15 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 			return gtserror.Newf("error processing new status %s: %w", bareStatus.URI, err)
 		}
 
-	case fMsg.APIri != nil:
+	case fMsg.APIRI != nil:
 		// Model was not set, deref with IRI (this is a forward).
 		// This will also cause the status to be inserted into the db.
 		status, statusable, err = p.federate.GetStatusByURI(ctx,
-			fMsg.ReceivingAccount.Username,
-			fMsg.APIri,
+			fMsg.Receiving.Username,
+			fMsg.APIRI,
 		)
 		if err != nil {
-			return gtserror.Newf("error dereferencing forwarded status %s: %w", fMsg.APIri, err)
+			return gtserror.Newf("error dereferencing forwarded status %s: %w", fMsg.APIRI, err)
 		}
 
 	default:
@@ -230,7 +217,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.incrementStatusesCount(ctx, fMsg.RequestingAccount, status); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, fMsg.Requesting, status); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -248,7 +235,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	return nil
 }
 
-func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Cast poll vote type from the worker message.
 	vote, ok := fMsg.GTSModel.(*gtsmodel.PollVote)
 	if !ok {
@@ -293,7 +280,7 @@ func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg messages.FromFediAPI)
 	return nil
 }
 
-func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	followRequest, ok := fMsg.GTSModel.(*gtsmodel.FollowRequest)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", fMsg.GTSModel)
@@ -310,7 +297,7 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI
 		}
 
 		// And update stats for the local account.
-		if err := p.utilF.incrementFollowRequestsCount(ctx, fMsg.ReceivingAccount); err != nil {
+		if err := p.utils.incrementFollowRequestsCount(ctx, fMsg.Receiving); err != nil {
 			log.Errorf(ctx, "error updating account stats: %v", err)
 		}
 
@@ -330,12 +317,12 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI
 	}
 
 	// Update stats for the local account.
-	if err := p.utilF.incrementFollowersCount(ctx, fMsg.ReceivingAccount); err != nil {
+	if err := p.utils.incrementFollowersCount(ctx, fMsg.Receiving); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.incrementFollowingCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.incrementFollowingCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -350,7 +337,7 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI
 	return nil
 }
 
-func (p *fediAPI) CreateLike(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateLike(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	fave, ok := fMsg.GTSModel.(*gtsmodel.StatusFave)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", fMsg.GTSModel)
@@ -372,7 +359,7 @@ func (p *fediAPI) CreateLike(ctx context.Context, fMsg messages.FromFediAPI) err
 	return nil
 }
 
-func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	boost, ok := fMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", fMsg.GTSModel)
@@ -386,7 +373,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI)
 	boost, err = p.federate.EnrichAnnounce(
 		ctx,
 		boost,
-		fMsg.ReceivingAccount.Username,
+		fMsg.Receiving.Username,
 	)
 	if err != nil {
 		if gtserror.IsUnretrievable(err) {
@@ -400,7 +387,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI)
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.incrementStatusesCount(ctx, fMsg.RequestingAccount, boost); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, fMsg.Requesting, boost); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -420,7 +407,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI)
 	return nil
 }
 
-func (p *fediAPI) CreateBlock(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateBlock(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	block, ok := fMsg.GTSModel.(*gtsmodel.Block)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Block", fMsg.GTSModel)
@@ -499,7 +486,7 @@ func (p *fediAPI) CreateBlock(ctx context.Context, fMsg messages.FromFediAPI) er
 	return nil
 }
 
-func (p *fediAPI) CreateFlag(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateFlag(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	incomingReport, ok := fMsg.GTSModel.(*gtsmodel.Report)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Report", fMsg.GTSModel)
@@ -515,7 +502,7 @@ func (p *fediAPI) CreateFlag(ctx context.Context, fMsg messages.FromFediAPI) err
 	return nil
 }
 
-func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Parse the old/existing account model.
 	account, ok := fMsg.GTSModel.(*gtsmodel.Account)
 	if !ok {
@@ -523,15 +510,15 @@ func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI)
 	}
 
 	// Because this was an Update, the new Accountable should be set on the message.
-	apubAcc, ok := fMsg.APObjectModel.(ap.Accountable)
+	apubAcc, ok := fMsg.APObject.(ap.Accountable)
 	if !ok {
-		return gtserror.Newf("cannot cast %T -> ap.Accountable", fMsg.APObjectModel)
+		return gtserror.Newf("cannot cast %T -> ap.Accountable", fMsg.APObject)
 	}
 
 	// Fetch up-to-date bio, avatar, header, etc.
 	_, _, err := p.federate.RefreshAccount(
 		ctx,
-		fMsg.ReceivingAccount.Username,
+		fMsg.Receiving.Username,
 		account,
 		apubAcc,
 		// Force refresh within 5min window.
@@ -544,25 +531,25 @@ func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI)
 	return nil
 }
 
-func (p *fediAPI) AcceptFollow(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) AcceptFollow(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Update stats for the remote account.
-	if err := p.utilF.decrementFollowRequestsCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.decrementFollowRequestsCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
-	if err := p.utilF.incrementFollowersCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.incrementFollowersCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the local account.
-	if err := p.utilF.incrementFollowingCount(ctx, fMsg.ReceivingAccount); err != nil {
+	if err := p.utils.incrementFollowingCount(ctx, fMsg.Receiving); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	return nil
 }
 
-func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Cast the existing Status model attached to msg.
 	existing, ok := fMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
@@ -570,12 +557,12 @@ func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	}
 
 	// Cast the updated ActivityPub statusable object .
-	apStatus, _ := fMsg.APObjectModel.(ap.Statusable)
+	apStatus, _ := fMsg.APObject.(ap.Statusable)
 
 	// Fetch up-to-date attach status attachments, etc.
 	status, _, err := p.federate.RefreshStatus(
 		ctx,
-		fMsg.ReceivingAccount.Username,
+		fMsg.Receiving.Username,
 		existing,
 		apStatus,
 		// Force refresh within 5min window.
@@ -605,7 +592,7 @@ func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	return nil
 }
 
-func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Delete attachments from this status, since this request
 	// comes from the federating API, and there's no way the
 	// poster can do a delete + redraft for it on our instance.
@@ -616,12 +603,34 @@ func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", fMsg.GTSModel)
 	}
 
-	if err := p.utilF.wipeStatus(ctx, status, deleteAttachments); err != nil {
+	// Try to populate status structs if possible,
+	// in order to more thoroughly remove them.
+	if err := p.state.DB.PopulateStatus(
+		ctx, status,
+	); err != nil && !errors.Is(err, db.ErrNoEntries) {
+		return gtserror.Newf("db error populating status: %w", err)
+	}
+
+	// Drop any outgoing queued AP requests about / targeting
+	// this status, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", status.URI)
+
+	// Drop any incoming queued client messages about / targeting
+	// status, (stops processing of local origin data for status).
+	p.state.Workers.Client.Queue.Delete("TargetURI", status.URI)
+
+	// Drop any incoming queued federator messages targeting status,
+	// (stops processing of remote origin data targeting this status).
+	p.state.Workers.Federator.Queue.Delete("TargetURI", status.URI)
+
+	// First perform the actual status deletion.
+	if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil {
 		log.Errorf(ctx, "error wiping status: %v", err)
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.decrementStatusesCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.decrementStatusesCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -634,12 +643,28 @@ func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	return nil
 }
 
-func (p *fediAPI) DeleteAccount(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) DeleteAccount(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	account, ok := fMsg.GTSModel.(*gtsmodel.Account)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Account", fMsg.GTSModel)
 	}
 
+	// Drop any outgoing queued AP requests to / from / targeting
+	// this account, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", account.URI)
+
+	// Drop any incoming queued client messages to / from this
+	// account, (stops processing of local origin data for acccount).
+	p.state.Workers.Client.Queue.Delete("Target.ID", account.ID)
+	p.state.Workers.Client.Queue.Delete("TargetURI", account.URI)
+
+	// Drop any incoming queued federator messages to this account,
+	// (stops processing of remote origin data targeting this account).
+	p.state.Workers.Federator.Queue.Delete("Requesting.ID", account.ID)
+	p.state.Workers.Federator.Queue.Delete("TargetURI", account.URI)
+
+	// First perform the actual account deletion.
 	if err := p.account.Delete(ctx, account, account.ID); err != nil {
 		log.Errorf(ctx, "error deleting account: %v", err)
 	}
diff --git a/internal/processing/workers/fromfediapi_move.go b/internal/processing/workers/fromfediapi_move.go
index 0188a5d14..d1e43c0c7 100644
--- a/internal/processing/workers/fromfediapi_move.go
+++ b/internal/processing/workers/fromfediapi_move.go
@@ -220,10 +220,7 @@ func (p *fediAPI) GetOrCreateMove(
 //	APActivityType:   "Move"
 //	GTSModel:         stub *gtsmodel.Move.
 //	ReceivingAccount: Account of inbox owner receiving the Move.
-func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) error {
-	// The account who received the Move message.
-	receiver := fMsg.ReceivingAccount
-
+func (p *fediAPI) MoveAccount(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// *gtsmodel.Move activity.
 	stubMove, ok := fMsg.GTSModel.(*gtsmodel.Move)
 	if !ok {
@@ -236,7 +233,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 	// Move origin and target info.
 	var (
 		originAcctURIStr = stubMove.OriginURI
-		originAcct       = fMsg.RequestingAccount
+		originAcct       = fMsg.Requesting
 		targetAcctURIStr = stubMove.TargetURI
 		targetAcctURI    = stubMove.Target
 	)
@@ -308,7 +305,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 	// Account to which the Move is taking place.
 	targetAcct, targetAcctable, err := p.federate.GetAccountByURI(
 		ctx,
-		receiver.Username,
+		fMsg.Receiving.Username,
 		targetAcctURI,
 	)
 	if err != nil {
@@ -340,7 +337,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 		// Force refresh Move target account
 		// to ensure we have up-to-date version.
 		targetAcct, _, err = p.federate.RefreshAccount(ctx,
-			receiver.Username,
+			fMsg.Receiving.Username,
 			targetAcct,
 			targetAcctable,
 			dereferencing.Freshest,
@@ -379,7 +376,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 
 	// Transfer originAcct's followers
 	// on this instance to targetAcct.
-	redirectOK := p.utilF.redirectFollowers(
+	redirectOK := p.utils.redirectFollowers(
 		ctx,
 		originAcct,
 		targetAcct,
diff --git a/internal/processing/workers/fromfediapi_test.go b/internal/processing/workers/fromfediapi_test.go
index eb3d73e0c..b8acca94d 100644
--- a/internal/processing/workers/fromfediapi_test.go
+++ b/internal/processing/workers/fromfediapi_test.go
@@ -54,12 +54,12 @@ func (suite *FromFediAPITestSuite) TestProcessFederationAnnounce() {
 	announceStatus.Account = boostingAccount
 	announceStatus.Visibility = boostedStatus.Visibility
 
-	err := suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{
-		APObjectType:      ap.ActivityAnnounce,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          announceStatus,
-		ReceivingAccount:  suite.testAccounts["local_account_1"],
-		RequestingAccount: boostingAccount,
+	err := suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{
+		APObjectType:   ap.ActivityAnnounce,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       announceStatus,
+		Receiving:      suite.testAccounts["local_account_1"],
+		Requesting:     boostingAccount,
 	})
 	suite.NoError(err)
 
@@ -115,12 +115,12 @@ func (suite *FromFediAPITestSuite) TestProcessReplyMention() {
 	suite.NoError(errWithCode)
 
 	// Send the replied status off to the fedi worker to be further processed.
-	err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{
-		APObjectType:      ap.ObjectNote,
-		APActivityType:    ap.ActivityCreate,
-		APObjectModel:     replyingStatusable,
-		ReceivingAccount:  repliedAccount,
-		RequestingAccount: replyingAccount,
+	err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{
+		APObjectType:   ap.ObjectNote,
+		APActivityType: ap.ActivityCreate,
+		APObject:       replyingStatusable,
+		Receiving:      repliedAccount,
+		Requesting:     replyingAccount,
 	})
 	suite.NoError(err)
 
@@ -179,12 +179,12 @@ func (suite *FromFediAPITestSuite) TestProcessFave() {
 	err := suite.db.Put(context.Background(), fave)
 	suite.NoError(err)
 
-	err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{
-		APObjectType:      ap.ActivityLike,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          fave,
-		ReceivingAccount:  favedAccount,
-		RequestingAccount: favingAccount,
+	err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{
+		APObjectType:   ap.ActivityLike,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       fave,
+		Receiving:      favedAccount,
+		Requesting:     favingAccount,
 	})
 	suite.NoError(err)
 
@@ -249,12 +249,12 @@ func (suite *FromFediAPITestSuite) TestProcessFaveWithDifferentReceivingAccount(
 	err := suite.db.Put(context.Background(), fave)
 	suite.NoError(err)
 
-	err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{
-		APObjectType:      ap.ActivityLike,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          fave,
-		ReceivingAccount:  receivingAccount,
-		RequestingAccount: favingAccount,
+	err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{
+		APObjectType:   ap.ActivityLike,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       fave,
+		Receiving:      receivingAccount,
+		Requesting:     favingAccount,
 	})
 	suite.NoError(err)
 
@@ -321,12 +321,12 @@ func (suite *FromFediAPITestSuite) TestProcessAccountDelete() {
 	suite.NoError(err)
 
 	// now they are mufos!
-	err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ObjectProfile,
-		APActivityType:    ap.ActivityDelete,
-		GTSModel:          deletedAccount,
-		ReceivingAccount:  receivingAccount,
-		RequestingAccount: deletedAccount,
+	err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{
+		APObjectType:   ap.ObjectProfile,
+		APActivityType: ap.ActivityDelete,
+		GTSModel:       deletedAccount,
+		Receiving:      receivingAccount,
+		Requesting:     deletedAccount,
 	})
 	suite.NoError(err)
 
@@ -402,12 +402,12 @@ func (suite *FromFediAPITestSuite) TestProcessFollowRequestLocked() {
 	err := suite.db.Put(ctx, satanFollowRequestTurtle)
 	suite.NoError(err)
 
-	err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ActivityFollow,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          satanFollowRequestTurtle,
-		ReceivingAccount:  targetAccount,
-		RequestingAccount: originAccount,
+	err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{
+		APObjectType:   ap.ActivityFollow,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       satanFollowRequestTurtle,
+		Receiving:      targetAccount,
+		Requesting:     originAccount,
 	})
 	suite.NoError(err)
 
@@ -456,12 +456,12 @@ func (suite *FromFediAPITestSuite) TestProcessFollowRequestUnlocked() {
 	err := suite.db.Put(ctx, satanFollowRequestTurtle)
 	suite.NoError(err)
 
-	err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ActivityFollow,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          satanFollowRequestTurtle,
-		ReceivingAccount:  targetAccount,
-		RequestingAccount: originAccount,
+	err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{
+		APObjectType:   ap.ActivityFollow,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       satanFollowRequestTurtle,
+		Receiving:      targetAccount,
+		Requesting:     originAccount,
 	})
 	suite.NoError(err)
 
@@ -532,13 +532,13 @@ func (suite *FromFediAPITestSuite) TestCreateStatusFromIRI() {
 	receivingAccount := suite.testAccounts["local_account_1"]
 	statusCreator := suite.testAccounts["remote_account_2"]
 
-	err := suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ObjectNote,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          nil, // gtsmodel is nil because this is a forwarded status -- we want to dereference it using the iri
-		ReceivingAccount:  receivingAccount,
-		RequestingAccount: statusCreator,
-		APIri:             testrig.URLMustParse("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1"),
+	err := suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{
+		APObjectType:   ap.ObjectNote,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       nil, // gtsmodel is nil because this is a forwarded status -- we want to dereference it using the iri
+		Receiving:      receivingAccount,
+		Requesting:     statusCreator,
+		APIRI:          testrig.URLMustParse("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1"),
 	})
 	suite.NoError(err)
 
@@ -585,7 +585,7 @@ func (suite *FromFediAPITestSuite) TestMoveAccount() {
 	}
 
 	// Process the Move.
-	err := suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{
+	err := suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{
 		APObjectType:   ap.ObjectProfile,
 		APActivityType: ap.ActivityMove,
 		GTSModel: &gtsmodel.Move{
@@ -595,8 +595,8 @@ func (suite *FromFediAPITestSuite) TestMoveAccount() {
 			Target:    testrig.URLMustParse(targetAcct.URI),
 			URI:       "https://fossbros-anonymous.io/users/foss_satan/moves/01HRA064871MR8HGVSAFJ333GM",
 		},
-		ReceivingAccount:  receivingAcct,
-		RequestingAccount: requestingAcct,
+		Receiving:  receivingAcct,
+		Requesting: requestingAcct,
 	})
 	suite.NoError(err)
 
diff --git a/internal/processing/workers/util.go b/internal/processing/workers/util.go
index cd936f428..a01982e1a 100644
--- a/internal/processing/workers/util.go
+++ b/internal/processing/workers/util.go
@@ -32,9 +32,9 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 )
 
-// utilF wraps util functions used by both
+// util provides util functions used by both
 // the fromClientAPI and fromFediAPI functions.
-type utilF struct {
+type utils struct {
 	state   *state.State
 	media   *media.Processor
 	account *account.Processor
@@ -45,7 +45,7 @@ type utilF struct {
 // used to totally delete a status + all
 // its attachments, notifications, boosts,
 // and timeline entries.
-func (u *utilF) wipeStatus(
+func (u *utils) wipeStatus(
 	ctx context.Context,
 	statusToDelete *gtsmodel.Status,
 	deleteAttachments bool,
@@ -152,7 +152,7 @@ func (u *utilF) wipeStatus(
 // already, and the Move must be valid.
 //
 // Return bool will be true if all goes OK.
-func (u *utilF) redirectFollowers(
+func (u *utils) redirectFollowers(
 	ctx context.Context,
 	originAcct *gtsmodel.Account,
 	targetAcct *gtsmodel.Account,
@@ -239,7 +239,7 @@ func (u *utilF) redirectFollowers(
 	return true
 }
 
-func (u *utilF) incrementStatusesCount(
+func (u *utils) incrementStatusesCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 	status *gtsmodel.Status,
@@ -271,7 +271,7 @@ func (u *utilF) incrementStatusesCount(
 	return nil
 }
 
-func (u *utilF) decrementStatusesCount(
+func (u *utils) decrementStatusesCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -305,7 +305,7 @@ func (u *utilF) decrementStatusesCount(
 	return nil
 }
 
-func (u *utilF) incrementFollowersCount(
+func (u *utils) incrementFollowersCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -334,7 +334,7 @@ func (u *utilF) incrementFollowersCount(
 	return nil
 }
 
-func (u *utilF) decrementFollowersCount(
+func (u *utils) decrementFollowersCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -368,7 +368,7 @@ func (u *utilF) decrementFollowersCount(
 	return nil
 }
 
-func (u *utilF) incrementFollowingCount(
+func (u *utils) incrementFollowingCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -397,7 +397,7 @@ func (u *utilF) incrementFollowingCount(
 	return nil
 }
 
-func (u *utilF) decrementFollowingCount(
+func (u *utils) decrementFollowingCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -431,7 +431,7 @@ func (u *utilF) decrementFollowingCount(
 	return nil
 }
 
-func (u *utilF) incrementFollowRequestsCount(
+func (u *utils) incrementFollowRequestsCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -460,7 +460,7 @@ func (u *utilF) incrementFollowRequestsCount(
 	return nil
 }
 
-func (u *utilF) decrementFollowRequestsCount(
+func (u *utils) decrementFollowRequestsCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
diff --git a/internal/processing/workers/workers.go b/internal/processing/workers/workers.go
index 8488e501c..1159b61a5 100644
--- a/internal/processing/workers/workers.go
+++ b/internal/processing/workers/workers.go
@@ -30,9 +30,9 @@ import (
 )
 
 type Processor struct {
+	clientAPI clientAPI
+	fediAPI   fediAPI
 	workers   *workers.Workers
-	clientAPI *clientAPI
-	fediAPI   *fediAPI
 }
 
 func New(
@@ -45,6 +45,14 @@ func New(
 	media *media.Processor,
 	stream *stream.Processor,
 ) Processor {
+	// Init federate logic
+	// wrapper struct.
+	federate := &federate{
+		Federator: federator,
+		state:     state,
+		converter: converter,
+	}
+
 	// Init surface logic
 	// wrapper struct.
 	surface := &surface{
@@ -55,16 +63,8 @@ func New(
 		emailSender: emailSender,
 	}
 
-	// Init federate logic
-	// wrapper struct.
-	federate := &federate{
-		Federator: federator,
-		state:     state,
-		converter: converter,
-	}
-
 	// Init shared util funcs.
-	utilF := &utilF{
+	utils := &utils{
 		state:   state,
 		media:   media,
 		account: account,
@@ -73,20 +73,20 @@ func New(
 
 	return Processor{
 		workers: &state.Workers,
-		clientAPI: &clientAPI{
+		clientAPI: clientAPI{
 			state:     state,
 			converter: converter,
 			surface:   surface,
 			federate:  federate,
 			account:   account,
-			utilF:     utilF,
+			utils:     utils,
 		},
-		fediAPI: &fediAPI{
+		fediAPI: fediAPI{
 			state:    state,
 			surface:  surface,
 			federate: federate,
 			account:  account,
-			utilF:    utilF,
+			utils:    utils,
 		},
 	}
 }
diff --git a/internal/processing/workers/workers_test.go b/internal/processing/workers/workers_test.go
index 3ab19d38d..af1428e64 100644
--- a/internal/processing/workers/workers_test.go
+++ b/internal/processing/workers/workers_test.go
@@ -127,9 +127,6 @@ func (suite *WorkersTestSuite) SetupTest() {
 	suite.processor = processing.NewProcessor(cleaner.New(&suite.state), suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, &suite.state, suite.emailSender)
 	testrig.StartWorkers(&suite.state, suite.processor.Workers())
 
-	suite.state.Workers.EnqueueClientAPI = suite.processor.Workers().EnqueueClientAPI
-	suite.state.Workers.EnqueueFediAPI = suite.processor.Workers().EnqueueFediAPI
-
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 	testrig.StandardStorageSetup(suite.storage, "../../../testrig/media")
 }