ActivityPub background actor validation, update and cleanup process (#4765)

* feat(ap): add validation+update+cleanup AP actor job

* fix(test): use time.Time instead of sql.NullTime

* feat(test): add integration test for AP follower cleanup/validation job

activitypub/activitypub.go

@@ -3,6 +3,7 @@ package activitypub
 import (
 	"github.com/owncast/owncast/activitypub/crypto"
 	"github.com/owncast/owncast/activitypub/inbox"
+	"github.com/owncast/owncast/activitypub/jobs"
 	"github.com/owncast/owncast/activitypub/outbox"
 	"github.com/owncast/owncast/activitypub/persistence"
 	"github.com/owncast/owncast/activitypub/persistence/followersrepository"
@@ -32,6 +33,9 @@ func Start(datastore *data.Datastore) {
 			log.Errorln("Unable to get private key", err)
 		}
 	}
+
+	// Start follower validation background job.
+	jobs.StartFollowerValidationJob()
 }
 
 func getOutboundWorkerPoolSize() int {

activitypub/jobs/followervalidation.go

@@ -0,0 +1,117 @@
+package jobs
+
+import (
+	"time"
+
+	"github.com/owncast/owncast/activitypub/persistence/followersrepository"
+	"github.com/owncast/owncast/activitypub/resolvers"
+	"github.com/owncast/owncast/config"
+	"github.com/owncast/owncast/models"
+	"github.com/owncast/owncast/persistence/configrepository"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	// ValidationInterval is how often the validation job runs.
+	ValidationInterval = 1 * time.Hour
+	// FollowersPerRun is how many followers to validate per job run.
+	FollowersPerRun = 5
+	// FailureDurationThreshold is how long a follower must be unreachable before removal.
+	FailureDurationThreshold = 7 * 24 * time.Hour // 7 days
+	// DelayBetweenFollowers is the delay between validating individual followers.
+	DelayBetweenFollowers = 2 * time.Second
+)
+
+// GetValidationInterval returns the configured validation interval, or the default.
+func GetValidationInterval() time.Duration {
+	if config.FollowerValidationInterval > 0 {
+		return config.FollowerValidationInterval
+	}
+	return ValidationInterval
+}
+
+// StartFollowerValidationJob starts the background job that periodically validates followers.
+func StartFollowerValidationJob() {
+	interval := GetValidationInterval()
+	ticker := time.NewTicker(interval)
+	go func() {
+		for range ticker.C {
+			runFollowerValidation()
+		}
+	}()
+	log.Debugf("Follower validation job scheduled with interval %v", interval)
+}
+
+func runFollowerValidation() {
+	configRepo := configrepository.Get()
+	if !configRepo.GetFederationEnabled() {
+		return
+	}
+
+	followersRepo := followersrepository.Get()
+	followers, err := followersRepo.GetFollowersToValidate(FollowersPerRun)
+	if err != nil {
+		log.Errorln("Failed to get followers for validation:", err)
+		return
+	}
+
+	for _, follower := range followers {
+		validateAndUpdateFollower(followersRepo, follower)
+		time.Sleep(DelayBetweenFollowers)
+	}
+}
+
+func validateAndUpdateFollower(repo followersrepository.FollowersRepository, follower models.Follower) {
+	resolvedActor, err := resolvers.GetResolvedActorFromIRI(follower.ActorIRI)
+	if err != nil {
+		handleValidationFailure(repo, follower, err)
+		return
+	}
+
+	// Success - clear failure timestamp and update data
+	if err := repo.UpdateFollowerValidationSuccess(follower.ActorIRI); err != nil {
+		log.Errorln("Failed to update validation success:", err)
+	}
+
+	// Update follower data
+	if err := repo.Update(
+		resolvedActor.ActorIriString(),
+		resolvedActor.InboxString(),
+		resolvedActor.SharedInboxString(),
+		resolvedActor.Name,
+		resolvedActor.FullUsername,
+		resolvedActor.ImageString(),
+	); err != nil {
+		log.Errorln("Failed to update follower data:", err)
+	}
+}
+
+func handleValidationFailure(repo followersrepository.FollowersRepository, follower models.Follower, resolveErr error) {
+	log.Debugf("Follower validation failed for %s: %v", follower.ActorIRI, resolveErr)
+
+	// Check removal eligibility BEFORE updating failure timestamp.
+	// We use the existing FirstValidationFailureAt from the database to determine
+	// if this follower has been failing long enough to be removed.
+	// If FirstValidationFailureAt is not set, this is a new failure and we just record it.
+	shouldRemove := false
+	if follower.FirstValidationFailureAt.Valid {
+		failureDuration := time.Since(follower.FirstValidationFailureAt.Time)
+		shouldRemove = failureDuration >= FailureDurationThreshold
+	}
+
+	// Update failure timestamp (sets first_validation_failure_at if not already set)
+	if err := repo.UpdateFollowerValidationFailure(follower.ActorIRI); err != nil {
+		log.Errorln("Failed to update validation failure:", err)
+		return
+	}
+
+	// Remove follower if they've exceeded the failure threshold
+	if shouldRemove {
+		failureDuration := time.Since(follower.FirstValidationFailureAt.Time)
+		log.Infof("Removing follower %s after %v of consecutive failures",
+			follower.ActorIRI, failureDuration.Round(time.Hour))
+		if err := repo.RemoveByIRI(follower.ActorIRI); err != nil {
+			log.Errorln("Failed to remove invalid follower:", err)
+		}
+	}
+}

activitypub/jobs/followervalidation_test.go

@@ -0,0 +1,351 @@
+package jobs
+
+import (
+	"net/url"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/go-fed/activity/streams"
+	"github.com/owncast/owncast/activitypub/apmodels"
+	"github.com/owncast/owncast/activitypub/persistence"
+	"github.com/owncast/owncast/activitypub/persistence/followersrepository"
+	"github.com/owncast/owncast/core/data"
+	"github.com/owncast/owncast/utils"
+)
+
+func TestMain(m *testing.M) {
+	setup()
+	code := m.Run()
+	os.Exit(code)
+}
+
+var datastore *data.Datastore
+
+func setup() {
+	resetTestDatabase()
+}
+
+// resetTestDatabase initializes a fresh in-memory database for testing.
+func resetTestDatabase() {
+	data.SetupPersistence(":memory:")
+	datastore = data.GetDatastore()
+	persistence.Setup(datastore)
+}
+
+// setupTestWithRepo resets the database and returns a new repository instance.
+func setupTestWithRepo(t *testing.T) followersrepository.FollowersRepository {
+	t.Helper()
+	resetTestDatabase()
+	return followersrepository.New(datastore)
+}
+
+func createTestFollower(repo followersrepository.FollowersRepository, iri, inbox, name, username string) {
+	actorIRI, _ := url.Parse(iri)
+	inboxURL, _ := url.Parse(inbox)
+	requestIRI, _ := url.Parse("https://fake.server/follow/request")
+	fakeRequest := streams.NewActivityStreamsFollow()
+
+	repo.Add(apmodels.ActivityPubActor{
+		ActorIri:         actorIRI,
+		Inbox:            inboxURL,
+		Name:             name,
+		Username:         username,
+		FullUsername:     username + "@fake.server",
+		FollowRequestIri: requestIRI,
+		RequestObject:    fakeRequest,
+	}, true)
+}
+
+func TestGetFollowersToValidate(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create some test followers
+	for i := 0; i < 10; i++ {
+		user, _ := utils.GenerateRandomString(10)
+		createTestFollower(repo, "https://fake.server/user/"+user, "https://fake.server/user/"+user+"/inbox", user, user)
+	}
+
+	// Get followers to validate
+	followers, err := repo.GetFollowersToValidate(5)
+	if err != nil {
+		t.Fatalf("Error getting followers to validate: %s", err)
+	}
+
+	if len(followers) != 5 {
+		t.Errorf("Expected 5 followers to validate, got %d", len(followers))
+	}
+}
+
+func TestUpdateFollowerValidationSuccess(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create a test follower
+	testIRI := "https://fake.server/user/testuser"
+	createTestFollower(repo, testIRI, "https://fake.server/user/testuser/inbox", "Test User", "testuser")
+
+	// Mark validation as successful
+	err := repo.UpdateFollowerValidationSuccess(testIRI)
+	if err != nil {
+		t.Fatalf("Error updating follower validation success: %s", err)
+	}
+
+	// Verify the follower was updated
+	followers, err := repo.GetFollowersToValidate(10)
+	if err != nil {
+		t.Fatalf("Error getting followers: %s", err)
+	}
+
+	// After validation success, FirstValidationFailureAt should be NULL/invalid
+	for _, f := range followers {
+		if f.ActorIRI == testIRI {
+			if f.FirstValidationFailureAt.Valid {
+				t.Error("Expected FirstValidationFailureAt to be NULL after success")
+			}
+		}
+	}
+}
+
+func TestUpdateFollowerValidationFailure(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create a test follower
+	testIRI := "https://fake.server/user/testuser"
+	createTestFollower(repo, testIRI, "https://fake.server/user/testuser/inbox", "Test User", "testuser")
+
+	// Mark validation as failed
+	err := repo.UpdateFollowerValidationFailure(testIRI)
+	if err != nil {
+		t.Fatalf("Error updating follower validation failure: %s", err)
+	}
+
+	// Verify the FirstValidationFailureAt is set
+	followers, err := repo.GetFollowersToValidate(10)
+	if err != nil {
+		t.Fatalf("Error getting followers: %s", err)
+	}
+
+	found := false
+	for _, f := range followers {
+		if f.ActorIRI == testIRI {
+			found = true
+			if !f.FirstValidationFailureAt.Valid {
+				t.Error("Expected FirstValidationFailureAt to be set after failure")
+			}
+		}
+	}
+
+	if !found {
+		t.Error("Test follower not found in results")
+	}
+}
+
+func TestValidationFailureClearedOnSuccess(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create a test follower
+	testIRI := "https://fake.server/user/testuser"
+	createTestFollower(repo, testIRI, "https://fake.server/user/testuser/inbox", "Test User", "testuser")
+
+	// Mark validation as failed first
+	err := repo.UpdateFollowerValidationFailure(testIRI)
+	if err != nil {
+		t.Fatalf("Error updating follower validation failure: %s", err)
+	}
+
+	// Then mark as successful
+	err = repo.UpdateFollowerValidationSuccess(testIRI)
+	if err != nil {
+		t.Fatalf("Error updating follower validation success: %s", err)
+	}
+
+	// Verify FirstValidationFailureAt is cleared
+	followers, err := repo.GetFollowersToValidate(10)
+	if err != nil {
+		t.Fatalf("Error getting followers: %s", err)
+	}
+
+	for _, f := range followers {
+		if f.ActorIRI == testIRI {
+			if f.FirstValidationFailureAt.Valid {
+				t.Error("Expected FirstValidationFailureAt to be NULL after success")
+			}
+		}
+	}
+}
+
+func TestRemoveByIRI(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create a test follower
+	testIRI := "https://fake.server/user/testuser"
+	createTestFollower(repo, testIRI, "https://fake.server/user/testuser/inbox", "Test User", "testuser")
+
+	// Verify follower exists
+	count, err := repo.GetCount()
+	if err != nil {
+		t.Fatalf("Error getting count: %s", err)
+	}
+	if count != 1 {
+		t.Errorf("Expected 1 follower, got %d", count)
+	}
+
+	// Remove the follower
+	err = repo.RemoveByIRI(testIRI)
+	if err != nil {
+		t.Fatalf("Error removing follower: %s", err)
+	}
+
+	// Verify follower is removed
+	count, err = repo.GetCount()
+	if err != nil {
+		t.Fatalf("Error getting count: %s", err)
+	}
+	if count != 0 {
+		t.Errorf("Expected 0 followers after removal, got %d", count)
+	}
+}
+
+func TestFailureDurationThresholdLogic(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create a test follower
+	testIRI := "https://fake.server/user/testfailure"
+	createTestFollower(repo, testIRI, "https://fake.server/user/testfailure/inbox", "Test User", "testfailure")
+
+	// Set first_validation_failure_at to 8 days ago (past threshold)
+	eightDaysAgo := time.Now().Add(-8 * 24 * time.Hour)
+	_, err := datastore.DB.Exec(
+		"UPDATE ap_followers SET first_validation_failure_at = ? WHERE iri = ?",
+		eightDaysAgo, testIRI,
+	)
+	if err != nil {
+		t.Fatalf("Error setting first_validation_failure_at: %s", err)
+	}
+
+	// Fetch the follower
+	followers, err := repo.GetFollowersToValidate(1)
+	if err != nil {
+		t.Fatalf("Error getting followers: %s", err)
+	}
+
+	if len(followers) != 1 {
+		t.Fatalf("Expected 1 follower, got %d", len(followers))
+	}
+
+	// Verify the failure duration is calculated correctly
+	failureDuration := time.Since(followers[0].FirstValidationFailureAt.Time)
+	if failureDuration < FailureDurationThreshold {
+		t.Errorf("Expected failure duration (%v) to be >= threshold (%v)", failureDuration, FailureDurationThreshold)
+	}
+
+	// Test removal - this directly tests the removal logic without network calls
+	err = repo.RemoveByIRI(testIRI)
+	if err != nil {
+		t.Fatalf("Error removing follower: %s", err)
+	}
+
+	// Verify the follower was removed
+	count, err := repo.GetCount()
+	if err != nil {
+		t.Fatalf("Error getting count: %s", err)
+	}
+	if count != 0 {
+		t.Errorf("Expected follower to be removed, but count is %d", count)
+	}
+}
+
+func TestFailureNotRemovedBeforeThreshold(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create a test follower
+	testIRI := "https://fake.server/user/testnotremoved"
+	createTestFollower(repo, testIRI, "https://fake.server/user/testnotremoved/inbox", "Test User", "testnotremoved")
+
+	// Set first_validation_failure_at to 1 day ago (not past threshold)
+	oneDayAgo := time.Now().Add(-1 * 24 * time.Hour)
+	_, err := datastore.DB.Exec(
+		"UPDATE ap_followers SET first_validation_failure_at = ? WHERE iri = ?",
+		oneDayAgo, testIRI,
+	)
+	if err != nil {
+		t.Fatalf("Error setting first_validation_failure_at: %s", err)
+	}
+
+	// Fetch the follower
+	followers, err := repo.GetFollowersToValidate(1)
+	if err != nil {
+		t.Fatalf("Error getting followers: %s", err)
+	}
+
+	if len(followers) != 1 {
+		t.Fatalf("Expected 1 follower, got %d", len(followers))
+	}
+
+	// Verify the failure duration is below threshold
+	failureDuration := time.Since(followers[0].FirstValidationFailureAt.Time)
+	if failureDuration >= FailureDurationThreshold {
+		t.Errorf("Expected failure duration (%v) to be < threshold (%v)", failureDuration, FailureDurationThreshold)
+	}
+
+	// Follower should NOT be removed yet
+	count, err := repo.GetCount()
+	if err != nil {
+		t.Fatalf("Error getting count: %s", err)
+	}
+	if count != 1 {
+		t.Errorf("Expected follower to NOT be removed yet, but count is %d", count)
+	}
+}
+
+func TestFollowersOrderedByOldestValidatedFirst(t *testing.T) {
+	repo := setupTestWithRepo(t)
+
+	// Create followers
+	iri1 := "https://fake.server/user/first"
+	iri2 := "https://fake.server/user/second"
+	iri3 := "https://fake.server/user/third"
+
+	createTestFollower(repo, iri1, "https://fake.server/user/first/inbox", "First", "first")
+	createTestFollower(repo, iri2, "https://fake.server/user/second/inbox", "Second", "second")
+	createTestFollower(repo, iri3, "https://fake.server/user/third/inbox", "Third", "third")
+
+	// Set different last_validated_at times
+	now := time.Now()
+	_, err := datastore.DB.Exec("UPDATE ap_followers SET last_validated_at = ? WHERE iri = ?",
+		now.Add(-3*time.Hour), iri1)
+	if err != nil {
+		t.Fatalf("Error updating: %s", err)
+	}
+	_, err = datastore.DB.Exec("UPDATE ap_followers SET last_validated_at = ? WHERE iri = ?",
+		now.Add(-1*time.Hour), iri2)
+	if err != nil {
+		t.Fatalf("Error updating: %s", err)
+	}
+	// iri3 has NULL last_validated_at (never validated)
+
+	// Get followers - should return in order: iri3 (NULL), iri1 (oldest), iri2 (newest)
+	followers, err := repo.GetFollowersToValidate(3)
+	if err != nil {
+		t.Fatalf("Error getting followers: %s", err)
+	}
+
+	if len(followers) != 3 {
+		t.Fatalf("Expected 3 followers, got %d", len(followers))
+	}
+
+	// NULL values should come first (NULLS FIRST in query)
+	if followers[0].ActorIRI != iri3 {
+		t.Errorf("Expected first follower to be %s (never validated), got %s", iri3, followers[0].ActorIRI)
+	}
+
+	// Then oldest validated
+	if followers[1].ActorIRI != iri1 {
+		t.Errorf("Expected second follower to be %s (oldest validated), got %s", iri1, followers[1].ActorIRI)
+	}
+
+	// Then newest validated
+	if followers[2].ActorIRI != iri2 {
+		t.Errorf("Expected third follower to be %s (newest validated), got %s", iri2, followers[2].ActorIRI)
+	}
+}

activitypub/persistence/followers.go

@@ -19,6 +19,8 @@ func createFederationFollowersTable() {
 		"approved_at" TIMESTAMP,
     "disabled_at" TIMESTAMP,
     "request_object" BLOB,
+    "last_validated_at" TIMESTAMP,
+    "first_validation_failure_at" TIMESTAMP,
 		PRIMARY KEY (iri));`
 	_datastore.MustExec(createTableSQL)
 	_datastore.MustExec(`CREATE INDEX IF NOT EXISTS idx_iri ON ap_followers (iri);`)

activitypub/persistence/followersrepository/followersrepository.go

@@ -42,6 +42,14 @@ type FollowersRepository interface {
 	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
+	// GetFollowersToValidate returns followers needing validation, ordered by oldest validated first.
+	GetFollowersToValidate(limit int) ([]models.Follower, error)
+	// UpdateFollowerValidationSuccess marks a follower as successfully validated and clears failure timestamp.
+	UpdateFollowerValidationSuccess(iri string) error
+	// UpdateFollowerValidationFailure marks a validation failure, setting first failure time if not already set.
+	UpdateFollowerValidationFailure(iri string) error
+	// RemoveByIRI removes a follower directly by IRI string.
+	RemoveByIRI(iri string) error
 }
 
 // SqlFollowersRepository is the SQL-based implementation of FollowersRepository.
@@ -367,3 +375,98 @@ func (r *SqlFollowersRepository) removeFollow(actor *url.URL) error {
 
 	return tx.Commit()
 }
+
+// GetFollowersToValidate returns followers needing validation, ordered by oldest validated first.
+func (r *SqlFollowersRepository) GetFollowersToValidate(limit int) ([]models.Follower, error) {
+	ctx := context.Background()
+	followersResult, err := r.datastore.GetQueries().GetFollowersToValidate(ctx, utils.SafeIntToInt32(limit))
+	if err != nil {
+		return nil, 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,
+			FirstValidationFailureAt: utils.NullTime(row.FirstValidationFailureAt),
+		}
+
+		followers = append(followers, singleFollower)
+	}
+
+	return followers, nil
+}
+
+// UpdateFollowerValidationSuccess marks a follower as successfully validated and clears failure timestamp.
+func (r *SqlFollowersRepository) UpdateFollowerValidationSuccess(iri string) error {
+	r.datastore.DbLock.Lock()
+	defer r.datastore.DbLock.Unlock()
+
+	ctx := context.Background()
+	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).UpdateFollowerValidationSuccess(ctx, db.UpdateFollowerValidationSuccessParams{
+		LastValidatedAt: sql.NullTime{Time: time.Now(), Valid: true},
+		Iri:             iri,
+	}); err != nil {
+		return errors.Wrap(err, "error updating follower validation success")
+	}
+
+	return tx.Commit()
+}
+
+// UpdateFollowerValidationFailure marks a validation failure, setting first failure time if not already set.
+func (r *SqlFollowersRepository) UpdateFollowerValidationFailure(iri string) error {
+	r.datastore.DbLock.Lock()
+	defer r.datastore.DbLock.Unlock()
+
+	ctx := context.Background()
+	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).UpdateFollowerValidationFailure(ctx, db.UpdateFollowerValidationFailureParams{
+		LastValidatedAt: sql.NullTime{Time: time.Now(), Valid: true},
+		Iri:             iri,
+	}); err != nil {
+		return errors.Wrap(err, "error updating follower validation failure")
+	}
+
+	return tx.Commit()
+}
+
+// RemoveByIRI removes a follower directly by IRI string.
+func (r *SqlFollowersRepository) RemoveByIRI(iri string) 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(), iri); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}