[feature] Store admin actions in the db, prevent conflicting actions (#2167)

2025-06-05 21:59:39 +02:00 · 2023-09-04 15:55:17 +02:00
parent 4f3daeebcb
commit 3ed1ca68e5
23 changed files with 1393 additions and 272 deletions
--- a/internal/processing/admin/account.go
+++ b/internal/processing/admin/account.go
@@ -29,36 +29,74 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/messages"
 )

-func (p *Processor) AccountAction(ctx context.Context, account *gtsmodel.Account, form *apimodel.AdminAccountActionRequest) gtserror.WithCode {
-	targetAccount, err := p.state.DB.GetAccountByID(ctx, form.TargetAccountID)
+func (p *Processor) AccountAction(
+	ctx context.Context,
+	adminAcct *gtsmodel.Account,
+	request *apimodel.AdminActionRequest,
+) (string, gtserror.WithCode) {
+	targetAcct, err := p.state.DB.GetAccountByID(ctx, request.TargetID)
 	if err != nil {
-		return gtserror.NewErrorInternalError(err)
+		err := gtserror.Newf("db error getting target account: %w", err)
+		return "", gtserror.NewErrorInternalError(err)
 	}

-	adminAction := &gtsmodel.AdminAccountAction{
-		ID:              id.NewULID(),
-		AccountID:       account.ID,
-		TargetAccountID: targetAccount.ID,
-		Text:            form.Text,
-	}
+	switch gtsmodel.NewAdminActionType(request.Type) {
+	case gtsmodel.AdminActionSuspend:
+		return p.accountActionSuspend(ctx, adminAcct, targetAcct, request.Text)

-	switch form.Type {
-	case string(gtsmodel.AdminActionSuspend):
-		adminAction.Type = gtsmodel.AdminActionSuspend
-		// pass the account delete through the client api channel for processing
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
-			APObjectType:   ap.ActorPerson,
-			APActivityType: ap.ActivityDelete,
-			OriginAccount:  account,
-			TargetAccount:  targetAccount,
-		})
 	default:
-		return gtserror.NewErrorBadRequest(fmt.Errorf("admin action type %s is not supported for this endpoint", form.Type))
-	}
+		// TODO: add more types to this slice when adding
+		//       more types to the switch statement above.
+		supportedTypes := []string{
+			gtsmodel.AdminActionSuspend.String(),
+		}

-	if err := p.state.DB.Put(ctx, adminAction); err != nil {
-		return gtserror.NewErrorInternalError(err)
-	}
+		err := fmt.Errorf(
+			"admin action type %s is not supported for this endpoint, "+
+				"currently supported types are: %q",
+			request.Type, supportedTypes)

-	return nil
+		return "", gtserror.NewErrorBadRequest(err, err.Error())
+	}
+}
+
+func (p *Processor) accountActionSuspend(
+	ctx context.Context,
+	adminAcct *gtsmodel.Account,
+	targetAcct *gtsmodel.Account,
+	text string,
+) (string, gtserror.WithCode) {
+	actionID := id.NewULID()
+
+	errWithCode := p.actions.Run(
+		ctx,
+		&gtsmodel.AdminAction{
+			ID:             actionID,
+			TargetCategory: gtsmodel.AdminActionCategoryAccount,
+			TargetID:       targetAcct.ID,
+			Target:         targetAcct,
+			Type:           gtsmodel.AdminActionSuspend,
+			AccountID:      adminAcct.ID,
+			Text:           text,
+		},
+		func(ctx context.Context) gtserror.MultiError {
+			if err := p.state.Workers.ProcessFromClientAPI(
+				ctx,
+				messages.FromClientAPI{
+					APObjectType:   ap.ActorPerson,
+					APActivityType: ap.ActivityDelete,
+					OriginAccount:  adminAcct,
+					TargetAccount:  targetAcct,
+				},
+			); err != nil {
+				errs := gtserror.NewMultiError(1)
+				errs.Append(err)
+				return errs
+			}
+
+			return nil
+		},
+	)
+
+	return actionID, errWithCode
 }
