diff --git a/app/Jobs/ProcessIncomingEmail.php b/app/Jobs/ProcessIncomingEmail.php
new file mode 100644
index 0000000..b834a45
--- /dev/null
+++ b/app/Jobs/ProcessIncomingEmail.php
@@ -0,0 +1,150 @@
+<?php
+
+namespace App\Jobs;
+
+use App\Events\NewEmailReceived;
+use App\Models\Email;
+use App\Models\EmailBody;
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\Middleware\WithoutOverlapping;
+use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\Cache;
+use Illuminate\Support\Facades\Log;
+use MongoDB\Laravel\Collection;
+
+class ProcessIncomingEmail implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+
+    /**
+     * The number of times the job may be attempted.
+     */
+    public int $tries = 3;
+
+    /**
+     * The backoff strategy (seconds) between retries.
+     *
+     * @var array<int, int>
+     */
+    public array $backoff = [5, 15, 30];
+
+    /**
+     * Create a new job instance.
+     *
+     * @param  array<string, mixed>  $payload  The validated webhook payload.
+     */
+    public function __construct(
+        public array $payload,
+    ) {
+        $this->onQueue('emails');
+        $this->onConnection('redis');
+    }
+
+    /**
+     * Get the middleware the job should pass through.
+     *
+     * @return array<int, object>
+     */
+    public function middleware(): array
+    {
+        return [
+            (new WithoutOverlapping($this->payload['hash']))->dontRelease(),
+        ];
+    }
+
+    /**
+     * Execute the job.
+     */
+    public function handle(): void
+    {
+        $metadata = $this->payload['metadata'];
+        $bodyText = $this->payload['bodyText'] ?? null;
+        $bodyHtml = $this->payload['bodyHtml'] ?? null;
+
+        $preview = $this->generatePreview($bodyText, $bodyHtml);
+
+        $email = Email::updateOrCreate(
+            ['unique_id_hash' => $this->payload['hash']],
+            [
+                'recipient_email' => $metadata['recipientEmail'],
+                'recipient_name' => $metadata['recipientName'] ?? '',
+                'sender_email' => $metadata['senderEmail'],
+                'sender_name' => $metadata['senderName'] ?? '',
+                'domain' => $metadata['domain'],
+                'subject' => $metadata['subject'] ?? '',
+                'preview' => $preview,
+                'attachments_json' => $metadata['attachments'] ?? [],
+                'attachment_size' => $metadata['attachmentSize'] ?? 0,
+                'received_at' => $metadata['received_at'],
+            ]
+        );
+
+        EmailBody::updateOrCreate(
+            ['unique_id_hash' => $this->payload['hash']],
+            [
+                'body_text' => $bodyText,
+                'body_html' => $bodyHtml,
+            ]
+        );
+
+        $this->ensureTtlIndex();
+
+        NewEmailReceived::dispatch($email);
+    }
+
+    /**
+     * Generate an excerpt from the email body for the preview column.
+     *
+     * Prefers body_text. Falls back to body_html with tags stripped.
+     */
+    private function generatePreview(?string $bodyText, ?string $bodyHtml): string
+    {
+        if (! empty($bodyText)) {
+            return mb_substr(trim($bodyText), 0, 500);
+        }
+
+        if (! empty($bodyHtml)) {
+            $stripped = strip_tags($bodyHtml);
+            $stripped = preg_replace('/\s+/', ' ', $stripped);
+
+            return mb_substr(trim($stripped), 0, 500);
+        }
+
+        return '';
+    }
+
+    /**
+     * Ensure the MongoDB TTL index exists on the `recent_email_bodies` collection.
+     *
+     * Uses a cache flag to avoid checking on every job execution.
+     */
+    private function ensureTtlIndex(): void
+    {
+        Cache::rememberForever('mongodb_ttl_index_ensured', function () {
+            $ttlSeconds = config('services.mailops.email_body_ttl_seconds', 259200);
+
+            /** @var Collection $collection */
+            $collection = (new EmailBody)->getCollection();
+            $collection->createIndex(
+                ['created_at' => 1],
+                ['expireAfterSeconds' => $ttlSeconds, 'name' => 'ttl_created_at']
+            );
+
+            return true;
+        });
+    }
+
+    /**
+     * Handle a job failure.
+     */
+    public function failed(?\Throwable $exception): void
+    {
+        Log::error('ProcessIncomingEmail failed', [
+            'hash' => $this->payload['hash'] ?? 'unknown',
+            'error' => $exception?->getMessage(),
+        ]);
+    }
+}