Add shared inbox support for ActivityPub delivery (#4755)

* feat(ap): add support for shared inboxes to reduce outbound load

* feat(db): refactor ap followers db into followers repository

* fix(ap): use the updated activity library to pull out the shared inbox endpoint

* chore(deps): point at updated build of owncast/activity

* fix(ap): typeless endpoints

* feat(test): update ActivityPub test to support shared inboxes

* chore(test): remove unused variable

* fix: feedback from review. Guard against SSRF/non-HTTPS/local and handle transaction errors

activitypub/persistence/followers.go

@@ -1,12 +1,6 @@
 package persistence
 
 import (
-	"context"
-
-	"github.com/owncast/owncast/db"
-	"github.com/owncast/owncast/models"
-	"github.com/owncast/owncast/utils"
-	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -16,6 +10,7 @@ func createFederationFollowersTable() {
 	createTableSQL := `CREATE TABLE IF NOT EXISTS ap_followers (
 		"iri" TEXT NOT NULL,
 		"inbox" TEXT NOT NULL,
+		"shared_inbox" TEXT,
 		"name" TEXT,
 		"username" TEXT NOT NULL,
 		"image" TEXT,
@@ -29,91 +24,3 @@ func createFederationFollowersTable() {
 	_datastore.MustExec(`CREATE INDEX IF NOT EXISTS idx_iri ON ap_followers (iri);`)
 	_datastore.MustExec(`CREATE INDEX IF NOT EXISTS idx_approved_at ON ap_followers (approved_at);`)
 }
-
-// GetFollowerCount will return the number of followers we're keeping track of.
-func GetFollowerCount() (int64, error) {
-	ctx := context.Background()
-	return _datastore.GetQueries().GetFollowerCount(ctx)
-}
-
-// GetFederationFollowers will return a slice of the followers we keep track of locally.
-func GetFederationFollowers(limit int, offset int) ([]models.Follower, int, error) {
-	ctx := context.Background()
-	total, err := _datastore.GetQueries().GetFollowerCount(ctx)
-	if err != nil {
-		return nil, 0, errors.Wrap(err, "unable to fetch total number of followers")
-	}
-
-	followersResult, err := _datastore.GetQueries().GetFederationFollowersWithOffset(ctx, db.GetFederationFollowersWithOffsetParams{
-		Limit:  limit,
-		Offset: offset,
-	})
-	if err != nil {
-		return nil, 0, err
-	}
-
-	followers := make([]models.Follower, 0)
-
-	for _, row := range followersResult {
-		singleFollower := models.Follower{
-			Name:      row.Name.String,
-			Username:  row.Username,
-			Image:     row.Image.String,
-			ActorIRI:  row.Iri,
-			Inbox:     row.Inbox,
-			Timestamp: utils.NullTime(row.CreatedAt),
-		}
-
-		followers = append(followers, singleFollower)
-	}
-
-	return followers, int(total), nil
-}
-
-// GetPendingFollowRequests will return pending follow requests.
-func GetPendingFollowRequests() ([]models.Follower, error) {
-	pendingFollowersResult, err := _datastore.GetQueries().GetFederationFollowerApprovalRequests(context.Background())
-	if err != nil {
-		return nil, err
-	}
-
-	followers := make([]models.Follower, 0)
-
-	for _, row := range pendingFollowersResult {
-		singleFollower := models.Follower{
-			Name:      row.Name.String,
-			Username:  row.Username,
-			Image:     row.Image.String,
-			ActorIRI:  row.Iri,
-			Inbox:     row.Inbox,
-			Timestamp: utils.NullTime{Time: row.CreatedAt.Time, Valid: true},
-		}
-		followers = append(followers, singleFollower)
-	}
-
-	return followers, nil
-}
-
-// GetBlockedAndRejectedFollowers will return blocked and rejected followers.
-func GetBlockedAndRejectedFollowers() ([]models.Follower, error) {
-	pendingFollowersResult, err := _datastore.GetQueries().GetRejectedAndBlockedFollowers(context.Background())
-	if err != nil {
-		return nil, err
-	}
-
-	followers := make([]models.Follower, 0)
-
-	for _, row := range pendingFollowersResult {
-		singleFollower := models.Follower{
-			Name:       row.Name.String,
-			Username:   row.Username,
-			Image:      row.Image.String,
-			ActorIRI:   row.Iri,
-			DisabledAt: utils.NullTime{Time: row.DisabledAt.Time, Valid: true},
-			Timestamp:  utils.NullTime{Time: row.CreatedAt.Time, Valid: true},
-		}
-		followers = append(followers, singleFollower)
-	}
-
-	return followers, nil
-}

activitypub/persistence/followers_test.go

@@ -1,9 +1,14 @@
 package persistence
 
 import (
+	"net/url"
 	"os"
+	"strings"
 	"testing"
 
+	"github.com/go-fed/activity/streams"
+	"github.com/owncast/owncast/activitypub/apmodels"
+	"github.com/owncast/owncast/activitypub/persistence/followersrepository"
 	"github.com/owncast/owncast/core/data"
 	"github.com/owncast/owncast/models"
 	"github.com/owncast/owncast/utils"
@@ -22,16 +27,30 @@ func setup() {
 	_datastore = data.GetDatastore()
 	createFederationFollowersTable()
 
+	followersRepo := followersrepository.New(_datastore)
+
 	number := 100
 	for i := 0; i < number; i++ {
 		u := createFakeFollower()
-		createFollow(u.ActorIRI, u.Inbox, "https://fake.fediverse.server/some/request", u.Name, u.Username, u.Image, nil, true)
+		actorIRI, _ := url.Parse(u.ActorIRI)
+		inboxURL, _ := url.Parse(u.Inbox)
+		requestIRI, _ := url.Parse("https://fake.fediverse.server/some/request")
+		fakeRequest := streams.NewActivityStreamsFollow()
+		followersRepo.Add(apmodels.ActivityPubActor{
+			ActorIri:         actorIRI,
+			Inbox:            inboxURL,
+			Name:             u.Name,
+			Username:         u.Username,
+			FollowRequestIri: requestIRI,
+			RequestObject:    fakeRequest,
+		}, true)
 		followers = append(followers, u)
 	}
 }
 
 func TestQueryFollowers(t *testing.T) {
-	f, total, err := GetFederationFollowers(10, 0)
+	followersRepo := followersrepository.New(_datastore)
+	f, total, err := followersRepo.GetFollowers(10, 0)
 	if err != nil {
 		t.Errorf("Error querying followers: %s", err)
 	}
@@ -46,7 +65,8 @@ func TestQueryFollowers(t *testing.T) {
 }
 
 func TestQueryFollowersWithOffset(t *testing.T) {
-	f, total, err := GetFederationFollowers(10, 10)
+	followersRepo := followersrepository.New(_datastore)
+	f, total, err := followersRepo.GetFollowers(10, 10)
 	if err != nil {
 		t.Errorf("Error querying followers: %s", err)
 	}
@@ -61,7 +81,8 @@ func TestQueryFollowersWithOffset(t *testing.T) {
 }
 
 func TestQueryFollowersWithOffsetAndLimit(t *testing.T) {
-	f, total, err := GetFederationFollowers(10, 90)
+	followersRepo := followersrepository.New(_datastore)
+	f, total, err := followersRepo.GetFollowers(10, 90)
 	if err != nil {
 		t.Errorf("Error querying followers: %s", err)
 	}
@@ -76,7 +97,8 @@ func TestQueryFollowersWithOffsetAndLimit(t *testing.T) {
 }
 
 func TestQueryFollowersWithPagination(t *testing.T) {
-	f, _, err := GetFederationFollowers(15, 10)
+	followersRepo := followersrepository.New(_datastore)
+	f, _, err := followersRepo.GetFollowers(15, 10)
 	if err != nil {
 		t.Errorf("Error querying followers: %s", err)
 	}
@@ -105,3 +127,194 @@ func createFakeFollower() models.Follower {
 		Timestamp: utils.NullTime{},
 	}
 }
+
+func createTestFollower(followersRepo followersrepository.FollowersRepository, actor, inbox, sharedInbox, request, name, username string) {
+	actorIRI, _ := url.Parse(actor)
+	inboxURL, _ := url.Parse(inbox)
+	var sharedInboxURL *url.URL
+	if sharedInbox != "" {
+		sharedInboxURL, _ = url.Parse(sharedInbox)
+	}
+	requestIRI, _ := url.Parse(request)
+	fakeRequest := streams.NewActivityStreamsFollow()
+	followersRepo.Add(apmodels.ActivityPubActor{
+		ActorIri:         actorIRI,
+		Inbox:            inboxURL,
+		SharedInbox:      sharedInboxURL,
+		Name:             name,
+		Username:         username,
+		FollowRequestIri: requestIRI,
+		RequestObject:    fakeRequest,
+	}, true)
+}
+
+func TestGetUniqueDeliveryInboxes(t *testing.T) {
+	// Set up a fresh database for this test
+	data.SetupPersistence(":memory:")
+	ds := data.GetDatastore()
+	_datastore = ds
+	createFederationFollowersTable()
+	followersRepo := followersrepository.New(ds)
+
+	// Create followers from server1 with a shared inbox (3 users, 1 shared inbox)
+	server1SharedInbox := "https://server1.example.com/inbox"
+	for i := 0; i < 3; i++ {
+		user, _ := utils.GenerateRandomString(10)
+		createTestFollower(
+			followersRepo,
+			"https://server1.example.com/user/"+user,
+			"https://server1.example.com/user/"+user+"/inbox",
+			server1SharedInbox,
+			"https://server1.example.com/follow/"+user,
+			user,
+			user,
+		)
+	}
+
+	// Create followers from server2 with a shared inbox (2 users, 1 shared inbox)
+	server2SharedInbox := "https://server2.example.com/inbox"
+	for i := 0; i < 2; i++ {
+		user, _ := utils.GenerateRandomString(10)
+		createTestFollower(
+			followersRepo,
+			"https://server2.example.com/user/"+user,
+			"https://server2.example.com/user/"+user+"/inbox",
+			server2SharedInbox,
+			"https://server2.example.com/follow/"+user,
+			user,
+			user,
+		)
+	}
+
+	// Create followers from server3 WITHOUT a shared inbox (2 users, 2 individual inboxes)
+	for i := 0; i < 2; i++ {
+		user, _ := utils.GenerateRandomString(10)
+		createTestFollower(
+			followersRepo,
+			"https://server3.example.com/user/"+user,
+			"https://server3.example.com/user/"+user+"/inbox",
+			"",
+			"https://server3.example.com/follow/"+user,
+			user,
+			user,
+		)
+	}
+
+	// Total: 7 followers, but should result in 4 unique delivery inboxes:
+	// - 1 shared inbox for server1
+	// - 1 shared inbox for server2
+	// - 2 individual inboxes for server3
+
+	inboxes, err := followersRepo.GetUniqueDeliveryInboxes()
+	if err != nil {
+		t.Fatalf("Error getting unique delivery inboxes: %s", err)
+	}
+
+	if len(inboxes) != 4 {
+		t.Errorf("Expected 4 unique delivery inboxes, got %d: %v", len(inboxes), inboxes)
+	}
+
+	// Verify the shared inboxes are included
+	hasServer1Shared := false
+	hasServer2Shared := false
+	server3IndividualCount := 0
+
+	for _, inbox := range inboxes {
+		if inbox == server1SharedInbox {
+			hasServer1Shared = true
+		}
+		if inbox == server2SharedInbox {
+			hasServer2Shared = true
+		}
+		if len(inbox) > 0 && inbox != server1SharedInbox && inbox != server2SharedInbox {
+			// Should be one of server3's individual inboxes
+			if !strings.Contains(inbox, "server3.example.com") {
+				t.Errorf("Unexpected inbox in results: %s", inbox)
+			}
+			server3IndividualCount++
+		}
+	}
+
+	if !hasServer1Shared {
+		t.Error("Expected server1 shared inbox to be in results")
+	}
+	if !hasServer2Shared {
+		t.Error("Expected server2 shared inbox to be in results")
+	}
+	if server3IndividualCount != 2 {
+		t.Errorf("Expected 2 individual inboxes from server3, got %d", server3IndividualCount)
+	}
+}
+
+func TestSharedInboxPreferredOverIndividual(t *testing.T) {
+	// Set up a fresh database for this test
+	data.SetupPersistence(":memory:")
+	ds := data.GetDatastore()
+	_datastore = ds
+	createFederationFollowersTable()
+	followersRepo := followersrepository.New(ds)
+
+	// Create a single follower with both individual and shared inbox
+	sharedInbox := "https://mastodon.social/inbox"
+	individualInbox := "https://mastodon.social/users/testuser/inbox"
+
+	createTestFollower(
+		followersRepo,
+		"https://mastodon.social/users/testuser",
+		individualInbox,
+		sharedInbox,
+		"https://mastodon.social/follow/123",
+		"Test User",
+		"testuser",
+	)
+
+	inboxes, err := followersRepo.GetUniqueDeliveryInboxes()
+	if err != nil {
+		t.Fatalf("Error getting unique delivery inboxes: %s", err)
+	}
+
+	if len(inboxes) != 1 {
+		t.Errorf("Expected 1 unique delivery inbox, got %d", len(inboxes))
+	}
+
+	// The shared inbox should be returned, not the individual inbox
+	if inboxes[0] != sharedInbox {
+		t.Errorf("Expected shared inbox %s, got %s", sharedInbox, inboxes[0])
+	}
+}
+
+func TestIndividualInboxUsedWhenNoSharedInbox(t *testing.T) {
+	// Set up a fresh database for this test
+	data.SetupPersistence(":memory:")
+	ds := data.GetDatastore()
+	_datastore = ds
+	createFederationFollowersTable()
+	followersRepo := followersrepository.New(ds)
+
+	// Create a follower without a shared inbox
+	individualInbox := "https://pleroma.example.com/users/testuser/inbox"
+
+	createTestFollower(
+		followersRepo,
+		"https://pleroma.example.com/users/testuser",
+		individualInbox,
+		"",
+		"https://pleroma.example.com/follow/123",
+		"Test User",
+		"testuser",
+	)
+
+	inboxes, err := followersRepo.GetUniqueDeliveryInboxes()
+	if err != nil {
+		t.Fatalf("Error getting unique delivery inboxes: %s", err)
+	}
+
+	if len(inboxes) != 1 {
+		t.Errorf("Expected 1 unique delivery inbox, got %d", len(inboxes))
+	}
+
+	// The individual inbox should be returned when no shared inbox exists
+	if inboxes[0] != individualInbox {
+		t.Errorf("Expected individual inbox %s, got %s", individualInbox, inboxes[0])
+	}
+}

activitypub/persistence/followersrepository/followersrepository.go

@@ -0,0 +1,366 @@
+package followersrepository
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"net/url"
+	"time"
+
+	"github.com/go-fed/activity/streams"
+	"github.com/go-fed/activity/streams/vocab"
+	"github.com/owncast/owncast/activitypub/apmodels"
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/db"
+	"github.com/owncast/owncast/models"
+	"github.com/owncast/owncast/utils"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+// FollowersRepository handles persistence of ActivityPub followers.
+type FollowersRepository interface {
+	// GetCount returns the number of followers.
+	GetCount() (int64, error)
+	// GetFollowers returns a paginated list of followers.
+	GetFollowers(limit int, offset int) ([]models.Follower, int, error)
+	// GetPendingFollowRequests returns pending follow requests.
+	GetPendingFollowRequests() ([]models.Follower, error)
+	// GetBlockedAndRejected returns blocked and rejected followers.
+	GetBlockedAndRejected() ([]models.Follower, error)
+	// GetUniqueDeliveryInboxes returns unique inbox URLs for delivery.
+	GetUniqueDeliveryInboxes() ([]string, error)
+	// GetByIRI returns a single follower by IRI.
+	GetByIRI(iri string) (*apmodels.ActivityPubActor, error)
+	// Add saves a new follow to the datastore.
+	Add(follow apmodels.ActivityPubActor, approved bool) error
+	// Remove removes a follow from the datastore.
+	Remove(unfollow apmodels.ActivityPubActor) error
+	// ApprovePreviousRequest approves a pending follow request.
+	ApprovePreviousRequest(iri string) error
+	// BlockOrReject blocks an existing follower or rejects a follow request.
+	BlockOrReject(iri string) error
+	// Update updates the details of a stored follower.
+	Update(actorIRI string, inbox string, sharedInbox string, name string, username string, image string) error
+}
+
+// SqlFollowersRepository is the SQL-based implementation of FollowersRepository.
+type SqlFollowersRepository struct {
+	datastore *data.Datastore
+}
+
+// NOTE: This is temporary during the transition period.
+var temporaryGlobalInstance FollowersRepository
+
+// Get returns the followers repository singleton.
+func Get() FollowersRepository {
+	if temporaryGlobalInstance == nil {
+		i := New(data.GetDatastore())
+		temporaryGlobalInstance = i
+	}
+	return temporaryGlobalInstance
+}
+
+// New creates a new instance of the FollowersRepository.
+func New(datastore *data.Datastore) FollowersRepository {
+	r := SqlFollowersRepository{
+		datastore: datastore,
+	}
+	return &r
+}
+
+// GetCount returns the number of followers.
+func (r *SqlFollowersRepository) GetCount() (int64, error) {
+	ctx := context.Background()
+	return r.datastore.GetQueries().GetFollowerCount(ctx)
+}
+
+// GetFollowers returns a paginated list of followers.
+func (r *SqlFollowersRepository) GetFollowers(limit int, offset int) ([]models.Follower, int, error) {
+	ctx := context.Background()
+	total, err := r.datastore.GetQueries().GetFollowerCount(ctx)
+	if err != nil {
+		return nil, 0, errors.Wrap(err, "unable to fetch total number of followers")
+	}
+
+	followersResult, err := r.datastore.GetQueries().GetFederationFollowersWithOffset(ctx, db.GetFederationFollowersWithOffsetParams{
+		Limit:  utils.SafeIntToInt32(limit),
+		Offset: utils.SafeIntToInt32(offset),
+	})
+	if err != nil {
+		return nil, 0, err
+	}
+
+	followers := make([]models.Follower, 0)
+
+	for _, row := range followersResult {
+		singleFollower := models.Follower{
+			Name:        row.Name.String,
+			Username:    row.Username,
+			Image:       row.Image.String,
+			ActorIRI:    row.Iri,
+			Inbox:       row.Inbox,
+			SharedInbox: row.SharedInbox.String,
+			Timestamp:   utils.NullTime(row.CreatedAt),
+		}
+
+		followers = append(followers, singleFollower)
+	}
+
+	return followers, int(total), nil
+}
+
+// GetPendingFollowRequests returns pending follow requests.
+func (r *SqlFollowersRepository) GetPendingFollowRequests() ([]models.Follower, error) {
+	pendingFollowersResult, err := r.datastore.GetQueries().GetFederationFollowerApprovalRequests(context.Background())
+	if err != nil {
+		return nil, err
+	}
+
+	followers := make([]models.Follower, 0)
+
+	for _, row := range pendingFollowersResult {
+		singleFollower := models.Follower{
+			Name:        row.Name.String,
+			Username:    row.Username,
+			Image:       row.Image.String,
+			ActorIRI:    row.Iri,
+			Inbox:       row.Inbox,
+			SharedInbox: row.SharedInbox.String,
+			Timestamp:   utils.NullTime{Time: row.CreatedAt.Time, Valid: true},
+		}
+		followers = append(followers, singleFollower)
+	}
+
+	return followers, nil
+}
+
+// GetBlockedAndRejected returns blocked and rejected followers.
+func (r *SqlFollowersRepository) GetBlockedAndRejected() ([]models.Follower, error) {
+	pendingFollowersResult, err := r.datastore.GetQueries().GetRejectedAndBlockedFollowers(context.Background())
+	if err != nil {
+		return nil, err
+	}
+
+	followers := make([]models.Follower, 0)
+
+	for _, row := range pendingFollowersResult {
+		singleFollower := models.Follower{
+			Name:       row.Name.String,
+			Username:   row.Username,
+			Image:      row.Image.String,
+			ActorIRI:   row.Iri,
+			DisabledAt: utils.NullTime{Time: row.DisabledAt.Time, Valid: true},
+			Timestamp:  utils.NullTime{Time: row.CreatedAt.Time, Valid: true},
+		}
+		followers = append(followers, singleFollower)
+	}
+
+	return followers, nil
+}
+
+// GetUniqueDeliveryInboxes returns unique inbox URLs for delivery.
+func (r *SqlFollowersRepository) GetUniqueDeliveryInboxes() ([]string, error) {
+	ctx := context.Background()
+	return r.datastore.GetQueries().GetUniqueDeliveryInboxes(ctx)
+}
+
+// GetByIRI returns a single follower by IRI.
+func (r *SqlFollowersRepository) GetByIRI(iri string) (*apmodels.ActivityPubActor, error) {
+	result, err := r.datastore.GetQueries().GetFollowerByIRI(context.Background(), iri)
+	if err != nil {
+		return nil, err
+	}
+
+	followIRI, err := url.Parse(result.Request)
+	if err != nil {
+		return nil, errors.Wrap(err, "error parsing follow request IRI")
+	}
+
+	iriURL, err := url.Parse(result.Iri)
+	if err != nil {
+		return nil, errors.Wrap(err, "error parsing actor IRI")
+	}
+
+	inbox, err := url.Parse(result.Inbox)
+	if err != nil {
+		return nil, errors.Wrap(err, "error parsing acting inbox")
+	}
+
+	var sharedInbox *url.URL
+	if result.SharedInbox.Valid && result.SharedInbox.String != "" {
+		sharedInbox, err = url.Parse(result.SharedInbox.String)
+		if err != nil {
+			log.Warnln("error parsing shared inbox, ignoring:", err)
+		}
+	}
+
+	requestObjectBytes := result.RequestObject
+	var followRequestObject vocab.ActivityStreamsFollow
+
+	resolver, err := streams.NewJSONResolver(func(c context.Context, followObject vocab.ActivityStreamsFollow) error {
+		followRequestObject = followObject
+		return nil
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating JSON resolver")
+	}
+	jsonMap := make(map[string]interface{})
+	err = json.Unmarshal(requestObjectBytes, &jsonMap)
+	if err != nil {
+		return nil, errors.Wrap(err, "error unmarshaling follow request object")
+	}
+
+	err = resolver.Resolve(context.Background(), jsonMap)
+	if err != nil {
+		return nil, errors.Wrap(err, "error resolving follow request object")
+	}
+
+	image, _ := url.Parse(result.Image.String)
+
+	var disabledAt *time.Time
+	if result.DisabledAt.Valid {
+		disabledAt = &result.DisabledAt.Time
+	}
+
+	follower := apmodels.ActivityPubActor{
+		ActorIri:         iriURL,
+		Inbox:            inbox,
+		SharedInbox:      sharedInbox,
+		Name:             result.Name.String,
+		Username:         result.Username,
+		Image:            image,
+		FollowRequestIri: followIRI,
+		DisabledAt:       disabledAt,
+		RequestObject:    followRequestObject,
+	}
+
+	return &follower, nil
+}
+
+// Add saves a new follow to the datastore.
+func (r *SqlFollowersRepository) Add(follow apmodels.ActivityPubActor, approved bool) error {
+	if err := follow.Validate(); err != nil {
+		return errors.Wrap(err, "cannot add invalid follow")
+	}
+
+	log.Traceln("Saving", follow.ActorIriString(), "as a follower.")
+
+	followRequestObject, err := apmodels.Serialize(follow.RequestObject)
+	if err != nil {
+		return errors.Wrap(err, "error serializing follow request object")
+	}
+
+	return r.createFollow(follow.ActorIriString(), follow.InboxString(), follow.SharedInboxString(), follow.FollowRequestIriString(), follow.Name, follow.Username, follow.ImageString(), followRequestObject, approved)
+}
+
+// Remove removes a follow from the datastore.
+func (r *SqlFollowersRepository) Remove(unfollow apmodels.ActivityPubActor) error {
+	if err := unfollow.Validate(); err != nil {
+		return errors.Wrap(err, "cannot remove invalid follow")
+	}
+	log.Traceln("Removing", unfollow.ActorIriString(), "as a follower.")
+	return r.removeFollow(unfollow.ActorIri)
+}
+
+// ApprovePreviousRequest approves a pending follow request.
+func (r *SqlFollowersRepository) ApprovePreviousRequest(iri string) error {
+	return r.datastore.GetQueries().ApproveFederationFollower(context.Background(), db.ApproveFederationFollowerParams{
+		Iri: iri,
+		ApprovedAt: sql.NullTime{
+			Time:  time.Now(),
+			Valid: true,
+		},
+	})
+}
+
+// BlockOrReject blocks an existing follower or rejects a follow request.
+func (r *SqlFollowersRepository) BlockOrReject(iri string) error {
+	return r.datastore.GetQueries().RejectFederationFollower(context.Background(), db.RejectFederationFollowerParams{
+		Iri: iri,
+		DisabledAt: sql.NullTime{
+			Time:  time.Now(),
+			Valid: true,
+		},
+	})
+}
+
+// Update updates the details of a stored follower.
+func (r *SqlFollowersRepository) Update(actorIRI string, inbox string, sharedInbox string, name string, username string, image string) error {
+	r.datastore.DbLock.Lock()
+	defer r.datastore.DbLock.Unlock()
+
+	tx, err := r.datastore.DB.Begin()
+	if err != nil {
+		return errors.Wrap(err, "error beginning transaction")
+	}
+	defer func() {
+		_ = tx.Rollback()
+	}()
+
+	if err = r.datastore.GetQueries().WithTx(tx).UpdateFollowerByIRI(context.Background(), db.UpdateFollowerByIRIParams{
+		Inbox:       inbox,
+		SharedInbox: sql.NullString{String: sharedInbox, Valid: sharedInbox != ""},
+		Name:        sql.NullString{String: name, Valid: true},
+		Username:    username,
+		Image:       sql.NullString{String: image, Valid: true},
+		Iri:         actorIRI,
+	}); err != nil {
+		return errors.Wrap(err, "error updating follower "+actorIRI)
+	}
+
+	return tx.Commit()
+}
+
+func (r *SqlFollowersRepository) createFollow(actor, inbox, sharedInbox, request, name, username, image string, requestObject []byte, approved bool) error {
+	tx, err := r.datastore.DB.Begin()
+	if err != nil {
+		return errors.Wrap(err, "error beginning transaction")
+	}
+	defer func() {
+		_ = tx.Rollback()
+	}()
+
+	var approvedAt sql.NullTime
+	if approved {
+		approvedAt = sql.NullTime{
+			Time:  time.Now(),
+			Valid: true,
+		}
+	}
+
+	if err = r.datastore.GetQueries().WithTx(tx).AddFollower(context.Background(), db.AddFollowerParams{
+		Iri:           actor,
+		Inbox:         inbox,
+		SharedInbox:   sql.NullString{String: sharedInbox, Valid: sharedInbox != ""},
+		Name:          sql.NullString{String: name, Valid: true},
+		Username:      username,
+		Image:         sql.NullString{String: image, Valid: true},
+		ApprovedAt:    approvedAt,
+		Request:       request,
+		RequestObject: requestObject,
+	}); err != nil {
+		log.Errorln("error creating new federation follow: ", err)
+	}
+
+	return tx.Commit()
+}
+
+func (r *SqlFollowersRepository) removeFollow(actor *url.URL) error {
+	r.datastore.DbLock.Lock()
+	defer r.datastore.DbLock.Unlock()
+
+	tx, err := r.datastore.DB.Begin()
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = tx.Rollback()
+	}()
+
+	if err := r.datastore.GetQueries().WithTx(tx).RemoveFollowerByIRI(context.Background(), actor.String()); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}

activitypub/persistence/persistence.go

@@ -3,18 +3,16 @@ package persistence
 import (
 	"context"
 	"database/sql"
-	"encoding/json"
 	"fmt"
-	"net/url"
 	"time"
 
 	"github.com/go-fed/activity/streams"
 	"github.com/go-fed/activity/streams/vocab"
-	"github.com/owncast/owncast/activitypub/apmodels"
 	"github.com/owncast/owncast/activitypub/resolvers"
 	"github.com/owncast/owncast/core/data"
 	"github.com/owncast/owncast/db"
 	"github.com/owncast/owncast/models"
+	"github.com/owncast/owncast/utils"
 	"github.com/pkg/errors"
 
 	log "github.com/sirupsen/logrus"
@@ -31,193 +29,9 @@ func Setup(datastore *data.Datastore) {
 	addFollowersFixtureData()
 }
 
-// AddFollow will save a follow to the datastore.
-func AddFollow(follow apmodels.ActivityPubActor, approved bool) error {
-	if err := follow.Validate(); err != nil {
-		return errors.Wrap(err, "cannot add invalid follow")
-	}
-
-	log.Traceln("Saving", follow.ActorIriString(), "as a follower.")
-
-	followRequestObject, err := apmodels.Serialize(follow.RequestObject)
-	if err != nil {
-		return errors.Wrap(err, "error serializing follow request object")
-	}
-
-	return createFollow(follow.ActorIriString(), follow.InboxString(), follow.FollowRequestIriString(), follow.Name, follow.Username, follow.ImageString(), followRequestObject, approved)
-}
-
-// RemoveFollow will remove a follow from the datastore.
-func RemoveFollow(unfollow apmodels.ActivityPubActor) error {
-	if err := unfollow.Validate(); err != nil {
-		return errors.Wrap(err, "cannot remove invalid follow")
-	}
-	log.Traceln("Removing", unfollow.ActorIriString(), "as a follower.")
-	return removeFollow(unfollow.ActorIri)
-}
-
-// GetFollower will return a single follower/request given an IRI.
-func GetFollower(iri string) (*apmodels.ActivityPubActor, error) {
-	result, err := _datastore.GetQueries().GetFollowerByIRI(context.Background(), iri)
-	if err != nil {
-		return nil, err
-	}
-
-	followIRI, err := url.Parse(result.Request)
-	if err != nil {
-		return nil, errors.Wrap(err, "error parsing follow request IRI")
-	}
-
-	iriURL, err := url.Parse(result.Iri)
-	if err != nil {
-		return nil, errors.Wrap(err, "error parsing actor IRI")
-	}
-
-	inbox, err := url.Parse(result.Inbox)
-	if err != nil {
-		return nil, errors.Wrap(err, "error parsing acting inbox")
-	}
-
-	requestObjectBytes := result.RequestObject
-	var followRequestObject vocab.ActivityStreamsFollow
-
-	resolver, err := streams.NewJSONResolver(func(c context.Context, followObject vocab.ActivityStreamsFollow) error {
-		followRequestObject = followObject
-		return nil
-	})
-	if err != nil {
-		return nil, errors.Wrap(err, "error creating JSON resolver")
-	}
-	jsonMap := make(map[string]interface{})
-	err = json.Unmarshal(requestObjectBytes, &jsonMap)
-	if err != nil {
-		return nil, errors.Wrap(err, "error unmarshaling follow request object")
-	}
-
-	err = resolver.Resolve(context.Background(), jsonMap)
-	if err != nil {
-		return nil, errors.Wrap(err, "error resolving follow request object")
-	}
-
-	image, _ := url.Parse(result.Image.String)
-
-	var disabledAt *time.Time
-	if result.DisabledAt.Valid {
-		disabledAt = &result.DisabledAt.Time
-	}
-
-	follower := apmodels.ActivityPubActor{
-		ActorIri:         iriURL,
-		Inbox:            inbox,
-		Name:             result.Name.String,
-		Username:         result.Username,
-		Image:            image,
-		FollowRequestIri: followIRI,
-		DisabledAt:       disabledAt,
-		RequestObject:    followRequestObject,
-	}
-
-	return &follower, nil
-}
-
-// ApprovePreviousFollowRequest will approve a follow request.
-func ApprovePreviousFollowRequest(iri string) error {
-	return _datastore.GetQueries().ApproveFederationFollower(context.Background(), db.ApproveFederationFollowerParams{
-		Iri: iri,
-		ApprovedAt: sql.NullTime{
-			Time:  time.Now(),
-			Valid: true,
-		},
-	})
-}
-
-// BlockOrRejectFollower will block an existing follower or reject a follow request.
-func BlockOrRejectFollower(iri string) error {
-	return _datastore.GetQueries().RejectFederationFollower(context.Background(), db.RejectFederationFollowerParams{
-		Iri: iri,
-		DisabledAt: sql.NullTime{
-			Time:  time.Now(),
-			Valid: true,
-		},
-	})
-}
-
-func createFollow(actor, inbox, request, name, username, image string, requestObject []byte, approved bool) error {
-	tx, err := _datastore.DB.Begin()
-	if err != nil {
-		log.Debugln(err)
-	}
-	defer func() {
-		_ = tx.Rollback()
-	}()
-
-	var approvedAt sql.NullTime
-	if approved {
-		approvedAt = sql.NullTime{
-			Time:  time.Now(),
-			Valid: true,
-		}
-	}
-
-	if err = _datastore.GetQueries().WithTx(tx).AddFollower(context.Background(), db.AddFollowerParams{
-		Iri:           actor,
-		Inbox:         inbox,
-		Name:          sql.NullString{String: name, Valid: true},
-		Username:      username,
-		Image:         sql.NullString{String: image, Valid: true},
-		ApprovedAt:    approvedAt,
-		Request:       request,
-		RequestObject: requestObject,
-	}); err != nil {
-		log.Errorln("error creating new federation follow: ", err)
-	}
-
-	return tx.Commit()
-}
-
-// UpdateFollower will update the details of a stored follower given an IRI.
-func UpdateFollower(actorIRI string, inbox string, name string, username string, image string) error {
-	_datastore.DbLock.Lock()
-	defer _datastore.DbLock.Unlock()
-
-	tx, err := _datastore.DB.Begin()
-	if err != nil {
-		log.Debugln(err)
-	}
-	defer func() {
-		_ = tx.Rollback()
-	}()
-
-	if err = _datastore.GetQueries().WithTx(tx).UpdateFollowerByIRI(context.Background(), db.UpdateFollowerByIRIParams{
-		Inbox:    inbox,
-		Name:     sql.NullString{String: name, Valid: true},
-		Username: username,
-		Image:    sql.NullString{String: image, Valid: true},
-		Iri:      actorIRI,
-	}); err != nil {
-		return fmt.Errorf("error updating follower %s %s", actorIRI, err)
-	}
-
-	return tx.Commit()
-}
-
-func removeFollow(actor *url.URL) error {
-	_datastore.DbLock.Lock()
-	defer _datastore.DbLock.Unlock()
-
-	tx, err := _datastore.DB.Begin()
-	if err != nil {
-		return err
-	}
-	defer func() {
-		_ = tx.Rollback()
-	}()
-
-	if err := _datastore.GetQueries().WithTx(tx).RemoveFollowerByIRI(context.Background(), actor.String()); err != nil {
-		return err
-	}
-
-	return tx.Commit()
+// GetDatastore returns the datastore instance for use by sub-repositories.
+func GetDatastore() *data.Datastore {
+	return _datastore
 }
 
 // createFederatedActivitiesTable will create the accepted
@@ -263,7 +77,7 @@ func GetOutbox(limit int, offset int) (vocab.ActivityStreamsOrderedCollection, e
 	orderedItems := streams.NewActivityStreamsOrderedItemsProperty()
 	rows, err := _datastore.GetQueries().GetOutboxWithOffset(
 		context.Background(),
-		db.GetOutboxWithOffsetParams{Limit: limit, Offset: offset},
+		db.GetOutboxWithOffsetParams{Limit: utils.SafeIntToInt32(limit), Offset: utils.SafeIntToInt32(offset)},
 	)
 	if err != nil {
 		return collection, err
@@ -335,8 +149,8 @@ func SaveInboundFediverseActivity(objectIRI string, actorIRI string, eventType s
 func GetInboundActivities(limit int, offset int) ([]models.FederatedActivity, int, error) {
 	ctx := context.Background()
 	rows, err := _datastore.GetQueries().GetInboundActivitiesWithOffset(ctx, db.GetInboundActivitiesWithOffsetParams{
-		Limit:  limit,
-		Offset: offset,
+		Limit:  utils.SafeIntToInt32(limit),
+		Offset: utils.SafeIntToInt32(offset),
 	})
 	if err != nil {
 		return nil, 0, err