Federated user activity following: Isolated model changes (#8078)

This PR is part of https://codeberg.org/forgejo/forgejo/pulls/4767

This should not have an outside impact but bring all model changes needed & bring migrations.

Reviewed-on: https://codeberg.org/forgejo/forgejo/pulls/8078
Reviewed-by: Earl Warren <earl-warren@noreply.codeberg.org>
Co-authored-by: Michael Jerger <michael.jerger@meissa-gmbh.de>
Co-committed-by: Michael Jerger <michael.jerger@meissa-gmbh.de>

.deadcode-out

@@ -13,6 +13,13 @@ forgejo.org/models
 	IsErrSHANotFound
 	IsErrMergeDivergingFastForwardOnly
 
+forgejo.org/models/activities
+	GetActivityByID
+	NewFederatedUserActivity
+	CreateUserActivity
+	GetFollowingFeeds
+	FederatedUserActivity.loadActor
+
 forgejo.org/models/auth
 	WebAuthnCredentials
 
@@ -54,9 +61,17 @@ forgejo.org/models/user
 	IsErrExternalLoginUserAlreadyExist
 	IsErrExternalLoginUserNotExist
 	NewFederatedUser
+	NewFederatedUserFollower
 	IsErrUserSettingIsNotExist
 	GetUserAllSettings
 	DeleteUserSetting
+	GetFederatedUser
+	GetFederatedUserByUserID
+	UpdateFederatedUser
+	GetFollowersForUser
+	AddFollower
+	RemoveFollower
+	IsFollowingAp
 
 forgejo.org/modules/activitypub
 	NewContext

models/activities/action.go

@@ -442,6 +442,12 @@ func (a *Action) GetIssueContent(ctx context.Context) string {
 	return a.Issue.Content
 }
 
+func GetActivityByID(ctx context.Context, id int64) (*Action, error) {
+	var act Action
+	_, err := db.GetEngine(ctx).ID(id).Get(&act)
+	return &act, err
+}
+
 // GetFeedsOptions options for retrieving feeds
 type GetFeedsOptions struct {
 	db.ListOptions
@@ -595,13 +601,14 @@ func DeleteOldActions(ctx context.Context, olderThan time.Duration) (err error)
 }
 
 // NotifyWatchers creates batch of actions for every watcher.
-func NotifyWatchers(ctx context.Context, actions ...*Action) error {
+func NotifyWatchers(ctx context.Context, actions ...*Action) ([]Action, error) {
 	var watchers []*repo_model.Watch
 	var repo *repo_model.Repository
 	var err error
 	var permCode []bool
 	var permIssue []bool
 	var permPR []bool
+	var out []Action
 
 	e := db.GetEngine(ctx)
 
@@ -612,14 +619,14 @@ func NotifyWatchers(ctx context.Context, actions ...*Action) error {
 			// Add feeds for user self and all watchers.
 			watchers, err = repo_model.GetWatchers(ctx, act.RepoID)
 			if err != nil {
-				return fmt.Errorf("get watchers: %w", err)
+				return nil, fmt.Errorf("get watchers: %w", err)
 			}
 
 			// Be aware that optimizing this correctly into the `GetWatchers` SQL
 			// query is for most cases less performant than doing this.
 			blockedDoerUserIDs, err := user_model.ListBlockedByUsersID(ctx, act.ActUserID)
 			if err != nil {
-				return fmt.Errorf("user_model.ListBlockedByUsersID: %w", err)
+				return nil, fmt.Errorf("user_model.ListBlockedByUsersID: %w", err)
 			}
 
 			if len(blockedDoerUserIDs) > 0 {
@@ -634,8 +641,9 @@ func NotifyWatchers(ctx context.Context, actions ...*Action) error {
 		// Add feed for actioner.
 		act.UserID = act.ActUserID
 		if _, err = e.Insert(act); err != nil {
-			return fmt.Errorf("insert new actioner: %w", err)
+			return nil, fmt.Errorf("insert new actioner: %w", err)
 		}
+		out = append(out, *act)
 
 		if repoChanged {
 			act.loadRepo(ctx)
@@ -643,7 +651,7 @@ func NotifyWatchers(ctx context.Context, actions ...*Action) error {
 
 			// check repo owner exist.
 			if err := act.Repo.LoadOwner(ctx); err != nil {
-				return fmt.Errorf("can't get repo owner: %w", err)
+				return nil, fmt.Errorf("can't get repo owner: %w", err)
 			}
 		} else if act.Repo == nil {
 			act.Repo = repo
@@ -654,7 +662,7 @@ func NotifyWatchers(ctx context.Context, actions ...*Action) error {
 			act.ID = 0
 			act.UserID = act.Repo.Owner.ID
 			if err = db.Insert(ctx, act); err != nil {
-				return fmt.Errorf("insert new actioner: %w", err)
+				return nil, fmt.Errorf("insert new actioner: %w", err)
 			}
 		}
 
@@ -707,26 +715,29 @@ func NotifyWatchers(ctx context.Context, actions ...*Action) error {
 			}
 
 			if err = db.Insert(ctx, act); err != nil {
-				return fmt.Errorf("insert new action: %w", err)
+				return nil, fmt.Errorf("insert new action: %w", err)
 			}
 		}
 	}
-	return nil
+	return out, nil
 }
 
 // NotifyWatchersActions creates batch of actions for every watcher.
-func NotifyWatchersActions(ctx context.Context, acts []*Action) error {
+func NotifyWatchersActions(ctx context.Context, acts []*Action) ([]Action, error) {
 	ctx, committer, err := db.TxContext(ctx)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	defer committer.Close()
+	var out []Action
 	for _, act := range acts {
-		if err := NotifyWatchers(ctx, act); err != nil {
-			return err
+		as, err := NotifyWatchers(ctx, act)
+		if err != nil {
+			return nil, err
 		}
+		out = append(out, as...)
 	}
-	return committer.Commit()
+	return out, committer.Commit()
 }
 
 // DeleteIssueActions delete all actions related with issueID

models/activities/action_test.go

@@ -197,7 +197,8 @@ func TestNotifyWatchers(t *testing.T) {
 		RepoID:    1,
 		OpType:    activities_model.ActionStarRepo,
 	}
-	require.NoError(t, activities_model.NotifyWatchers(db.DefaultContext, action))
+	_, err := activities_model.NotifyWatchers(db.DefaultContext, action)
+	require.NoError(t, err)
 
 	// One watchers are inactive, thus action is only created for user 8, 1, 4, 11
 	unittest.AssertExistsAndLoadBean(t, &activities_model.Action{

models/activities/federated_user_activity.go

@@ -0,0 +1,106 @@
+// Copyright 2024 The Forgejo Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package activities
+
+import (
+	"context"
+	"fmt"
+
+	"forgejo.org/models/db"
+	user_model "forgejo.org/models/user"
+	"forgejo.org/modules/json"
+	"forgejo.org/modules/log"
+	"forgejo.org/modules/timeutil"
+	"forgejo.org/modules/validation"
+
+	ap "github.com/go-ap/activitypub"
+)
+
+type FederatedUserActivity struct {
+	ID           int64 `xorm:"pk autoincr"`
+	UserID       int64 `xorm:"NOT NULL"`
+	ActorID      int64
+	ActorURI     string
+	Actor        *user_model.User   `xorm:"-"` // transient
+	NoteContent  string             `xorm:"TEXT"`
+	NoteURL      string             `xorm:"VARCHAR(255)"`
+	OriginalNote string             `xorm:"TEXT"`
+	Created      timeutil.TimeStamp `xorm:"created"`
+}
+
+func init() {
+	db.RegisterModel(new(FederatedUserActivity))
+}
+
+func NewFederatedUserActivity(userID, actorID int64, actorURI, noteContent, noteURL string, originalNote ap.Activity) (FederatedUserActivity, error) {
+	jsonString, err := json.Marshal(originalNote)
+	if err != nil {
+		return FederatedUserActivity{}, err
+	}
+	result := FederatedUserActivity{
+		UserID:       userID,
+		ActorID:      actorID,
+		ActorURI:     actorURI,
+		NoteContent:  noteContent,
+		NoteURL:      noteURL,
+		OriginalNote: string(jsonString),
+	}
+	if valid, err := validation.IsValid(result); !valid {
+		return FederatedUserActivity{}, err
+	}
+	return result, nil
+}
+
+func (federatedUserActivity FederatedUserActivity) Validate() []string {
+	var result []string
+	result = append(result, validation.ValidateNotEmpty(federatedUserActivity.UserID, "UserID")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUserActivity.ActorID, "ActorID")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUserActivity.ActorURI, "ActorURI")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUserActivity.NoteContent, "NoteContent")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUserActivity.NoteURL, "NoteURL")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUserActivity.OriginalNote, "OriginalNote")...)
+	return result
+}
+
+func CreateUserActivity(ctx context.Context, federatedUserActivity *FederatedUserActivity) error {
+	if valid, err := validation.IsValid(federatedUserActivity); !valid {
+		return err
+	}
+	_, err := db.GetEngine(ctx).Insert(federatedUserActivity)
+	return err
+}
+
+type GetFollowingFeedsOptions struct {
+	db.ListOptions
+}
+
+func GetFollowingFeeds(ctx context.Context, actorID int64, opts GetFollowingFeedsOptions) ([]*FederatedUserActivity, int64, error) {
+	log.Debug("user_id = %s", actorID)
+	sess := db.GetEngine(ctx).Where("user_id = ?", actorID)
+	opts.SetDefaultValues()
+	sess = db.SetSessionPagination(sess, &opts)
+
+	actions := make([]*FederatedUserActivity, 0, opts.PageSize)
+	count, err := sess.FindAndCount(&actions)
+	if err != nil {
+		return nil, 0, fmt.Errorf("FindAndCount: %w", err)
+	}
+	for _, act := range actions {
+		if err := act.loadActor(ctx); err != nil {
+			return nil, 0, err
+		}
+	}
+	return actions, count, err
+}
+
+func (federatedUserActivity *FederatedUserActivity) loadActor(ctx context.Context) error {
+	log.Debug("for activity %s", federatedUserActivity)
+	actorUser, _, err := user_model.GetFederatedUserByUserID(ctx, federatedUserActivity.ActorID)
+	if err != nil {
+		return err
+	}
+	federatedUserActivity.Actor = actorUser
+
+	return nil
+}

models/activities/federated_user_activity_test.go

@@ -0,0 +1,24 @@
+// Copyright 2024 The Forgejo Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package activities
+
+import (
+	"testing"
+
+	"forgejo.org/modules/validation"
+)
+
+func Test_FederatedUserActivityValidation(t *testing.T) {
+	sut := FederatedUserActivity{}
+	sut.UserID = 13
+	sut.ActorID = 33
+	sut.ActorURI = "33"
+	sut.NoteContent = "Any content!"
+	sut.NoteURL = "https://example.org/note/17"
+	sut.OriginalNote = "federatedUserActivityNote-17"
+
+	if res, _ := validation.IsValid(sut); !res {
+		t.Errorf("sut expected to be valid: %v\n", sut.Validate())
+	}
+}

models/forgefed/federationhost.go

@@ -6,6 +6,7 @@ package forgefed
 import (
 	"database/sql"
 	"fmt"
+	"net/url"
 	"strings"
 	"time"
 
@@ -17,9 +18,9 @@ import (
 // swagger:model
 type FederationHost struct {
 	ID             int64                  `xorm:"pk autoincr"`
-	HostFqdn       string                 `xorm:"host_fqdn UNIQUE INDEX VARCHAR(255) NOT NULL"`
+	HostFqdn       string                 `xorm:"host_fqdn UNIQUE(federation_host) INDEX VARCHAR(255) NOT NULL"`
+	HostPort       uint16                 `xorm:" UNIQUE(federation_host) INDEX NOT NULL DEFAULT 443"`
 	NodeInfo       NodeInfo               `xorm:"extends NOT NULL"`
-	HostPort       uint16                 `xorm:"NOT NULL DEFAULT 443"`
 	HostSchema     string                 `xorm:"NOT NULL DEFAULT 'https'"`
 	LatestActivity time.Time              `xorm:"NOT NULL"`
 	KeyID          sql.NullString         `xorm:"key_id UNIQUE"`
@@ -42,6 +43,13 @@ func NewFederationHost(hostFqdn string, nodeInfo NodeInfo, port uint16, schema s
 	return result, nil
 }
 
+func (host FederationHost) AsURL() url.URL {
+	return url.URL{
+		Scheme: host.HostSchema,
+		Host:   fmt.Sprintf("%v:%v", host.HostFqdn, host.HostPort),
+	}
+}
+
 // Validate collects error strings in a slice and returns this
 func (host FederationHost) Validate() []string {
 	var result []string

models/forgefed/nodeinfo.go

@@ -19,10 +19,12 @@ type (
 const (
 	ForgejoSourceType    SoftwareNameType = "forgejo"
 	GiteaSourceType      SoftwareNameType = "gitea"
+	MastodonSourceType   SoftwareNameType = "mastodon"
+	GoToSocialSourceType SoftwareNameType = "gotosocial"
 )
 
 var KnownSourceTypes = []any{
-	ForgejoSourceType, GiteaSourceType,
+	ForgejoSourceType, GiteaSourceType, MastodonSourceType, GoToSocialSourceType,
 }
 
 // ------------------------------------------------ NodeInfoWellKnown ------------------------------------------------

models/forgejo_migrations/migrate.go

@@ -103,6 +103,8 @@ var migrations = []*Migration{
 	NewMigration("Normalize repository.topics to empty slice instead of null", SetTopicsAsEmptySlice),
 	// v31 -> v32
 	NewMigration("Migrate maven package name concatenation", ChangeMavenArtifactConcatenation),
+	// v32 -> v33
+	NewMigration("Add federated user activity tables, update the `federated_user` table & add indexes", FederatedUserActivityMigration),
 }
 
 // GetCurrentDBVersion returns the current Forgejo database version.

models/forgejo_migrations/v33.go

@@ -0,0 +1,126 @@
+// Copyright 2025 The Forgejo Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package forgejo_migrations //nolint:revive
+
+import (
+	"fmt"
+
+	"forgejo.org/modules/log"
+	"forgejo.org/modules/timeutil"
+
+	"xorm.io/xorm"
+)
+
+func dropOldFederationHostIndexes(x *xorm.Engine) {
+	// drop unique index on HostFqdn
+	type FederationHost struct {
+		ID       int64  `xorm:"pk autoincr"`
+		HostFqdn string `xorm:"host_fqdn UNIQUE INDEX VARCHAR(255) NOT NULL"`
+	}
+
+	err := x.DropIndexes(FederationHost{})
+	if err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+}
+
+func addFederatedUserActivityTables(x *xorm.Engine) {
+	type FederatedUserActivity struct {
+		ID           int64 `xorm:"pk autoincr"`
+		UserID       int64 `xorm:"NOT NULL INDEX user_id"`
+		ActorID      int64
+		ActorURI     string
+		NoteContent  string             `xorm:"TEXT"`
+		NoteURL      string             `xorm:"VARCHAR(255)"`
+		OriginalNote string             `xorm:"TEXT"`
+		Created      timeutil.TimeStamp `xorm:"created"`
+	}
+
+	// add unique index on HostFqdn+HostPort
+	type FederationHost struct {
+		ID       int64  `xorm:"pk autoincr"`
+		HostFqdn string `xorm:"host_fqdn UNIQUE(federation_host) INDEX VARCHAR(255) NOT NULL"`
+		HostPort uint16 `xorm:"UNIQUE(federation_host) INDEX NOT NULL DEFAULT 443"`
+	}
+
+	type FederatedUserFollower struct {
+		ID int64 `xorm:"pk autoincr"`
+
+		FollowedUserID  int64 `xorm:"NOT NULL unique(fuf_rel)"`
+		FollowingUserID int64 `xorm:"NOT NULL unique(fuf_rel)"`
+	}
+
+	// Add InboxPath to FederatedUser & add index fo UserID
+	type FederatedUser struct {
+		ID        int64 `xorm:"pk autoincr"`
+		UserID    int64 `xorm:"NOT NULL INDEX user_id"`
+		InboxPath string
+	}
+
+	var err error
+
+	err = x.Sync(&FederationHost{})
+	if err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+
+	err = x.Sync(&FederatedUserActivity{})
+	if err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+
+	err = x.Sync(&FederatedUserFollower{})
+	if err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+
+	err = x.Sync(&FederatedUser{})
+	if err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+
+	// Migrate
+	sessMigration := x.NewSession()
+	defer sessMigration.Close()
+	if err := sessMigration.Begin(); err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+	federatedUsers := make([]*FederatedUser, 0)
+	err = sessMigration.OrderBy("id").Find(&federatedUsers)
+	if err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+
+	for _, federatedUser := range federatedUsers {
+		if federatedUser.InboxPath != "" {
+			log.Info("migration[33]: This user was already migrated: %v", federatedUser)
+		} else {
+			// Migrate User.InboxPath
+			sql := "UPDATE `federated_user` SET `inbox_path` = ? WHERE `id` = ?"
+			if _, err := sessMigration.Exec(sql, fmt.Sprintf("/api/v1/activitypub/user-id/%v/inbox", federatedUser.UserID), federatedUser.ID); err != nil {
+				log.Warn("migration[33]: There was an issue: %v", err)
+				return
+			}
+		}
+	}
+
+	err = sessMigration.Commit()
+	if err != nil {
+		log.Warn("migration[33]: There was an issue: %v", err)
+		return
+	}
+}
+
+func FederatedUserActivityMigration(x *xorm.Engine) error {
+	dropOldFederationHostIndexes(x)
+	addFederatedUserActivityTables(x)
+	return nil
+}

models/forgejo_migrations/v33_test.go

@@ -0,0 +1,46 @@
+// Copyright 2025 The Forgejo Authors.
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package forgejo_migrations //nolint:revive
+
+import (
+	"testing"
+	"time"
+
+	migration_tests "forgejo.org/models/migrations/test"
+	"forgejo.org/modules/log"
+	ft "forgejo.org/modules/test"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_FederatedUserActivityMigration(t *testing.T) {
+	lc, cl := ft.NewLogChecker(log.DEFAULT, log.WARN)
+	lc.Filter("migration[33]")
+	defer cl()
+
+	// intentionally conflicting definition
+	type FederatedUser struct {
+		ID     int64 `xorm:"pk autoincr"`
+		UserID string
+	}
+
+	// Prepare TestEnv
+	x, deferable := migration_tests.PrepareTestEnv(t, 0,
+		new(FederatedUser),
+	)
+	sessTest := x.NewSession()
+	sessTest.Insert(FederatedUser{UserID: "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" +
+		"1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" +
+		"1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890"})
+	sessTest.Commit()
+	defer deferable()
+	if x == nil || t.Failed() {
+		return
+	}
+
+	require.NoError(t, FederatedUserActivityMigration(x))
+	logFiltered, _ := lc.Check(5 * time.Second)
+	assert.NotEmpty(t, logFiltered)
+}

models/user/federated_user.go

@@ -11,19 +11,21 @@ import (
 
 type FederatedUser struct {
 	ID                    int64                  `xorm:"pk autoincr"`
-	UserID                int64                  `xorm:"NOT NULL"`
+	UserID                int64                  `xorm:"NOT NULL INDEX user_id"`
 	ExternalID            string                 `xorm:"UNIQUE(federation_user_mapping) NOT NULL"`
 	FederationHostID      int64                  `xorm:"UNIQUE(federation_user_mapping) NOT NULL"`
 	KeyID                 sql.NullString         `xorm:"key_id UNIQUE"`
 	PublicKey             sql.Null[sql.RawBytes] `xorm:"BLOB"`
+	InboxPath             string
 	NormalizedOriginalURL string // This field is just to keep original information. Pls. do not use for search or as ID!
 }
 
-func NewFederatedUser(userID int64, externalID string, federationHostID int64, normalizedOriginalURL string) (FederatedUser, error) {
+func NewFederatedUser(userID int64, externalID string, federationHostID int64, inboxPath, normalizedOriginalURL string) (FederatedUser, error) {
 	result := FederatedUser{
 		UserID:                userID,
 		ExternalID:            externalID,
 		FederationHostID:      federationHostID,
+		InboxPath:             inboxPath,
 		NormalizedOriginalURL: normalizedOriginalURL,
 	}
 	if valid, err := validation.IsValid(result); !valid {
@@ -32,10 +34,11 @@ func NewFederatedUser(userID int64, externalID string, federationHostID int64, n
 	return result, nil
 }
 
-func (user FederatedUser) Validate() []string {
+func (federatedUser FederatedUser) Validate() []string {
 	var result []string
-	result = append(result, validation.ValidateNotEmpty(user.UserID, "UserID")...)
-	result = append(result, validation.ValidateNotEmpty(user.ExternalID, "ExternalID")...)
-	result = append(result, validation.ValidateNotEmpty(user.FederationHostID, "FederationHostID")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUser.UserID, "UserID")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUser.ExternalID, "ExternalID")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUser.FederationHostID, "FederationHostID")...)
+	result = append(result, validation.ValidateNotEmpty(federatedUser.InboxPath, "InboxPath")...)
 	return result
 }

models/user/federated_user_follower.go

@@ -0,0 +1,30 @@
+// Copyright 2025 The Forgejo Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package user
+
+import "forgejo.org/modules/validation"
+
+type FederatedUserFollower struct {
+	ID              int64 `xorm:"pk autoincr"`
+	FollowedUserID  int64 `xorm:"NOT NULL unique(fuf_rel)"`
+	FollowingUserID int64 `xorm:"NOT NULL unique(fuf_rel)"`
+}
+
+func NewFederatedUserFollower(followedUserID, federatedUserID int64) (FederatedUserFollower, error) {
+	result := FederatedUserFollower{
+		FollowedUserID:  followedUserID,
+		FollowingUserID: federatedUserID,
+	}
+	if valid, err := validation.IsValid(result); !valid {
+		return FederatedUserFollower{}, err
+	}
+	return result, nil
+}
+
+func (user FederatedUserFollower) Validate() []string {
+	var result []string
+	result = append(result, validation.ValidateNotEmpty(user.FollowedUserID, "FollowedUserID")...)
+	result = append(result, validation.ValidateNotEmpty(user.FollowingUserID, "FollowingUserID")...)
+	return result
+}

models/user/federated_user_follower_test.go

@@ -0,0 +1,27 @@
+// Copyright 2024 The Forgejo Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package user
+
+import (
+	"testing"
+
+	"forgejo.org/modules/validation"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_FederatedUserFollowerValidation(t *testing.T) {
+	sut := FederatedUserFollower{
+		FollowedUserID:  12,
+		FollowingUserID: 1,
+	}
+	res, err := validation.IsValid(sut)
+	assert.Truef(t, res, "sut should be valid but was %q", err)
+
+	sut = FederatedUserFollower{
+		FollowedUserID: 1,
+	}
+	res, _ = validation.IsValid(sut)
+	assert.False(t, res, "sut should be invalid")
+}

models/user/federated_user_test.go

@@ -14,6 +14,7 @@ func Test_FederatedUserValidation(t *testing.T) {
 		UserID:           12,
 		ExternalID:       "12",
 		FederationHostID: 1,
+		InboxPath:        "/api/v1/activitypub/user-id/12/inbox",
 	}
 	if res, err := validation.IsValid(sut); !res {
 		t.Errorf("sut should be valid but was %q", err)
@@ -22,6 +23,7 @@ func Test_FederatedUserValidation(t *testing.T) {
 	sut = FederatedUser{
 		ExternalID:       "12",
 		FederationHostID: 1,
+		InboxPath:        "/api/v1/activitypub/user-id/12/inbox",
 	}
 	if res, _ := validation.IsValid(sut); res {
 		t.Error("sut should be invalid")

models/user/follow.go

@@ -11,6 +11,7 @@ import (
 )
 
 // Follow represents relations of user and their followers.
+// TODO: We should unify Activity-pub-following and classical following (see models/user/user_repository.go#IsFollowingAp)
 type Follow struct {
 	ID          int64              `xorm:"pk autoincr"`
 	UserID      int64              `xorm:"UNIQUE(follow)"`

models/user/user_repository.go

@@ -1,4 +1,4 @@
-// Copyright 2024 The Forgejo Authors. All rights reserved.
+// Copyright 2024, 2025 The Forgejo Authors. All rights reserved.
 // SPDX-License-Identifier: MIT
 
 package user
@@ -8,12 +8,14 @@ import (
 	"fmt"
 
 	"forgejo.org/models/db"
+	"forgejo.org/modules/log"
 	"forgejo.org/modules/optional"
 	"forgejo.org/modules/validation"
 )
 
 func init() {
 	db.RegisterModel(new(FederatedUser))
+	db.RegisterModel(new(FederatedUserFollower))
 }
 
 func CreateFederatedUser(ctx context.Context, user *User, federatedUser *FederatedUser) error {
@@ -30,7 +32,12 @@ func CreateFederatedUser(ctx context.Context, user *User, federatedUser *Federat
 	if err != nil {
 		return err
 	}
-	defer committer.Close()
+	defer func() {
+		err := committer.Close()
+		if err != nil {
+			log.Error("Error closing committer: %v", err)
+		}
+	}()
 
 	if err := CreateUser(ctx, user, &overwrite); err != nil {
 		return err
@@ -50,6 +57,14 @@ func CreateFederatedUser(ctx context.Context, user *User, federatedUser *Federat
 	return committer.Commit()
 }
 
+func (federatedUser *FederatedUser) UpdateFederatedUser(ctx context.Context) error {
+	if _, err := validation.IsValid(federatedUser); err != nil {
+		return err
+	}
+	_, err := db.GetEngine(ctx).ID(federatedUser.ID).Cols("inbox_path").Update(federatedUser)
+	return err
+}
+
 func FindFederatedUser(ctx context.Context, externalID string, federationHostID int64) (*User, *FederatedUser, error) {
 	federatedUser := new(FederatedUser)
 	user := new(User)
@@ -75,6 +90,41 @@ func FindFederatedUser(ctx context.Context, externalID string, federationHostID
 	return user, federatedUser, nil
 }
 
+func GetFederatedUser(ctx context.Context, externalID string, federationHostID int64) (*User, *FederatedUser, error) {
+	user, federatedUser, err := FindFederatedUser(ctx, externalID, federationHostID)
+	if err != nil {
+		return nil, nil, err
+	} else if federatedUser == nil {
+		return nil, nil, fmt.Errorf("FederatedUser for externalId = %v and federationHostId = %v does not exist", externalID, federationHostID)
+	}
+	return user, federatedUser, nil
+}
+
+func GetFederatedUserByUserID(ctx context.Context, userID int64) (*User, *FederatedUser, error) {
+	federatedUser := new(FederatedUser)
+	user := new(User)
+	has, err := db.GetEngine(ctx).Where("user_id=?", userID).Get(federatedUser)
+	if err != nil {
+		return nil, nil, err
+	} else if !has {
+		return nil, nil, fmt.Errorf("Federated user %v does not exist", federatedUser.UserID)
+	}
+	has, err = db.GetEngine(ctx).ID(federatedUser.UserID).Get(user)
+	if err != nil {
+		return nil, nil, err
+	} else if !has {
+		return nil, nil, fmt.Errorf("User %v for federated user is missing", federatedUser.UserID)
+	}
+
+	if res, err := validation.IsValid(*user); !res {
+		return nil, nil, err
+	}
+	if res, err := validation.IsValid(*federatedUser); !res {
+		return nil, nil, err
+	}
+	return user, federatedUser, nil
+}
+
 func FindFederatedUserByKeyID(ctx context.Context, keyID string) (*User, *FederatedUser, error) {
 	federatedUser := new(FederatedUser)
 	user := new(User)
@@ -101,7 +151,85 @@ func FindFederatedUserByKeyID(ctx context.Context, keyID string) (*User, *Federa
 	return user, federatedUser, nil
 }
 
+func UpdateFederatedUser(ctx context.Context, federatedUser *FederatedUser) error {
+	if res, err := validation.IsValid(federatedUser); !res {
+		return err
+	}
+	_, err := db.GetEngine(ctx).ID(federatedUser.ID).Update(federatedUser)
+	return err
+}
+
 func DeleteFederatedUser(ctx context.Context, userID int64) error {
 	_, err := db.GetEngine(ctx).Delete(&FederatedUser{UserID: userID})
 	return err
 }
+
+func GetFollowersForUser(ctx context.Context, user *User) ([]*FederatedUserFollower, error) {
+	if res, err := validation.IsValid(user); !res {
+		return nil, err
+	}
+	followers := make([]*FederatedUserFollower, 0, 8)
+
+	err := db.GetEngine(ctx).
+		Where("followed_user_id = ?", user.ID).
+		Find(&followers)
+	if err != nil {
+		return nil, err
+	}
+	for _, element := range followers {
+		if res, err := validation.IsValid(*element); !res {
+			return nil, err
+		}
+	}
+	return followers, nil
+}
+
+func AddFollower(ctx context.Context, followedUser *User, followingUser *FederatedUser) (*FederatedUserFollower, error) {
+	if res, err := validation.IsValid(followedUser); !res {
+		return nil, err
+	}
+	if res, err := validation.IsValid(followingUser); !res {
+		return nil, err
+	}
+
+	federatedUserFollower, err := NewFederatedUserFollower(followedUser.ID, followingUser.UserID)
+	if err != nil {
+		return nil, err
+	}
+	_, err = db.GetEngine(ctx).Insert(&federatedUserFollower)
+	if err != nil {
+		return nil, err
+	}
+
+	return &federatedUserFollower, err
+}
+
+func RemoveFollower(ctx context.Context, followedUser *User, followingUser *FederatedUser) error {
+	if res, err := validation.IsValid(followedUser); !res {
+		return err
+	}
+	if res, err := validation.IsValid(followingUser); !res {
+		return err
+	}
+
+	_, err := db.GetEngine(ctx).Delete(&FederatedUserFollower{
+		FollowedUserID:  followedUser.ID,
+		FollowingUserID: followingUser.UserID,
+	})
+	return err
+}
+
+// TODO: We should unify Activity-pub-following and classical following (see models/user/follow.go)
+func IsFollowingAp(ctx context.Context, followedUser *User, followingUser *FederatedUser) (bool, error) {
+	if res, err := validation.IsValid(followedUser); !res {
+		return false, err
+	}
+	if res, err := validation.IsValid(followingUser); !res {
+		return false, err
+	}
+
+	return db.GetEngine(ctx).Get(&FederatedUserFollower{
+		FollowedUserID:  followedUser.ID,
+		FollowingUserID: followingUser.UserID,
+	})
+}

models/user/user_test.go

@@ -148,7 +148,7 @@ func TestAPActorID_APActorID(t *testing.T) {
 	assert.Equal(t, expected, url)
 }
 
-func TestAPActorKeyID(t *testing.T) {
+func TestKeyID(t *testing.T) {
 	user := user_model.User{ID: 1}
 	url := user.APActorKeyID()
 	expected := "https://try.gitea.io/api/v1/activitypub/user-id/1#main-key"

services/federation/federation_service.go

@@ -211,6 +211,11 @@ func CreateUserFromAP(ctx context.Context, personID fm.PersonID, federationHostI
 		return nil, nil, err
 	}
 
+	inbox, err := url.ParseRequestURI(person.Inbox.GetLink().String())
+	if err != nil {
+		return nil, nil, err
+	}
+
 	newUser := user.User{
 		LowerName:                    strings.ToLower(name),
 		Name:                         name,
@@ -227,6 +232,7 @@ func CreateUserFromAP(ctx context.Context, personID fm.PersonID, federationHostI
 	federatedUser := user.FederatedUser{
 		ExternalID:            personID.ID,
 		FederationHostID:      federationHostID,
+		InboxPath:             inbox.Path,
 		NormalizedOriginalURL: personID.AsURI(),
 	}
 

services/feed/action.go

@@ -39,6 +39,24 @@ func NewNotifier() notify_service.Notifier {
 	return &actionNotifier{}
 }
 
+func notifyAll(ctx context.Context, action *activities_model.Action) error {
+	_, err := activities_model.NotifyWatchers(ctx, action)
+	if err != nil {
+		return err
+	}
+	return err
+	// return federation_service.NotifyActivityPubFollowers(ctx, out)
+}
+
+func notifyAllActions(ctx context.Context, acts []*activities_model.Action) error {
+	_, err := activities_model.NotifyWatchersActions(ctx, acts)
+	if err != nil {
+		return err
+	}
+	return nil
+	// return federation_service.NotifyActivityPubFollowers(ctx, out)
+}
+
 func (a *actionNotifier) NewIssue(ctx context.Context, issue *issues_model.Issue, mentions []*user_model.User) {
 	if err := issue.LoadPoster(ctx); err != nil {
 		log.Error("issue.LoadPoster: %v", err)
@@ -50,7 +68,7 @@ func (a *actionNotifier) NewIssue(ctx context.Context, issue *issues_model.Issue
 	}
 	repo := issue.Repo
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: issue.Poster.ID,
 		ActUser:   issue.Poster,
 		OpType:    activities_model.ActionCreateIssue,
@@ -91,7 +109,7 @@ func (a *actionNotifier) IssueChangeStatus(ctx context.Context, doer *user_model
 	}
 
 	// Notify watchers for whatever action comes in, ignore if no action type.
-	if err := activities_model.NotifyWatchers(ctx, act); err != nil {
+	if err := notifyAll(ctx, act); err != nil {
 		log.Error("NotifyWatchers: %v", err)
 	}
 }
@@ -127,7 +145,7 @@ func (a *actionNotifier) CreateIssueComment(ctx context.Context, doer *user_mode
 	}
 
 	// Notify watchers for whatever action comes in, ignore if no action type.
-	if err := activities_model.NotifyWatchers(ctx, act); err != nil {
+	if err := notifyAll(ctx, act); err != nil {
 		log.Error("NotifyWatchers: %v", err)
 	}
 }
@@ -146,7 +164,7 @@ func (a *actionNotifier) NewPullRequest(ctx context.Context, pull *issues_model.
 		return
 	}
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: pull.Issue.Poster.ID,
 		ActUser:   pull.Issue.Poster,
 		OpType:    activities_model.ActionCreatePullRequest,
@@ -160,7 +178,7 @@ func (a *actionNotifier) NewPullRequest(ctx context.Context, pull *issues_model.
 }
 
 func (a *actionNotifier) RenameRepository(ctx context.Context, doer *user_model.User, repo *repo_model.Repository, oldRepoName string) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionRenameRepo,
@@ -174,7 +192,7 @@ func (a *actionNotifier) RenameRepository(ctx context.Context, doer *user_model.
 }
 
 func (a *actionNotifier) TransferRepository(ctx context.Context, doer *user_model.User, repo *repo_model.Repository, oldOwnerName string) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionTransferRepo,
@@ -188,7 +206,7 @@ func (a *actionNotifier) TransferRepository(ctx context.Context, doer *user_mode
 }
 
 func (a *actionNotifier) CreateRepository(ctx context.Context, doer, u *user_model.User, repo *repo_model.Repository) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionCreateRepo,
@@ -201,7 +219,7 @@ func (a *actionNotifier) CreateRepository(ctx context.Context, doer, u *user_mod
 }
 
 func (a *actionNotifier) ForkRepository(ctx context.Context, doer *user_model.User, oldRepo, repo *repo_model.Repository) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionCreateRepo,
@@ -266,13 +284,13 @@ func (a *actionNotifier) PullRequestReview(ctx context.Context, pr *issues_model
 		actions = append(actions, action)
 	}
 
-	if err := activities_model.NotifyWatchersActions(ctx, actions); err != nil {
+	if err := notifyAllActions(ctx, actions); err != nil {
 		log.Error("notify watchers '%d/%d': %v", review.Reviewer.ID, review.Issue.RepoID, err)
 	}
 }
 
 func (*actionNotifier) MergePullRequest(ctx context.Context, doer *user_model.User, pr *issues_model.PullRequest) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionMergePullRequest,
@@ -286,7 +304,7 @@ func (*actionNotifier) MergePullRequest(ctx context.Context, doer *user_model.Us
 }
 
 func (*actionNotifier) AutoMergePullRequest(ctx context.Context, doer *user_model.User, pr *issues_model.PullRequest) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionAutoMergePullRequest,
@@ -304,7 +322,7 @@ func (*actionNotifier) NotifyPullRevieweDismiss(ctx context.Context, doer *user_
 	if len(review.OriginalAuthor) > 0 {
 		reviewerName = review.OriginalAuthor
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    activities_model.ActionPullReviewDismissed,
@@ -342,7 +360,7 @@ func (a *actionNotifier) PushCommits(ctx context.Context, pusher *user_model.Use
 		opType = activities_model.ActionDeleteBranch
 	}
 
-	if err = activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err = notifyAll(ctx, &activities_model.Action{
 		ActUserID: pusher.ID,
 		ActUser:   pusher,
 		OpType:    opType,
@@ -362,7 +380,7 @@ func (a *actionNotifier) CreateRef(ctx context.Context, doer *user_model.User, r
 		// has sent same action in `PushCommits`, so skip it.
 		return
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    opType,
@@ -381,7 +399,7 @@ func (a *actionNotifier) DeleteRef(ctx context.Context, doer *user_model.User, r
 		// has sent same action in `PushCommits`, so skip it.
 		return
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: doer.ID,
 		ActUser:   doer,
 		OpType:    opType,
@@ -405,7 +423,7 @@ func (a *actionNotifier) SyncPushCommits(ctx context.Context, pusher *user_model
 		return
 	}
 
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: repo.OwnerID,
 		ActUser:   repo.MustOwner(ctx),
 		OpType:    activities_model.ActionMirrorSyncPush,
@@ -420,7 +438,7 @@ func (a *actionNotifier) SyncPushCommits(ctx context.Context, pusher *user_model
 }
 
 func (a *actionNotifier) SyncCreateRef(ctx context.Context, doer *user_model.User, repo *repo_model.Repository, refFullName git.RefName, refID string) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: repo.OwnerID,
 		ActUser:   repo.MustOwner(ctx),
 		OpType:    activities_model.ActionMirrorSyncCreate,
@@ -434,7 +452,7 @@ func (a *actionNotifier) SyncCreateRef(ctx context.Context, doer *user_model.Use
 }
 
 func (a *actionNotifier) SyncDeleteRef(ctx context.Context, doer *user_model.User, repo *repo_model.Repository, refFullName git.RefName) {
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: repo.OwnerID,
 		ActUser:   repo.MustOwner(ctx),
 		OpType:    activities_model.ActionMirrorSyncDelete,
@@ -452,7 +470,7 @@ func (a *actionNotifier) NewRelease(ctx context.Context, rel *repo_model.Release
 		log.Error("LoadAttributes: %v", err)
 		return
 	}
-	if err := activities_model.NotifyWatchers(ctx, &activities_model.Action{
+	if err := notifyAll(ctx, &activities_model.Action{
 		ActUserID: rel.PublisherID,
 		ActUser:   rel.Publisher,
 		OpType:    activities_model.ActionPublishRelease,