mirror of
https://github.com/superseriousbusiness/gotosocial
synced 2025-06-05 21:59:39 +02:00
Timeline manager (#40)
* start messing about with timeline manager * i have no idea what i'm doing * i continue to not know what i'm doing * it's coming along * bit more progress * update timeline with new posts as they come in * lint and fmt * Select accounts where empty string * restructure a bunch, get unfaves working * moving stuff around * federate status deletes properly * mention regex better but not 100% there * fix regex * some more hacking away at the timeline code phew * fix up some little things * i can't even * more timeline stuff * move to ulid * fiddley * some lil fixes for kibou compatibility * timelines working pretty alright! * tidy + lint
This commit is contained in:
@ -20,45 +20,70 @@ package processing
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"sync"
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
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"
|
||||
"github.com/superseriousbusiness/gotosocial/internal/oauth"
|
||||
)
|
||||
|
||||
func (p *processor) HomeTimelineGet(authed *oauth.Auth, maxID string, sinceID string, minID string, limit int, local bool) ([]apimodel.Status, ErrorWithCode) {
|
||||
statuses, err := p.db.GetHomeTimelineForAccount(authed.Account.ID, maxID, sinceID, minID, limit, local)
|
||||
if err != nil {
|
||||
return nil, NewErrorInternalError(err)
|
||||
func (p *processor) HomeTimelineGet(authed *oauth.Auth, maxID string, sinceID string, minID string, limit int, local bool) (*apimodel.StatusTimelineResponse, gtserror.WithCode) {
|
||||
resp := &apimodel.StatusTimelineResponse{
|
||||
Statuses: []*apimodel.Status{},
|
||||
}
|
||||
|
||||
s, err := p.filterStatuses(authed, statuses)
|
||||
apiStatuses, err := p.timelineManager.HomeTimeline(authed.Account.ID, maxID, sinceID, minID, limit, local)
|
||||
if err != nil {
|
||||
return nil, NewErrorInternalError(err)
|
||||
return nil, gtserror.NewErrorInternalError(err)
|
||||
}
|
||||
resp.Statuses = apiStatuses
|
||||
|
||||
// prepare the next and previous links
|
||||
if len(apiStatuses) != 0 {
|
||||
nextLink := &url.URL{
|
||||
Scheme: p.config.Protocol,
|
||||
Host: p.config.Host,
|
||||
Path: "/api/v1/timelines/home",
|
||||
RawPath: url.PathEscape("api/v1/timelines/home"),
|
||||
RawQuery: fmt.Sprintf("limit=%d&max_id=%s", limit, apiStatuses[len(apiStatuses)-1].ID),
|
||||
}
|
||||
next := fmt.Sprintf("<%s>; rel=\"next\"", nextLink.String())
|
||||
|
||||
prevLink := &url.URL{
|
||||
Scheme: p.config.Protocol,
|
||||
Host: p.config.Host,
|
||||
Path: "/api/v1/timelines/home",
|
||||
RawQuery: fmt.Sprintf("limit=%d&min_id=%s", limit, apiStatuses[0].ID),
|
||||
}
|
||||
prev := fmt.Sprintf("<%s>; rel=\"prev\"", prevLink.String())
|
||||
resp.LinkHeader = fmt.Sprintf("%s, %s", next, prev)
|
||||
}
|
||||
|
||||
return s, nil
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (p *processor) PublicTimelineGet(authed *oauth.Auth, maxID string, sinceID string, minID string, limit int, local bool) ([]apimodel.Status, ErrorWithCode) {
|
||||
func (p *processor) PublicTimelineGet(authed *oauth.Auth, maxID string, sinceID string, minID string, limit int, local bool) ([]*apimodel.Status, gtserror.WithCode) {
|
||||
statuses, err := p.db.GetPublicTimelineForAccount(authed.Account.ID, maxID, sinceID, minID, limit, local)
|
||||
if err != nil {
|
||||
return nil, NewErrorInternalError(err)
|
||||
return nil, gtserror.NewErrorInternalError(err)
|
||||
}
|
||||
|
||||
s, err := p.filterStatuses(authed, statuses)
|
||||
if err != nil {
|
||||
return nil, NewErrorInternalError(err)
|
||||
return nil, gtserror.NewErrorInternalError(err)
|
||||
}
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (p *processor) filterStatuses(authed *oauth.Auth, statuses []*gtsmodel.Status) ([]apimodel.Status, error) {
|
||||
func (p *processor) filterStatuses(authed *oauth.Auth, statuses []*gtsmodel.Status) ([]*apimodel.Status, error) {
|
||||
l := p.log.WithField("func", "filterStatuses")
|
||||
|
||||
apiStatuses := []apimodel.Status{}
|
||||
apiStatuses := []*apimodel.Status{}
|
||||
for _, s := range statuses {
|
||||
targetAccount := >smodel.Account{}
|
||||
if err := p.db.GetByID(s.AccountID, targetAccount); err != nil {
|
||||
@ -66,7 +91,7 @@ func (p *processor) filterStatuses(authed *oauth.Auth, statuses []*gtsmodel.Stat
|
||||
l.Debugf("skipping status %s because account %s can't be found in the db", s.ID, s.AccountID)
|
||||
continue
|
||||
}
|
||||
return nil, NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error getting status author: %s", err))
|
||||
return nil, gtserror.NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error getting status author: %s", err))
|
||||
}
|
||||
|
||||
relevantAccounts, err := p.db.PullRelevantAccountsFromStatus(s)
|
||||
@ -75,9 +100,9 @@ func (p *processor) filterStatuses(authed *oauth.Auth, statuses []*gtsmodel.Stat
|
||||
continue
|
||||
}
|
||||
|
||||
visible, err := p.db.StatusVisible(s, targetAccount, authed.Account, relevantAccounts)
|
||||
visible, err := p.db.StatusVisible(s, authed.Account, relevantAccounts)
|
||||
if err != nil {
|
||||
return nil, NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error checking status visibility: %s", err))
|
||||
return nil, gtserror.NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error checking status visibility: %s", err))
|
||||
}
|
||||
if !visible {
|
||||
continue
|
||||
@ -91,7 +116,7 @@ func (p *processor) filterStatuses(authed *oauth.Auth, statuses []*gtsmodel.Stat
|
||||
l.Debugf("skipping status %s because status %s can't be found in the db", s.ID, s.BoostOfID)
|
||||
continue
|
||||
}
|
||||
return nil, NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error getting boosted status: %s", err))
|
||||
return nil, gtserror.NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error getting boosted status: %s", err))
|
||||
}
|
||||
boostedRelevantAccounts, err := p.db.PullRelevantAccountsFromStatus(bs)
|
||||
if err != nil {
|
||||
@ -99,9 +124,9 @@ func (p *processor) filterStatuses(authed *oauth.Auth, statuses []*gtsmodel.Stat
|
||||
continue
|
||||
}
|
||||
|
||||
boostedVisible, err := p.db.StatusVisible(bs, relevantAccounts.BoostedAccount, authed.Account, boostedRelevantAccounts)
|
||||
boostedVisible, err := p.db.StatusVisible(bs, authed.Account, boostedRelevantAccounts)
|
||||
if err != nil {
|
||||
return nil, NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error checking boosted status visibility: %s", err))
|
||||
return nil, gtserror.NewErrorInternalError(fmt.Errorf("HomeTimelineGet: error checking boosted status visibility: %s", err))
|
||||
}
|
||||
|
||||
if boostedVisible {
|
||||
@ -115,8 +140,113 @@ func (p *processor) filterStatuses(authed *oauth.Auth, statuses []*gtsmodel.Stat
|
||||
continue
|
||||
}
|
||||
|
||||
apiStatuses = append(apiStatuses, *apiStatus)
|
||||
apiStatuses = append(apiStatuses, apiStatus)
|
||||
}
|
||||
|
||||
return apiStatuses, nil
|
||||
}
|
||||
|
||||
func (p *processor) initTimelines() error {
|
||||
// get all local accounts (ie., domain = nil) that aren't suspended (suspended_at = nil)
|
||||
localAccounts := []*gtsmodel.Account{}
|
||||
where := []db.Where{
|
||||
{
|
||||
Key: "domain", Value: nil,
|
||||
},
|
||||
{
|
||||
Key: "suspended_at", Value: nil,
|
||||
},
|
||||
}
|
||||
if err := p.db.GetWhere(where, &localAccounts); err != nil {
|
||||
if _, ok := err.(db.ErrNoEntries); ok {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("initTimelines: db error initializing timelines: %s", err)
|
||||
}
|
||||
|
||||
// we want to wait until all timelines are populated so created a waitgroup here
|
||||
wg := &sync.WaitGroup{}
|
||||
wg.Add(len(localAccounts))
|
||||
|
||||
for _, localAccount := range localAccounts {
|
||||
// to save time we can populate the timelines asynchronously
|
||||
// this will go heavy on the database, but since we're not actually serving yet it doesn't really matter
|
||||
go p.initTimelineFor(localAccount, wg)
|
||||
}
|
||||
|
||||
// wait for all timelines to be populated before we exit
|
||||
wg.Wait()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *processor) initTimelineFor(account *gtsmodel.Account, wg *sync.WaitGroup) {
|
||||
defer wg.Done()
|
||||
|
||||
l := p.log.WithFields(logrus.Fields{
|
||||
"func": "initTimelineFor",
|
||||
"accountID": account.ID,
|
||||
})
|
||||
|
||||
desiredIndexLength := p.timelineManager.GetDesiredIndexLength()
|
||||
|
||||
statuses, err := p.db.GetStatusesWhereFollowing(account.ID, "", "", "", desiredIndexLength, false)
|
||||
if err != nil {
|
||||
if _, ok := err.(db.ErrNoEntries); !ok {
|
||||
l.Error(fmt.Errorf("initTimelineFor: error getting statuses: %s", err))
|
||||
}
|
||||
return
|
||||
}
|
||||
p.indexAndIngest(statuses, account, desiredIndexLength)
|
||||
|
||||
lengthNow := p.timelineManager.GetIndexedLength(account.ID)
|
||||
if lengthNow < desiredIndexLength {
|
||||
// try and get more posts from the last ID onwards
|
||||
rearmostStatusID, err := p.timelineManager.GetOldestIndexedID(account.ID)
|
||||
if err != nil {
|
||||
l.Error(fmt.Errorf("initTimelineFor: error getting id of rearmost status: %s", err))
|
||||
return
|
||||
}
|
||||
|
||||
if rearmostStatusID != "" {
|
||||
moreStatuses, err := p.db.GetStatusesWhereFollowing(account.ID, rearmostStatusID, "", "", desiredIndexLength/2, false)
|
||||
if err != nil {
|
||||
l.Error(fmt.Errorf("initTimelineFor: error getting more statuses: %s", err))
|
||||
return
|
||||
}
|
||||
p.indexAndIngest(moreStatuses, account, desiredIndexLength)
|
||||
}
|
||||
}
|
||||
|
||||
l.Debugf("prepared timeline of length %d for account %s", lengthNow, account.ID)
|
||||
}
|
||||
|
||||
func (p *processor) indexAndIngest(statuses []*gtsmodel.Status, timelineAccount *gtsmodel.Account, desiredIndexLength int) {
|
||||
l := p.log.WithFields(logrus.Fields{
|
||||
"func": "indexAndIngest",
|
||||
"accountID": timelineAccount.ID,
|
||||
})
|
||||
|
||||
for _, s := range statuses {
|
||||
relevantAccounts, err := p.db.PullRelevantAccountsFromStatus(s)
|
||||
if err != nil {
|
||||
l.Error(fmt.Errorf("initTimelineFor: error getting relevant accounts from status %s: %s", s.ID, err))
|
||||
continue
|
||||
}
|
||||
visible, err := p.db.StatusVisible(s, timelineAccount, relevantAccounts)
|
||||
if err != nil {
|
||||
l.Error(fmt.Errorf("initTimelineFor: error checking visibility of status %s: %s", s.ID, err))
|
||||
continue
|
||||
}
|
||||
if visible {
|
||||
if err := p.timelineManager.Ingest(s, timelineAccount.ID); err != nil {
|
||||
l.Error(fmt.Errorf("initTimelineFor: error ingesting status %s: %s", s.ID, err))
|
||||
continue
|
||||
}
|
||||
|
||||
// check if we have enough posts now and return if we do
|
||||
if p.timelineManager.GetIndexedLength(timelineAccount.ID) >= desiredIndexLength {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
Reference in New Issue
Block a user