mirror of
				https://github.com/superseriousbusiness/gotosocial.git
				synced 2025-10-31 10:22:25 -05:00 
			
		
		
		
	* 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
		
	
			
		
			
				
	
	
		
			350 lines
		
	
	
	
		
			11 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			350 lines
		
	
	
	
		
			11 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
| // 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"
 | |
| 	"fmt"
 | |
| 	"time"
 | |
| 
 | |
| 	"codeberg.org/gruf/go-kv"
 | |
| 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 | |
| 	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
 | |
| 	"github.com/superseriousbusiness/gotosocial/internal/config"
 | |
| 	"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/internal/log"
 | |
| 	"github.com/superseriousbusiness/gotosocial/internal/messages"
 | |
| 	"github.com/superseriousbusiness/gotosocial/internal/text"
 | |
| )
 | |
| 
 | |
| func (p *Processor) createDomainBlock(
 | |
| 	ctx context.Context,
 | |
| 	adminAcct *gtsmodel.Account,
 | |
| 	domain string,
 | |
| 	obfuscate bool,
 | |
| 	publicComment string,
 | |
| 	privateComment string,
 | |
| 	subscriptionID string,
 | |
| ) (*apimodel.DomainPermission, 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)
 | |
| 	}
 | |
| 
 | |
| 	if domainBlock == nil {
 | |
| 		// No block exists yet, create it.
 | |
| 		domainBlock = >smodel.DomainBlock{
 | |
| 			ID:                 id.NewULID(),
 | |
| 			Domain:             domain,
 | |
| 			CreatedByAccountID: adminAcct.ID,
 | |
| 			PrivateComment:     text.SanitizeToPlaintext(privateComment),
 | |
| 			PublicComment:      text.SanitizeToPlaintext(publicComment),
 | |
| 			Obfuscate:          &obfuscate,
 | |
| 			SubscriptionID:     subscriptionID,
 | |
| 		}
 | |
| 
 | |
| 		// 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: %w", domain, err)
 | |
| 			return nil, "", gtserror.NewErrorInternalError(err)
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	actionID := id.NewULID()
 | |
| 
 | |
| 	// Process domain block side
 | |
| 	// effects asynchronously.
 | |
| 	if errWithCode := p.actions.Run(
 | |
| 		ctx,
 | |
| 		>smodel.AdminAction{
 | |
| 			ID:             actionID,
 | |
| 			TargetCategory: gtsmodel.AdminActionCategoryDomain,
 | |
| 			TargetID:       domain,
 | |
| 			Type:           gtsmodel.AdminActionSuspend,
 | |
| 			AccountID:      adminAcct.ID,
 | |
| 			Text:           domainBlock.PrivateComment,
 | |
| 		},
 | |
| 		func(ctx context.Context) gtserror.MultiError {
 | |
| 			// Log start + finish.
 | |
| 			l := log.WithFields(kv.Fields{
 | |
| 				{"domain", domain},
 | |
| 				{"actionID", actionID},
 | |
| 			}...).WithContext(ctx)
 | |
| 
 | |
| 			skip, err := p.skipBlockSideEffects(ctx, domain)
 | |
| 			if err != nil {
 | |
| 				return err
 | |
| 			}
 | |
| 			if skip != "" {
 | |
| 				l.Infof("skipping domain block side effects: %s", skip)
 | |
| 				return nil
 | |
| 			}
 | |
| 
 | |
| 			l.Info("processing domain block side effects")
 | |
| 			defer func() { l.Info("finished processing domain block side effects") }()
 | |
| 
 | |
| 			return p.domainBlockSideEffects(ctx, domainBlock)
 | |
| 		},
 | |
| 	); errWithCode != nil {
 | |
| 		return nil, actionID, errWithCode
 | |
| 	}
 | |
| 
 | |
| 	apiDomainBlock, errWithCode := p.apiDomainPerm(ctx, domainBlock, false)
 | |
| 	if errWithCode != nil {
 | |
| 		return nil, actionID, errWithCode
 | |
| 	}
 | |
| 
 | |
| 	return apiDomainBlock, actionID, nil
 | |
| }
 | |
