dpl/app/Services/MailFlow/Dispatcher.php

223 lines
12 KiB
PHP

<?php
namespace App\Services\MailFlow;
use App\Models\MailMessage;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Str;
class Dispatcher
{
public function __construct(private ResendClient $resend, private QuotaService $quota, private ConventionalTransport $conventional) {}
public function send(array $params, string $subject, string $html): array
{
$params['purpose'] = AutomaticMailPolicy::purpose($params);
if (!filter_var($params['email'] ?? '', FILTER_VALIDATE_EMAIL)) {
return ['status' => 'failed', 'error' => 'invalid_recipient', 'retryable' => false];
}
$payload = ['to' => $params['email'], 'subject' => $subject, 'html' => $html,
'headers' => $params['headers'] ?? []];
if (!empty($params['attachments'])) { $payload['attachments'] = $params['attachments']; }
if ($params['purpose'] === 'service_expired') {
$payload['policy_context'] = ['user_id' => $params['user_id'] ?? null,
'subscription_expired_at' => $params['subscription_expired_at'] ?? null];
}
if (Settings::get('reply_to')) { $payload['reply_to'] = Settings::get('reply_to'); }
$persisted = $params['persisted_message'] ?? false;
if (!$persisted && !Settings::get('enabled') && isset($params['message_key'])
&& in_array($params['purpose'], ['traffic_reminder', 'expiry_reminder', 'automatic_broadcast'], true)
&& \Illuminate\Support\Facades\Schema::hasTable('v2_mail_messages')) {
$persisted = MailMessage::where('message_key', $params['message_key'])->exists();
}
if (!Settings::get('enabled') && $params['purpose'] !== 'service_expired' && !$persisted) {
return AutomaticMailPolicy::decision($params['purpose'], $params) ?? $this->sendLegacy($payload);
}
$key = $params['message_key'] ?? (string) Str::uuid();
$lock = Cache::lock('mailflow:message:' . $key, 60);
if (!$lock->get()) {
return ['status' => 'pending', 'error' => 'mail_busy', 'retryable' => true];
}
try {
$message = MailMessage::firstOrCreate(['message_key' => $key], [
'purpose' => $params['purpose'] ?? 'other', 'email' => $params['email'], 'subject' => $subject,
'template_name' => $params['template_name'] ?? null, 'user_id' => $params['user_id'] ?? null,
'parent_id' => $params['parent_id'] ?? null, 'payload' => $payload,
'expires_at' => $params['expires_at'] ?? null,
]);
if (in_array($message->status, ['submitted', 'delivered', 'delayed', 'bounced', 'complained', 'suppressed', 'simulated', 'expired', 'failed', 'policy_blocked'], true)) {
return $this->result($message);
}
if ($message->status === 'sending') {
$message->update(['status' => 'unknown', 'error_code' => 'worker_interrupted']);
}
if ($message->status === 'pending' && isset($message->payload['email_params'])) {
$message->update(['payload' => $payload]);
}
if ($blocked = $this->checkPolicy($message)) { return $blocked; }
// SMTP has no cross-request idempotency. Never blindly repeat an uncertain SMTP submission.
if ($message->status === 'unknown' && $message->channel === 'conventional') {
return $this->result($message);
}
if ($message->expires_at?->isPast() && $message->status === 'unknown') {
$message->update(['payload' => null]);
return $this->result($message);
}
if ($message->expires_at?->isPast() && $message->status !== 'unknown') {
return $this->finish($message, 'expired', 'mail_expired', false);
}
if ($message->status === 'unknown' && $message->first_attempt_at?->lt(now()->subHours(23))) {
// Resend idempotency retention is 24h; stop before it expires.
$message->update(['payload' => null]);
return $this->result($message);
}
$message->increment('attempts');
$payload = $message->payload;
if (!$payload) {
return $this->result($message);
}
$eligible = in_array($message->purpose, ['verification', 'login_link', 'ticket_reply'], true);
$useResend = $message->channel === 'resend' || ($eligible && Settings::get('resend_enabled') && Settings::get('resend_key') && Settings::get('resend_from'));
if ($useResend && !$message->reservation && $message->status !== 'unknown') {
try {
$useResend = $this->quota->reserve($message);
if (!$useResend) {
$message->fallback_reason = 'quota_exhausted';
}
} catch (\Throwable $e) {
$useResend = false;
$message->fallback_reason = 'quota_unavailable';
}
}
if ($useResend) {
return $this->sendResend($message, $payload);
}
return $this->sendConventional($message, $payload);
} finally {
$lock->release();
}
}
private function sendLegacy(array $payload): array
{
try {
return $this->conventional->send($payload) + ['error' => null, 'retryable' => false];
} catch (\Throwable $e) {
if ($e instanceof \Symfony\Component\Mailer\Exception\UnexpectedResponseException && $e->getCode() >= 400 && $e->getCode() <= 599) {
return ['status' => 'failed', 'error' => 'smtp_rejected_' . $e->getCode(), 'retryable' => $e->getCode() < 500];
}
// Even before enabling tracking, a transport exception is not proof of non-acceptance.
return ['status' => 'unknown', 'error' => 'conventional_transport_error', 'retryable' => false];
}
}
private function sendResend(MailMessage $message, array $payload): array
{
if ($blocked = $this->checkPolicy($message)) { return $blocked; }
if (empty($payload['from'])) {
$from = (string) Settings::get('resend_from');
$payload['from'] = filter_var($from, FILTER_VALIDATE_EMAIL)
? (new \Symfony\Component\Mime\Address($from, (string) admin_setting('app_name', config('mail.from.name'))))->toString()
: $from;
$payload['tags'] = [['name' => 'mail_key', 'value' => $message->message_key]];
$message->payload = $payload;
}
$message->channel = 'resend';
$message->status = 'sending';
$message->first_attempt_at ??= now();
$message->save();
try {
$response = $this->resend->send($payload, 'dpl-mail/' . $message->message_key);
} catch (\Throwable $e) {
$message->update(['status' => 'unknown', 'error_code' => 'resend_transport_unknown']);
return $this->result($message, true);
}
if ($response->successful() && is_string($response->json('id'))) {
return DB::transaction(function () use ($message, $response) {
// A delivery webhook may have arrived before this response.
$current = MailMessage::whereKey($message->id)->lockForUpdate()->firstOrFail();
$this->quota->settle($current, true);
$current->provider_id = $response->json('id');
$current->submitted_at ??= now();
if (in_array($current->status, ['sending', 'unknown', 'pending'], true)) {
$current->status = 'submitted';
}
$current->payload = null;
$current->error_code = null;
$current->save();
return $this->result($current);
}, 3);
}
$name = (string) $response->json('name', 'resend_rejected');
if ($response->status() >= 500 || $response->status() === 408 || $response->status() === 409 || $response->successful()) {
$message->update(['status' => 'unknown', 'error_code' => 'resend_response_unknown']);
return $this->result($message, true);
}
// A definite HTTP rejection releases the reservation. Never route hard bounces through another sender.
DB::transaction(fn () => $this->quota->settle($message, false), 3);
if (in_array($name, ['daily_quota_exceeded', 'monthly_quota_exceeded'], true)) {
$this->quota->exhausted($name === 'daily_quota_exceeded' ? 'daily' : 'monthly');
$message->fallback_reason = $name;
return $this->sendConventional($message, $payload);
}
if ($response->status() === 429) {
$message->update(['status' => 'pending', 'error_code' => 'resend_rate_limited']);
return $this->result($message, true);
}
return $this->finish($message, 'failed', 'resend_http_' . $response->status(), false);
}
private function sendConventional(MailMessage $message, array $payload): array
{
if ($blocked = $this->checkPolicy($message)) { return $blocked; }
$message->channel = 'conventional';
$message->status = 'sending';
$message->first_attempt_at ??= now();
$message->save();
try {
$result = $this->conventional->send($payload);
} catch (\Throwable $e) {
if ($e instanceof \Symfony\Component\Mailer\Exception\UnexpectedResponseException && $e->getCode() >= 400 && $e->getCode() <= 599) {
$code = 'smtp_rejected_' . $e->getCode();
if ($e->getCode() < 500) {
$message->update(['status' => 'pending', 'error_code' => $code]);
return $this->result($message, true);
}
return $this->finish($message, 'failed', $code, false);
}
$message->update(['status' => 'unknown', 'error_code' => 'conventional_transport_unknown']);
return $this->result($message);
}
$message->update(['status' => $result['status'], 'provider_id' => $result['provider_id'],
'submitted_at' => $result['status'] === 'submitted' ? now() : null, 'error_code' => null, 'payload' => null]);
return $this->result($message);
}
private function finish(MailMessage $message, string $status, string $error, bool $accepted): array
{
return DB::transaction(function () use ($message, $status, $error, $accepted) {
$this->quota->settle($message, $accepted);
$message->update(['status' => $status, 'error_code' => $error, 'payload' => null]);
return $this->result($message);
}, 3);
}
private function checkPolicy(MailMessage $message): ?array
{
$context = $message->payload['policy_context'] ?? $message->payload['email_params'] ?? [];
$decision = AutomaticMailPolicy::decision($message->purpose, $context);
if (!$decision) { return null; }
// Unknown submissions retain their original status/quota until acceptance can be established.
if ($message->status === 'unknown') { return $this->result($message); }
if ($decision['retryable']) { return $decision + ['message_id' => $message->id]; }
return $this->finish($message, 'policy_blocked', $decision['error'], false);
}
private function result(MailMessage $message, bool $retryable = false): array
{
return ['status' => $message->status, 'error' => $message->error_code,
'retryable' => $retryable, 'message_id' => $message->id, 'channel' => $message->channel];
}
}