[feature] Rework timeline code to make it useful for more than just statuses (#373)

* add preparable and timelineable interfaces

* initialize timeline manager within the processor

* generic renaming

* move status-specific timeline logic into the processor

* refactor timeline to make it useful for more than statuses

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

@@ -213,10 +213,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
 	mediaHandler    media.Handler
 	storage         *kv.KVStore
-	timelineManager timeline.Manager
+	statusTimelines timeline.Manager
 	db              db.DB
 	filter          visibility.Filter
 
@@ -261,7 +261,6 @@ func NewProcessor(
 	oauthServer oauth.Server,
 	mediaHandler media.Handler,
 	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, mediaHandler, 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,
 		mediaHandler:    mediaHandler,
 		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.federator = testrig.NewTestFederator(suite.db, suite.transportController, suite.storage)
 	suite.oauthServer = testrig.NewTestOauthServer(suite.db)
 	suite.mediaHandler = testrig.NewTestMediaHandler(suite.db, suite.storage)
-	suite.timelineManager = testrig.NewTestTimelineManager(suite.db)
 	suite.emailSender = testrig.NewEmailSender("../../web/template/", nil)
 
-	suite.processor = processing.NewProcessor(suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaHandler, suite.storage, suite.timelineManager, suite.db, suite.emailSender)
+	suite.processor = processing.NewProcessor(suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaHandler, suite.storage, suite.db, suite.emailSender)
 
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 	testrig.StandardStorageSetup(suite.storage, "../../testrig/media")

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) {