diff --git a/app/Jobs/MovePipeline/MoveMigrateFollowersPipeline.php b/app/Jobs/MovePipeline/MoveMigrateFollowersPipeline.php index ba34b3fee..99e03989a 100644 --- a/app/Jobs/MovePipeline/MoveMigrateFollowersPipeline.php +++ b/app/Jobs/MovePipeline/MoveMigrateFollowersPipeline.php @@ -14,50 +14,26 @@ use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Foundation\Queue\Queueable; use Illuminate\Queue\Middleware\ThrottlesExceptionsWithRedis; use Illuminate\Queue\Middleware\WithoutOverlapping; +use Illuminate\Support\Facades\Log; +use GuzzleHttp\Psr7\Request; class MoveMigrateFollowersPipeline implements ShouldQueue { use Queueable; - public $target; - - public $activity; - - /** - * The number of times the job may be attempted. - * - * @var int - */ - public $tries = 15; - - /** - * The maximum number of unhandled exceptions to allow before failing. - * - * @var int - */ - public $maxExceptions = 5; - - /** - * The number of seconds the job can run before timing out. - * - * @var int - */ - public $timeout = 900; - - /** - * Create a new job instance. - */ - public function __construct($target, $activity) + public string $target; + public string $activity; + + public int $tries = 15; + public int $maxExceptions = 5; + public int $timeout = 900; + + public function __construct(string $target, string $activity) { $this->target = $target; $this->activity = $activity; } - /** - * Get the middleware the job should pass through. - * - * @return array - */ public function middleware(): array { return [ @@ -66,40 +42,111 @@ class MoveMigrateFollowersPipeline implements ShouldQueue ]; } - /** - * Determine the time at which the job should timeout. - */ public function retryUntil(): DateTime { return now()->addMinutes(15); } - /** - * Execute the job. - */ public function handle(): void { - if (config('app.env') !== 'production' || (bool) config_cache('federation.activitypub.enabled') == false) { - throw new Exception('Activitypub not enabled'); + try { + $this->validateEnvironment(); + + $targetAccount = $this->fetchProfile($this->target); + $actorAccount = $this->fetchProfile($this->activity); + + if (!$targetAccount || !$actorAccount) { + throw new Exception('Invalid move accounts'); + } + + $client = $this->createHttpClient(); + $targetInbox = $targetAccount['sharedInbox'] ?? $targetAccount['inbox_url']; + $targetPid = $targetAccount['id']; + + DB::table('followers') + ->join('profiles', 'followers.profile_id', '=', 'profiles.id') + ->where('followers.following_id', $actorAccount['id']) + ->whereNotNull('profiles.user_id') + ->whereNull('profiles.deleted_at') + ->select('profiles.id', 'profiles.user_id', 'profiles.username', 'profiles.private_key', 'profiles.status') + ->chunkById(100, function ($followers) use ($client, $targetInbox, $targetPid) { + $this->processFollowerChunk($followers, $client, $targetInbox, $targetPid); + }, 'id'); + } catch (Exception $e) { + Log::error('MoveMigrateFollowersPipeline failed', [ + 'target' => $this->target, + 'activity' => $this->activity, + 'error' => $e->getMessage() + ]); + throw $e; } + } - $target = $this->target; - $actor = $this->activity; + private function validateEnvironment(): void + { + if (config('app.env') !== 'production' || !(bool)config('federation.activitypub.enabled')) { + throw new Exception('ActivityPub not enabled'); + } + } - $targetAccount = Helpers::profileFetch($target); - $actorAccount = Helpers::profileFetch($actor); + private function fetchProfile(string $url): ?array + { + return Helpers::profileFetch($url); + } + + private function createHttpClient(): Client + { + return new Client([ + 'timeout' => config('federation.activitypub.delivery.timeout'), + ]); + } - if (! $targetAccount || ! $actorAccount) { - throw new Exception('Invalid move accounts'); + private function processFollowerChunk($followers, Client $client, string $targetInbox, int $targetPid): void + { + $requests = $this->generateRequests($followers, $targetInbox, $targetPid); + + $pool = new Pool($client, $requests, [ + 'concurrency' => config('federation.activitypub.delivery.concurrency'), + 'fulfilled' => function ($response, $index) { + // Log success if needed + }, + 'rejected' => function ($reason, $index) { + Log::error('Failed to process follower', ['reason' => $reason, 'index' => $index]); + }, + ]); + + $pool->promise()->wait(); + } + + private function generateRequests($followers, string $targetInbox, int $targetPid): \Generator + { + foreach ($followers as $follower) { + if (!$this->isValidFollower($follower)) { + continue; + } + + yield $this->createFollowRequest($follower, $targetInbox, $targetPid); } + } + + private function isValidFollower($follower): bool + { + return $follower->private_key && $follower->username && $follower->user_id && $follower->status !== 'delete'; + } + private function createFollowRequest($follower, string $targetInbox, int $targetPid): \GuzzleHttp\Psr7\Request + { + $permalink = 'https://'.config('pixelfed.domain.app').'/users/'.$follower->username; $activity = [ '@context' => 'https://www.w3.org/ns/activitystreams', 'type' => 'Follow', - 'actor' => null, - 'object' => $target, + 'actor' => $permalink, + 'object' => $this->target, ]; + $keyId = $permalink.'#main-key'; + $payload = json_encode($activity); + $version = config('pixelfed.version'); $appUrl = config('app.url'); $userAgent = "(Pixelfed/{$version}; +{$appUrl})"; @@ -107,59 +154,14 @@ class MoveMigrateFollowersPipeline implements ShouldQueue 'Content-Type' => 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"', 'User-Agent' => $userAgent, ]; - $targetInbox = $targetAccount['sharedInbox'] ?? $targetAccount['inbox_url']; - $targetPid = $targetAccount['id']; - - DB::table('followers') - ->join('profiles', 'followers.profile_id', '=', 'profiles.id') - ->where('followers.following_id', $actorAccount['id']) - ->whereNotNull('profiles.user_id') - ->whereNull('profiles.deleted_at') - ->select('profiles.id', 'profiles.user_id', 'profiles.username', 'profiles.private_key', 'profiles.status') - ->chunkById(100, function ($followers) use ($addlHeaders, $targetInbox, $targetPid, $target) { - $client = new Client([ - 'timeout' => config('federation.activitypub.delivery.timeout'), - ]); - $requests = function ($followers) use ($client, $target, $addlHeaders, $targetInbox, $targetPid) { - $activity = [ - '@context' => 'https://www.w3.org/ns/activitystreams', - 'type' => 'Follow', - 'actor' => null, - 'object' => $target, - ]; - foreach ($followers as $follower) { - if (! $follower->private_key || ! $follower->username || ! $follower->user_id || $follower->status === 'delete') { - continue; - } - $permalink = 'https://'.config('pixelfed.domain.app').'/users/'.$follower->username; - $activity['actor'] = $permalink; - $keyId = $permalink.'#main-key'; - $payload = json_encode($activity); - $url = $targetInbox; - $headers = HttpSignature::signRaw($follower->private_key, $keyId, $targetInbox, $activity, $addlHeaders); - Follower::updateOrCreate([ - 'profile_id' => $follower->id, - 'following_id' => $targetPid, - ]); - yield new $client->postAsync($url, [ - 'curl' => [ - CURLOPT_HTTPHEADER => $headers, - CURLOPT_POSTFIELDS => $payload, - CURLOPT_HEADER => true, - ], - ]); - } - }; - - $pool = new Pool($client, $requests($followers), [ - 'concurrency' => config('federation.activitypub.delivery.concurrency'), - 'fulfilled' => function ($response, $index) {}, - 'rejected' => function ($reason, $index) {}, - ]); - - $promise = $pool->promise(); - - $promise->wait(); - }, 'id'); + + $headers = HttpSignature::signRaw($follower->private_key, $keyId, $targetInbox, $activity, $addlHeaders); + + Follower::updateOrCreate([ + 'profile_id' => $follower->id, + 'following_id' => $targetPid, + ]); + + return new Request('POST', $targetInbox, $headers, $payload); } } diff --git a/app/Jobs/MovePipeline/UnfollowLegacyAccountMovePipeline.php b/app/Jobs/MovePipeline/UnfollowLegacyAccountMovePipeline.php index 385c1ea7e..e7b303763 100644 --- a/app/Jobs/MovePipeline/UnfollowLegacyAccountMovePipeline.php +++ b/app/Jobs/MovePipeline/UnfollowLegacyAccountMovePipeline.php @@ -9,47 +9,29 @@ use DB; use Exception; use GuzzleHttp\Client; use GuzzleHttp\Pool; +use GuzzleHttp\Psr7\Request; use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Foundation\Queue\Queueable; use Illuminate\Queue\Middleware\ThrottlesExceptions; use Illuminate\Queue\Middleware\WithoutOverlapping; +use Illuminate\Support\Facades\Log; class UnfollowLegacyAccountMovePipeline implements ShouldQueue { use Queueable; - public $target; + public string $target; + public string $activity; - public $activity; + public int $tries = 6; + public int $maxExceptions = 3; - /** - * The number of times the job may be attempted. - * - * @var int - */ - public $tries = 6; - - /** - * The maximum number of unhandled exceptions to allow before failing. - * - * @var int - */ - public $maxExceptions = 3; - - /** - * Create a new job instance. - */ - public function __construct($target, $activity) + public function __construct(string $target, string $activity) { $this->target = $target; $this->activity = $activity; } - /** - * Get the middleware the job should pass through. - * - * @return array - */ public function middleware(): array { return [ @@ -58,32 +40,121 @@ class UnfollowLegacyAccountMovePipeline implements ShouldQueue ]; } - /** - * Determine the time at which the job should timeout. - */ public function retryUntil(): DateTime { return now()->addMinutes(5); } - /** - * Execute the job. - */ public function handle(): void { - if (config('app.env') !== 'production' || (bool) config_cache('federation.activitypub.enabled') == false) { - throw new Exception('Activitypub not enabled'); + try { + $this->validateEnvironment(); + + $targetAccount = $this->fetchProfile($this->target); + $actorAccount = $this->fetchProfile($this->activity); + + if (!$targetAccount || !$actorAccount) { + throw new Exception('Invalid move accounts'); + } + + $client = $this->createHttpClient(); + $targetInbox = $actorAccount['sharedInbox'] ?? $actorAccount['inbox_url']; + $targetPid = $actorAccount['id']; + + $this->processFollowers($client, $targetInbox, $targetPid); + } catch (Exception $e) { + Log::error('UnfollowLegacyAccountMovePipeline failed', [ + 'target' => $this->target, + 'activity' => $this->activity, + 'error' => $e->getMessage() + ]); + throw $e; + } + } + + private function validateEnvironment(): void + { + if (config('app.env') !== 'production' || !(bool)config('federation.activitypub.enabled')) { + throw new Exception('ActivityPub not enabled'); } + } + + private function fetchProfile(string $url): ?array + { + return Helpers::profileFetch($url); + } + + private function createHttpClient(): Client + { + return new Client([ + 'timeout' => config('federation.activitypub.delivery.timeout'), + ]); + } + + private function processFollowers(Client $client, string $targetInbox, int $targetPid): void + { + DB::table('followers') + ->join('profiles', 'followers.profile_id', '=', 'profiles.id') + ->where('followers.following_id', $targetPid) + ->whereNotNull('profiles.user_id') + ->whereNull('profiles.deleted_at') + ->select('profiles.id', 'profiles.user_id', 'profiles.username', 'profiles.private_key', 'profiles.status') + ->chunkById(100, function ($followers) use ($client, $targetInbox, $targetPid) { + $this->processFollowerChunk($followers, $client, $targetInbox, $targetPid); + }, 'id'); + } - $target = $this->target; - $actor = $this->activity; + private function processFollowerChunk($followers, Client $client, string $targetInbox, int $targetPid): void + { + $requests = $this->generateRequests($followers, $targetInbox, $targetPid); + + $pool = new Pool($client, $requests, [ + 'concurrency' => config('federation.activitypub.delivery.concurrency'), + 'fulfilled' => function ($response, $index) { + // Log success if needed + }, + 'rejected' => function ($reason, $index) { + Log::error('Failed to process unfollow', ['reason' => $reason, 'index' => $index]); + }, + ]); + + $pool->promise()->wait(); + } - $targetAccount = Helpers::profileFetch($target); - $actorAccount = Helpers::profileFetch($actor); + private function generateRequests($followers, string $targetInbox, int $targetPid): \Generator + { + foreach ($followers as $follower) { + if (!$this->isValidFollower($follower)) { + continue; + } - if (! $targetAccount || ! $actorAccount) { - throw new Exception('Invalid move accounts'); + yield $this->createUnfollowRequest($follower, $targetInbox, $targetPid); } + } + + private function isValidFollower($follower): bool + { + return $follower->private_key && $follower->username && $follower->user_id && $follower->status !== 'delete'; + } + + private function createUnfollowRequest($follower, string $targetInbox, int $targetPid): Request + { + $permalink = 'https://'.config('pixelfed.domain.app').'/users/'.$follower->username; + $activity = [ + '@context' => 'https://www.w3.org/ns/activitystreams', + 'type' => 'Undo', + 'id' => $permalink.'#follow/'.$targetPid.'/undo', + 'actor' => $permalink, + 'object' => [ + 'type' => 'Follow', + 'id' => $permalink.'#follows/'.$targetPid, + 'object' => $this->activity, + 'actor' => $permalink, + ], + ]; + + $keyId = $permalink.'#main-key'; + $payload = json_encode($activity); $version = config('pixelfed.version'); $appUrl = config('app.url'); @@ -92,64 +163,9 @@ class UnfollowLegacyAccountMovePipeline implements ShouldQueue 'Content-Type' => 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"', 'User-Agent' => $userAgent, ]; - $targetInbox = $actorAccount['sharedInbox'] ?? $actorAccount['inbox_url']; - $targetPid = $actorAccount['id']; - DB::table('followers') - ->join('profiles', 'followers.profile_id', '=', 'profiles.id') - ->where('followers.following_id', $actorAccount['id']) - ->whereNotNull('profiles.user_id') - ->whereNull('profiles.deleted_at') - ->select('profiles.id', 'profiles.user_id', 'profiles.username', 'profiles.private_key', 'profiles.status') - ->chunkById(100, function ($followers) use ($actor, $addlHeaders, $targetInbox, $targetPid) { - $client = new Client([ - 'timeout' => config('federation.activitypub.delivery.timeout'), - ]); - $requests = function ($followers) use ($client, $actor, $addlHeaders, $targetInbox, $targetPid) { - $activity = [ - '@context' => 'https://www.w3.org/ns/activitystreams', - 'type' => 'Undo', - 'id' => null, - 'actor' => null, - 'object' => [ - 'type' => 'Follow', - 'id' => null, - 'object' => $actor, - 'actor' => null, - ], - ]; - foreach ($followers as $follower) { - if (! $follower->private_key || ! $follower->username || ! $follower->user_id || $follower->status === 'delete') { - continue; - } - $permalink = 'https://'.config('pixelfed.domain.app').'/users/'.$follower->username; - $activity['id'] = $permalink.'#follow/'.$targetPid.'/undo'; - $activity['actor'] = $permalink; - $activity['object']['id'] = $permalink.'#follows/'.$targetPid; - $activity['object']['actor'] = $permalink; - $keyId = $permalink.'#main-key'; - $payload = json_encode($activity); - $url = $targetInbox; - $headers = HttpSignature::signRaw($follower->private_key, $keyId, $targetInbox, $activity, $addlHeaders); - yield new $client->postAsync($url, [ - 'curl' => [ - CURLOPT_HTTPHEADER => $headers, - CURLOPT_POSTFIELDS => $payload, - CURLOPT_HEADER => true, - ], - ]); - } - }; - - $pool = new Pool($client, $requests($followers), [ - 'concurrency' => config('federation.activitypub.delivery.concurrency'), - 'fulfilled' => function ($response, $index) {}, - 'rejected' => function ($reason, $index) {}, - ]); - - $promise = $pool->promise(); - - $promise->wait(); - }, 'id'); + $headers = HttpSignature::signRaw($follower->private_key, $keyId, $targetInbox, $activity, $addlHeaders); + + return new Request('POST', $targetInbox, $headers, $payload); } }