Merge branch 'main' into media_refactor

internal/processing/federation/getwebfinger.go

@@ -44,6 +44,9 @@ func (p *processor) GetWebfingerAccount(ctx context.Context, requestedUsername s
 	}
 
 	accountDomain := viper.GetString(config.Keys.AccountDomain)
+	if accountDomain == "" {
+		accountDomain = viper.GetString(config.Keys.Host)
+	}
 
 	// return the webfinger representation
 	return &apimodel.WellKnownResponse{

internal/processing/fromclientapi.go

@@ -192,10 +192,10 @@ func (p *processor) processCreateBlockFromClientAPI(ctx context.Context, clientM
 	}
 
 	// remove any of the blocking account's statuses from the blocked account's timeline, and vice versa
-	if err := p.timelineManager.WipeStatusesFromAccountID(ctx, block.AccountID, block.TargetAccountID); err != nil {
+	if err := p.statusTimelines.WipeItemsFromAccountID(ctx, block.AccountID, block.TargetAccountID); err != nil {
 		return err
 	}
-	if err := p.timelineManager.WipeStatusesFromAccountID(ctx, block.TargetAccountID, block.AccountID); err != nil {
+	if err := p.statusTimelines.WipeItemsFromAccountID(ctx, block.TargetAccountID, block.AccountID); err != nil {
 		return err
 	}
 

internal/processing/fromcommon.go

@@ -413,7 +413,7 @@ func (p *processor) timelineStatusForAccount(ctx context.Context, status *gtsmod
 	}
 
 	// stick the status in the timeline for the account and then immediately prepare it so they can see it right away
-	inserted, err := p.timelineManager.IngestAndPrepare(ctx, status, timelineAccount.ID)
+	inserted, err := p.statusTimelines.IngestAndPrepare(ctx, status, timelineAccount.ID)
 	if err != nil {
 		errors <- fmt.Errorf("timelineStatusForAccount: error ingesting status %s: %s", status.ID, err)
 		return
@@ -436,7 +436,7 @@ func (p *processor) timelineStatusForAccount(ctx context.Context, status *gtsmod
 // deleteStatusFromTimelines completely removes the given status from all timelines.
 // It will also stream deletion of the status to all open streams.
 func (p *processor) deleteStatusFromTimelines(ctx context.Context, status *gtsmodel.Status) error {
-	if err := p.timelineManager.WipeStatusFromAllTimelines(ctx, status.ID); err != nil {
+	if err := p.statusTimelines.WipeItemFromAllTimelines(ctx, status.ID); err != nil {
 		return err
 	}
 

internal/processing/fromfederator.go

@@ -309,10 +309,10 @@ func (p *processor) processCreateBlockFromFederator(ctx context.Context, federat
 	}
 
 	// remove any of the blocking account's statuses from the blocked account's timeline, and vice versa
-	if err := p.timelineManager.WipeStatusesFromAccountID(ctx, block.AccountID, block.TargetAccountID); err != nil {
+	if err := p.statusTimelines.WipeItemsFromAccountID(ctx, block.AccountID, block.TargetAccountID); err != nil {
 		return err
 	}
-	if err := p.timelineManager.WipeStatusesFromAccountID(ctx, block.TargetAccountID, block.AccountID); err != nil {
+	if err := p.statusTimelines.WipeItemsFromAccountID(ctx, block.TargetAccountID, block.AccountID); err != nil {
 		return err
 	}
 	// TODO: same with notifications

internal/processing/processor.go

@@ -237,7 +237,7 @@ type processor struct {
 	oauthServer     oauth.Server
 	mediaManager    media.Manager
 	storage         *kv.KVStore
-	timelineManager timeline.Manager
+	statusTimelines timeline.Manager
 	db              db.DB
 	filter          visibility.Filter
 
@@ -261,7 +261,6 @@ func NewProcessor(
 	oauthServer oauth.Server,
 	mediaManager media.Manager,
 	storage *kv.KVStore,
-	timelineManager timeline.Manager,
 	db db.DB,
 	emailSender email.Sender) Processor {
 	fromClientAPI := make(chan messages.FromClientAPI, 1000)
@@ -274,6 +273,7 @@ func NewProcessor(
 	mediaProcessor := mediaProcessor.New(db, tc, mediaManager, storage)
 	userProcessor := user.New(db, emailSender)
 	federationProcessor := federationProcessor.New(db, tc, federator, fromFederator)
+	filter := visibility.NewFilter(db)
 
 	return &processor{
 		fromClientAPI:   fromClientAPI,
@@ -284,7 +284,7 @@ func NewProcessor(
 		oauthServer:     oauthServer,
 		mediaManager:    mediaManager,
 		storage:         storage,
-		timelineManager: timelineManager,
+		statusTimelines: timeline.NewManager(StatusGrabFunction(db), StatusFilterFunction(db, filter), StatusPrepareFunction(db, tc), StatusSkipInsertFunction()),
 		db:              db,
 		filter:          visibility.NewFilter(db),
 

internal/processing/processor_test.go

@@ -219,10 +219,9 @@ func (suite *ProcessingStandardTestSuite) SetupTest() {
 	suite.mediaManager = testrig.NewTestMediaManager(suite.db, suite.storage)
 	suite.federator = testrig.NewTestFederator(suite.db, suite.transportController, suite.storage, suite.mediaManager)
 	suite.oauthServer = testrig.NewTestOauthServer(suite.db)
-	suite.timelineManager = testrig.NewTestTimelineManager(suite.db)
 	suite.emailSender = testrig.NewEmailSender("../../web/template/", nil)
 
-	suite.processor = processing.NewProcessor(suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, suite.storage, suite.timelineManager, suite.db, suite.emailSender)
+	suite.processor = processing.NewProcessor(suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, suite.storage, suite.db, suite.emailSender)
 
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 	testrig.StandardStorageSetup(suite.storage, "../../testrig/media")

internal/processing/status/util.go

@@ -223,8 +223,11 @@ func (p *processor) ProcessTags(ctx context.Context, form *apimodel.AdvancedStat
 		return fmt.Errorf("error generating hashtags from status: %s", err)
 	}
 	for _, tag := range gtsTags {
-		if err := p.db.Put(ctx, tag); err != nil && err != db.ErrAlreadyExists {
-			return fmt.Errorf("error putting tags in db: %s", err)
+		if err := p.db.Put(ctx, tag); err != nil {
+			var alreadyExistsError *db.ErrAlreadyExists
+			if !errors.As(err, &alreadyExistsError) {
+				return fmt.Errorf("error putting tags in db: %s", err)
+			}
 		}
 		tags = append(tags, tag.ID)
 	}

internal/processing/statustimeline.go

@@ -20,6 +20,7 @@ package processing
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/url"
 
@@ -32,8 +33,113 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
+	"github.com/superseriousbusiness/gotosocial/internal/timeline"
+	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
+	"github.com/superseriousbusiness/gotosocial/internal/visibility"
 )
 
+const boostReinsertionDepth = 50
+
+// StatusGrabFunction returns a function that satisfies the GrabFunction interface in internal/timeline.
+func StatusGrabFunction(database db.DB) timeline.GrabFunction {
+	return func(ctx context.Context, timelineAccountID string, maxID string, sinceID string, minID string, limit int) ([]timeline.Timelineable, bool, error) {
+		statuses, err := database.GetHomeTimeline(ctx, timelineAccountID, maxID, sinceID, minID, limit, false)
+		if err != nil {
+			if err == db.ErrNoEntries {
+				return nil, true, nil // we just don't have enough statuses left in the db so return stop = true
+			}
+			return nil, false, fmt.Errorf("statusGrabFunction: error getting statuses from db: %s", err)
+		}
+
+		items := []timeline.Timelineable{}
+		for _, s := range statuses {
+			items = append(items, s)
+		}
+
+		return items, false, nil
+	}
+}
+
+// StatusFilterFunction returns a function that satisfies the FilterFunction interface in internal/timeline.
+func StatusFilterFunction(database db.DB, filter visibility.Filter) timeline.FilterFunction {
+	return func(ctx context.Context, timelineAccountID string, item timeline.Timelineable) (shouldIndex bool, err error) {
+		status, ok := item.(*gtsmodel.Status)
+		if !ok {
+			return false, errors.New("statusFilterFunction: could not convert item to *gtsmodel.Status")
+		}
+
+		requestingAccount, err := database.GetAccountByID(ctx, timelineAccountID)
+		if err != nil {
+			return false, fmt.Errorf("statusFilterFunction: error getting account with id %s", timelineAccountID)
+		}
+
+		timelineable, err := filter.StatusHometimelineable(ctx, status, requestingAccount)
+		if err != nil {
+			logrus.Warnf("error checking hometimelineability of status %s for account %s: %s", status.ID, timelineAccountID, err)
+		}
+
+		return timelineable, nil // we don't return the error here because we want to just skip this item if something goes wrong
+	}
+}
+
+// StatusPrepareFunction returns a function that satisfies the PrepareFunction interface in internal/timeline.
+func StatusPrepareFunction(database db.DB, tc typeutils.TypeConverter) timeline.PrepareFunction {
+	return func(ctx context.Context, timelineAccountID string, itemID string) (timeline.Preparable, error) {
+		status, err := database.GetStatusByID(ctx, itemID)
+		if err != nil {
+			return nil, fmt.Errorf("statusPrepareFunction: error getting status with id %s", itemID)
+		}
+
+		requestingAccount, err := database.GetAccountByID(ctx, timelineAccountID)
+		if err != nil {
+			return nil, fmt.Errorf("statusPrepareFunction: error getting account with id %s", timelineAccountID)
+		}
+
+		return tc.StatusToAPIStatus(ctx, status, requestingAccount)
+	}
+}
+
+// StatusSkipInsertFunction returns a function that satisifes the SkipInsertFunction interface in internal/timeline.
+func StatusSkipInsertFunction() timeline.SkipInsertFunction {
+	return func(
+		ctx context.Context,
+		newItemID string,
+		newItemAccountID string,
+		newItemBoostOfID string,
+		newItemBoostOfAccountID string,
+		nextItemID string,
+		nextItemAccountID string,
+		nextItemBoostOfID string,
+		nextItemBoostOfAccountID string,
+		depth int) (bool, error) {
+
+		// make sure we don't insert a duplicate
+		if newItemID == nextItemID {
+			return true, nil
+		}
+
+		// check if it's a boost
+		if newItemBoostOfID != "" {
+			// skip if we've recently put another boost of this status in the timeline
+			if newItemBoostOfID == nextItemBoostOfID {
+				if depth < boostReinsertionDepth {
+					return true, nil
+				}
+			}
+
+			// skip if we've recently put the original status in the timeline
+			if newItemBoostOfID == nextItemID {
+				if depth < boostReinsertionDepth {
+					return true, nil
+				}
+			}
+		}
+
+		// insert the item
+		return false, nil
+	}
+}
+
 func (p *processor) packageStatusResponse(statuses []*apimodel.Status, path string, nextMaxID string, prevMinID string, limit int) (*apimodel.StatusTimelineResponse, gtserror.WithCode) {
 	resp := &apimodel.StatusTimelineResponse{
 		Statuses: []*apimodel.Status{},
@@ -67,18 +173,27 @@ func (p *processor) packageStatusResponse(statuses []*apimodel.Status, path stri
 }
 
 func (p *processor) HomeTimelineGet(ctx context.Context, authed *oauth.Auth, maxID string, sinceID string, minID string, limit int, local bool) (*apimodel.StatusTimelineResponse, gtserror.WithCode) {
-	statuses, err := p.timelineManager.HomeTimeline(ctx, authed.Account.ID, maxID, sinceID, minID, limit, local)
+	preparedItems, err := p.statusTimelines.GetTimeline(ctx, authed.Account.ID, maxID, sinceID, minID, limit, local)
 	if err != nil {
 		return nil, gtserror.NewErrorInternalError(err)
 	}
 
-	if len(statuses) == 0 {
+	if len(preparedItems) == 0 {
 		return &apimodel.StatusTimelineResponse{
 			Statuses: []*apimodel.Status{},
 		}, nil
 	}
 
-	return p.packageStatusResponse(statuses, "api/v1/timelines/home", statuses[len(statuses)-1].ID, statuses[0].ID, limit)
+	statuses := []*apimodel.Status{}
+	for _, i := range preparedItems {
+		status, ok := i.(*apimodel.Status)
+		if !ok {
+			return nil, gtserror.NewErrorInternalError(errors.New("error converting prepared timeline entry to api status"))
+		}
+		statuses = append(statuses, status)
+	}
+
+	return p.packageStatusResponse(statuses, "api/v1/timelines/home", statuses[len(preparedItems)-1].ID, statuses[0].ID, limit)
 }
 
 func (p *processor) PublicTimelineGet(ctx context.Context, authed *oauth.Auth, maxID string, sinceID string, minID string, limit int, local bool) (*apimodel.StatusTimelineResponse, gtserror.WithCode) {

internal/processing/user/emailconfirm_test.go

@@ -54,7 +54,7 @@ func (suite *EmailConfirmTestSuite) TestSendConfirmEmail() {
 	suite.NotEmpty(token)
 
 	// email should contain the token
-	emailShould := fmt.Sprintf("Subject: GoToSocial Email Confirmation\r\nFrom: GoToSocial <test@example.org>\r\nTo: some.email@example.org\r\nMIME-version: 1.0;\nContent-Type: text/html;\r\n<!DOCTYPE html>\n<html>\n    </head>\n    <body>\n        <div>\n            <h1>\n                Hello the_mighty_zork!\n            </h1>\n        </div>\n        <div>\n            <p>\n                You are receiving this mail because you've requested an account on <a href=\"http://localhost:8080\">localhost:8080</a>.\n            </p>\n            <p>\n                We just need to confirm that this is your email address. To confirm your email, <a href=\"http://localhost:8080/confirm_email?token=%s\">click here</a> or paste the following in your browser's address bar:\n            </p>\n            <p>\n                <code>\n                    http://localhost:8080/confirm_email?token=%s\n                </code>\n            </p>\n        </div>\n        <div>\n            <p>\n                If you believe you've been sent this email in error, feel free to ignore it, or contact the administrator of <a href=\"http://localhost:8080\">localhost:8080</a>.\n            </p>\n        </div>\n    </body>\n</html>\r\n", token, token)
+	emailShould := fmt.Sprintf("To: some.email@example.org\r\nSubject: GoToSocial Email Confirmation\r\n\r\nHello the_mighty_zork!\r\n\r\nYou are receiving this mail because you've requested an account on http://localhost:8080.\r\n\r\nWe just need to confirm that this is your email address. To confirm your email, paste the following in your browser's address bar:\r\n\r\nhttp://localhost:8080/confirm_email?token=%s\r\n\r\nIf you believe you've been sent this email in error, feel free to ignore it, or contact the administrator of http://localhost:8080\r\n\r\n", token)
 	suite.Equal(emailShould, email)
 
 	// confirmationSentAt should be recent