--- a/internal/processing/admin/account_test.go
+++ b/internal/processing/admin/account_test.go
@@ -0,0 +1,103 @@
+// GoToSocial
+// Copyright (C) GoToSocial Authors admin@gotosocial.org
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package admin_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type AccountTestSuite struct {
+	AdminStandardTestSuite
+}
+
+func (suite *AccountTestSuite) TestAccountActionSuspend() {
+	var (
+		ctx       = context.Background()
+		adminAcct = suite.testAccounts["admin_account"]
+		request   = &apimodel.AdminActionRequest{
+			Category: gtsmodel.AdminActionCategoryAccount.String(),
+			Type:     gtsmodel.AdminActionSuspend.String(),
+			Text:     "stinky",
+			TargetID: suite.testAccounts["local_account_1"].ID,
+		}
+	)
+
+	actionID, errWithCode := suite.adminProcessor.AccountAction(
+		ctx,
+		adminAcct,
+		request,
+	)
+	suite.NoError(errWithCode)
+	suite.NotEmpty(actionID)
+
+	// Wait for action to finish.
+	if !testrig.WaitFor(func() bool {
+		return suite.adminProcessor.Actions().TotalRunning() == 0
+	}) {
+		suite.FailNow("timed out waiting for admin action(s) to finish")
+	}
+
+	// Ensure action marked as
+	// completed in the database.
+	adminAction, err := suite.db.GetAdminAction(ctx, actionID)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.NotZero(adminAction.CompletedAt)
+	suite.Empty(adminAction.Errors)
+
+	// Ensure target account suspended.
+	targetAcct, err := suite.db.GetAccountByID(ctx, request.TargetID)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.NotZero(targetAcct.SuspendedAt)
+}
+
+func (suite *AccountTestSuite) TestAccountActionUnsupported() {
+	var (
+		ctx       = context.Background()
+		adminAcct = suite.testAccounts["admin_account"]
+		request   = &apimodel.AdminActionRequest{
+			Category: gtsmodel.AdminActionCategoryAccount.String(),
+			Type:     "pee pee poo poo",
+			Text:     "stinky",
+			TargetID: suite.testAccounts["local_account_1"].ID,
+		}
+	)
+
+	actionID, errWithCode := suite.adminProcessor.AccountAction(
+		ctx,
+		adminAcct,
+		request,
+	)
+	suite.EqualError(errWithCode, "admin action type pee pee poo poo is not supported for this endpoint, currently supported types are: [\"suspend\"]")
+	suite.Empty(actionID)
+}
+
+func TestAccountTestSuite(t *testing.T) {
+	suite.Run(t, new(AccountTestSuite))
+}
--- a/internal/processing/admin/actions.go
+++ b/internal/processing/admin/actions.go
@@ -0,0 +1,159 @@
+// GoToSocial
+// Copyright (C) GoToSocial Authors admin@gotosocial.org
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package admin
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/log"
+	"github.com/superseriousbusiness/gotosocial/internal/state"
+	"golang.org/x/exp/slices"
+)
+
+func errActionConflict(action *gtsmodel.AdminAction) gtserror.WithCode {
+	err := gtserror.NewfAt(
+		4, // Include caller's function name.
+		"an action (%s) is currently running (duration %s) which conflicts with the attempted action",
+		action.Key(), time.Since(action.CreatedAt),
+	)
+
+	const help = "wait until this action is complete and try again"
+	return gtserror.NewErrorConflict(err, err.Error(), help)
+}
+
+type Actions struct {
+	r     map[string]*gtsmodel.AdminAction
+	state *state.State
+
+	// Not embedded struct,
+	// to shield from access
+	// by outside packages.
+	m sync.Mutex
+}
+
+// Run runs the given admin action by executing the supplied function.
+//
+// Run handles locking, action insertion and updating, so you don't have to!
+//
+// If an action is already running which overlaps/conflicts with the
+// given action, an ErrorWithCode 409 will be returned.
+//
+// If execution of the provided function returns errors, the errors
+// will be updated on the provided admin action in the database.
+func (a *Actions) Run(
+	ctx context.Context,
+	action *gtsmodel.AdminAction,
+	f func(context.Context) gtserror.MultiError,
+) gtserror.WithCode {
+	actionKey := action.Key()
+
+	// LOCK THE MAP HERE, since we're
+	// going to do some operations on it.
+	a.m.Lock()
+
+	// Bail if an action with
+	// this key is already running.
+	running, ok := a.r[actionKey]
+	if ok {
+		a.m.Unlock()
+		return errActionConflict(running)
+	}
+
+	// Action with this key not
+	// yet running, create it.
+	if err := a.state.DB.PutAdminAction(ctx, action); err != nil {
+		err = gtserror.Newf("db error putting admin action %s: %w", actionKey, err)
+
+		// Don't store in map
+		// if there's an error.
+		a.m.Unlock()
+		return gtserror.NewErrorInternalError(err)
+	}
+
+	// Action was inserted,
+	// store in map.
+	a.r[actionKey] = action
+
+	// UNLOCK THE MAP HERE, since
+	// 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) {
+		// Run the thing and collect errors.
+		if errs := f(ctx); errs != nil {
+			action.Errors = make([]string, 0, len(errs))
+			for _, err := range errs {
+				action.Errors = append(action.Errors, err.Error())
+			}
+		}
+
+		// Action is no longer running:
+		// remove from running map.
+		a.m.Lock()
+		delete(a.r, actionKey)
+		a.m.Unlock()
+
+		// Mark as completed in the db,
+		// storing errors for later review.
+		action.CompletedAt = time.Now()
+		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
+}
+
+// GetRunning sounds like a threat, but it actually just
+// returns all of the currently running actions held by
+// the Actions struct, ordered by ID descending.
+func (a *Actions) GetRunning() []*gtsmodel.AdminAction {
+	a.m.Lock()
+	defer a.m.Unlock()
+
+	// Assemble all currently running actions.
+	running := make([]*gtsmodel.AdminAction, 0, len(a.r))
+	for _, action := range a.r {
+		running = append(running, action)
+	}
+
+	// Order by ID descending (creation date).
+	slices.SortFunc(
+		running,
+		func(a *gtsmodel.AdminAction, b *gtsmodel.AdminAction) bool {
+			return a.ID > b.ID
+		},
+	)
+
+	return running
+}
+
+// TotalRunning is a sequel to the classic
+// 1972 environmental-themed science fiction
+// film Silent Running, starring Bruce Dern.
+func (a *Actions) TotalRunning() int {
+	a.m.Lock()
+	defer a.m.Unlock()
+
+	return len(a.r)
+}
--- a/internal/processing/admin/actions_test.go
+++ b/internal/processing/admin/actions_test.go
@@ -0,0 +1,162 @@
+// GoToSocial
+// Copyright (C) GoToSocial Authors admin@gotosocial.org
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package admin_test
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/suite"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
+	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/id"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type ActionsTestSuite struct {
+	AdminStandardTestSuite
+}
+
+func (suite *ActionsTestSuite) TestActionOverlap() {
+	ctx := context.Background()
+
+	// Suspend account.
+	action1 := &gtsmodel.AdminAction{
+		ID:             id.NewULID(),
+		TargetCategory: gtsmodel.AdminActionCategoryAccount,
+		TargetID:       "01H90S1CXQ97J9625C5YBXZWGT",
+		Type:           gtsmodel.AdminActionSuspend,
+		AccountID:      "01H90S1ZZXP4N74H4A9RVW1MRP",
+	}
+	key1 := action1.Key()
+	suite.Equal("account/01H90S1CXQ97J9625C5YBXZWGT", key1)
+
+	// Unsuspend account.
+	action2 := &gtsmodel.AdminAction{
+		ID:             id.NewULID(),
+		TargetCategory: gtsmodel.AdminActionCategoryAccount,
+		TargetID:       "01H90S1CXQ97J9625C5YBXZWGT",
+		Type:           gtsmodel.AdminActionUnsuspend,
+		AccountID:      "01H90S1ZZXP4N74H4A9RVW1MRP",
+	}
+	key2 := action2.Key()
+	suite.Equal("account/01H90S1CXQ97J9625C5YBXZWGT", key2)
+
+	errWithCode := suite.adminProcessor.Actions().Run(
+		ctx,
+		action1,
+		func(ctx context.Context) gtserror.MultiError {
+			// Noop, just sleep (mood).
+			time.Sleep(3 * time.Second)
+			return nil
+		},
+	)
+	suite.NoError(errWithCode)
+
+	// While first action is sleeping, try to
+	// process another with the same key.
+	errWithCode = suite.adminProcessor.Actions().Run(
+		ctx,
+		action2,
+		func(ctx context.Context) gtserror.MultiError {
+			return nil
+		},
+	)
+	if errWithCode == nil {
+		suite.FailNow("expected error with code, but error was nil")
+	}
+
+	// Code should be 409.
+	suite.Equal(http.StatusConflict, errWithCode.Code())
+
+	// Wait for action to finish.
+	if !testrig.WaitFor(func() bool {
+		return suite.adminProcessor.Actions().TotalRunning() == 0
+	}) {
+		suite.FailNow("timed out waiting for admin action(s) to finish")
+	}
+
+	// Try again.
+	errWithCode = suite.adminProcessor.Actions().Run(
+		ctx,
+		action2,
+		func(ctx context.Context) gtserror.MultiError {
+			return nil
+		},
+	)
+	suite.NoError(errWithCode)
+
+	// Wait for action to finish.
+	if !testrig.WaitFor(func() bool {
+		return suite.adminProcessor.Actions().TotalRunning() == 0
+	}) {
+		suite.FailNow("timed out waiting for admin action(s) to finish")
+	}
+}
+
+func (suite *ActionsTestSuite) TestActionWithErrors() {
+	ctx := context.Background()
+
+	// Suspend a domain.
+	action := &gtsmodel.AdminAction{
+		ID:             id.NewULID(),
+		TargetCategory: gtsmodel.AdminActionCategoryDomain,
+		TargetID:       "example.org",
+		Type:           gtsmodel.AdminActionSuspend,
+		AccountID:      "01H90S1ZZXP4N74H4A9RVW1MRP",
+	}
+
+	errWithCode := suite.adminProcessor.Actions().Run(
+		ctx,
+		action,
+		func(ctx context.Context) gtserror.MultiError {
+			// Noop, just return some errs.
+			return gtserror.MultiError{
+				db.ErrNoEntries,
+				errors.New("fucky wucky"),
+			}
+		},
+	)
+	suite.NoError(errWithCode)
+
+	// Wait for action to finish.
+	if !testrig.WaitFor(func() bool {
+		return suite.adminProcessor.Actions().TotalRunning() == 0
+	}) {
+		suite.FailNow("timed out waiting for admin action(s) to finish")
+	}
+
+	// Get action from the db.
+	dbAction, err := suite.db.GetAdminAction(ctx, action.ID)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.EqualValues([]string{
+		"sql: no rows in result set",
+		"fucky wucky",
+	}, dbAction.Errors)
+}
+
+func TestActionsTestSuite(t *testing.T) {
+	suite.Run(t, new(ActionsTestSuite))
+}
--- a/internal/processing/admin/admin.go
+++ b/internal/processing/admin/admin.go
@@ -20,6 +20,7 @@ package admin
 import (
 	"github.com/superseriousbusiness/gotosocial/internal/cleaner"
 	"github.com/superseriousbusiness/gotosocial/internal/email"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/media"
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/transport"
@@ -33,6 +34,14 @@ type Processor struct {
 	mediaManager        *media.Manager
 	transportController transport.Controller
 	emailSender         email.Sender
+
+	// admin Actions currently
+	// undergoing processing
+	actions *Actions
+}
+
+func (p *Processor) Actions() *Actions {
+	return p.actions
 }

 // New returns a new admin processor.
@@ -44,5 +53,10 @@ func New(state *state.State, tc typeutils.TypeConverter, mediaManager *media.Man
 		mediaManager:        mediaManager,
 		transportController: transportController,
 		emailSender:         emailSender,
+
+		actions: &Actions{
+			r:     make(map[string]*gtsmodel.AdminAction),
+			state: state,
+		},
 	}
 }
