diff options
Diffstat (limited to 'internal/processing')
49 files changed, 527 insertions, 252 deletions
diff --git a/internal/processing/account/create.go b/internal/processing/account/create.go index 37c742b45..9eb618994 100644 --- a/internal/processing/account/create.go +++ b/internal/processing/account/create.go @@ -60,7 +60,7 @@ func (p *processor) Create(ctx context.Context, applicationToken oauth2.TokenInf } l.Tracef("generating a token for user %s with account %s and application %s", user.ID, user.AccountID, application.ID) - accessToken, err := p.oauthServer.GenerateUserAccessToken(applicationToken, application.ClientSecret, user.ID) + accessToken, err := p.oauthServer.GenerateUserAccessToken(ctx, applicationToken, application.ClientSecret, user.ID) if err != nil { return nil, fmt.Errorf("error creating new access token for user %s: %s", user.ID, err) } diff --git a/internal/processing/account/get.go b/internal/processing/account/get.go index 5f039127c..dd56df356 100644 --- a/internal/processing/account/get.go +++ b/internal/processing/account/get.go @@ -45,13 +45,13 @@ func (p *processor) Get(ctx context.Context, requestingAccount *gtsmodel.Account } } - var mastoAccount *apimodel.Account + var apiAccount *apimodel.Account if blocked { - mastoAccount, err = p.tc.AccountToMastoBlocked(ctx, targetAccount) + apiAccount, err = p.tc.AccountToAPIAccountBlocked(ctx, targetAccount) if err != nil { return nil, fmt.Errorf("error converting account: %s", err) } - return mastoAccount, nil + return apiAccount, nil } // last-minute check to make sure we have remote account header/avi cached @@ -63,12 +63,12 @@ func (p *processor) Get(ctx context.Context, requestingAccount *gtsmodel.Account } if requestingAccount != nil && targetAccount.ID == requestingAccount.ID { - mastoAccount, err = p.tc.AccountToMastoSensitive(ctx, targetAccount) + apiAccount, err = p.tc.AccountToAPIAccountSensitive(ctx, targetAccount) } else { - mastoAccount, err = p.tc.AccountToMastoPublic(ctx, targetAccount) + apiAccount, err = p.tc.AccountToAPIAccountPublic(ctx, targetAccount) } if err != nil { return nil, fmt.Errorf("error converting account: %s", err) } - return mastoAccount, nil + return apiAccount, nil } diff --git a/internal/processing/account/getfollowers.go b/internal/processing/account/getfollowers.go index 517467085..bcc607290 100644 --- a/internal/processing/account/getfollowers.go +++ b/internal/processing/account/getfollowers.go @@ -64,7 +64,7 @@ func (p *processor) FollowersGet(ctx context.Context, requestingAccount *gtsmode f.Account = a } - account, err := p.tc.AccountToMastoPublic(ctx, f.Account) + account, err := p.tc.AccountToAPIAccountPublic(ctx, f.Account) if err != nil { return nil, gtserror.NewErrorInternalError(err) } diff --git a/internal/processing/account/getfollowing.go b/internal/processing/account/getfollowing.go index 543213f90..d7e9d5f63 100644 --- a/internal/processing/account/getfollowing.go +++ b/internal/processing/account/getfollowing.go @@ -64,7 +64,7 @@ func (p *processor) FollowingGet(ctx context.Context, requestingAccount *gtsmode f.TargetAccount = a } - account, err := p.tc.AccountToMastoPublic(ctx, f.TargetAccount) + account, err := p.tc.AccountToAPIAccountPublic(ctx, f.TargetAccount) if err != nil { return nil, gtserror.NewErrorInternalError(err) } diff --git a/internal/processing/account/getrelationship.go b/internal/processing/account/getrelationship.go index ebfd9b479..9f23cd070 100644 --- a/internal/processing/account/getrelationship.go +++ b/internal/processing/account/getrelationship.go @@ -38,7 +38,7 @@ func (p *processor) RelationshipGet(ctx context.Context, requestingAccount *gtsm return nil, gtserror.NewErrorInternalError(fmt.Errorf("error getting relationship: %s", err)) } - r, err := p.tc.RelationshipToMasto(ctx, gtsR) + r, err := p.tc.RelationshipToAPIRelationship(ctx, gtsR) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting relationship: %s", err)) } diff --git a/internal/processing/account/getstatuses.go b/internal/processing/account/getstatuses.go index dc157e43c..56b5b0eae 100644 --- a/internal/processing/account/getstatuses.go +++ b/internal/processing/account/getstatuses.go @@ -51,9 +51,9 @@ func (p *processor) StatusesGet(ctx context.Context, requestingAccount *gtsmodel continue } - apiStatus, err := p.tc.StatusToMasto(ctx, s, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, s, requestingAccount) if err != nil { - return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status to masto: %s", err)) + return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status to api: %s", err)) } apiStatuses = append(apiStatuses, *apiStatus) diff --git a/internal/processing/account/update.go b/internal/processing/account/update.go index 6dc288849..1ab25787f 100644 --- a/internal/processing/account/update.go +++ b/internal/processing/account/update.go @@ -105,7 +105,7 @@ func (p *processor) Update(ctx context.Context, account *gtsmodel.Account, form if err := validate.Privacy(*form.Source.Privacy); err != nil { return nil, err } - privacy := p.tc.MastoVisToVis(apimodel.Visibility(*form.Source.Privacy)) + privacy := p.tc.APIVisToVis(apimodel.Visibility(*form.Source.Privacy)) account.Privacy = privacy } } @@ -122,9 +122,9 @@ func (p *processor) Update(ctx context.Context, account *gtsmodel.Account, form OriginAccount: updatedAccount, } - acctSensitive, err := p.tc.AccountToMastoSensitive(ctx, updatedAccount) + acctSensitive, err := p.tc.AccountToAPIAccountSensitive(ctx, updatedAccount) if err != nil { - return nil, fmt.Errorf("could not convert account into mastosensitive account: %s", err) + return nil, fmt.Errorf("could not convert account into apisensitive account: %s", err) } return acctSensitive, nil } diff --git a/internal/processing/admin/createdomainblock.go b/internal/processing/admin/createdomainblock.go index 50df056e5..399007fe0 100644 --- a/internal/processing/admin/createdomainblock.go +++ b/internal/processing/admin/createdomainblock.go @@ -73,12 +73,12 @@ func (p *processor) DomainBlockCreate(ctx context.Context, account *gtsmodel.Acc go p.initiateDomainBlockSideEffects(ctx, account, domainBlock) // TODO: add this to a queuing system so it can retry/resume } - mastoDomainBlock, err := p.tc.DomainBlockToMasto(ctx, domainBlock, false) + apiDomainBlock, err := p.tc.DomainBlockToAPIDomainBlock(ctx, domainBlock, false) if err != nil { - return nil, gtserror.NewErrorInternalError(fmt.Errorf("DomainBlockCreate: error converting domain block to frontend/masto representation %s: %s", domain, err)) + return nil, gtserror.NewErrorInternalError(fmt.Errorf("DomainBlockCreate: error converting domain block to frontend/api representation %s: %s", domain, err)) } - return mastoDomainBlock, nil + return apiDomainBlock, nil } // initiateDomainBlockSideEffects should be called asynchronously, to process the side effects of a domain block: diff --git a/internal/processing/admin/deletedomainblock.go b/internal/processing/admin/deletedomainblock.go index d11374c78..156c156ec 100644 --- a/internal/processing/admin/deletedomainblock.go +++ b/internal/processing/admin/deletedomainblock.go @@ -42,7 +42,7 @@ func (p *processor) DomainBlockDelete(ctx context.Context, account *gtsmodel.Acc } // prepare the domain block to return - mastoDomainBlock, err := p.tc.DomainBlockToMasto(ctx, domainBlock, false) + apiDomainBlock, err := p.tc.DomainBlockToAPIDomainBlock(ctx, domainBlock, false) if err != nil { return nil, gtserror.NewErrorInternalError(err) } @@ -80,5 +80,5 @@ func (p *processor) DomainBlockDelete(ctx context.Context, account *gtsmodel.Acc return nil, gtserror.NewErrorInternalError(fmt.Errorf("database error removing suspension_origin from accounts: %s", err)) } - return mastoDomainBlock, nil + return apiDomainBlock, nil } diff --git a/internal/processing/admin/emoji.go b/internal/processing/admin/emoji.go index f56bde8e0..de36b3e50 100644 --- a/internal/processing/admin/emoji.go +++ b/internal/processing/admin/emoji.go @@ -61,14 +61,14 @@ func (p *processor) EmojiCreate(ctx context.Context, account *gtsmodel.Account, } emoji.ID = emojiID - mastoEmoji, err := p.tc.EmojiToMasto(ctx, emoji) + apiEmoji, err := p.tc.EmojiToAPIEmoji(ctx, emoji) if err != nil { - return nil, fmt.Errorf("error converting emoji to mastotype: %s", err) + return nil, fmt.Errorf("error converting emoji to apitype: %s", err) } if err := p.db.Put(ctx, emoji); err != nil { return nil, fmt.Errorf("database error while processing emoji: %s", err) } - return &mastoEmoji, nil + return &apiEmoji, nil } diff --git a/internal/processing/admin/getdomainblock.go b/internal/processing/admin/getdomainblock.go index 19bc9fe09..4cb626a4e 100644 --- a/internal/processing/admin/getdomainblock.go +++ b/internal/processing/admin/getdomainblock.go @@ -40,10 +40,10 @@ func (p *processor) DomainBlockGet(ctx context.Context, account *gtsmodel.Accoun return nil, gtserror.NewErrorNotFound(fmt.Errorf("no entry for ID %s", id)) } - mastoDomainBlock, err := p.tc.DomainBlockToMasto(ctx, domainBlock, export) + apiDomainBlock, err := p.tc.DomainBlockToAPIDomainBlock(ctx, domainBlock, export) if err != nil { return nil, gtserror.NewErrorInternalError(err) } - return mastoDomainBlock, nil + return apiDomainBlock, nil } diff --git a/internal/processing/admin/getdomainblocks.go b/internal/processing/admin/getdomainblocks.go index 0ec33cfff..13cc2ed19 100644 --- a/internal/processing/admin/getdomainblocks.go +++ b/internal/processing/admin/getdomainblocks.go @@ -37,14 +37,14 @@ func (p *processor) DomainBlocksGet(ctx context.Context, account *gtsmodel.Accou } } - mastoDomainBlocks := []*apimodel.DomainBlock{} + apiDomainBlocks := []*apimodel.DomainBlock{} for _, b := range domainBlocks { - mastoDomainBlock, err := p.tc.DomainBlockToMasto(ctx, b, export) + apiDomainBlock, err := p.tc.DomainBlockToAPIDomainBlock(ctx, b, export) if err != nil { return nil, gtserror.NewErrorInternalError(err) } - mastoDomainBlocks = append(mastoDomainBlocks, mastoDomainBlock) + apiDomainBlocks = append(apiDomainBlocks, apiDomainBlock) } - return mastoDomainBlocks, nil + return apiDomainBlocks, nil } diff --git a/internal/processing/app.go b/internal/processing/app.go index d6ded6efa..d0aba3636 100644 --- a/internal/processing/app.go +++ b/internal/processing/app.go @@ -29,7 +29,7 @@ import ( ) func (p *processor) AppCreate(ctx context.Context, authed *oauth.Auth, form *apimodel.ApplicationCreateRequest) (*apimodel.Application, error) { - // set default 'read' for scopes if it's not set, this follows the default of the mastodon api https://docs.joinmastodon.org/methods/apps/ + // set default 'read' for scopes if it's not set var scopes string if form.Scopes == "" { scopes = "read" @@ -78,10 +78,10 @@ func (p *processor) AppCreate(ctx context.Context, authed *oauth.Auth, form *api return nil, err } - mastoApp, err := p.tc.AppToMastoSensitive(ctx, app) + apiApp, err := p.tc.AppToAPIAppSensitive(ctx, app) if err != nil { return nil, err } - return mastoApp, nil + return apiApp, nil } diff --git a/internal/processing/blocks.go b/internal/processing/blocks.go index 7c8371989..1144579a4 100644 --- a/internal/processing/blocks.go +++ b/internal/processing/blocks.go @@ -44,7 +44,7 @@ func (p *processor) BlocksGet(ctx context.Context, authed *oauth.Auth, maxID str apiAccounts := []*apimodel.Account{} for _, a := range accounts { - apiAccount, err := p.tc.AccountToMastoBlocked(ctx, a) + apiAccount, err := p.tc.AccountToAPIAccountBlocked(ctx, a) if err != nil { continue } diff --git a/internal/processing/federation.go b/internal/processing/federation.go index 1fccfa36e..1336a6e46 100644 --- a/internal/processing/federation.go +++ b/internal/processing/federation.go @@ -120,7 +120,7 @@ func (p *processor) GetFediFollowers(ctx context.Context, requestedUsername stri return nil, gtserror.NewErrorInternalError(fmt.Errorf("error parsing url %s: %s", requestedAccount.URI, err)) } - requestedFollowers, err := p.federator.FederatingDB().Followers(context.Background(), requestedAccountURI) + requestedFollowers, err := p.federator.FederatingDB().Followers(ctx, requestedAccountURI) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error fetching followers for uri %s: %s", requestedAccountURI.String(), err)) } @@ -165,7 +165,7 @@ func (p *processor) GetFediFollowing(ctx context.Context, requestedUsername stri return nil, gtserror.NewErrorInternalError(fmt.Errorf("error parsing url %s: %s", requestedAccount.URI, err)) } - requestedFollowing, err := p.federator.FederatingDB().Following(context.Background(), requestedAccountURI) + requestedFollowing, err := p.federator.FederatingDB().Following(ctx, requestedAccountURI) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error fetching following for uri %s: %s", requestedAccountURI.String(), err)) } diff --git a/internal/processing/followrequest.go b/internal/processing/followrequest.go index b313e42f8..74bffd693 100644 --- a/internal/processing/followrequest.go +++ b/internal/processing/followrequest.go @@ -47,11 +47,11 @@ func (p *processor) FollowRequestsGet(ctx context.Context, auth *oauth.Auth) ([] fr.Account = frAcct } - mastoAcct, err := p.tc.AccountToMastoPublic(ctx, fr.Account) + apiAcct, err := p.tc.AccountToAPIAccountPublic(ctx, fr.Account) if err != nil { return nil, gtserror.NewErrorInternalError(err) } - accts = append(accts, *mastoAcct) + accts = append(accts, *apiAcct) } return accts, nil } @@ -91,7 +91,7 @@ func (p *processor) FollowRequestAccept(ctx context.Context, auth *oauth.Auth, a return nil, gtserror.NewErrorInternalError(err) } - r, err := p.tc.RelationshipToMasto(ctx, gtsR) + r, err := p.tc.RelationshipToAPIRelationship(ctx, gtsR) if err != nil { return nil, gtserror.NewErrorInternalError(err) } diff --git a/internal/processing/fromcommon.go b/internal/processing/fromcommon.go index e14c36fd4..88f9994d4 100644 --- a/internal/processing/fromcommon.go +++ b/internal/processing/fromcommon.go @@ -96,12 +96,12 @@ func (p *processor) notifyStatus(ctx context.Context, status *gtsmodel.Status) e } // now stream the notification to the user - mastoNotif, err := p.tc.NotificationToMasto(ctx, notif) + apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif) if err != nil { - return fmt.Errorf("notifyStatus: error converting notification to masto representation: %s", err) + return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err) } - if err := p.streamingProcessor.StreamNotificationToAccount(mastoNotif, m.TargetAccount); err != nil { + if err := p.streamingProcessor.StreamNotificationToAccount(apiNotif, m.TargetAccount); err != nil { return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err) } } @@ -143,12 +143,12 @@ func (p *processor) notifyFollowRequest(ctx context.Context, followRequest *gtsm } // now stream the notification to the user - mastoNotif, err := p.tc.NotificationToMasto(ctx, notif) + apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif) if err != nil { - return fmt.Errorf("notifyStatus: error converting notification to masto representation: %s", err) + return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err) } - if err := p.streamingProcessor.StreamNotificationToAccount(mastoNotif, targetAccount); err != nil { + if err := p.streamingProcessor.StreamNotificationToAccount(apiNotif, targetAccount); err != nil { return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err) } @@ -189,12 +189,12 @@ func (p *processor) notifyFollow(ctx context.Context, follow *gtsmodel.Follow, t } // now stream the notification to the user - mastoNotif, err := p.tc.NotificationToMasto(ctx, notif) + apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif) if err != nil { - return fmt.Errorf("notifyStatus: error converting notification to masto representation: %s", err) + return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err) } - if err := p.streamingProcessor.StreamNotificationToAccount(mastoNotif, targetAccount); err != nil { + if err := p.streamingProcessor.StreamNotificationToAccount(apiNotif, targetAccount); err != nil { return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err) } @@ -237,12 +237,12 @@ func (p *processor) notifyFave(ctx context.Context, fave *gtsmodel.StatusFave) e } // now stream the notification to the user - mastoNotif, err := p.tc.NotificationToMasto(ctx, notif) + apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif) if err != nil { - return fmt.Errorf("notifyStatus: error converting notification to masto representation: %s", err) + return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err) } - if err := p.streamingProcessor.StreamNotificationToAccount(mastoNotif, targetAccount); err != nil { + if err := p.streamingProcessor.StreamNotificationToAccount(apiNotif, targetAccount); err != nil { return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err) } @@ -316,12 +316,12 @@ func (p *processor) notifyAnnounce(ctx context.Context, status *gtsmodel.Status) } // now stream the notification to the user - mastoNotif, err := p.tc.NotificationToMasto(ctx, notif) + apiNotif, err := p.tc.NotificationToAPINotification(ctx, notif) if err != nil { - return fmt.Errorf("notifyStatus: error converting notification to masto representation: %s", err) + return fmt.Errorf("notifyStatus: error converting notification to api representation: %s", err) } - if err := p.streamingProcessor.StreamNotificationToAccount(mastoNotif, status.BoostOfAccount); err != nil { + if err := p.streamingProcessor.StreamNotificationToAccount(apiNotif, status.BoostOfAccount); err != nil { return fmt.Errorf("notifyStatus: error streaming notification to account: %s", err) } @@ -414,21 +414,21 @@ func (p *processor) timelineStatusForAccount(ctx context.Context, status *gtsmod // the status was inserted to stream it to the user if inserted { - mastoStatus, err := p.tc.StatusToMasto(ctx, status, timelineAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, status, timelineAccount) if err != nil { errors <- fmt.Errorf("timelineStatusForAccount: error converting status %s to frontend representation: %s", status.ID, err) } else { - if err := p.streamingProcessor.StreamStatusToAccount(mastoStatus, timelineAccount); err != nil { + if err := p.streamingProcessor.StreamUpdateToAccount(apiStatus, timelineAccount); err != nil { errors <- fmt.Errorf("timelineStatusForAccount: error streaming status %s: %s", status.ID, err) } } } - mastoStatus, err := p.tc.StatusToMasto(ctx, status, timelineAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, status, timelineAccount) if err != nil { errors <- fmt.Errorf("timelineStatusForAccount: error converting status %s to frontend representation: %s", status.ID, err) } else { - if err := p.streamingProcessor.StreamStatusToAccount(mastoStatus, timelineAccount); err != nil { + if err := p.streamingProcessor.StreamUpdateToAccount(apiStatus, timelineAccount); err != nil { errors <- fmt.Errorf("timelineStatusForAccount: error streaming status %s: %s", status.ID, err) } } diff --git a/internal/processing/instance.go b/internal/processing/instance.go index 41139c491..75d17d13a 100644 --- a/internal/processing/instance.go +++ b/internal/processing/instance.go @@ -36,7 +36,7 @@ func (p *processor) InstanceGet(ctx context.Context, domain string) (*apimodel.I return nil, gtserror.NewErrorInternalError(fmt.Errorf("db error fetching instance %s: %s", p.config.Host, err)) } - ai, err := p.tc.InstanceToMasto(ctx, i) + ai, err := p.tc.InstanceToAPIInstance(ctx, i) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting instance to api representation: %s", err)) } @@ -151,7 +151,7 @@ func (p *processor) InstancePatch(ctx context.Context, form *apimodel.InstanceSe return nil, gtserror.NewErrorInternalError(fmt.Errorf("db error updating instance %s: %s", p.config.Host, err)) } - ai, err := p.tc.InstanceToMasto(ctx, i) + ai, err := p.tc.InstanceToAPIInstance(ctx, i) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting instance to api representation: %s", err)) } diff --git a/internal/processing/media/create.go b/internal/processing/media/create.go index 43162f3f6..0783bfae8 100644 --- a/internal/processing/media/create.go +++ b/internal/processing/media/create.go @@ -73,7 +73,7 @@ func (p *processor) Create(ctx context.Context, account *gtsmodel.Account, form // prepare the frontend representation now -- if there are any errors here at least we can bail without // having already put something in the database and then having to clean it up again (eugh) - mastoAttachment, err := p.tc.AttachmentToMasto(ctx, attachment) + apiAttachment, err := p.tc.AttachmentToAPIAttachment(ctx, attachment) if err != nil { return nil, fmt.Errorf("error parsing media attachment to frontend type: %s", err) } @@ -83,5 +83,5 @@ func (p *processor) Create(ctx context.Context, account *gtsmodel.Account, form return nil, fmt.Errorf("error storing media attachment in db: %s", err) } - return &mastoAttachment, nil + return &apiAttachment, nil } diff --git a/internal/processing/media/getmedia.go b/internal/processing/media/getmedia.go index 91608e90d..763aaf8f6 100644 --- a/internal/processing/media/getmedia.go +++ b/internal/processing/media/getmedia.go @@ -43,7 +43,7 @@ func (p *processor) GetMedia(ctx context.Context, account *gtsmodel.Account, med return nil, gtserror.NewErrorNotFound(errors.New("attachment not owned by requesting account")) } - a, err := p.tc.AttachmentToMasto(ctx, attachment) + a, err := p.tc.AttachmentToAPIAttachment(ctx, attachment) if err != nil { return nil, gtserror.NewErrorNotFound(fmt.Errorf("error converting attachment: %s", err)) } diff --git a/internal/processing/media/update.go b/internal/processing/media/update.go index e6c78563d..b3455bc91 100644 --- a/internal/processing/media/update.go +++ b/internal/processing/media/update.go @@ -63,7 +63,7 @@ func (p *processor) Update(ctx context.Context, account *gtsmodel.Account, media } } - a, err := p.tc.AttachmentToMasto(ctx, attachment) + a, err := p.tc.AttachmentToAPIAttachment(ctx, attachment) if err != nil { return nil, gtserror.NewErrorNotFound(fmt.Errorf("error converting attachment: %s", err)) } diff --git a/internal/processing/notification.go b/internal/processing/notification.go index f91d2f2cd..27280a973 100644 --- a/internal/processing/notification.go +++ b/internal/processing/notification.go @@ -34,15 +34,15 @@ func (p *processor) NotificationsGet(ctx context.Context, authed *oauth.Auth, li return nil, gtserror.NewErrorInternalError(err) } - mastoNotifs := []*apimodel.Notification{} + apiNotifs := []*apimodel.Notification{} for _, n := range notifs { - mastoNotif, err := p.tc.NotificationToMasto(ctx, n) + apiNotif, err := p.tc.NotificationToAPINotification(ctx, n) if err != nil { - l.Debugf("got an error converting a notification to masto, will skip it: %s", err) + l.Debugf("got an error converting a notification to api, will skip it: %s", err) continue } - mastoNotifs = append(mastoNotifs, mastoNotif) + apiNotifs = append(apiNotifs, apiNotif) } - return mastoNotifs, nil + return apiNotifs, nil } diff --git a/internal/processing/processor.go b/internal/processing/processor.go index 6f36f6d21..c643738c2 100644 --- a/internal/processing/processor.go +++ b/internal/processing/processor.go @@ -256,7 +256,7 @@ func NewProcessor(config *config.Config, tc typeutils.TypeConverter, federator f fromFederator := make(chan messages.FromFederator, 1000) statusProcessor := status.New(db, tc, config, fromClientAPI, log) - streamingProcessor := streaming.New(db, tc, oauthServer, config, log) + streamingProcessor := streaming.New(db, oauthServer, log) accountProcessor := account.New(db, tc, mediaHandler, oauthServer, fromClientAPI, federator, config, log) adminProcessor := admin.New(db, tc, mediaHandler, fromClientAPI, config, log) mediaProcessor := mediaProcessor.New(db, tc, mediaHandler, storage, config, log) diff --git a/internal/processing/search.go b/internal/processing/search.go index 2fb1f6062..c5bfd722b 100644 --- a/internal/processing/search.go +++ b/internal/processing/search.go @@ -93,8 +93,8 @@ func (p *processor) SearchGet(ctx context.Context, authed *oauth.Auth, searchQue // make sure there's no block in either direction between the account and the requester if blocked, err := p.db.IsBlocked(ctx, authed.Account.ID, foundAccount.ID, true); err == nil && !blocked { // all good, convert it and add it to the results - if acctMasto, err := p.tc.AccountToMastoPublic(ctx, foundAccount); err == nil && acctMasto != nil { - results.Accounts = append(results.Accounts, *acctMasto) + if apiAcct, err := p.tc.AccountToAPIAccountPublic(ctx, foundAccount); err == nil && apiAcct != nil { + results.Accounts = append(results.Accounts, *apiAcct) } } } @@ -104,12 +104,12 @@ func (p *processor) SearchGet(ctx context.Context, authed *oauth.Auth, searchQue continue } - statusMasto, err := p.tc.StatusToMasto(ctx, foundStatus, authed.Account) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, foundStatus, authed.Account) if err != nil { continue } - results.Statuses = append(results.Statuses, *statusMasto) + results.Statuses = append(results.Statuses, *apiStatus) } return results, nil diff --git a/internal/processing/status/boost.go b/internal/processing/status/boost.go index 4276ca9fa..2ee6acd55 100644 --- a/internal/processing/status/boost.go +++ b/internal/processing/status/boost.go @@ -74,10 +74,10 @@ func (p *processor) Boost(ctx context.Context, requestingAccount *gtsmodel.Accou } // return the frontend representation of the new status to the submitter - mastoStatus, err := p.tc.StatusToMasto(ctx, boostWrapperStatus, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, boostWrapperStatus, requestingAccount) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", targetStatus.ID, err)) } - return mastoStatus, nil + return apiStatus, nil } diff --git a/internal/processing/status/boostedby.go b/internal/processing/status/boostedby.go index 46f41039f..00256d2dd 100644 --- a/internal/processing/status/boostedby.go +++ b/internal/processing/status/boostedby.go @@ -64,15 +64,15 @@ func (p *processor) BoostedBy(ctx context.Context, requestingAccount *gtsmodel.A // TODO: filter other things here? suspended? muted? silenced? - // now we can return the masto representation of those accounts - mastoAccounts := []*apimodel.Account{} + // now we can return the api representation of those accounts + apiAccounts := []*apimodel.Account{} for _, acc := range filteredAccounts { - mastoAccount, err := p.tc.AccountToMastoPublic(ctx, acc) + apiAccount, err := p.tc.AccountToAPIAccountPublic(ctx, acc) if err != nil { return nil, gtserror.NewErrorNotFound(fmt.Errorf("StatusFavedBy: error converting account to api model: %s", err)) } - mastoAccounts = append(mastoAccounts, mastoAccount) + apiAccounts = append(apiAccounts, apiAccount) } - return mastoAccounts, nil + return apiAccounts, nil } diff --git a/internal/processing/status/context.go b/internal/processing/status/context.go index 3e8e93d09..75c756cfe 100644 --- a/internal/processing/status/context.go +++ b/internal/processing/status/context.go @@ -58,9 +58,9 @@ func (p *processor) Context(ctx context.Context, requestingAccount *gtsmodel.Acc for _, status := range parents { if v, err := p.filter.StatusVisible(ctx, status, requestingAccount); err == nil && v { - mastoStatus, err := p.tc.StatusToMasto(ctx, status, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, status, requestingAccount) if err == nil { - context.Ancestors = append(context.Ancestors, *mastoStatus) + context.Ancestors = append(context.Ancestors, *apiStatus) } } } @@ -76,9 +76,9 @@ func (p *processor) Context(ctx context.Context, requestingAccount *gtsmodel.Acc for _, status := range children { if v, err := p.filter.StatusVisible(ctx, status, requestingAccount); err == nil && v { - mastoStatus, err := p.tc.StatusToMasto(ctx, status, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, status, requestingAccount) if err == nil { - context.Descendants = append(context.Descendants, *mastoStatus) + context.Descendants = append(context.Descendants, *apiStatus) } } } diff --git a/internal/processing/status/create.go b/internal/processing/status/create.go index a87dbc7fe..655be5b17 100644 --- a/internal/processing/status/create.go +++ b/internal/processing/status/create.go @@ -105,10 +105,10 @@ func (p *processor) Create(ctx context.Context, account *gtsmodel.Account, appli } // return the frontend representation of the new status to the submitter - mastoStatus, err := p.tc.StatusToMasto(ctx, newStatus, account) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, newStatus, account) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", newStatus.ID, err)) } - return mastoStatus, nil + return apiStatus, nil } diff --git a/internal/processing/status/delete.go b/internal/processing/status/delete.go index dfb2c3626..822e1559c 100644 --- a/internal/processing/status/delete.go +++ b/internal/processing/status/delete.go @@ -43,7 +43,7 @@ func (p *processor) Delete(ctx context.Context, requestingAccount *gtsmodel.Acco return nil, gtserror.NewErrorForbidden(errors.New("status doesn't belong to requesting account")) } - mastoStatus, err := p.tc.StatusToMasto(ctx, targetStatus, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, targetStatus, requestingAccount) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", targetStatus.ID, err)) } @@ -61,5 +61,5 @@ func (p *processor) Delete(ctx context.Context, requestingAccount *gtsmodel.Acco TargetAccount: requestingAccount, } - return mastoStatus, nil + return apiStatus, nil } diff --git a/internal/processing/status/fave.go b/internal/processing/status/fave.go index f3f10c43c..571e0715c 100644 --- a/internal/processing/status/fave.go +++ b/internal/processing/status/fave.go @@ -93,11 +93,11 @@ func (p *processor) Fave(ctx context.Context, requestingAccount *gtsmodel.Accoun } } - // return the mastodon representation of the target status - mastoStatus, err := p.tc.StatusToMasto(ctx, targetStatus, requestingAccount) + // return the apidon representation of the target status + apiStatus, err := p.tc.StatusToAPIStatus(ctx, targetStatus, requestingAccount) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", targetStatus.ID, err)) } - return mastoStatus, nil + return apiStatus, nil } diff --git a/internal/processing/status/favedby.go b/internal/processing/status/favedby.go index 227fb669d..e8681e379 100644 --- a/internal/processing/status/favedby.go +++ b/internal/processing/status/favedby.go @@ -62,15 +62,15 @@ func (p *processor) FavedBy(ctx context.Context, requestingAccount *gtsmodel.Acc } } - // now we can return the masto representation of those accounts - mastoAccounts := []*apimodel.Account{} + // now we can return the api representation of those accounts + apiAccounts := []*apimodel.Account{} for _, acc := range filteredAccounts { - mastoAccount, err := p.tc.AccountToMastoPublic(ctx, acc) + apiAccount, err := p.tc.AccountToAPIAccountPublic(ctx, acc) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", targetStatus.ID, err)) } - mastoAccounts = append(mastoAccounts, mastoAccount) + apiAccounts = append(apiAccounts, apiAccount) } - return mastoAccounts, nil + return apiAccounts, nil } diff --git a/internal/processing/status/get.go b/internal/processing/status/get.go index 258210faf..fb9a3ea77 100644 --- a/internal/processing/status/get.go +++ b/internal/processing/status/get.go @@ -45,10 +45,10 @@ func (p *processor) Get(ctx context.Context, requestingAccount *gtsmodel.Account return nil, gtserror.NewErrorNotFound(errors.New("status is not visible")) } - mastoStatus, err := p.tc.StatusToMasto(ctx, targetStatus, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, targetStatus, requestingAccount) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", targetStatus.ID, err)) } - return mastoStatus, nil + return apiStatus, nil } diff --git a/internal/processing/status/unboost.go b/internal/processing/status/unboost.go index 13c24d638..94e74f8e0 100644 --- a/internal/processing/status/unboost.go +++ b/internal/processing/status/unboost.go @@ -100,10 +100,10 @@ func (p *processor) Unboost(ctx context.Context, requestingAccount *gtsmodel.Acc } } - mastoStatus, err := p.tc.StatusToMasto(ctx, targetStatus, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, targetStatus, requestingAccount) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", targetStatus.ID, err)) } - return mastoStatus, nil + return apiStatus, nil } diff --git a/internal/processing/status/unfave.go b/internal/processing/status/unfave.go index 27ce9b156..a8ddea39d 100644 --- a/internal/processing/status/unfave.go +++ b/internal/processing/status/unfave.go @@ -82,10 +82,10 @@ func (p *processor) Unfave(ctx context.Context, requestingAccount *gtsmodel.Acco } } - mastoStatus, err := p.tc.StatusToMasto(ctx, targetStatus, requestingAccount) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, targetStatus, requestingAccount) if err != nil { return nil, gtserror.NewErrorInternalError(fmt.Errorf("error converting status %s to frontend representation: %s", targetStatus.ID, err)) } - return mastoStatus, nil + return apiStatus, nil } diff --git a/internal/processing/status/util.go b/internal/processing/status/util.go index edbb9a31a..e655632db 100644 --- a/internal/processing/status/util.go +++ b/internal/processing/status/util.go @@ -42,7 +42,7 @@ func (p *processor) ProcessVisibility(ctx context.Context, form *apimodel.Advanc // If visibility isn't set on the form, then just take the account default. // If that's also not set, take the default for the whole instance. if form.Visibility != "" { - vis = p.tc.MastoVisToVis(form.Visibility) + vis = p.tc.APIVisToVis(form.Visibility) } else if accountDefaultVis != "" { vis = accountDefaultVis } else { diff --git a/internal/processing/streaming/authorize.go b/internal/processing/streaming/authorize.go index f938a0c0c..1a5724f51 100644 --- a/internal/processing/streaming/authorize.go +++ b/internal/processing/streaming/authorize.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming import ( @@ -8,7 +26,7 @@ import ( ) func (p *processor) AuthorizeStreamingRequest(ctx context.Context, accessToken string) (*gtsmodel.Account, error) { - ti, err := p.oauthServer.LoadAccessToken(context.Background(), accessToken) + ti, err := p.oauthServer.LoadAccessToken(ctx, accessToken) if err != nil { return nil, fmt.Errorf("AuthorizeStreamingRequest: error loading access token: %s", err) } diff --git a/internal/processing/streaming/authorize_test.go b/internal/processing/streaming/authorize_test.go new file mode 100644 index 000000000..f52396250 --- /dev/null +++ b/internal/processing/streaming/authorize_test.go @@ -0,0 +1,48 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/suite" +) + +type AuthorizeTestSuite struct { + StreamingTestSuite +} + +func (suite *AuthorizeTestSuite) TestAuthorize() { + account1, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), suite.testTokens["local_account_1"].Access) + suite.NoError(err) + suite.Equal(suite.testAccounts["local_account_1"].ID, account1.ID) + + account2, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), suite.testTokens["local_account_2"].Access) + suite.NoError(err) + suite.Equal(suite.testAccounts["local_account_2"].ID, account2.ID) + + noAccount, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), "aaaaaaaaaaaaaaaaaaaaa!!") + suite.EqualError(err, "AuthorizeStreamingRequest: error loading access token: no entries") + suite.Nil(noAccount) +} + +func TestAuthorizeTestSuite(t *testing.T) { + suite.Run(t, &AuthorizeTestSuite{}) +} diff --git a/internal/processing/streaming/notification.go b/internal/processing/streaming/notification.go new file mode 100644 index 000000000..870490be4 --- /dev/null +++ b/internal/processing/streaming/notification.go @@ -0,0 +1,37 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming + +import ( + "encoding/json" + "fmt" + + apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" + "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" + "github.com/superseriousbusiness/gotosocial/internal/stream" +) + +func (p *processor) StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error { + bytes, err := json.Marshal(n) + if err != nil { + return fmt.Errorf("error marshalling notification to json: %s", err) + } + + return p.streamToAccount(string(bytes), stream.EventTypeNotification, account.ID) +} diff --git a/internal/processing/streaming/notification_test.go b/internal/processing/streaming/notification_test.go new file mode 100644 index 000000000..fa77a8f92 --- /dev/null +++ b/internal/processing/streaming/notification_test.go @@ -0,0 +1,60 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/suite" + apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" + "github.com/superseriousbusiness/gotosocial/testrig" +) + +type NotificationTestSuite struct { + StreamingTestSuite +} + +func (suite *NotificationTestSuite) TestStreamNotification() { + account := suite.testAccounts["local_account_1"] + + openStream, errWithCode := suite.streamingProcessor.OpenStreamForAccount(context.Background(), account, "user") + suite.NoError(errWithCode) + + followAccount := suite.testAccounts["remote_account_1"] + followAccountAPIModel, err := testrig.NewTestTypeConverter(suite.db).AccountToAPIAccountPublic(context.Background(), followAccount) + suite.NoError(err) + + notification := &apimodel.Notification{ + ID: "01FH57SJCMDWQGEAJ0X08CE3WV", + Type: "follow", + CreatedAt: "2021-10-04T10:52:36+02:00", + Account: followAccountAPIModel, + } + + err = suite.streamingProcessor.StreamNotificationToAccount(notification, account) + suite.NoError(err) + + msg := <-openStream.Messages + suite.Equal(`{"id":"01FH57SJCMDWQGEAJ0X08CE3WV","type":"follow","created_at":"2021-10-04T10:52:36+02:00","account":{"id":"01F8MH5ZK5VRH73AKHQM6Y9VNX","username":"foss_satan","acct":"foss_satan@fossbros-anonymous.io","display_name":"big gerald","locked":false,"bot":false,"created_at":"2021-09-26T12:52:36+02:00","note":"i post about like, i dunno, stuff, or whatever!!!!","url":"http://fossbros-anonymous.io/@foss_satan","avatar":"","avatar_static":"","header":"","header_static":"","followers_count":0,"following_count":0,"statuses_count":0,"last_status_at":"","emojis":[],"fields":[]}}`, msg.Payload) +} + +func TestNotificationTestSuite(t *testing.T) { + suite.Run(t, &NotificationTestSuite{}) +} diff --git a/internal/processing/streaming/openstream.go b/internal/processing/streaming/openstream.go index d4e4eef9f..74b6486f5 100644 --- a/internal/processing/streaming/openstream.go +++ b/internal/processing/streaming/openstream.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming import ( diff --git a/internal/processing/streaming/openstream_test.go b/internal/processing/streaming/openstream_test.go new file mode 100644 index 000000000..6c4134997 --- /dev/null +++ b/internal/processing/streaming/openstream_test.go @@ -0,0 +1,41 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/suite" +) + +type OpenStreamTestSuite struct { + StreamingTestSuite +} + +func (suite *OpenStreamTestSuite) TestOpenStream() { + account := suite.testAccounts["local_account_1"] + + _, errWithCode := suite.streamingProcessor.OpenStreamForAccount(context.Background(), account, "user") + suite.NoError(errWithCode) +} + +func TestOpenStreamTestSuite(t *testing.T) { + suite.Run(t, &OpenStreamTestSuite{}) +} diff --git a/internal/processing/streaming/streamdelete.go b/internal/processing/streaming/streamdelete.go index cd541bc57..8332c37dc 100644 --- a/internal/processing/streaming/streamdelete.go +++ b/internal/processing/streaming/streamdelete.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming import ( @@ -10,39 +28,20 @@ import ( func (p *processor) StreamDelete(statusID string) error { errs := []string{} - // we want to range through ALL streams for ALL accounts here to make sure it's very clear to everyone that the status has been deleted - p.streamMap.Range(func(k interface{}, v interface{}) bool { - // the key of this map should be an accountID (string) - accountID, ok := k.(string) - if !ok { - errs = append(errs, "key in streamMap was not a string!") - return false - } - - // the value of the map should be a buncha streams - streamsForAccount, ok := v.(*stream.StreamsForAccount) - if !ok { - errs = append(errs, fmt.Sprintf("stream map error for account stream %s", accountID)) - } - - // lock the streams while we work on them - streamsForAccount.Lock() - defer streamsForAccount.Unlock() - for _, s := range streamsForAccount.Streams { - // lock each individual stream as we work on it - s.Lock() - defer s.Unlock() - if s.Connected { - s.Messages <- &stream.Message{ - Stream: []string{s.Type}, - Event: "delete", - Payload: statusID, - } - } - } + // get all account IDs with open streams + accountIDs := []string{} + p.streamMap.Range(func(k interface{}, _ interface{}) bool { + accountIDs = append(accountIDs, k.(string)) return true }) + // stream the delete to every account + for _, accountID := range accountIDs { + if err := p.streamToAccount(statusID, stream.EventTypeDelete, accountID); err != nil { + errs = append(errs, err.Error()) + } + } + if len(errs) != 0 { return fmt.Errorf("one or more errors streaming status delete: %s", strings.Join(errs, ";")) } diff --git a/internal/processing/streaming/streaming.go b/internal/processing/streaming/streaming.go index 610d4a9d2..abce30cd1 100644 --- a/internal/processing/streaming/streaming.go +++ b/internal/processing/streaming/streaming.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming import ( @@ -6,14 +24,11 @@ import ( "github.com/sirupsen/logrus" 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/oauth" "github.com/superseriousbusiness/gotosocial/internal/stream" - "github.com/superseriousbusiness/gotosocial/internal/typeutils" - "github.com/superseriousbusiness/gotosocial/internal/visibility" ) // Processor wraps a bunch of functions for processing streaming. @@ -22,8 +37,8 @@ type Processor interface { AuthorizeStreamingRequest(ctx context.Context, accessToken string) (*gtsmodel.Account, error) // OpenStreamForAccount returns a new Stream for the given account, which will contain a channel for passing messages back to the caller. OpenStreamForAccount(ctx context.Context, account *gtsmodel.Account, streamType string) (*stream.Stream, gtserror.WithCode) - // StreamStatusToAccount streams the given status to any open, appropriate streams belonging to the given account. - StreamStatusToAccount(s *apimodel.Status, account *gtsmodel.Account) error + // StreamUpdateToAccount streams the given update to any open, appropriate streams belonging to the given account. + StreamUpdateToAccount(s *apimodel.Status, account *gtsmodel.Account) error // StreamNotificationToAccount streams the given notification to any open, appropriate streams belonging to the given account. StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error // StreamDelete streams the delete of the given statusID to *ALL* open streams. @@ -31,22 +46,16 @@ type Processor interface { } type processor struct { - tc typeutils.TypeConverter - config *config.Config db db.DB - filter visibility.Filter log *logrus.Logger oauthServer oauth.Server streamMap *sync.Map } // New returns a new status processor. -func New(db db.DB, tc typeutils.TypeConverter, oauthServer oauth.Server, config *config.Config, log *logrus.Logger) Processor { +func New(db db.DB, oauthServer oauth.Server, log *logrus.Logger) Processor { return &processor{ - tc: tc, - config: config, db: db, - filter: visibility.NewFilter(db, log), log: log, oauthServer: oauthServer, streamMap: &sync.Map{}, diff --git a/internal/processing/streaming/streaming_test.go b/internal/processing/streaming/streaming_test.go new file mode 100644 index 000000000..acc090b06 --- /dev/null +++ b/internal/processing/streaming/streaming_test.go @@ -0,0 +1,55 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming_test + +import ( + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/suite" + "github.com/superseriousbusiness/gotosocial/internal/db" + "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" + "github.com/superseriousbusiness/gotosocial/internal/oauth" + "github.com/superseriousbusiness/gotosocial/internal/processing/streaming" + "github.com/superseriousbusiness/gotosocial/testrig" +) + +type StreamingTestSuite struct { + suite.Suite + testAccounts map[string]*gtsmodel.Account + testTokens map[string]*gtsmodel.Token + db db.DB + oauthServer oauth.Server + log *logrus.Logger + + streamingProcessor streaming.Processor +} + +func (suite *StreamingTestSuite) SetupTest() { + suite.testAccounts = testrig.NewTestAccounts() + suite.testTokens = testrig.NewTestTokens() + suite.db = testrig.NewTestDB() + suite.oauthServer = testrig.NewTestOauthServer(suite.db) + suite.log = testrig.NewTestLog() + suite.streamingProcessor = streaming.New(suite.db, suite.oauthServer, suite.log) + + testrig.StandardDBSetup(suite.db, suite.testAccounts) +} + +func (suite *StreamingTestSuite) TearDownTest() { + testrig.StandardDBTeardown(suite.db) +} diff --git a/internal/processing/streaming/streamnotification.go b/internal/processing/streaming/streamnotification.go deleted file mode 100644 index d8460874f..000000000 --- a/internal/processing/streaming/streamnotification.go +++ /dev/null @@ -1,51 +0,0 @@ -package streaming - -import ( - "encoding/json" - "errors" - "fmt" - - "github.com/sirupsen/logrus" - apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" - "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" - "github.com/superseriousbusiness/gotosocial/internal/stream" -) - -func (p *processor) StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error { - l := p.log.WithFields(logrus.Fields{ - "func": "StreamNotificationToAccount", - "account": account.ID, - }) - v, ok := p.streamMap.Load(account.ID) - if !ok { - // no open connections so nothing to stream - return nil - } - - streamsForAccount, ok := v.(*stream.StreamsForAccount) - if !ok { - return errors.New("stream map error") - } - - notificationBytes, err := json.Marshal(n) - if err != nil { - return fmt.Errorf("error marshalling notification to json: %s", err) - } - - streamsForAccount.Lock() - defer streamsForAccount.Unlock() - for _, s := range streamsForAccount.Streams { - s.Lock() - defer s.Unlock() - if s.Connected { - l.Debugf("streaming notification to stream id %s", s.ID) - s.Messages <- &stream.Message{ - Stream: []string{s.Type}, - Event: "notification", - Payload: string(notificationBytes), - } - } - } - - return nil -} diff --git a/internal/processing/streaming/streamstatus.go b/internal/processing/streaming/streamstatus.go deleted file mode 100644 index f4d6b2629..000000000 --- a/internal/processing/streaming/streamstatus.go +++ /dev/null @@ -1,51 +0,0 @@ -package streaming - -import ( - "encoding/json" - "errors" - "fmt" - - "github.com/sirupsen/logrus" - apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" - "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" - "github.com/superseriousbusiness/gotosocial/internal/stream" -) - -func (p *processor) StreamStatusToAccount(s *apimodel.Status, account *gtsmodel.Account) error { - l := p.log.WithFields(logrus.Fields{ - "func": "StreamStatusForAccount", - "account": account.ID, - }) - v, ok := p.streamMap.Load(account.ID) - if !ok { - // no open connections so nothing to stream - return nil - } - - streamsForAccount, ok := v.(*stream.StreamsForAccount) - if !ok { - return errors.New("stream map error") - } - - statusBytes, err := json.Marshal(s) - if err != nil { - return fmt.Errorf("error marshalling status to json: %s", err) - } - - streamsForAccount.Lock() - defer streamsForAccount.Unlock() - for _, s := range streamsForAccount.Streams { - s.Lock() - defer s.Unlock() - if s.Connected { - l.Debugf("streaming status to stream id %s", s.ID) - s.Messages <- &stream.Message{ - Stream: []string{s.Type}, - Event: "update", - Payload: string(statusBytes), - } - } - } - - return nil -} diff --git a/internal/processing/streaming/streamtoaccount.go b/internal/processing/streaming/streamtoaccount.go new file mode 100644 index 000000000..140910ab7 --- /dev/null +++ b/internal/processing/streaming/streamtoaccount.go @@ -0,0 +1,55 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming + +import ( + "errors" + + "github.com/superseriousbusiness/gotosocial/internal/stream" +) + +// streamToAccount streams the given payload with the given event type to any streams currently open for the given account ID. +func (p *processor) streamToAccount(payload string, event stream.EventType, accountID string) error { + v, ok := p.streamMap.Load(accountID) + if !ok { + // no open connections so nothing to stream + return nil + } + + streamsForAccount, ok := v.(*stream.StreamsForAccount) + if !ok { + return errors.New("stream map error") + } + + streamsForAccount.Lock() + defer streamsForAccount.Unlock() + for _, s := range streamsForAccount.Streams { + s.Lock() + defer s.Unlock() + if s.Connected { + s.Messages <- &stream.Message{ + Stream: []string{s.Type}, + Event: string(event), + Payload: payload, + } + } + } + + return nil +} diff --git a/internal/processing/streaming/update.go b/internal/processing/streaming/update.go new file mode 100644 index 000000000..da7dcb6ce --- /dev/null +++ b/internal/processing/streaming/update.go @@ -0,0 +1,37 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + 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 streaming + +import ( + "encoding/json" + "fmt" + + apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" + "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" + "github.com/superseriousbusiness/gotosocial/internal/stream" +) + +func (p *processor) StreamUpdateToAccount(s *apimodel.Status, account *gtsmodel.Account) error { + bytes, err := json.Marshal(s) + if err != nil { + return fmt.Errorf("error marshalling status to json: %s", err) + } + + return p.streamToAccount(string(bytes), stream.EventTypeUpdate, account.ID) +} diff --git a/internal/processing/timeline.go b/internal/processing/timeline.go index 6a409a6cc..249ef542d 100644 --- a/internal/processing/timeline.go +++ b/internal/processing/timeline.go @@ -151,9 +151,9 @@ func (p *processor) filterPublicStatuses(ctx context.Context, authed *oauth.Auth continue } - apiStatus, err := p.tc.StatusToMasto(ctx, s, authed.Account) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, s, authed.Account) if err != nil { - l.Debugf("filterPublicStatuses: skipping status %s because it couldn't be converted to its mastodon representation: %s", s.ID, err) + l.Debugf("filterPublicStatuses: skipping status %s because it couldn't be converted to its api representation: %s", s.ID, err) continue } @@ -186,9 +186,9 @@ func (p *processor) filterFavedStatuses(ctx context.Context, authed *oauth.Auth, continue } - apiStatus, err := p.tc.StatusToMasto(ctx, s, authed.Account) + apiStatus, err := p.tc.StatusToAPIStatus(ctx, s, authed.Account) if err != nil { - l.Debugf("filterFavedStatuses: skipping status %s because it couldn't be converted to its mastodon representation: %s", s.ID, err) + l.Debugf("filterFavedStatuses: skipping status %s because it couldn't be converted to its api representation: %s", s.ID, err) continue } |