$d Validated data from Validator::planVisit * @return array{lead:array, duplicate:bool} */ public static function persist(array $d, string $ip, string $userAgent): array { $pdo = Db::pdo(); $existing = self::find($d['submission_id']); if ($existing) { return ['lead' => $existing, 'duplicate' => true]; } $now = gmdate('Y-m-d\TH:i:s\Z'); $st = $pdo->prepare('INSERT INTO leads ( submission_id, created_at, updated_at, first_name, last_name, email, phone_e164, campus_id, visit_date, service_time, adults, children, child_age_ranges, questions, email_consent, sms_consent, consent_version, consent_at, consent_ip_hash, attribution, variant, referrer, landing_url, user_agent, mailchimp_status, sms_status ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)'); $attr = $d['attribution']; try { $st->execute([ $d['submission_id'], $now, $now, $d['first_name'], $d['last_name'], $d['email'], $d['phone_e164'], $d['campus_id'], $d['visit_date'], $d['service_time'], $d['adults'], $d['children'], $d['child_age_ranges'], $d['questions'], (int)$d['email_consent'], (int)$d['sms_consent'], $d['consent_version'], $now, hash('sha256', $ip . '|' . (Config::get('HASH_SALT') ?? 'visit')), json_encode($attr, JSON_UNESCAPED_SLASHES), $attr['variant'] ?? null, $attr['referrer'] ?? null, $attr['landing_url'] ?? null, mb_substr($userAgent, 0, 300), $d['email_consent'] ? 'pending' : 'skipped', $d['sms_consent'] ? 'pending' : 'skipped', ]); } catch (PDOException $e) { // Unique constraint race: another request with the same id got there first. if (str_contains($e->getMessage(), 'UNIQUE') || str_contains($e->getMessage(), 'Duplicate')) { $row = self::find($d['submission_id']); if ($row) { return ['lead' => $row, 'duplicate' => true]; } } throw $e; } $lead = self::find($d['submission_id']); if (!$lead) { throw new RuntimeException('Lead insert failed'); } return ['lead' => $lead, 'duplicate' => false]; } /** @return array|null */ public static function find(string $submissionId): ?array { $st = Db::pdo()->prepare('SELECT * FROM leads WHERE submission_id = ?'); $st->execute([$submissionId]); $row = $st->fetch(); return $row ?: null; } /** * Runs Mailchimp then Sinch for a stored lead. Each step is guarded by a * status transition so concurrent requests/cron cannot double-send. * * @param array $lead * @return array{email:string, sms:string} */ public static function runIntegrations(array $lead): array { $campus = Campuses::get($lead['campus_id']) ?? ['id' => $lead['campus_id'], 'name' => $lead['campus_id'], 'address' => ['line1' => '', 'city' => ''], 'serviceTimes' => []]; $out = ['email' => $lead['mailchimp_status'], 'sms' => $lead['sms_status']]; if (self::claim($lead['id'], 'mailchimp')) { try { $r = Mailchimp::upsertVisitor($lead, $campus); } catch (Throwable $e) { Log::error('Mailchimp adapter threw', ['error' => $e->getMessage()]); $r = ['status' => 'failed', 'detail' => 'exception: ' . $e->getMessage(), 'transient' => true]; } $out['email'] = self::record($lead['id'], 'mailchimp', $r['status'], $r['detail'] ?? null, $r['transient'] ?? false); } if (self::claim($lead['id'], 'sms')) { try { $r = Sinch::sendVisitConfirmation($lead, $campus); } catch (Throwable $e) { Log::error('Sinch adapter threw', ['error' => $e->getMessage()]); $r = ['status' => 'failed', 'detail' => 'exception: ' . $e->getMessage(), 'message_id' => null, 'transient' => true]; } $out['sms'] = self::record($lead['id'], 'sms', $r['status'], $r['detail'] ?? null, $r['transient'] ?? false, $r['message_id'] ?? null); if (in_array($r['status'], ['sent', 'mock'], true)) { try { Sinch::sendInternalLeadNotification($lead, $campus); } catch (Throwable $e) { Log::warn('Internal notify failed', ['error' => $e->getMessage()]); } } } return $out; } /** Atomically move pending/retry → sending. Returns false if someone else owns it. */ private static function claim(int $id, string $which): bool { $col = $which === 'sms' ? 'sms_status' : 'mailchimp_status'; $att = $which === 'sms' ? 'sms_attempts' : 'mailchimp_attempts'; $st = Db::pdo()->prepare("UPDATE leads SET $col = 'sending', $att = $att + 1, updated_at = ? WHERE id = ? AND $col IN ('pending','retry')"); $st->execute([gmdate('Y-m-d\TH:i:s\Z'), $id]); return $st->rowCount() === 1; } /** Persists the outcome and returns the final stored status ('sent', 'mock', 'skipped', 'failed' or 'retry'). */ private static function record(int $id, string $which, string $status, ?string $detail, bool $transient, ?string $messageId = null): string { $col = $which === 'sms' ? 'sms_status' : 'mailchimp_status'; $det = $which === 'sms' ? 'sms_detail' : 'mailchimp_detail'; $att = $which === 'sms' ? 'sms_attempts' : 'mailchimp_attempts'; $pdo = Db::pdo(); $attempts = (int)$pdo->query("SELECT $att FROM leads WHERE id = $id")->fetchColumn(); $final = $status; $nextRetry = null; if ($status === 'failed' && $transient && $attempts < self::MAX_ATTEMPTS) { $final = 'retry'; $nextRetry = gmdate('Y-m-d\TH:i:s\Z', time() + self::backoffSeconds($attempts)); } $sql = "UPDATE leads SET $col = ?, $det = ?, updated_at = ?" . ($which === 'sms' ? ', sms_message_id = COALESCE(?, sms_message_id)' : '') . ', next_retry_at = ? WHERE id = ?'; $params = [$final, $detail ? mb_substr($detail, 0, 500) : null, gmdate('Y-m-d\TH:i:s\Z')]; if ($which === 'sms') { $params[] = $messageId; } $params[] = $nextRetry; $params[] = $id; $pdo->prepare($sql)->execute($params); // Keep the earliest pending retry across both integrations $pdo->prepare("UPDATE leads SET next_retry_at = (SELECT MIN(x) FROM (SELECT next_retry_at AS x FROM leads WHERE id = ?)) WHERE id = ?")->execute([$id, $id]); Log::info("Integration $which → $final", ['lead' => $id, 'attempts' => $attempts]); return $final; } public static function backoffSeconds(int $attempt): int { return min(6 * 3600, (int)(60 * (2 ** max(0, $attempt - 1)))); // 1m, 2m, 4m, 8m … capped at 6h } /** * Cron entry point: retry rows whose integrations are in 'retry' (or stuck in 'sending' for >15 min). * @return array{checked:int, processed:int} */ public static function retryDue(int $limit = 25): array { $pdo = Db::pdo(); $now = gmdate('Y-m-d\TH:i:s\Z'); $stale = gmdate('Y-m-d\TH:i:s\Z', time() - 900); // Unstick crashed "sending" rows $pdo->prepare("UPDATE leads SET mailchimp_status = 'retry' WHERE mailchimp_status = 'sending' AND updated_at < ?")->execute([$stale]); $pdo->prepare("UPDATE leads SET sms_status = 'retry' WHERE sms_status = 'sending' AND updated_at < ?")->execute([$stale]); $st = $pdo->prepare("SELECT * FROM leads WHERE (mailchimp_status = 'retry' OR sms_status = 'retry') AND (next_retry_at IS NULL OR next_retry_at <= ?) ORDER BY created_at ASC LIMIT $limit"); $st->execute([$now]); $rows = $st->fetchAll(); $processed = 0; foreach ($rows as $row) { self::runIntegrations($row); $processed++; } return ['checked' => count($rows), 'processed' => $processed]; } /** Data minimisation: delete leads older than LEAD_RETENTION_DAYS (0 disables). */ public static function purgeExpired(): int { $days = Config::int('LEAD_RETENTION_DAYS', 0); if ($days <= 0) { return 0; } $cutoff = gmdate('Y-m-d\TH:i:s\Z', time() - $days * 86400); $st = Db::pdo()->prepare("DELETE FROM leads WHERE created_at < ? AND mailchimp_status NOT IN ('pending','retry','sending') AND sms_status NOT IN ('pending','retry','sending')"); $st->execute([$cutoff]); return $st->rowCount(); } }