--- a/internal/processing/admin/admin_test.go
+++ b/internal/processing/admin/admin_test.go
@@ -0,0 +1,127 @@
+// GoToSocial
+// Copyright (C) GoToSocial Authors admin@gotosocial.org
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package admin_test
+
+import (
+	"github.com/stretchr/testify/suite"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
+	"github.com/superseriousbusiness/gotosocial/internal/email"
+	"github.com/superseriousbusiness/gotosocial/internal/federation"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/media"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
+	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/processing"
+	"github.com/superseriousbusiness/gotosocial/internal/processing/admin"
+	"github.com/superseriousbusiness/gotosocial/internal/state"
+	"github.com/superseriousbusiness/gotosocial/internal/storage"
+	"github.com/superseriousbusiness/gotosocial/internal/transport"
+	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
+	"github.com/superseriousbusiness/gotosocial/internal/visibility"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type AdminStandardTestSuite struct {
+	// standard suite interfaces
+	suite.Suite
+	db                  db.DB
+	tc                  typeutils.TypeConverter
+	storage             *storage.Driver
+	state               state.State
+	mediaManager        *media.Manager
+	oauthServer         oauth.Server
+	fromClientAPIChan   chan messages.FromClientAPI
+	transportController transport.Controller
+	federator           federation.Federator
+	emailSender         email.Sender
+	sentEmails          map[string]string
+	processor           *processing.Processor
+
+	// standard suite models
+	testTokens       map[string]*gtsmodel.Token
+	testClients      map[string]*gtsmodel.Client
+	testApplications map[string]*gtsmodel.Application
+	testUsers        map[string]*gtsmodel.User
+	testAccounts     map[string]*gtsmodel.Account
+	testFollows      map[string]*gtsmodel.Follow
+	testAttachments  map[string]*gtsmodel.MediaAttachment
+	testStatuses     map[string]*gtsmodel.Status
+
+	// module being tested
+	adminProcessor *admin.Processor
+}
+
+func (suite *AdminStandardTestSuite) SetupSuite() {
+	suite.testTokens = testrig.NewTestTokens()
+	suite.testClients = testrig.NewTestClients()
+	suite.testApplications = testrig.NewTestApplications()
+	suite.testUsers = testrig.NewTestUsers()
+	suite.testAccounts = testrig.NewTestAccounts()
+	suite.testFollows = testrig.NewTestFollows()
+	suite.testAttachments = testrig.NewTestAttachments()
+	suite.testStatuses = testrig.NewTestStatuses()
+}
+
+func (suite *AdminStandardTestSuite) SetupTest() {
+	suite.state.Caches.Init()
+	testrig.StartWorkers(&suite.state)
+
+	testrig.InitTestConfig()
+	testrig.InitTestLog()
+
+	suite.db = testrig.NewTestDB(&suite.state)
+	suite.state.DB = suite.db
+	suite.tc = testrig.NewTestTypeConverter(suite.db)
+
+	testrig.StartTimelines(
+		&suite.state,
+		visibility.NewFilter(&suite.state),
+		suite.tc,
+	)
+
+	suite.storage = testrig.NewInMemoryStorage()
+	suite.state.Storage = suite.storage
+	suite.mediaManager = testrig.NewTestMediaManager(&suite.state)
+	suite.oauthServer = testrig.NewTestOauthServer(suite.db)
+
+	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)
+	suite.emailSender = testrig.NewEmailSender("../../../web/template/", suite.sentEmails)
+
+	suite.processor = processing.NewProcessor(
+		suite.tc,
+		suite.federator,
+		suite.oauthServer,
+		suite.mediaManager,
+		&suite.state,
+		suite.emailSender,
+	)
+
+	suite.state.Workers.ProcessFromClientAPI = suite.processor.Workers().ProcessFromClientAPI
+	suite.adminProcessor = suite.processor.Admin()
+
+	testrig.StandardDBSetup(suite.db, nil)
+	testrig.StandardStorageSetup(suite.storage, "../../../testrig/media")
+}
+
+func (suite *AdminStandardTestSuite) TearDownTest() {
+	testrig.StandardDBTeardown(suite.db)
+	testrig.StandardStorageTeardown(suite.storage)
+	testrig.StopWorkers(&suite.state)
+}
--- a/internal/processing/admin/domainblock.go
+++ b/internal/processing/admin/domainblock.go
@@ -44,21 +44,24 @@ import (
 // and then processes side effects of that block (deleting accounts, media, etc).
 //
 // If a domain block already exists for the domain, side effects will be retried.
+//
+// Return values for this function are the (new) domain block, the ID of the admin
+// action resulting from this call, and/or an error if something goes wrong.
 func (p *Processor) DomainBlockCreate(
 	ctx context.Context,
-	account *gtsmodel.Account,
+	adminAcct *gtsmodel.Account,
 	domain string,
 	obfuscate bool,
 	publicComment string,
 	privateComment string,
 	subscriptionID string,
-) (*apimodel.DomainBlock, gtserror.WithCode) {
+) (*apimodel.DomainBlock, string, gtserror.WithCode) {
 	// Check if a block already exists for this domain.
 	domainBlock, err := p.state.DB.GetDomainBlock(ctx, domain)
 	if err != nil && !errors.Is(err, db.ErrNoEntries) {
 		// Something went wrong in the DB.
 		err = gtserror.Newf("db error getting domain block %s: %w", domain, err)
-		return nil, gtserror.NewErrorInternalError(err)
+		return nil, "", gtserror.NewErrorInternalError(err)
 	}

 	if domainBlock == nil {
@@ -66,7 +69,7 @@ func (p *Processor) DomainBlockCreate(
 		domainBlock = &gtsmodel.DomainBlock{
 			ID:                 id.NewULID(),
 			Domain:             domain,
-			CreatedByAccountID: account.ID,
+			CreatedByAccountID: adminAcct.ID,
 			PrivateComment:     text.SanitizeToPlaintext(privateComment),
 			PublicComment:      text.SanitizeToPlaintext(publicComment),
 			Obfuscate:          &obfuscate,
@@ -75,18 +78,100 @@ func (p *Processor) DomainBlockCreate(

 		// Insert the new block into the database.
 		if err := p.state.DB.CreateDomainBlock(ctx, domainBlock); err != nil {
-			err = gtserror.Newf("db error putting domain block %s: %s", domain, err)
-			return nil, gtserror.NewErrorInternalError(err)
+			err = gtserror.Newf("db error putting domain block %s: %w", domain, err)
+			return nil, "", gtserror.NewErrorInternalError(err)
 		}
 	}

-	// Process the side effects of the domain block
-	// asynchronously since it might take a while.
-	p.state.Workers.ClientAPI.Enqueue(func(ctx context.Context) {
-		p.domainBlockSideEffects(ctx, account, domainBlock)
-	})
+	actionID := id.NewULID()

-	return p.apiDomainBlock(ctx, domainBlock)
+	// Process domain block side
+	// effects asynchronously.
+	if errWithCode := p.actions.Run(
+		ctx,
+		&gtsmodel.AdminAction{
+			ID:             actionID,
+			TargetCategory: gtsmodel.AdminActionCategoryDomain,
+			TargetID:       domain,
+			Type:           gtsmodel.AdminActionSuspend,
+			AccountID:      adminAcct.ID,
+			Text:           domainBlock.PrivateComment,
+		},
+		func(ctx context.Context) gtserror.MultiError {
+			return p.domainBlockSideEffects(ctx, domainBlock)
+		},
+	); errWithCode != nil {
+		return nil, actionID, errWithCode
+	}
+
+	apiDomainBlock, errWithCode := p.apiDomainBlock(ctx, domainBlock)
+	if errWithCode != nil {
+		return nil, actionID, errWithCode
+	}
+
+	return apiDomainBlock, actionID, nil
+}
+
+// DomainBlockDelete removes one domain block with the given ID,
+// and processes side effects of removing the block asynchronously.
+//
+// Return values for this function are the deleted domain block, the ID of the admin
+// action resulting from this call, and/or an error if something goes wrong.
+func (p *Processor) DomainBlockDelete(
+	ctx context.Context,
+	adminAcct *gtsmodel.Account,
+	domainBlockID string,
+) (*apimodel.DomainBlock, string, gtserror.WithCode) {
+	domainBlock, err := p.state.DB.GetDomainBlockByID(ctx, domainBlockID)
+	if err != nil {
+		if !errors.Is(err, db.ErrNoEntries) {
+			// Real error.
+			err = gtserror.Newf("db error getting domain block: %w", err)
+			return nil, "", gtserror.NewErrorInternalError(err)
+		}
+
+		// There are just no entries for this ID.
+		err = fmt.Errorf("no domain block entry exists with ID %s", domainBlockID)
+		return nil, "", gtserror.NewErrorNotFound(err, err.Error())
+	}
+
+	// Prepare the domain block to return, *before* the deletion goes through.
+	apiDomainBlock, errWithCode := p.apiDomainBlock(ctx, domainBlock)
+	if errWithCode != nil {
+		return nil, "", errWithCode
+	}
+
+	// Copy value of the domain block.
+	domainBlockC := new(gtsmodel.DomainBlock)
+	*domainBlockC = *domainBlock
+
+	// Delete the original domain block.
+	if err := p.state.DB.DeleteDomainBlock(ctx, domainBlock.Domain); err != nil {
+		err = gtserror.Newf("db error deleting domain block: %w", err)
+		return nil, "", gtserror.NewErrorInternalError(err)
+	}
+
+	actionID := id.NewULID()
+
+	// Process domain unblock side
+	// effects asynchronously.
+	if errWithCode := p.actions.Run(
+		ctx,
+		&gtsmodel.AdminAction{
+			ID:             actionID,
+			TargetCategory: gtsmodel.AdminActionCategoryDomain,
+			TargetID:       domainBlockC.Domain,
+			Type:           gtsmodel.AdminActionUnsuspend,
+			AccountID:      adminAcct.ID,
+		},
+		func(ctx context.Context) gtserror.MultiError {
+			return p.domainUnblockSideEffects(ctx, domainBlock)
+		},
+	); errWithCode != nil {
+		return nil, actionID, errWithCode
+	}
+
+	return apiDomainBlock, actionID, nil
 }

 // DomainBlocksImport handles the import of multiple domain blocks,
@@ -153,7 +238,7 @@ func (p *Processor) DomainBlocksImport(
 			errWithCode    gtserror.WithCode
 		)

-		domainBlock, errWithCode = p.DomainBlockCreate(
+		domainBlock, _, errWithCode = p.DomainBlockCreate(
 			ctx,
 			account,
 			domain,
@@ -227,131 +312,6 @@ func (p *Processor) DomainBlockGet(ctx context.Context, id string, export bool)
 	return p.apiDomainBlock(ctx, domainBlock)
 }

-// DomainBlockDelete removes one domain block with the given ID,
-// and processes side effects of removing the block asynchronously.
-func (p *Processor) DomainBlockDelete(ctx context.Context, account *gtsmodel.Account, id string) (*apimodel.DomainBlock, gtserror.WithCode) {
-	domainBlock, err := p.state.DB.GetDomainBlockByID(ctx, id)
-	if err != nil {
-		if !errors.Is(err, db.ErrNoEntries) {
-			// Real error.
-			err = gtserror.Newf("db error getting domain block: %w", err)
-			return nil, gtserror.NewErrorInternalError(err)
-		}
-
-		// There are just no entries for this ID.
-		err = fmt.Errorf("no domain block entry exists with ID %s", id)
-		return nil, gtserror.NewErrorNotFound(err, err.Error())
-	}
-
-	// Prepare the domain block to return, *before* the deletion goes through.
-	apiDomainBlock, errWithCode := p.apiDomainBlock(ctx, domainBlock)
-	if errWithCode != nil {
-		return nil, errWithCode
-	}
-
-	// Copy value of the domain block.
-	domainBlockC := new(gtsmodel.DomainBlock)
-	*domainBlockC = *domainBlock
-
-	// Delete the original domain block.
-	if err := p.state.DB.DeleteDomainBlock(ctx, domainBlock.Domain); err != nil {
-		err = gtserror.Newf("db error deleting domain block: %w", err)
-		return nil, gtserror.NewErrorInternalError(err)
-	}
-
-	// Process the side effects of the domain unblock
-	// asynchronously since it might take a while.
-	p.state.Workers.ClientAPI.Enqueue(func(ctx context.Context) {
-		p.domainUnblockSideEffects(ctx, domainBlockC) // Use the copy.
-	})
-
-	return apiDomainBlock, nil
-}
-
-// stubbifyInstance renders the given instance as a stub,
-// removing most information from it and marking it as
-// suspended.
-//
-// For caller's convenience, this function returns the db
-// names of all columns that are updated by it.
-func stubbifyInstance(instance *gtsmodel.Instance, domainBlockID string) []string {
-	instance.Title = ""
-	instance.SuspendedAt = time.Now()
-	instance.DomainBlockID = domainBlockID
-	instance.ShortDescription = ""
-	instance.Description = ""
-	instance.Terms = ""
-	instance.ContactEmail = ""
-	instance.ContactAccountUsername = ""
-	instance.ContactAccountID = ""
-	instance.Version = ""
-
-	return []string{
-		"title",
-		"suspended_at",
-		"domain_block_id",
-		"short_description",
-		"description",
-		"terms",
-		"contact_email",
-		"contact_account_username",
-		"contact_account_id",
-		"version",
-	}
-}
-
-// apiDomainBlock is a cheeky shortcut function for returning the API
-// version of the given domainBlock, or an appropriate error if
-// something goes wrong.
-func (p *Processor) apiDomainBlock(ctx context.Context, domainBlock *gtsmodel.DomainBlock) (*apimodel.DomainBlock, gtserror.WithCode) {
-	apiDomainBlock, err := p.tc.DomainBlockToAPIDomainBlock(ctx, domainBlock, false)
-	if err != nil {
-		err = gtserror.Newf("error converting domain block for %s to api model : %w", domainBlock.Domain, err)
-		return nil, gtserror.NewErrorInternalError(err)
-	}
-
-	return apiDomainBlock, nil
-}
-
-// rangeAccounts iterates through all accounts originating from the
-// given domain, and calls the provided range function on each account.
-// If an error is returned from the range function, the loop will stop
-// and return the error.
-func (p *Processor) rangeAccounts(
-	ctx context.Context,
-	domain string,
-	rangeF func(*gtsmodel.Account) error,
-) error {
-	var (
-		limit = 50   // Limit selection to avoid spiking mem/cpu.
-		maxID string // Start with empty string to select from top.
-	)
-
-	for {
-		// Get (next) page of accounts.
-		accounts, err := p.state.DB.GetInstanceAccounts(ctx, domain, maxID, limit)
-		if err != nil && !errors.Is(err, db.ErrNoEntries) {
-			// Real db error.
-			return gtserror.Newf("db error getting instance accounts: %w", err)
-		}
-
-		if len(accounts) == 0 {
-			// No accounts left, we're done.
-			return nil
-		}
-
-		// Set next max ID for paging down.
-		maxID = accounts[len(accounts)-1].ID
-
-		// Call provided range function.
-		for _, account := range accounts {
-			if err := rangeF(account); err != nil {
-				return err
-			}
-		}
-	}
-}
-
 // domainBlockSideEffects processes the side effects of a domain block:
 //
 //  1. Strip most info away from the instance entry for the domain.
@@ -359,7 +319,10 @@ func (p *Processor) rangeAccounts(
 //
 // It should be called asynchronously, since it can take a while when
 // there are many accounts present on the given domain.
-func (p *Processor) domainBlockSideEffects(ctx context.Context, account *gtsmodel.Account, block *gtsmodel.DomainBlock) {
+func (p *Processor) domainBlockSideEffects(
+	ctx context.Context,
+	block *gtsmodel.DomainBlock,
+) gtserror.MultiError {
 	l := log.
 		WithContext(ctx).
 		WithFields(kv.Fields{
@@ -367,43 +330,46 @@ func (p *Processor) domainBlockSideEffects(ctx context.Context, account *gtsmode
 		}...)
 	l.Debug("processing domain block side effects")

+	var errs gtserror.MultiError
+
 	// If we have an instance entry for this domain,
 	// update it with the new block ID and clear all fields
 	instance, err := p.state.DB.GetInstance(ctx, block.Domain)
 	if err != nil && !errors.Is(err, db.ErrNoEntries) {
-		l.Errorf("db error getting instance %s: %q", block.Domain, err)
+		errs.Appendf("db error getting instance %s: %w", block.Domain, err)
+		return errs
 	}

 	if instance != nil {
 		// We had an entry for this domain.
 		columns := stubbifyInstance(instance, block.ID)
 		if err := p.state.DB.UpdateInstance(ctx, instance, columns...); err != nil {
-			l.Errorf("db error updating instance: %s", err)
-		} else {
-			l.Debug("instance entry updated")
+			errs.Appendf("db error updating instance: %w", err)
+			return errs
 		}
+		l.Debug("instance entry updated")
 	}

-	// For each account that belongs to this domain, create
-	// an account delete message to process via the client API
-	// worker pool, to remove that account's posts, media, etc.
-	msgs := []messages.FromClientAPI{}
-	if err := p.rangeAccounts(ctx, block.Domain, func(account *gtsmodel.Account) error {
-		msgs = append(msgs, messages.FromClientAPI{
+	// For each account that belongs to this domain,
+	// 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{
 			APObjectType:   ap.ActorPerson,
 			APActivityType: ap.ActivityDelete,
 			GTSModel:       block,
 			OriginAccount:  account,
 			TargetAccount:  account,
-		})
+		}

-		return nil
+		if err := p.state.Workers.ProcessFromClientAPI(ctx, cMsg); err != nil {
+			errs.Append(err)
+		}
 	}); err != nil {
-		l.Errorf("error while ranging through accounts: %q", err)
+		errs.Appendf("db error ranging through accounts: %w", err)
 	}

-	// Batch process all accreted messages.
-	p.state.Workers.EnqueueClientAPI(ctx, msgs...)
+	return errs
 }

 // domainUnblockSideEffects processes the side effects of undoing a
@@ -415,7 +381,10 @@ func (p *Processor) domainBlockSideEffects(ctx context.Context, account *gtsmode
 //
 // It should be called asynchronously, since it can take a while when
 // there are many accounts present on the given domain.
-func (p *Processor) domainUnblockSideEffects(ctx context.Context, block *gtsmodel.DomainBlock) {
+func (p *Processor) domainUnblockSideEffects(
+	ctx context.Context,
+	block *gtsmodel.DomainBlock,
+) gtserror.MultiError {
 	l := log.
 		WithContext(ctx).
 		WithFields(kv.Fields{
@@ -423,10 +392,12 @@ func (p *Processor) domainUnblockSideEffects(ctx context.Context, block *gtsmode
 		}...)
 	l.Debug("processing domain unblock side effects")

+	var errs gtserror.MultiError
+
 	// Update instance entry for this domain, if we have it.
 	instance, err := p.state.DB.GetInstance(ctx, block.Domain)
 	if err != nil && !errors.Is(err, db.ErrNoEntries) {
-		l.Errorf("db error getting instance %s: %q", block.Domain, err)
+		errs.Appendf("db error getting instance %s: %w", block.Domain, err)
 	}

 	if instance != nil {
@@ -440,23 +411,23 @@ func (p *Processor) domainUnblockSideEffects(ctx context.Context, block *gtsmode
 			"suspended_at",
 			"domain_block_id",
 		); err != nil {
-			l.Errorf("db error updating instance: %s", err)
-		} else {
-			l.Debug("instance entry updated")
+			errs.Appendf("db error updating instance: %w", err)
+			return errs
 		}
+		l.Debug("instance entry updated")
 	}

 	// Unsuspend all accounts whose suspension origin was this domain block.
-	if err := p.rangeAccounts(ctx, block.Domain, func(account *gtsmodel.Account) error {
+	if err := p.rangeDomainAccounts(ctx, block.Domain, func(account *gtsmodel.Account) {
 		if account.SuspensionOrigin == "" || account.SuspendedAt.IsZero() {
 			// Account wasn't suspended, nothing to do.
-			return nil
+			return
 		}

 		if account.SuspensionOrigin != block.ID {
 			// Account was suspended, but not by
 			// this domain block, leave it alone.
-			return nil
+			return
 		}

 		// Account was suspended by this domain
@@ -470,11 +441,11 @@ func (p *Processor) domainUnblockSideEffects(ctx context.Context, block *gtsmode
 			"suspended_at",
 			"suspension_origin",
 		); err != nil {
-			return gtserror.Newf("db error updating account %s: %w", account.Username, err)
+			errs.Appendf("db error updating account %s: %w", account.Username, err)
 		}
-
-		return nil
 	}); err != nil {
-		l.Errorf("error while ranging through accounts: %q", err)
+		errs.Appendf("db error ranging through accounts: %w", err)
 	}
+
+	return errs
 }
--- a/internal/processing/admin/domainblock_test.go
+++ b/internal/processing/admin/domainblock_test.go
@@ -0,0 +1,76 @@
+// GoToSocial
+// Copyright (C) GoToSocial Authors admin@gotosocial.org
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package admin_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+	"github.com/superseriousbusiness/gotosocial/testrig"
+)
+
+type DomainBlockTestSuite struct {
+	AdminStandardTestSuite
+}
+
+func (suite *DomainBlockTestSuite) TestCreateDomainBlock() {
+	var (
+		ctx            = context.Background()
+		adminAcct      = suite.testAccounts["admin_account"]
+		domain         = "fossbros-anonymous.io"
+		obfuscate      = false
+		publicComment  = ""
+		privateComment = ""
+		subscriptionID = ""
+	)
+
+	apiBlock, actionID, errWithCode := suite.adminProcessor.DomainBlockCreate(
+		ctx,
+		adminAcct,
+		domain,
+		obfuscate,
+		publicComment,
+		privateComment,
+		subscriptionID,
+	)
+	suite.NoError(errWithCode)
+	suite.NotNil(apiBlock)
+	suite.NotEmpty(actionID)
+
+	// Wait for action to finish.
+	if !testrig.WaitFor(func() bool {
+		return suite.adminProcessor.Actions().TotalRunning() == 0
+	}) {
+		suite.FailNow("timed out waiting for admin action(s) to finish")
+	}
+
+	// Ensure action marked as
+	// completed in the database.
+	adminAction, err := suite.db.GetAdminAction(ctx, actionID)
+	if err != nil {
+		suite.FailNow(err.Error())
+	}
+
+	suite.NotZero(adminAction.CompletedAt)
+	suite.Empty(adminAction.Errors)
+}
+
+func TestDomainBlockTestSuite(t *testing.T) {
+	suite.Run(t, new(DomainBlockTestSuite))
+}
--- a/internal/processing/admin/util.go
+++ b/internal/processing/admin/util.go
@@ -0,0 +1,116 @@
+// GoToSocial
+// Copyright (C) GoToSocial Authors admin@gotosocial.org
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package admin
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
+	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+)
+
+// apiDomainBlock is a cheeky shortcut for returning
+// the API version of the given domainBlock, or an
+// appropriate error if something goes wrong.
+func (p *Processor) apiDomainBlock(
+	ctx context.Context,
+	domainBlock *gtsmodel.DomainBlock,
+) (*apimodel.DomainBlock, gtserror.WithCode) {
+	apiDomainBlock, err := p.tc.DomainBlockToAPIDomainBlock(ctx, domainBlock, false)
+	if err != nil {
+		err = gtserror.Newf("error converting domain block for %s to api model : %w", domainBlock.Domain, err)
+		return nil, gtserror.NewErrorInternalError(err)
+	}
+
+	return apiDomainBlock, nil
+}
+
+// stubbifyInstance renders the given instance as a stub,
+// removing most information from it and marking it as
+// suspended.
+//
+// For caller's convenience, this function returns the db
+// names of all columns that are updated by it.
+func stubbifyInstance(instance *gtsmodel.Instance, domainBlockID string) []string {
+	instance.Title = ""
+	instance.SuspendedAt = time.Now()
+	instance.DomainBlockID = domainBlockID
+	instance.ShortDescription = ""
+	instance.Description = ""
+	instance.Terms = ""
+	instance.ContactEmail = ""
+	instance.ContactAccountUsername = ""
+	instance.ContactAccountID = ""
+	instance.Version = ""
+
+	return []string{
+		"title",
+		"suspended_at",
+		"domain_block_id",
+		"short_description",
+		"description",
+		"terms",
+		"contact_email",
+		"contact_account_username",
+		"contact_account_id",
+		"version",
+	}
+}
+
+// rangeDomainAccounts iterates through all accounts
+// originating from the given domain, and calls the
+// provided range function on each account.
+//
+// If an error is returned while selecting accounts,
+// the loop will stop and return the error.
+func (p *Processor) rangeDomainAccounts(
+	ctx context.Context,
+	domain string,
+	rangeF func(*gtsmodel.Account),
+) error {
+	var (
+		limit = 50   // Limit selection to avoid spiking mem/cpu.
+		maxID string // Start with empty string to select from top.
+	)
+
+	for {
+		// Get (next) page of accounts.
+		accounts, err := p.state.DB.GetInstanceAccounts(ctx, domain, maxID, limit)
+		if err != nil && !errors.Is(err, db.ErrNoEntries) {
+			// Real db error.
+			return gtserror.Newf("db error getting instance accounts: %w", err)
+		}
+
+		if len(accounts) == 0 {
+			// No accounts left, we're done.
+			return nil
+		}
+
+		// Set next max ID for paging down.
+		maxID = accounts[len(accounts)-1].ID
+
+		// Call provided range function.
+		for _, account := range accounts {
+			rangeF(account)
+		}
+	}
+}