dpl/app/Services/MailFlow/DeliveryEvents.php

<?php

namespace App\Services\MailFlow;

use App\Models\MailMessage;
use Carbon\Carbon;
use Illuminate\Support\Facades\DB;

class DeliveryEvents
{
    private const STATUSES = [
        'email.sent' => 'submitted', 'email.delivered' => 'delivered',
        'email.delivery_delayed' => 'delayed', 'email.bounced' => 'bounced',
        'email.failed' => 'failed', 'email.complained' => 'complained', 'email.suppressed' => 'suppressed',
    ];

    public function __construct(private QuotaService $quota, private ResendClient $client) {}

    public function refreshStatus(MailMessage $message): void
    {
        if ($message->channel !== 'resend' || !$message->provider_id
            || !in_array($message->status, ['submitted', 'delayed'], true)) {
            return;
        }
        $email = $this->client->retrieve($message->provider_id);
        if (($email['id'] ?? null) !== $message->provider_id) {
            throw new \RuntimeException('resend_status_identity_mismatch');
        }
        $status = match ($email['last_event'] ?? '') {
            'sent' => 'submitted', 'delivery_delayed' => 'delayed',
            'delivered', 'opened', 'clicked' => 'delivered',
            'bounced' => 'bounced', 'failed' => 'failed',
            'complained' => 'complained', 'suppressed' => 'suppressed',
            default => null,
        };
        if (!$status) { return; }
        DB::transaction(function () use ($message, $status) {
            $current = MailMessage::whereKey($message->id)->lockForUpdate()->firstOrFail();
            if ($current->provider_id !== $message->provider_id) { return; }
            if ($this->rank($status) >= $this->rank($current->status)) {
                $current->status = $status;
                $current->error_code = in_array($status, ['bounced', 'failed', 'suppressed', 'complained'], true)
                    ? 'resend_' . $status : null;
            }
            // GET gives the latest status, not its event time. Only webhooks fill
            // provider_event_at/delivered_at; never fabricate a delivery timestamp.
            $current->touch();
        }, 3);
    }

    public function receive(string $id, array $event): void
    {
        if (!isset(self::STATUSES[$event['type'] ?? '']) || !is_string($event['data']['email_id'] ?? null)) {
            return;
        }
        $record = MailMessage::firstOrCreate(['message_key' => 'event:' . hash('sha256', $id)], [
            'direction' => 'event', 'purpose' => 'delivery_event', 'email' => '', 'subject' => '',
            'channel' => 'resend', 'provider_id' => $event['data']['email_id'],
            'payload' => ['type' => $event['type'], 'at' => $event['created_at'] ?? now()->toIso8601String()],
        ]);
        $this->apply($record);
    }

    public function apply(MailMessage $record, bool $lookup = false): void
    {
        if ($record->status !== 'pending') {
            return;
        }
        $message = MailMessage::where('direction', 'out')->where('channel', 'resend')
            ->where('provider_id', $record->provider_id)->first();
        if (!$message && $lookup) {
            // A webhook can precede the POST response, or recover a lost POST response.
            $email = $this->client->retrieve($record->provider_id);
            foreach ($email['tags'] ?? [] as $tag) {
                if (($tag['name'] ?? '') === 'mail_key') {
                    $message = MailMessage::where('direction', 'out')->where('channel', 'resend')
                        ->where('message_key', $tag['value'])->first();
                    break;
                }
            }
        }
        if (!$message) {
            return; // Persisted until the outbound identifier is known.
        }
        DB::transaction(function () use ($record, $message) {
            $record = MailMessage::whereKey($record->id)->lockForUpdate()->firstOrFail();
            if ($record->status !== 'pending') {
                return;
            }
            $message = MailMessage::whereKey($message->id)->lockForUpdate()->firstOrFail();
            $event = $record->payload;
            $at = Carbon::parse($event['at'])->setTimezone(config('app.timezone'));
            $status = self::STATUSES[$event['type']];
            $this->quota->settle($message, $status !== 'failed' && $status !== 'suppressed');
            $message->provider_id = $record->provider_id;
            if (!$message->provider_event_at || $at->gt($message->provider_event_at)
                || ($at->eq($message->provider_event_at) && $this->rank($status) > $this->rank($message->status))) {
                // Never downgrade a terminal complaint/bounce to a delayed/sent notification.
                if ($this->rank($status) >= $this->rank($message->status)) {
                    $message->status = $status;
                    $message->provider_event_at = $at;
                    $message->error_code = in_array($status, ['bounced', 'failed', 'suppressed', 'complained'], true) ? 'resend_' . $status : null;
                    if ($status === 'delivered') {
                        $message->delivered_at = $at;
                    }
                }
            }
            if (!in_array($status, ['failed', 'suppressed'], true)) {
                $message->submitted_at ??= $at;
            }
            $message->payload = null;
            $message->save();
            $record->update(['status' => 'processed', 'payload' => null]);
        }, 3);
    }

    private function rank(string $status): int
    {
        return match ($status) {
            'submitted' => 1, 'delayed' => 2, 'delivered' => 3,
            'failed', 'suppressed', 'bounced' => 4, 'complained' => 5, default => 0,
        };
    }
}