Refactor/tidy (#261)

* tidy up streaming

* cut down code duplication

* test get followers/following

* test streaming processor

* fix some test models

* add TimeMustParse

* fix uri / url typo

* make trace logging less verbose

* make logging more consistent

* disable quote on logging

* remove context.Background

* remove many extraneous mastodon references

* regenerate swagger

* don't log query on no rows result

* log latency first for easier reading

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

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
 }

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

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

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

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)

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
 }

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:

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
 }

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
 }

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
 }

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
 }

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
 }

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
 		}

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

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

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

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

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
 }

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

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

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
 }

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)

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

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
 }

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
 }

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

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
 }

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
 }

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
 }

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
 }

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
 }

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
 }

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
 }

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 {

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

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

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)
+}

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

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 (

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

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, ";"))
 	}

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{},

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)
+}

internal/processing/streaming/streamnotification.go

@@ -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
-}

internal/processing/streaming/streamstatus.go

@@ -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
-}

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
+}

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)
+}

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
 		}