Add Profile Migration federation

app/Http/Controllers/ProfileMigrationController.php

@@ -3,6 +3,7 @@
 namespace App\Http\Controllers;
 
 use App\Http\Requests\ProfileMigrationStoreRequest;
+use App\Jobs\ProfilePipeline\ProfileMigrationDeliverMoveActivityPipeline;
 use App\Jobs\ProfilePipeline\ProfileMigrationMoveFollowersPipeline;
 use App\Models\ProfileAlias;
 use App\Models\ProfileMigration;
@@ -10,6 +11,7 @@ use App\Services\AccountService;
 use App\Services\WebfingerService;
 use App\Util\ActivityPub\Helpers;
 use Illuminate\Http\Request;
+use Illuminate\Support\Facades\Bus;
 
 class ProfileMigrationController extends Controller
 {
@@ -20,6 +22,7 @@ class ProfileMigrationController extends Controller
 
     public function index(Request $request)
     {
+        abort_if((bool) config_cache('federation.activitypub.enabled') === false, 404);
         $hasExistingMigration = ProfileMigration::whereProfileId($request->user()->profile_id)
             ->where('created_at', '>', now()->subDays(30))
             ->exists();
@@ -29,6 +32,7 @@ class ProfileMigrationController extends Controller
 
     public function store(ProfileMigrationStoreRequest $request)
     {
+        abort_if((bool) config_cache('federation.activitypub.enabled') === false, 404);
         $acct = WebfingerService::rawGet($request->safe()->acct);
         if (! $acct) {
             return redirect()->back()->withErrors(['acct' => 'The new account you provided is not responding to our requests.']);
@@ -43,7 +47,7 @@ class ProfileMigrationController extends Controller
             'acct' => $request->safe()->acct,
             'uri' => $acct,
         ]);
-        ProfileMigration::create([
+        $migration = ProfileMigration::create([
             'profile_id' => $request->user()->profile_id,
             'acct' => $request->safe()->acct,
             'followers_count' => $request->user()->profile->followers_count,
@@ -55,7 +59,10 @@ class ProfileMigrationController extends Controller
         ]);
         AccountService::del($user->profile_id);
 
-        ProfileMigrationMoveFollowersPipeline::dispatch($user->profile_id, $newAccount->id);
+        Bus::batch([
+            new ProfileMigrationDeliverMoveActivityPipeline($migration, $user->profile, $newAccount),
+            new ProfileMigrationMoveFollowersPipeline($user->profile_id, $newAccount->id),
+        ])->onQueue('follow')->dispatch();
 
         return redirect()->back()->with(['status' => 'Succesfully migrated account!']);
     }

app/Jobs/ProfilePipeline/ProfileMigrationDeliverMoveActivityPipeline.php

@@ -0,0 +1,140 @@
+<?php
+
+namespace App\Jobs\ProfilePipeline;
+
+use App\Transformer\ActivityPub\Verb\Move;
+use App\Util\ActivityPub\HttpSignature;
+use GuzzleHttp\Client;
+use GuzzleHttp\Pool;
+use Illuminate\Bus\Batchable;
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\Middleware\WithoutOverlapping;
+use Illuminate\Queue\SerializesModels;
+use League\Fractal;
+use League\Fractal\Serializer\ArraySerializer;
+
+class ProfileMigrationDeliverMoveActivityPipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue
+{
+    use Batchable, Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+
+    public $migration;
+
+    public $oldAccount;
+
+    public $newAccount;
+
+    public $timeout = 1400;
+
+    public $tries = 3;
+
+    public $maxExceptions = 1;
+
+    public $failOnTimeout = true;
+
+    /**
+     * The number of seconds after which the job's unique lock will be released.
+     *
+     * @var int
+     */
+    public $uniqueFor = 3600;
+
+    /**
+     * Get the unique ID for the job.
+     */
+    public function uniqueId(): string
+    {
+        return 'profile:migration:deliver-move-followers:id:'.$this->migration->id;
+    }
+
+    /**
+     * Get the middleware the job should pass through.
+     *
+     * @return array<int, object>
+     */
+    public function middleware(): array
+    {
+        return [(new WithoutOverlapping('profile:migration:deliver-move-followers:id:'.$this->migration->id))->shared()->dontRelease()];
+    }
+
+    /**
+     * Create a new job instance.
+     */
+    public function __construct($migration, $oldAccount, $newAccount)
+    {
+        $this->migration = $migration;
+        $this->oldAccount = $oldAccount;
+        $this->newAccount = $newAccount;
+    }
+
+    /**
+     * Execute the job.
+     */
+    public function handle(): void
+    {
+        if ($this->batch()->cancelled()) {
+            return;
+        }
+
+        $migration = $this->migration;
+        $profile = $this->oldAccount;
+        $newAccount = $this->newAccount;
+
+        if ($profile->domain || ! $profile->private_key) {
+            return;
+        }
+
+        $audience = $profile->getAudienceInbox();
+        $activitypubObject = new Move();
+
+        $fractal = new Fractal\Manager();
+        $fractal->setSerializer(new ArraySerializer());
+        $resource = new Fractal\Resource\Item($migration, $activitypubObject);
+        $activity = $fractal->createData($resource)->toArray();
+
+        $payload = json_encode($activity);
+
+        $client = new Client([
+            'timeout' => config('federation.activitypub.delivery.timeout'),
+        ]);
+
+        $version = config('pixelfed.version');
+        $appUrl = config('app.url');
+        $userAgent = "(Pixelfed/{$version}; +{$appUrl})";
+
+        $requests = function ($audience) use ($client, $activity, $profile, $payload, $userAgent) {
+            foreach ($audience as $url) {
+                $headers = HttpSignature::sign($profile, $url, $activity, [
+                    'Content-Type' => 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"',
+                    'User-Agent' => $userAgent,
+                ]);
+                yield function () use ($client, $url, $headers, $payload) {
+                    return $client->postAsync($url, [
+                        'curl' => [
+                            CURLOPT_HTTPHEADER => $headers,
+                            CURLOPT_POSTFIELDS => $payload,
+                            CURLOPT_HEADER => true,
+                            CURLOPT_SSL_VERIFYPEER => false,
+                            CURLOPT_SSL_VERIFYHOST => false,
+                        ],
+                    ]);
+                };
+            }
+        };
+
+        $pool = new Pool($client, $requests($audience), [
+            'concurrency' => config('federation.activitypub.delivery.concurrency'),
+            'fulfilled' => function ($response, $index) {
+            },
+            'rejected' => function ($reason, $index) {
+            },
+        ]);
+
+        $promise = $pool->promise();
+
+        $promise->wait();
+    }
+}

app/Jobs/ProfilePipeline/ProfileMigrationMoveFollowersPipeline.php

@@ -2,22 +2,59 @@
 
 namespace App\Jobs\ProfilePipeline;
 
+use App\Follower;
+use App\Profile;
+use App\Services\AccountService;
+use Illuminate\Bus\Batchable;
 use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\Middleware\WithoutOverlapping;
 use Illuminate\Queue\SerializesModels;
-use App\Follower;
-use App\Profile;
-use App\Services\AccountService;
 
-class ProfileMigrationMoveFollowersPipeline implements ShouldQueue
+class ProfileMigrationMoveFollowersPipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue
 {
-    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+    use Batchable, Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
 
     public $oldPid;
+
     public $newPid;
 
+    public $timeout = 1400;
+
+    public $tries = 3;
+
+    public $maxExceptions = 1;
+
+    public $failOnTimeout = true;
+
+    /**
+     * The number of seconds after which the job's unique lock will be released.
+     *
+     * @var int
+     */
+    public $uniqueFor = 3600;
+
+    /**
+     * Get the unique ID for the job.
+     */
+    public function uniqueId(): string
+    {
+        return 'profile:migration:move-followers:oldpid-'.$this->oldPid.':newpid-'.$this->newPid;
+    }
+
+    /**
+     * Get the middleware the job should pass through.
+     *
+     * @return array<int, object>
+     */
+    public function middleware(): array
+    {
+        return [(new WithoutOverlapping('profile:migration:move-followers:oldpid-'.$this->oldPid.':newpid-'.$this->newPid))->shared()->dontRelease()];
+    }
+
     /**
      * Create a new job instance.
      */
@@ -32,9 +69,12 @@ class ProfileMigrationMoveFollowersPipeline implements ShouldQueue
      */
     public function handle(): void
     {
+        if ($this->batch()->cancelled()) {
+            return;
+        }
         $og = Profile::find($this->oldPid);
         $ne = Profile::find($this->newPid);
-        if(!$og || !$ne || $og == $ne) {
+        if (! $og || ! $ne || $og == $ne) {
             return;
         }
         $ne->followers_count = $og->followers_count;

app/Transformer/ActivityPub/Verb/Move.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace App\Transformer\ActivityPub\Verb;
+
+use App\Models\ProfileMigration;
+use League\Fractal;
+
+class Move extends Fractal\TransformerAbstract
+{
+    public function transform(ProfileMigration $migration)
+    {
+        $objUrl = $migration->target->permalink();
+        $id = $migration->target->permalink('#moves/'.$migration->id);
+        $to = $migration->target->permalink('/followers');
+
+        return [
+            '@context' => 'https://www.w3.org/ns/activitystreams',
+            'id' => $id,
+            'actor' => $objUrl,
+            'type' => 'Move',
+            'object' => $objUrl,
+            'target' => $migration->profile->permalink(),
+            'to' => $to,
+        ];
+    }
+}