Lizandro GuarnizoandClaude Sonnet 4.6
ea7560f9bf
- El número WA Business es global (configuración general), no por empresa - OutboundWorker usa fallback a WHATSAPP_DEFAULT_PHONE_NUMBER_ID si la empresa no tiene - CompanyRepository::save() ya no requiere phone_number_id - phone_number_id permite DEFAULT '' en schema Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
97 lines
3.6 KiB
PHP
97 lines
3.6 KiB
PHP
<?php
|
|
declare(strict_types=1);
|
|
|
|
class OutboundWorker
|
|
{
|
|
private const BATCH_SIZE = 10;
|
|
|
|
public static function processQueue(): array
|
|
{
|
|
$processed = 0;
|
|
$errors = [];
|
|
|
|
$items = self::dequeue();
|
|
|
|
foreach ($items as $item) {
|
|
try {
|
|
$result = self::process($item);
|
|
|
|
if ($result['success']) {
|
|
$stmt = db()->prepare("UPDATE outbound_queue SET status = 'sent', wam_id = ?, updated_at = NOW() WHERE id = ?");
|
|
$stmt->execute([$result['wam_id'], $item['id']]);
|
|
$processed++;
|
|
} else {
|
|
$attempts = (int)$item['attempts'] + 1;
|
|
$max = (int)$item['max_attempts'];
|
|
|
|
if ($attempts >= $max) {
|
|
$stmt = db()->prepare("UPDATE outbound_queue SET status = 'failed', attempts = ?, last_error = ?, updated_at = NOW() WHERE id = ?");
|
|
$stmt->execute([$attempts, $result['error'], $item['id']]);
|
|
} else {
|
|
$stmt = db()->prepare("UPDATE outbound_queue SET status = 'queued', attempts = ?, last_error = ?, updated_at = NOW() WHERE id = ?");
|
|
$stmt->execute([$attempts, $result['error'], $item['id']]);
|
|
}
|
|
|
|
$errors[] = "msg#{$item['id']}: {$result['error']}";
|
|
}
|
|
} catch (\Exception $e) {
|
|
$errors[] = "msg#{$item['id']}: " . $e->getMessage();
|
|
}
|
|
}
|
|
|
|
return ['processed' => $processed, 'errors' => $errors];
|
|
}
|
|
|
|
private static function dequeue(): array
|
|
{
|
|
$db = db();
|
|
$db->beginTransaction();
|
|
|
|
try {
|
|
$stmt = $db->prepare("
|
|
SELECT q.*, c.phone_number_id
|
|
FROM outbound_queue q
|
|
JOIN companies c ON c.id = q.company_id
|
|
WHERE q.status = 'queued' AND c.is_active = 1
|
|
ORDER BY q.id ASC
|
|
LIMIT ?
|
|
FOR UPDATE SKIP LOCKED
|
|
");
|
|
$stmt->execute([self::BATCH_SIZE]);
|
|
$items = $stmt->fetchAll();
|
|
|
|
if (!empty($items)) {
|
|
$ids = array_column($items, 'id');
|
|
$placeholders = implode(',', array_fill(0, count($ids), '?'));
|
|
$db->prepare("UPDATE outbound_queue SET status = 'sending', updated_at = NOW() WHERE id IN ({$placeholders})")->execute($ids);
|
|
}
|
|
|
|
$db->commit();
|
|
return $items;
|
|
} catch (\Exception $e) {
|
|
$db->rollBack();
|
|
throw $e;
|
|
}
|
|
}
|
|
|
|
private static function process(array $item): array
|
|
{
|
|
$payload = json_decode($item['payload'], true);
|
|
if (!is_array($payload)) {
|
|
return ['success' => false, 'error' => 'Payload inválido'];
|
|
}
|
|
|
|
$phoneNumberId = $item['phone_number_id'] ?: env('WHATSAPP_DEFAULT_PHONE_NUMBER_ID', '');
|
|
$type = $item['message_type'];
|
|
$to = $item['to_number'];
|
|
|
|
return match ($type) {
|
|
'text' => WhatsAppSender::sendText($to, $payload['text'] ?? '', $phoneNumberId),
|
|
'image' => WhatsAppSender::sendImage($to, $payload['media_id'] ?? $payload['url'] ?? '', $phoneNumberId, $payload['caption'] ?? null),
|
|
'template'=> WhatsAppSender::sendTemplate($to, $payload['template_name'] ?? '', $phoneNumberId, $payload['components'] ?? []),
|
|
'interactive' => WhatsAppSender::sendInteractive($to, $payload['interactive'] ?? [], $phoneNumberId),
|
|
default => ['success' => false, 'error' => "Tipo no soportado: {$type}"],
|
|
};
|
|
}
|
|
}
|