| 
 | |
| // skipBlockSideEffects checks if side effects of block creation
 | |
| // should be skipped for the given domain, taking account of
 | |
| // instance federation mode, and existence of any allows
 | |
| // which ought to "shield" this domain from being blocked.
 | |
| //
 | |
| // If the caller should skip, the returned string will be non-zero
 | |
| // and will be set to a reason why side effects should be skipped.
 | |
| //
 | |
| //   - blocklist mode + allow exists: "..." (skip)
 | |
| //   - blocklist mode + no allow:     ""    (don't skip)
 | |
| //   - allowlist mode + allow exists: ""    (don't skip)
 | |
| //   - allowlist mode + no allow:     ""    (don't skip)
 | |
| func (p *Processor) skipBlockSideEffects(
 | |
| 	ctx context.Context,
 | |
| 	domain string,
 | |
| ) (string, gtserror.MultiError) {
 | |
| 	var (
 | |
| 		skip string // Assume "" (don't skip).
 | |
| 		errs gtserror.MultiError
 | |
| 	)
 | |
| 
 | |
| 	// Never skip block side effects in allowlist mode.
 | |
| 	fediMode := config.GetInstanceFederationMode()
 | |
| 	if fediMode == config.InstanceFederationModeAllowlist {
 | |
| 		return skip, errs
 | |
| 	}
 | |
| 
 | |
| 	// We know we're in blocklist mode.
 | |
| 	//
 | |
| 	// We want to skip domain block side
 | |
| 	// effects if an allow is already
 | |
| 	// in place which overrides the block.
 | |
| 
 | |
| 	// Check if an explicit allow exists for this domain.
 | |
| 	domainAllow, err := p.state.DB.GetDomainAllow(ctx, domain)
 | |
| 	if err != nil && !errors.Is(err, db.ErrNoEntries) {
 | |
| 		errs.Appendf("error getting domain allow: %w", err)
 | |
| 		return skip, errs
 | |
| 	}
 | |
| 
 | |
| 	if domainAllow != nil {
 | |
| 		skip = "running in blocklist mode, and an explicit allow exists for this domain"
 | |
| 		return skip, errs
 | |
| 	}
 | |
| 
 | |
| 	return skip, errs
 | |
| }
 | |
| 
 | |
| // domainBlockSideEffects processes the side effects of a domain block:
 | |
| //
 | |
| //  1. Strip most info away from the instance entry for the domain.
 | |
| //  2. Pass each account from the domain to the processor for deletion.
 | |
| //
 | |
| // 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,
 | |
| 	block *gtsmodel.DomainBlock,
 | |
| ) gtserror.MultiError {
 | |
| 	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) {
 | |
| 		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 {
 | |
| 			errs.Appendf("db error updating instance: %w", err)
 | |
| 			return errs
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	// 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) {
 | |
| 		if err := p.state.Workers.Client.Process(ctx, &messages.FromClientAPI{
 | |
| 			APObjectType:   ap.ActorPerson,
 | |
| 			APActivityType: ap.ActivityDelete,
 | |
| 			GTSModel:       block,
 | |
| 			Origin:         account,
 | |
| 			Target:         account,
 | |
| 		}); err != nil {
 | |
| 			errs.Append(err)
 | |
| 		}
 | |
| 	}); err != nil {
 | |
| 		errs.Appendf("db error ranging through accounts: %w", err)
 | |
| 	}
 | |
| 
 | |
| 	return errs
 | |
| }
 | |
| 
 | |
