summaryrefslogtreecommitdiff
path: root/internal/federation
diff options
context:
space:
mode:
Diffstat (limited to 'internal/federation')
-rw-r--r--internal/federation/dereference.go8
-rw-r--r--internal/federation/dereferencing/account.go315
-rw-r--r--internal/federation/dereferencing/account_test.go3
-rw-r--r--internal/federation/dereferencing/attachment.go104
-rw-r--r--internal/federation/dereferencing/dereferencer.go64
-rw-r--r--internal/federation/dereferencing/dereferencer_test.go2
-rw-r--r--internal/federation/dereferencing/media.go55
-rw-r--r--internal/federation/dereferencing/media_test.go (renamed from internal/federation/dereferencing/attachment_test.go)90
-rw-r--r--internal/federation/dereferencing/status.go21
-rw-r--r--internal/federation/federatingprotocol.go2
-rw-r--r--internal/federation/federator.go11
-rw-r--r--internal/federation/federator_test.go4
12 files changed, 403 insertions, 276 deletions
diff --git a/internal/federation/dereference.go b/internal/federation/dereference.go
index 343ddadb7..8cb23a91f 100644
--- a/internal/federation/dereference.go
+++ b/internal/federation/dereference.go
@@ -26,12 +26,8 @@ import (
"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
)
-func (f *federator) GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, refresh bool) (*gtsmodel.Account, bool, error) {
- return f.dereferencer.GetRemoteAccount(ctx, username, remoteAccountID, refresh)
-}
-
-func (f *federator) EnrichRemoteAccount(ctx context.Context, username string, account *gtsmodel.Account) (*gtsmodel.Account, error) {
- return f.dereferencer.EnrichRemoteAccount(ctx, username, account)
+func (f *federator) GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, blocking bool, refresh bool) (*gtsmodel.Account, error) {
+ return f.dereferencer.GetRemoteAccount(ctx, username, remoteAccountID, blocking, refresh)
}
func (f *federator) GetRemoteStatus(ctx context.Context, username string, remoteStatusID *url.URL, refresh, includeParent bool) (*gtsmodel.Status, ap.Statusable, bool, error) {
diff --git a/internal/federation/dereferencing/account.go b/internal/federation/dereferencing/account.go
index d06ad21c1..d87192d3a 100644
--- a/internal/federation/dereferencing/account.go
+++ b/internal/federation/dereferencing/account.go
@@ -23,8 +23,11 @@ import (
"encoding/json"
"errors"
"fmt"
+ "io"
"net/url"
"strings"
+ "sync"
+ "time"
"github.com/sirupsen/logrus"
"github.com/superseriousbusiness/activity/streams"
@@ -32,6 +35,7 @@ import (
"github.com/superseriousbusiness/gotosocial/internal/ap"
"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
"github.com/superseriousbusiness/gotosocial/internal/id"
+ "github.com/superseriousbusiness/gotosocial/internal/media"
"github.com/superseriousbusiness/gotosocial/internal/transport"
)
@@ -42,94 +46,97 @@ func instanceAccount(account *gtsmodel.Account) bool {
(account.Username == "internal.fetch" && strings.Contains(account.Note, "internal service actor"))
}
-// EnrichRemoteAccount takes an account that's already been inserted into the database in a minimal form,
-// and populates it with additional fields, media, etc.
-//
-// EnrichRemoteAccount is mostly useful for calling after an account has been initially created by
-// the federatingDB's Create function, or during the federated authorization flow.
-func (d *deref) EnrichRemoteAccount(ctx context.Context, username string, account *gtsmodel.Account) (*gtsmodel.Account, error) {
- // if we're dealing with an instance account, we don't need to update anything
- if instanceAccount(account) {
- return account, nil
- }
-
- if err := d.PopulateAccountFields(ctx, account, username, false); err != nil {
- return nil, err
- }
-
- updated, err := d.db.UpdateAccount(ctx, account)
- if err != nil {
- logrus.Errorf("EnrichRemoteAccount: error updating account: %s", err)
- return account, nil
- }
-
- return updated, nil
-}
-
// GetRemoteAccount completely dereferences a remote account, converts it to a GtS model account,
-// puts it in the database, and returns it to a caller. The boolean indicates whether the account is new
-// to us or not. If we haven't seen the account before, bool will be true. If we have seen the account before,
-// it will be false.
+// puts it in the database, and returns it to a caller.
//
// Refresh indicates whether--if the account exists in our db already--it should be refreshed by calling
-// the remote instance again.
+// the remote instance again. Blocking indicates whether the function should block until processing of
+// the fetched account is complete.
//
// SIDE EFFECTS: remote account will be stored in the database, or updated if it already exists (and refresh is true).
-func (d *deref) GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, refresh bool) (*gtsmodel.Account, bool, error) {
+func (d *deref) GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, blocking bool, refresh bool) (*gtsmodel.Account, error) {
new := true
- // check if we already have the account in our db
- maybeAccount, err := d.db.GetAccountByURI(ctx, remoteAccountID.String())
+ // check if we already have the account in our db, and just return it unless we'd doing a refresh
+ remoteAccount, err := d.db.GetAccountByURI(ctx, remoteAccountID.String())
if err == nil {
- // we've seen this account before so it's not new
new = false
if !refresh {
- // we're not being asked to refresh, but just in case we don't have the avatar/header cached yet....
- maybeAccount, err = d.EnrichRemoteAccount(ctx, username, maybeAccount)
- return maybeAccount, new, err
+ // make sure the account fields are populated before returning:
+ // even if we're not doing a refresh, the caller might want to block
+ // until everything is loaded
+ changed, err := d.populateAccountFields(ctx, remoteAccount, username, refresh, blocking)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error populating remoteAccount fields: %s", err)
+ }
+
+ if changed {
+ updatedAccount, err := d.db.UpdateAccount(ctx, remoteAccount)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error updating remoteAccount: %s", err)
+ }
+ return updatedAccount, err
+ }
+
+ return remoteAccount, nil
}
}
- accountable, err := d.dereferenceAccountable(ctx, username, remoteAccountID)
- if err != nil {
- return nil, new, fmt.Errorf("FullyDereferenceAccount: error dereferencing accountable: %s", err)
- }
+ if new {
+ // we haven't seen this account before: dereference it from remote
+ accountable, err := d.dereferenceAccountable(ctx, username, remoteAccountID)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error dereferencing accountable: %s", err)
+ }
- gtsAccount, err := d.typeConverter.ASRepresentationToAccount(ctx, accountable, refresh)
- if err != nil {
- return nil, new, fmt.Errorf("FullyDereferenceAccount: error converting accountable to account: %s", err)
- }
+ newAccount, err := d.typeConverter.ASRepresentationToAccount(ctx, accountable, refresh)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error converting accountable to account: %s", err)
+ }
- if new {
- // generate a new id since we haven't seen this account before, and do a put
ulid, err := id.NewRandomULID()
if err != nil {
- return nil, new, fmt.Errorf("FullyDereferenceAccount: error generating new id for account: %s", err)
+ return nil, fmt.Errorf("GetRemoteAccount: error generating new id for account: %s", err)
}
- gtsAccount.ID = ulid
+ newAccount.ID = ulid
- if err := d.PopulateAccountFields(ctx, gtsAccount, username, refresh); err != nil {
- return nil, new, fmt.Errorf("FullyDereferenceAccount: error populating further account fields: %s", err)
+ if _, err := d.populateAccountFields(ctx, newAccount, username, refresh, blocking); err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error populating further account fields: %s", err)
}
- if err := d.db.Put(ctx, gtsAccount); err != nil {
- return nil, new, fmt.Errorf("FullyDereferenceAccount: error putting new account: %s", err)
+ if err := d.db.Put(ctx, newAccount); err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error putting new account: %s", err)
}
- } else {
- // take the id we already have and do an update
- gtsAccount.ID = maybeAccount.ID
- if err := d.PopulateAccountFields(ctx, gtsAccount, username, refresh); err != nil {
- return nil, new, fmt.Errorf("FullyDereferenceAccount: error populating further account fields: %s", err)
- }
+ return newAccount, nil
+ }
- gtsAccount, err = d.db.UpdateAccount(ctx, gtsAccount)
+ // we have seen this account before, but we have to refresh it
+ refreshedAccountable, err := d.dereferenceAccountable(ctx, username, remoteAccountID)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error dereferencing refreshedAccountable: %s", err)
+ }
+
+ refreshedAccount, err := d.typeConverter.ASRepresentationToAccount(ctx, refreshedAccountable, refresh)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error converting refreshedAccountable to refreshedAccount: %s", err)
+ }
+ refreshedAccount.ID = remoteAccount.ID
+
+ changed, err := d.populateAccountFields(ctx, refreshedAccount, username, refresh, blocking)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteAccount: error populating further refreshedAccount fields: %s", err)
+ }
+
+ if changed {
+ updatedAccount, err := d.db.UpdateAccount(ctx, refreshedAccount)
if err != nil {
- return nil, false, fmt.Errorf("EnrichRemoteAccount: error updating account: %s", err)
+ return nil, fmt.Errorf("GetRemoteAccount: error updating refreshedAccount: %s", err)
}
+ return updatedAccount, nil
}
- return gtsAccount, new, nil
+ return refreshedAccount, nil
}
// dereferenceAccountable calls remoteAccountID with a GET request, and tries to parse whatever
@@ -200,71 +207,191 @@ func (d *deref) dereferenceAccountable(ctx context.Context, username string, rem
return nil, fmt.Errorf("DereferenceAccountable: type name %s not supported", t.GetTypeName())
}
-// PopulateAccountFields populates any fields on the given account that weren't populated by the initial
+// populateAccountFields populates any fields on the given account that weren't populated by the initial
// dereferencing. This includes things like header and avatar etc.
-func (d *deref) PopulateAccountFields(ctx context.Context, account *gtsmodel.Account, requestingUsername string, refresh bool) error {
- l := logrus.WithFields(logrus.Fields{
- "func": "PopulateAccountFields",
- "requestingUsername": requestingUsername,
- })
+func (d *deref) populateAccountFields(ctx context.Context, account *gtsmodel.Account, requestingUsername string, blocking bool, refresh bool) (bool, error) {
+ // if we're dealing with an instance account, just bail, we don't need to do anything
+ if instanceAccount(account) {
+ return false, nil
+ }
accountURI, err := url.Parse(account.URI)
if err != nil {
- return fmt.Errorf("PopulateAccountFields: couldn't parse account URI %s: %s", account.URI, err)
+ return false, fmt.Errorf("populateAccountFields: couldn't parse account URI %s: %s", account.URI, err)
}
+
if blocked, err := d.db.IsDomainBlocked(ctx, accountURI.Host); blocked || err != nil {
- return fmt.Errorf("PopulateAccountFields: domain %s is blocked", accountURI.Host)
+ return false, fmt.Errorf("populateAccountFields: domain %s is blocked", accountURI.Host)
}
t, err := d.transportController.NewTransportForUsername(ctx, requestingUsername)
if err != nil {
- return fmt.Errorf("PopulateAccountFields: error getting transport for user: %s", err)
+ return false, fmt.Errorf("populateAccountFields: error getting transport for user: %s", err)
}
// fetch the header and avatar
- if err := d.fetchHeaderAndAviForAccount(ctx, account, t, refresh); err != nil {
- // if this doesn't work, just skip it -- we can do it later
- l.Debugf("error fetching header/avi for account: %s", err)
+ changed, err := d.fetchRemoteAccountMedia(ctx, account, t, refresh, blocking)
+ if err != nil {
+ return false, fmt.Errorf("populateAccountFields: error fetching header/avi for account: %s", err)
}
- return nil
+ return changed, nil
}
-// fetchHeaderAndAviForAccount fetches the header and avatar for a remote account, using a transport
-// on behalf of requestingUsername.
+// fetchRemoteAccountMedia fetches and stores the header and avatar for a remote account,
+// using a transport on behalf of requestingUsername.
+//
+// The returned boolean indicates whether anything changed -- in other words, whether the
+// account should be updated in the database.
//
// targetAccount's AvatarMediaAttachmentID and HeaderMediaAttachmentID will be updated as necessary.
//
-// SIDE EFFECTS: remote header and avatar will be stored in local storage.
-func (d *deref) fetchHeaderAndAviForAccount(ctx context.Context, targetAccount *gtsmodel.Account, t transport.Transport, refresh bool) error {
+// If refresh is true, then the media will be fetched again even if it's already been fetched before.
+//
+// If blocking is true, then the calls to the media manager made by this function will be blocking:
+// in other words, the function won't return until the header and the avatar have been fully processed.
+func (d *deref) fetchRemoteAccountMedia(ctx context.Context, targetAccount *gtsmodel.Account, t transport.Transport, blocking bool, refresh bool) (bool, error) {
+ changed := false
+
accountURI, err := url.Parse(targetAccount.URI)
if err != nil {
- return fmt.Errorf("fetchHeaderAndAviForAccount: couldn't parse account URI %s: %s", targetAccount.URI, err)
+ return changed, fmt.Errorf("fetchRemoteAccountMedia: couldn't parse account URI %s: %s", targetAccount.URI, err)
}
+
if blocked, err := d.db.IsDomainBlocked(ctx, accountURI.Host); blocked || err != nil {
- return fmt.Errorf("fetchHeaderAndAviForAccount: domain %s is blocked", accountURI.Host)
+ return changed, fmt.Errorf("fetchRemoteAccountMedia: domain %s is blocked", accountURI.Host)
}
if targetAccount.AvatarRemoteURL != "" && (targetAccount.AvatarMediaAttachmentID == "" || refresh) {
- a, err := d.mediaHandler.ProcessRemoteHeaderOrAvatar(ctx, t, &gtsmodel.MediaAttachment{
- RemoteURL: targetAccount.AvatarRemoteURL,
- Avatar: true,
- }, targetAccount.ID)
- if err != nil {
- return fmt.Errorf("error processing avatar for user: %s", err)
+ var processingMedia *media.ProcessingMedia
+
+ // first check if we're already processing this media
+ d.dereferencingAvatarsLock.Lock()
+ if alreadyProcessing, ok := d.dereferencingAvatars[targetAccount.ID]; ok {
+ // we're already on it, no worries
+ processingMedia = alreadyProcessing
}
- targetAccount.AvatarMediaAttachmentID = a.ID
+ d.dereferencingAvatarsLock.Unlock()
+
+ if processingMedia == nil {
+ // we're not already processing it so start now
+ avatarIRI, err := url.Parse(targetAccount.AvatarRemoteURL)
+ if err != nil {
+ return changed, err
+ }
+
+ data := func(innerCtx context.Context) (io.Reader, int, error) {
+ return t.DereferenceMedia(innerCtx, avatarIRI)
+ }
+
+ avatar := true
+ newProcessing, err := d.mediaManager.ProcessMedia(ctx, data, targetAccount.ID, &media.AdditionalMediaInfo{
+ RemoteURL: &targetAccount.AvatarRemoteURL,
+ Avatar: &avatar,
+ })
+ if err != nil {
+ return changed, err
+ }
+
+ // store it in our map to indicate it's in process
+ d.dereferencingAvatarsLock.Lock()
+ d.dereferencingAvatars[targetAccount.ID] = newProcessing
+ d.dereferencingAvatarsLock.Unlock()
+
+ processingMedia = newProcessing
+ }
+
+ // block until loaded if required...
+ if blocking {
+ if err := lockAndLoad(ctx, d.dereferencingAvatarsLock, processingMedia, d.dereferencingAvatars, targetAccount.ID); err != nil {
+ return changed, err
+ }
+ } else {
+ // ...otherwise do it async
+ go func() {
+ dlCtx, done := context.WithDeadline(context.Background(), time.Now().Add(1*time.Minute))
+ if err := lockAndLoad(dlCtx, d.dereferencingAvatarsLock, processingMedia, d.dereferencingAvatars, targetAccount.ID); err != nil {
+ logrus.Errorf("fetchRemoteAccountMedia: error during async lock and load of avatar: %s", err)
+ }
+ done()
+ }()
+ }
+
+ targetAccount.AvatarMediaAttachmentID = processingMedia.AttachmentID()
+ changed = true
}
if targetAccount.HeaderRemoteURL != "" && (targetAccount.HeaderMediaAttachmentID == "" || refresh) {
- a, err := d.mediaHandler.ProcessRemoteHeaderOrAvatar(ctx, t, &gtsmodel.MediaAttachment{
- RemoteURL: targetAccount.HeaderRemoteURL,
- Header: true,
- }, targetAccount.ID)
- if err != nil {
- return fmt.Errorf("error processing header for user: %s", err)
+ var processingMedia *media.ProcessingMedia
+
+ // first check if we're already processing this media
+ d.dereferencingHeadersLock.Lock()
+ if alreadyProcessing, ok := d.dereferencingHeaders[targetAccount.ID]; ok {
+ // we're already on it, no worries
+ processingMedia = alreadyProcessing
+ }
+ d.dereferencingHeadersLock.Unlock()
+
+ if processingMedia == nil {
+ // we're not already processing it so start now
+ headerIRI, err := url.Parse(targetAccount.HeaderRemoteURL)
+ if err != nil {
+ return changed, err
+ }
+
+ data := func(innerCtx context.Context) (io.Reader, int, error) {
+ return t.DereferenceMedia(innerCtx, headerIRI)
+ }
+
+ header := true
+ newProcessing, err := d.mediaManager.ProcessMedia(ctx, data, targetAccount.ID, &media.AdditionalMediaInfo{
+ RemoteURL: &targetAccount.HeaderRemoteURL,
+ Header: &header,
+ })
+ if err != nil {
+ return changed, err
+ }
+
+ // store it in our map to indicate it's in process
+ d.dereferencingHeadersLock.Lock()
+ d.dereferencingHeaders[targetAccount.ID] = newProcessing
+ d.dereferencingHeadersLock.Unlock()
+
+ processingMedia = newProcessing
}
- targetAccount.HeaderMediaAttachmentID = a.ID
+
+ // block until loaded if required...
+ if blocking {
+ if err := lockAndLoad(ctx, d.dereferencingHeadersLock, processingMedia, d.dereferencingHeaders, targetAccount.ID); err != nil {
+ return changed, err
+ }
+ } else {
+ // ...otherwise do it async
+ go func() {
+ dlCtx, done := context.WithDeadline(context.Background(), time.Now().Add(1*time.Minute))
+ if err := lockAndLoad(dlCtx, d.dereferencingHeadersLock, processingMedia, d.dereferencingHeaders, targetAccount.ID); err != nil {
+ logrus.Errorf("fetchRemoteAccountMedia: error during async lock and load of header: %s", err)
+ }
+ done()
+ }()
+ }
+
+ targetAccount.HeaderMediaAttachmentID = processingMedia.AttachmentID()
+ changed = true
}
- return nil
+
+ return changed, nil
+}
+
+func lockAndLoad(ctx context.Context, lock *sync.Mutex, processing *media.ProcessingMedia, processingMap map[string]*media.ProcessingMedia, accountID string) error {
+ // whatever happens, remove the in-process media from the map
+ defer func() {
+ lock.Lock()
+ delete(processingMap, accountID)
+ lock.Unlock()
+ }()
+
+ // try and load it
+ _, err := processing.LoadAttachment(ctx)
+ return err
}
diff --git a/internal/federation/dereferencing/account_test.go b/internal/federation/dereferencing/account_test.go
index 593ad341c..cb6f9588c 100644
--- a/internal/federation/dereferencing/account_test.go
+++ b/internal/federation/dereferencing/account_test.go
@@ -35,11 +35,10 @@ func (suite *AccountTestSuite) TestDereferenceGroup() {
fetchingAccount := suite.testAccounts["local_account_1"]
groupURL := testrig.URLMustParse("https://unknown-instance.com/groups/some_group")
- group, new, err := suite.dereferencer.GetRemoteAccount(context.Background(), fetchingAccount.Username, groupURL, false)
+ group, err := suite.dereferencer.GetRemoteAccount(context.Background(), fetchingAccount.Username, groupURL, false, false)
suite.NoError(err)
suite.NotNil(group)
suite.NotNil(group)
- suite.True(new)
// group values should be set
suite.Equal("https://unknown-instance.com/groups/some_group", group.URI)
diff --git a/internal/federation/dereferencing/attachment.go b/internal/federation/dereferencing/attachment.go
deleted file mode 100644
index 36ff2734c..000000000
--- a/internal/federation/dereferencing/attachment.go
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- GoToSocial
- Copyright (C) 2021-2022 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 dereferencing
-
-import (
- "context"
- "errors"
- "fmt"
- "net/url"
-
- "github.com/sirupsen/logrus"
- "github.com/superseriousbusiness/gotosocial/internal/db"
- "github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
-)
-
-func (d *deref) GetRemoteAttachment(ctx context.Context, requestingUsername string, minAttachment *gtsmodel.MediaAttachment) (*gtsmodel.MediaAttachment, error) {
- if minAttachment.RemoteURL == "" {
- return nil, fmt.Errorf("GetRemoteAttachment: minAttachment remote URL was empty")
- }
- remoteAttachmentURL := minAttachment.RemoteURL
-
- l := logrus.WithFields(logrus.Fields{
- "username": requestingUsername,
- "remoteAttachmentURL": remoteAttachmentURL,
- })
-
- // return early if we already have the attachment somewhere
- maybeAttachment := &gtsmodel.MediaAttachment{}
- where := []db.Where{
- {
- Key: "remote_url",
- Value: remoteAttachmentURL,
- },
- }
-
- if err := d.db.GetWhere(ctx, where, maybeAttachment); err == nil {
- // we already the attachment in the database
- l.Debugf("GetRemoteAttachment: attachment already exists with id %s", maybeAttachment.ID)
- return maybeAttachment, nil
- }
-
- a, err := d.RefreshAttachment(ctx, requestingUsername, minAttachment)
- if err != nil {
- return nil, fmt.Errorf("GetRemoteAttachment: error refreshing attachment: %s", err)
- }
-
- if err := d.db.Put(ctx, a); err != nil {
- var alreadyExistsError *db.ErrAlreadyExists
- if !errors.As(err, &alreadyExistsError) {
- return nil, fmt.Errorf("GetRemoteAttachment: error inserting attachment: %s", err)
- }
- }
-
- return a, nil
-}
-
-func (d *deref) RefreshAttachment(ctx context.Context, requestingUsername string, minAttachment *gtsmodel.MediaAttachment) (*gtsmodel.MediaAttachment, error) {
- // it just doesn't exist or we have to refresh
- if minAttachment.AccountID == "" {
- return nil, fmt.Errorf("RefreshAttachment: minAttachment account ID was empty")
- }
-
- if minAttachment.File.ContentType == "" {
- return nil, fmt.Errorf("RefreshAttachment: minAttachment.file.contentType was empty")
- }
-
- t, err := d.transportController.NewTransportForUsername(ctx, requestingUsername)
- if err != nil {
- return nil, fmt.Errorf("RefreshAttachment: error creating transport: %s", err)
- }
-
- derefURI, err := url.Parse(minAttachment.RemoteURL)
- if err != nil {
- return nil, err
- }
-
- attachmentBytes, err := t.DereferenceMedia(ctx, derefURI, minAttachment.File.ContentType)
- if err != nil {
- return nil, fmt.Errorf("RefreshAttachment: error dereferencing media: %s", err)
- }
-
- a, err := d.mediaHandler.ProcessAttachment(ctx, attachmentBytes, minAttachment)
- if err != nil {
- return nil, fmt.Errorf("RefreshAttachment: error processing attachment: %s", err)
- }
-
- return a, nil
-}
diff --git a/internal/federation/dereferencing/dereferencer.go b/internal/federation/dereferencing/dereferencer.go
index d0b653920..855c4baf8 100644
--- a/internal/federation/dereferencing/dereferencer.go
+++ b/internal/federation/dereferencing/dereferencer.go
@@ -33,42 +33,14 @@ import (
// Dereferencer wraps logic and functionality for doing dereferencing of remote accounts, statuses, etc, from federated instances.
type Dereferencer interface {
- GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, refresh bool) (*gtsmodel.Account, bool, error)
- EnrichRemoteAccount(ctx context.Context, username string, account *gtsmodel.Account) (*gtsmodel.Account, error)
+ GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, blocking bool, refresh bool) (*gtsmodel.Account, error)
GetRemoteStatus(ctx context.Context, username string, remoteStatusID *url.URL, refresh, includeParent bool) (*gtsmodel.Status, ap.Statusable, bool, error)
EnrichRemoteStatus(ctx context.Context, username string, status *gtsmodel.Status, includeParent bool) (*gtsmodel.Status, error)
GetRemoteInstance(ctx context.Context, username string, remoteInstanceURI *url.URL) (*gtsmodel.Instance, error)
- // GetRemoteAttachment takes a minimal attachment struct and converts it into a fully fleshed out attachment, stored in the database and instance storage.
- //
- // The parameter minAttachment must have at least the following fields defined:
- // * minAttachment.RemoteURL
- // * minAttachment.AccountID
- // * minAttachment.File.ContentType
- //
- // The returned attachment will have an ID generated for it, so no need to generate one beforehand.
- // A blurhash will also be generated for the attachment.
- //
- // Most other fields will be preserved on the passed attachment, including:
- // * minAttachment.StatusID
- // * minAttachment.CreatedAt
- // * minAttachment.UpdatedAt
- // * minAttachment.FileMeta
- // * minAttachment.AccountID
- // * minAttachment.Description
- // * minAttachment.ScheduledStatusID
- // * minAttachment.Thumbnail.RemoteURL
- // * minAttachment.Avatar
- // * minAttachment.Header
- //
- // GetRemoteAttachment will return early if an attachment with the same value as minAttachment.RemoteURL
- // is found in the database -- then that attachment will be returned and nothing else will be changed or stored.
- GetRemoteAttachment(ctx context.Context, requestingUsername string, minAttachment *gtsmodel.MediaAttachment) (*gtsmodel.MediaAttachment, error)
- // RefreshAttachment is like GetRemoteAttachment, but the attachment will always be dereferenced again,
- // whether or not it was already stored in the database.
- RefreshAttachment(ctx context.Context, requestingUsername string, minAttachment *gtsmodel.MediaAttachment) (*gtsmodel.MediaAttachment, error)
+ GetRemoteMedia(ctx context.Context, requestingUsername string, accountID string, remoteURL string, ai *media.AdditionalMediaInfo) (*media.ProcessingMedia, error)
DereferenceAnnounce(ctx context.Context, announce *gtsmodel.Status, requestingUsername string) error
DereferenceThread(ctx context.Context, username string, statusIRI *url.URL) error
@@ -77,21 +49,29 @@ type Dereferencer interface {
}
type deref struct {
- db db.DB
- typeConverter typeutils.TypeConverter
- transportController transport.Controller
- mediaHandler media.Handler
- handshakes map[string][]*url.URL
- handshakeSync *sync.Mutex // mutex to lock/unlock when checking or updating the handshakes map
+ db db.DB
+ typeConverter typeutils.TypeConverter
+ transportController transport.Controller
+ mediaManager media.Manager
+ dereferencingAvatars map[string]*media.ProcessingMedia
+ dereferencingAvatarsLock *sync.Mutex
+ dereferencingHeaders map[string]*media.ProcessingMedia
+ dereferencingHeadersLock *sync.Mutex
+ handshakes map[string][]*url.URL
+ handshakeSync *sync.Mutex // mutex to lock/unlock when checking or updating the handshakes map
}
// NewDereferencer returns a Dereferencer initialized with the given parameters.
-func NewDereferencer(db db.DB, typeConverter typeutils.TypeConverter, transportController transport.Controller, mediaHandler media.Handler) Dereferencer {
+func NewDereferencer(db db.DB, typeConverter typeutils.TypeConverter, transportController transport.Controller, mediaManager media.Manager) Dereferencer {
return &deref{
- db: db,
- typeConverter: typeConverter,
- transportController: transportController,
- mediaHandler: mediaHandler,
- handshakeSync: &sync.Mutex{},
+ db: db,
+ typeConverter: typeConverter,
+ transportController: transportController,
+ mediaManager: mediaManager,
+ dereferencingAvatars: make(map[string]*media.ProcessingMedia),
+ dereferencingAvatarsLock: &sync.Mutex{},
+ dereferencingHeaders: make(map[string]*media.ProcessingMedia),
+ dereferencingHeadersLock: &sync.Mutex{},
+ handshakeSync: &sync.Mutex{},
}
}
diff --git a/internal/federation/dereferencing/dereferencer_test.go b/internal/federation/dereferencing/dereferencer_test.go
index 569e8e93b..fe66abce4 100644
--- a/internal/federation/dereferencing/dereferencer_test.go
+++ b/internal/federation/dereferencing/dereferencer_test.go
@@ -64,7 +64,7 @@ func (suite *DereferencerStandardTestSuite) SetupTest() {
suite.db = testrig.NewTestDB()
suite.storage = testrig.NewTestStorage()
- suite.dereferencer = dereferencing.NewDereferencer(suite.db, testrig.NewTestTypeConverter(suite.db), suite.mockTransportController(), testrig.NewTestMediaHandler(suite.db, suite.storage))
+ suite.dereferencer = dereferencing.NewDereferencer(suite.db, testrig.NewTestTypeConverter(suite.db), suite.mockTransportController(), testrig.NewTestMediaManager(suite.db, suite.storage))
testrig.StandardDBSetup(suite.db, nil)
}
diff --git a/internal/federation/dereferencing/media.go b/internal/federation/dereferencing/media.go
new file mode 100644
index 000000000..0b19570f2
--- /dev/null
+++ b/internal/federation/dereferencing/media.go
@@ -0,0 +1,55 @@
+/*
+ GoToSocial
+ Copyright (C) 2021-2022 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 dereferencing
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "net/url"
+
+ "github.com/superseriousbusiness/gotosocial/internal/media"
+)
+
+func (d *deref) GetRemoteMedia(ctx context.Context, requestingUsername string, accountID string, remoteURL string, ai *media.AdditionalMediaInfo) (*media.ProcessingMedia, error) {
+ if accountID == "" {
+ return nil, fmt.Errorf("GetRemoteMedia: account ID was empty")
+ }
+
+ t, err := d.transportController.NewTransportForUsername(ctx, requestingUsername)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteMedia: error creating transport: %s", err)
+ }
+
+ derefURI, err := url.Parse(remoteURL)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteMedia: error parsing url: %s", err)
+ }
+
+ dataFunc := func(innerCtx context.Context) (io.Reader, int, error) {
+ return t.DereferenceMedia(innerCtx, derefURI)
+ }
+
+ processingMedia, err := d.mediaManager.ProcessMedia(ctx, dataFunc, accountID, ai)
+ if err != nil {
+ return nil, fmt.Errorf("GetRemoteMedia: error processing attachment: %s", err)
+ }
+
+ return processingMedia, nil
+}
diff --git a/internal/federation/dereferencing/attachment_test.go b/internal/federation/dereferencing/media_test.go
index d07cf1c6a..26d5c0c49 100644
--- a/internal/federation/dereferencing/attachment_test.go
+++ b/internal/federation/dereferencing/media_test.go
@@ -20,17 +20,22 @@ package dereferencing_test
import (
"context"
+ "fmt"
"testing"
+ "time"
"github.com/stretchr/testify/suite"
"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+ "github.com/superseriousbusiness/gotosocial/internal/media"
)
type AttachmentTestSuite struct {
DereferencerStandardTestSuite
}
-func (suite *AttachmentTestSuite) TestDereferenceAttachmentOK() {
+func (suite *AttachmentTestSuite) TestDereferenceAttachmentBlocking() {
+ ctx := context.Background()
+
fetchingAccount := suite.testAccounts["local_account_1"]
attachmentOwner := "01FENS9F666SEQ6TYQWEEY78GM"
@@ -38,19 +43,20 @@ func (suite *AttachmentTestSuite) TestDereferenceAttachmentOK() {
attachmentContentType := "image/jpeg"
attachmentURL := "https://s3-us-west-2.amazonaws.com/plushcity/media_attachments/files/106/867/380/219/163/828/original/88e8758c5f011439.jpg"
attachmentDescription := "It's a cute plushie."
+ attachmentBlurhash := "LwP?p=aK_4%N%MRjWXt7%hozM_a}"
+
+ media, err := suite.dereferencer.GetRemoteMedia(ctx, fetchingAccount.Username, attachmentOwner, attachmentURL, &media.AdditionalMediaInfo{
+ StatusID: &attachmentStatus,
+ RemoteURL: &attachmentURL,
+ Description: &attachmentDescription,
+ Blurhash: &attachmentBlurhash,
+ })
+ suite.NoError(err)
- minAttachment := &gtsmodel.MediaAttachment{
- RemoteURL: attachmentURL,
- AccountID: attachmentOwner,
- StatusID: attachmentStatus,
- File: gtsmodel.File{
- ContentType: attachmentContentType,
- },
- Description: attachmentDescription,
- }
-
- attachment, err := suite.dereferencer.GetRemoteAttachment(context.Background(), fetchingAccount.Username, minAttachment)
+ // make a blocking call to load the attachment from the in-process media
+ attachment, err := media.LoadAttachment(ctx)
suite.NoError(err)
+
suite.NotNil(attachment)
suite.Equal(attachmentOwner, attachment.AccountID)
@@ -65,7 +71,7 @@ func (suite *AttachmentTestSuite) TestDereferenceAttachmentOK() {
suite.Equal(2071680, attachment.FileMeta.Original.Size)
suite.Equal(1245, attachment.FileMeta.Original.Height)
suite.Equal(1664, attachment.FileMeta.Original.Width)
- suite.Equal("LwP?p=aK_4%N%MRjWXt7%hozM_a}", attachment.Blurhash)
+ suite.Equal(attachmentBlurhash, attachment.Blurhash)
suite.Equal(gtsmodel.ProcessingStatusProcessed, attachment.Processing)
suite.NotEmpty(attachment.File.Path)
suite.Equal(attachmentContentType, attachment.File.ContentType)
@@ -91,7 +97,7 @@ func (suite *AttachmentTestSuite) TestDereferenceAttachmentOK() {
suite.Equal(2071680, dbAttachment.FileMeta.Original.Size)
suite.Equal(1245, dbAttachment.FileMeta.Original.Height)
suite.Equal(1664, dbAttachment.FileMeta.Original.Width)
- suite.Equal("LwP?p=aK_4%N%MRjWXt7%hozM_a}", dbAttachment.Blurhash)
+ suite.Equal(attachmentBlurhash, dbAttachment.Blurhash)
suite.Equal(gtsmodel.ProcessingStatusProcessed, dbAttachment.Processing)
suite.NotEmpty(dbAttachment.File.Path)
suite.Equal(attachmentContentType, dbAttachment.File.ContentType)
@@ -101,6 +107,62 @@ func (suite *AttachmentTestSuite) TestDereferenceAttachmentOK() {
suite.NotEmpty(dbAttachment.Type)
}
+func (suite *AttachmentTestSuite) TestDereferenceAttachmentAsync() {
+ ctx := context.Background()
+
+ fetchingAccount := suite.testAccounts["local_account_1"]
+
+ attachmentOwner := "01FENS9F666SEQ6TYQWEEY78GM"
+ attachmentStatus := "01FENS9NTTVNEX1YZV7GB63MT8"
+ attachmentContentType := "image/jpeg"
+ attachmentURL := "https://s3-us-west-2.amazonaws.com/plushcity/media_attachments/files/106/867/380/219/163/828/original/88e8758c5f011439.jpg"
+ attachmentDescription := "It's a cute plushie."
+ attachmentBlurhash := "LwP?p=aK_4%N%MRjWXt7%hozM_a}"
+
+ processingMedia, err := suite.dereferencer.GetRemoteMedia(ctx, fetchingAccount.Username, attachmentOwner, attachmentURL, &media.AdditionalMediaInfo{
+ StatusID: &attachmentStatus,
+ RemoteURL: &attachmentURL,
+ Description: &attachmentDescription,
+ Blurhash: &attachmentBlurhash,
+ })
+ suite.NoError(err)
+ attachmentID := processingMedia.AttachmentID()
+
+ // wait for the media to finish processing
+ for finished := processingMedia.Finished(); !finished; finished = processingMedia.Finished() {
+ time.Sleep(10 * time.Millisecond)
+ fmt.Printf("\n\nnot finished yet...\n\n")
+ }
+ fmt.Printf("\n\nfinished!\n\n")
+
+ // now get the attachment from the database
+ attachment, err := suite.db.GetAttachmentByID(ctx, attachmentID)
+ suite.NoError(err)
+
+ suite.NotNil(attachment)
+
+ suite.Equal(attachmentOwner, attachment.AccountID)
+ suite.Equal(attachmentStatus, attachment.StatusID)
+ suite.Equal(attachmentURL, attachment.RemoteURL)
+ suite.NotEmpty(attachment.URL)
+ suite.NotEmpty(attachment.Blurhash)
+ suite.NotEmpty(attachment.ID)
+ suite.NotEmpty(attachment.CreatedAt)
+ suite.NotEmpty(attachment.UpdatedAt)
+ suite.Equal(1.336546184738956, attachment.FileMeta.Original.Aspect)
+ suite.Equal(2071680, attachment.FileMeta.Original.Size)
+ suite.Equal(1245, attachment.FileMeta.Original.Height)
+ suite.Equal(1664, attachment.FileMeta.Original.Width)
+ suite.Equal(attachmentBlurhash, attachment.Blurhash)
+ suite.Equal(gtsmodel.ProcessingStatusProcessed, attachment.Processing)
+ suite.NotEmpty(attachment.File.Path)
+ suite.Equal(attachmentContentType, attachment.File.ContentType)
+ suite.Equal(attachmentDescription, attachment.Description)
+
+ suite.NotEmpty(attachment.Thumbnail.Path)
+ suite.NotEmpty(attachment.Type)
+}
+
func TestAttachmentTestSuite(t *testing.T) {
suite.Run(t, new(AttachmentTestSuite))
}
diff --git a/internal/federation/dereferencing/status.go b/internal/federation/dereferencing/status.go
index d7de5936a..cacca91b2 100644
--- a/internal/federation/dereferencing/status.go
+++ b/internal/federation/dereferencing/status.go
@@ -32,6 +32,7 @@ import (
"github.com/superseriousbusiness/gotosocial/internal/ap"
"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
"github.com/superseriousbusiness/gotosocial/internal/id"
+ "github.com/superseriousbusiness/gotosocial/internal/media"
)
// EnrichRemoteStatus takes a status that's already been inserted into the database in a minimal form,
@@ -88,7 +89,7 @@ func (d *deref) GetRemoteStatus(ctx context.Context, username string, remoteStat
}
// do this so we know we have the remote account of the status in the db
- _, _, err = d.GetRemoteAccount(ctx, username, accountURI, false)
+ _, err = d.GetRemoteAccount(ctx, username, accountURI, true, false)
if err != nil {
return nil, statusable, new, fmt.Errorf("GetRemoteStatus: couldn't derive status author: %s", err)
}
@@ -331,7 +332,7 @@ func (d *deref) populateStatusMentions(ctx context.Context, status *gtsmodel.Sta
if targetAccount == nil {
// we didn't find the account in our database already
// check if we can get the account remotely (dereference it)
- if a, _, err := d.GetRemoteAccount(ctx, requestingUsername, targetAccountURI, false); err != nil {
+ if a, err := d.GetRemoteAccount(ctx, requestingUsername, targetAccountURI, false, false); err != nil {
errs = append(errs, err.Error())
} else {
logrus.Debugf("populateStatusMentions: got target account %s with id %s through GetRemoteAccount", targetAccountURI, a.ID)
@@ -393,9 +394,21 @@ func (d *deref) populateStatusAttachments(ctx context.Context, status *gtsmodel.
a.AccountID = status.AccountID
a.StatusID = status.ID
- attachment, err := d.GetRemoteAttachment(ctx, requestingUsername, a)
+ processingMedia, err := d.GetRemoteMedia(ctx, requestingUsername, a.AccountID, a.RemoteURL, &media.AdditionalMediaInfo{
+ CreatedAt: &a.CreatedAt,
+ StatusID: &a.StatusID,
+ RemoteURL: &a.RemoteURL,
+ Description: &a.Description,
+ Blurhash: &a.Blurhash,
+ })
if err != nil {
- logrus.Errorf("populateStatusAttachments: couldn't get remote attachment %s: %s", a.RemoteURL, err)
+ logrus.Errorf("populateStatusAttachments: couldn't get remote media %s: %s", a.RemoteURL, err)
+ continue
+ }
+
+ attachment, err := processingMedia.LoadAttachment(ctx)
+ if err != nil {
+ logrus.Errorf("populateStatusAttachments: couldn't load remote attachment %s: %s", a.RemoteURL, err)
continue
}
diff --git a/internal/federation/federatingprotocol.go b/internal/federation/federatingprotocol.go
index f5d42a8e3..789959810 100644
--- a/internal/federation/federatingprotocol.go
+++ b/internal/federation/federatingprotocol.go
@@ -153,7 +153,7 @@ func (f *federator) AuthenticatePostInbox(ctx context.Context, w http.ResponseWr
}
}
- requestingAccount, _, err := f.GetRemoteAccount(ctx, username, publicKeyOwnerURI, false)
+ requestingAccount, err := f.GetRemoteAccount(ctx, username, publicKeyOwnerURI, false, false)
if err != nil {
return nil, false, fmt.Errorf("couldn't get requesting account %s: %s", publicKeyOwnerURI, err)
}
diff --git a/internal/federation/federator.go b/internal/federation/federator.go
index 0a82f12bc..cb63084db 100644
--- a/internal/federation/federator.go
+++ b/internal/federation/federator.go
@@ -57,8 +57,7 @@ type Federator interface {
DereferenceRemoteThread(ctx context.Context, username string, statusURI *url.URL) error
DereferenceAnnounce(ctx context.Context, announce *gtsmodel.Status, requestingUsername string) error
- GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, refresh bool) (*gtsmodel.Account, bool, error)
- EnrichRemoteAccount(ctx context.Context, username string, account *gtsmodel.Account) (*gtsmodel.Account, error)
+ GetRemoteAccount(ctx context.Context, username string, remoteAccountID *url.URL, blocking bool, refresh bool) (*gtsmodel.Account, error)
GetRemoteStatus(ctx context.Context, username string, remoteStatusID *url.URL, refresh, includeParent bool) (*gtsmodel.Status, ap.Statusable, bool, error)
EnrichRemoteStatus(ctx context.Context, username string, status *gtsmodel.Status, includeParent bool) (*gtsmodel.Status, error)
@@ -78,13 +77,13 @@ type federator struct {
typeConverter typeutils.TypeConverter
transportController transport.Controller
dereferencer dereferencing.Dereferencer
- mediaHandler media.Handler
+ mediaManager media.Manager
actor pub.FederatingActor
}
// NewFederator returns a new federator
-func NewFederator(db db.DB, federatingDB federatingdb.DB, transportController transport.Controller, typeConverter typeutils.TypeConverter, mediaHandler media.Handler) Federator {
- dereferencer := dereferencing.NewDereferencer(db, typeConverter, transportController, mediaHandler)
+func NewFederator(db db.DB, federatingDB federatingdb.DB, transportController transport.Controller, typeConverter typeutils.TypeConverter, mediaManager media.Manager) Federator {
+ dereferencer := dereferencing.NewDereferencer(db, typeConverter, transportController, mediaManager)
clock := &Clock{}
f := &federator{
@@ -94,7 +93,7 @@ func NewFederator(db db.DB, federatingDB federatingdb.DB, transportController tr
typeConverter: typeConverter,
transportController: transportController,
dereferencer: dereferencer,
- mediaHandler: mediaHandler,
+ mediaManager: mediaManager,
}
actor := newFederatingActor(f, f, federatingDB, clock)
f.actor = actor
diff --git a/internal/federation/federator_test.go b/internal/federation/federator_test.go
index 43f4904a5..6dac76c05 100644
--- a/internal/federation/federator_test.go
+++ b/internal/federation/federator_test.go
@@ -78,7 +78,7 @@ func (suite *ProtocolTestSuite) TestPostInboxRequestBodyHook() {
return nil, nil
}), suite.db)
// setup module being tested
- federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db), tc, suite.typeConverter, testrig.NewTestMediaHandler(suite.db, suite.storage))
+ federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db), tc, suite.typeConverter, testrig.NewTestMediaManager(suite.db, suite.storage))
// setup request
ctx := context.Background()
@@ -107,7 +107,7 @@ func (suite *ProtocolTestSuite) TestAuthenticatePostInbox() {
tc := testrig.NewTestTransportController(testrig.NewMockHTTPClient(nil), suite.db)
// now setup module being tested, with the mock transport controller
- federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db), tc, suite.typeConverter, testrig.NewTestMediaHandler(suite.db, suite.storage))
+ federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db), tc, suite.typeConverter, testrig.NewTestMediaManager(suite.db, suite.storage))
request := httptest.NewRequest(http.MethodPost, "http://localhost:8080/users/the_mighty_zork/inbox", nil)
// we need these headers for the request to be validated
generated by cgit v1.2.3 (git 2.46.0) at 2025-07-24 07:46:04 +0000