From 6c13debd56d4119552113a81c716d480766f31f9 Mon Sep 17 00:00:00 2001
From: Daniel Supernault <danielsupernault@gmail.com>
Date: Mon, 10 Feb 2025 19:17:39 -0700
Subject: [PATCH] Update NewStatusPipeline, replaces #5706

---
 app/Jobs/StatusPipeline/NewStatusPipeline.php | 104 +++++++++++++++++-
 1 file changed, 99 insertions(+), 5 deletions(-)

diff --git a/app/Jobs/StatusPipeline/NewStatusPipeline.php b/app/Jobs/StatusPipeline/NewStatusPipeline.php
index 8ccb2926f..79a238836 100644
--- a/app/Jobs/StatusPipeline/NewStatusPipeline.php
+++ b/app/Jobs/StatusPipeline/NewStatusPipeline.php
@@ -2,6 +2,7 @@
 
 namespace App\Jobs\StatusPipeline;
 
+use App\Media;
 use App\Status;
 use Cache;
 use Illuminate\Bus\Queueable;
@@ -9,6 +10,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 Illuminate\Support\Facades\Redis;
 
 class NewStatusPipeline implements ShouldQueue
@@ -16,7 +18,7 @@ class NewStatusPipeline implements ShouldQueue
     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
 
     protected $status;
-    
+
     /**
      * Delete the job if its models no longer exist.
      *
@@ -24,9 +26,27 @@ class NewStatusPipeline implements ShouldQueue
      */
     public $deleteWhenMissingModels = true;
 
-    public $timeout = 5;
-    public $tries = 1;
-    
+    /**
+     * Increased timeout to handle cloud storage operations
+     *
+     * @var int
+     */
+    public $timeout = 30;
+
+    /**
+     * Number of times to attempt the job
+     *
+     * @var int
+     */
+    public $tries = 3;
+
+    /**
+     * Backoff periods between retries (in seconds)
+     *
+     * @var array
+     */
+    public $backoff = [30, 60, 120];
+
     /**
      * Create a new job instance.
      *
@@ -44,6 +64,80 @@ class NewStatusPipeline implements ShouldQueue
      */
     public function handle()
     {
-        StatusEntityLexer::dispatch($this->status);
+        // Check if status still exists
+        if (!Status::where('id', $this->status->id)->exists()) {
+            if(config('federation.activitypub.delivery.logger.enabled')) {
+                Log::info('Status ' . $this->status->id . ' was deleted before federation');
+            }
+            return;
+        }
+
+        // Skip media check if cloud storage isn't enabled or fast processing is on
+        if (!config_cache('pixelfed.cloud_storage') || config('pixelfed.media_fast_process')) {
+            $this->dispatchFederation();
+            return;
+        }
+
+        // Check for media still processing
+        $stillProcessing = Media::whereStatusId($this->status->id)
+            ->whereNull('cdn_url')
+            ->exists();
+
+        if ($stillProcessing) {
+            // Get the oldest processing media item
+            $oldestProcessingMedia = Media::whereStatusId($this->status->id)
+                ->whereNull('cdn_url')
+                ->oldest()
+                ->first();
+
+            // If media has been processing for more than 10 minutes, proceed anyway
+            if ($oldestProcessingMedia && now()->diffInMinutes($oldestProcessingMedia->created_at) > 10) {
+                if(config('federation.activitypub.delivery.logger.enabled')) {
+                    Log::warning('Media processing timeout for status ' . $this->status->id . '. Proceeding with federation.');
+                }
+                $this->dispatchFederation();
+                return;
+            }
+
+            // Release job back to queue with delay of 30 seconds
+            $this->release(30);
+            return;
+        }
+
+        // All media processed, proceed with federation
+        $this->dispatchFederation();
+    }
+
+    /**
+     * Dispatch the federation job
+     *
+     * @return void
+     */
+    protected function dispatchFederation()
+    {
+        try {
+            StatusEntityLexer::dispatch($this->status);
+        } catch (\Exception $e) {
+            if(config('federation.activitypub.delivery.logger.enabled')) {
+                Log::error('Federation dispatch failed for status ' . $this->status->id . ': ' . $e->getMessage());
+            }
+            throw $e;
+        }
+    }
+
+    /**
+     * Handle a job failure.
+     *
+     * @param  \Throwable  $exception
+     * @return void
+     */
+    public function failed(\Throwable $exception)
+    {
+        if(config('federation.activitypub.delivery.logger.enabled')) {
+            Log::error('NewStatusPipeline failed for status ' . $this->status->id, [
+                'exception' => $exception->getMessage(),
+                'trace' => $exception->getTraceAsString()
+            ]);
+        }
     }
 }