| func (p *Processor) deleteDomainBlock(
 | |
| 	ctx context.Context,
 | |
| 	adminAcct *gtsmodel.Account,
 | |
| 	domainBlockID string,
 | |
| ) (*apimodel.DomainPermission, 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.apiDomainPerm(ctx, domainBlock, false)
 | |
| 	if errWithCode != nil {
 | |
| 		return nil, "", errWithCode
 | |
| 	}
 | |
| 
 | |
| 	// 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,
 | |
| 		>smodel.AdminAction{
 | |
| 			ID:             actionID,
 | |
| 			TargetCategory: gtsmodel.AdminActionCategoryDomain,
 | |
| 			TargetID:       domainBlock.Domain,
 | |
| 			Type:           gtsmodel.AdminActionUnsuspend,
 | |
| 			AccountID:      adminAcct.ID,
 | |
| 		},
 | |
| 		func(ctx context.Context) gtserror.MultiError {
 | |
| 			// Log start + finish.
 | |
| 			l := log.WithFields(kv.Fields{
 | |
| 				{"domain", domainBlock.Domain},
 | |
| 				{"actionID", actionID},
 | |
| 			}...).WithContext(ctx)
 | |
| 
 | |
| 			l.Info("processing domain unblock side effects")
 | |
| 			defer func() { l.Info("finished processing domain unblock side effects") }()
 | |
| 
 | |
| 			return p.domainUnblockSideEffects(ctx, domainBlock)
 | |
| 		},
 | |
| 	); errWithCode != nil {
 | |
| 		return nil, actionID, errWithCode
 | |
| 	}
 | |
| 
 | |
| 	return apiDomainBlock, actionID, nil
 | |
| }
 | |
| 
 | |
| // domainUnblockSideEffects processes the side effects of undoing a
 | |
| // domain block:
 | |
| //
 | |
| //  1. Mark instance entry as no longer suspended.
 | |
| //  2. Mark each account from the domain as no longer suspended, if the
 | |
| //     suspension origin corresponds to the ID of the provided domain block.
 | |
| //
 | |
| // 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,
 | |
| ) gtserror.MultiError {
 | |
| 	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) {
 | |
| 		errs.Appendf("db error getting instance %s: %w", block.Domain, err)
 | |
| 	}
 | |
| 
 | |
| 	if instance != nil {
 | |
| 		// We had an entry, update it to signal
 | |
| 		// that it's no longer suspended.
 | |
| 		instance.SuspendedAt = time.Time{}
 | |
| 		instance.DomainBlockID = ""
 | |
| 		if err := p.state.DB.UpdateInstance(
 | |
| 			ctx,
 | |
| 			instance,
 | |
| 			"suspended_at",
 | |
| 			"domain_block_id",
 | |
| 		); err != nil {
 | |
| 			errs.Appendf("db error updating instance: %w", err)
 | |
| 			return errs
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	// Unsuspend all accounts whose suspension origin was this domain block.
 | |
| 	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
 | |
| 		}
 | |
| 
 | |
| 		if account.SuspensionOrigin != block.ID {
 | |
| 			// Account was suspended, but not by
 | |
| 			// this domain block, leave it alone.
 | |
| 			return
 | |
| 		}
 | |
| 
 | |
| 		// Account was suspended by this domain
 | |
| 		// block, mark it as unsuspended.
 | |
| 		account.SuspendedAt = time.Time{}
 | |
| 		account.SuspensionOrigin = ""
 | |
| 
 | |
| 		if err := p.state.DB.UpdateAccount(
 | |
| 			ctx,
 | |
| 			account,
 | |
| 			"suspended_at",
 | |
| 			"suspension_origin",
 | |
| 		); err != nil {
 | |
| 			errs.Appendf("db error updating account %s: %w", account.Username, err)
 | |
| 		}
 | |
| 	}); err != nil {
 | |
| 		errs.Appendf("db error ranging through accounts: %w", err)
 | |
| 	}
 | |
| 
 | |
| 	return errs
 | |
| }
 |