Merge pull request #6232 from shleeable/JobBatchv1-DeletePipeline

JobBatch v1- Delete pipelines defensive checks

app/Jobs/DeletePipeline/DeleteAccountPipeline.php

@@ -51,6 +51,7 @@ use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\Log;
 use Storage;
 
 class DeleteAccountPipeline implements ShouldQueue
@@ -75,6 +76,19 @@ class DeleteAccountPipeline implements ShouldQueue
     public function handle()
     {
         $user = $this->user;
+
+        // Verify user exists
+        if (!$user) {
+            Log::info("DeleteAccountPipeline: User no longer exists, skipping job");
+            return;
+        }
+
+        // Verify user has a profile
+        if (!$user->profile_id) {
+            Log::info("DeleteAccountPipeline: User {$user->id} has no profile_id, skipping job");
+            return;
+        }
+
         $profile = $user->profile;
         $id = $user->profile_id;
         $cloudStorageEnabled = (bool) config_cache('pixelfed.cloud_storage');

app/Jobs/DeletePipeline/DeleteRemoteProfilePipeline.php

@@ -7,6 +7,7 @@ use Illuminate\Queue\SerializesModels;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Support\Facades\Log;
 use DB;
 use Illuminate\Support\Str;
 use App\{
@@ -72,6 +73,13 @@ class DeleteRemoteProfilePipeline implements ShouldQueue
 	public function handle()
 	{
 		$profile = $this->profile;
+
+		// Verify profile exists
+		if (!$profile) {
+			Log::info("DeleteRemoteProfilePipeline: Profile no longer exists, skipping job");
+			return;
+		}
+
 		$pid = $profile->id;
 
 		if($profile->domain == null || $profile->private_key) {

app/Jobs/DeletePipeline/DeleteRemoteStatusPipeline.php

@@ -25,6 +25,7 @@ use App\Services\StatusService;
 use App\Jobs\MediaPipeline\MediaDeletePipeline;
 use Cache;
 use App\Services\Account\AccountStatService;
+use Illuminate\Support\Facades\Log;
 
 class DeleteRemoteStatusPipeline implements ShouldQueue
 {
@@ -56,9 +57,22 @@ class DeleteRemoteStatusPipeline implements ShouldQueue
     {
         $status = $this->status;
 
-        AccountStatService::decrementPostCount($status->profile_id);
-        NetworkTimelineService::del($status->id);
-        StatusService::del($status->id, true);
+        // Verify status exists
+        if (!$status) {
+            Log::info("DeleteRemoteStatusPipeline: Status no longer exists, skipping job");
+            return;
+        }
+
+        // Verify status has a profile
+        if (!$status->profile_id) {
+            Log::info("DeleteRemoteStatusPipeline: Status {$status->id} has no profile_id, skipping job");
+            return;
+        }
+
+        try {
+            AccountStatService::decrementPostCount($status->profile_id);
+            NetworkTimelineService::del($status->id);
+            StatusService::del($status->id, true);
         Bookmark::whereStatusId($status->id)->delete();
         Notification::whereItemType('App\Status')
             ->whereItemId($status->id)
@@ -77,6 +91,11 @@ class DeleteRemoteStatusPipeline implements ShouldQueue
         StatusView::whereStatusId($status->id)->delete();
         Status::whereReblogOfId($status->id)->forceDelete();
         $status->forceDelete();
+        } catch (\Exception $e) {
+            Log::warning("DeleteRemoteStatusPipeline: Failed to delete status {$status->id}: " . $e->getMessage());
+            throw $e;
+        }
+        
         return 1;
     }
 }

app/Jobs/DeletePipeline/FanoutDeletePipeline.php

@@ -16,6 +16,7 @@ use GuzzleHttp\Pool;
 use GuzzleHttp\Client;
 use GuzzleHttp\Promise;
 use App\Util\ActivityPub\HttpSignature;
+use Illuminate\Support\Facades\Log;
 
 class FanoutDeletePipeline implements ShouldQueue
 {
@@ -40,9 +41,22 @@ class FanoutDeletePipeline implements ShouldQueue
 	{
 		$profile = $this->profile;
 
-		$client = new Client([
-            'timeout'  => config('federation.activitypub.delivery.timeout')
-        ]);
+		// Verify profile exists
+		if (!$profile) {
+			Log::info("FanoutDeletePipeline: Profile no longer exists, skipping job");
+			return;
+		}
+
+		// Verify profile has required fields for ActivityPub
+		if (!$profile->permalink() || !$profile->private_key) {
+			Log::info("FanoutDeletePipeline: Profile {$profile->id} missing required fields for ActivityPub, skipping job");
+			return;
+		}
+
+		try {
+			$client = new Client([
+				'timeout'  => config('federation.activitypub.delivery.timeout')
+			]);
 
         $audience = Cache::remember('pf:ap:known_instances', now()->addHours(6), function() {
         	return Profile::whereNotNull('sharedInbox')->groupBy('sharedInbox')->pluck('sharedInbox')->toArray();
@@ -92,6 +106,10 @@ class FanoutDeletePipeline implements ShouldQueue
         $promise = $pool->promise();
 
         $promise->wait();
+		} catch (\Exception $e) {
+			Log::warning("FanoutDeletePipeline: Failed to fanout delete for profile {$profile->id}: " . $e->getMessage());
+			throw $e;
+		}
 
         return 1;
 	}