inserted notification ids */ public function enqueue(PDO $pdo, array $record): array { $validation = ($this->records ?? new NotificationRecord())->validate($record); if (!$validation['valid']) throw new InvalidArgumentException('Invalid notification: ' . implode(' ', $validation['errors'])); $title = is_scalar($record['title'] ?? null) ? trim((string)$record['title']) : ''; if ($title === '' || mb_strlen($title) > 190) throw new InvalidArgumentException('Notification title is required and must be 190 characters or fewer.'); $lookup = $pdo->prepare('SELECT id FROM users WHERE email = :email AND is_active = 1 LIMIT 1'); $insert = $pdo->prepare('INSERT INTO notifications (user_id, type, title, body, deduplication_key, read_at) VALUES (:user, :type, :title, :body, :dedup, :read_at) ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)'); $ids = []; $startedTransaction = false; try { if (!$pdo->inTransaction()) { $pdo->beginTransaction(); $startedTransaction = true; } foreach ($validation['recipients'] as $email) { $dto = ($this->records ?? new NotificationRecord())->toQueueDto($record, $email); $lookup->execute(['email' => $email]); $userId = $lookup->fetchColumn(); if ($userId === false) continue; $insert->execute(['user' => $userId, 'type' => $dto['type'], 'title' => $dto['title'], 'body' => $dto['body'], 'dedup' => $dto['deduplication_key'], 'read_at' => $dto['is_read'] ? date('Y-m-d H:i:s') : null]); $id = (int)$pdo->lastInsertId(); if ($id > 0 && !in_array($id, $ids, true)) $ids[] = $id; } if ($startedTransaction) $pdo->commit(); } catch (\Throwable $exception) { if ($startedTransaction && $pdo->inTransaction()) $pdo->rollBack(); throw $exception; } return $ids; } /** Map a multi-recipient notification into one queue DTO for one user. */ public function mapForUser(array $record, string $recipient): array { return ($this->records ?? new NotificationRecord())->toQueueDto($record, $recipient); } /** Mark a notification read only for the authenticated user's row. */ public function markRead(PDO $pdo, int|string $userId, array $command): bool { $user = filter_var($userId, FILTER_VALIDATE_INT, ['options' => ['min_range' => 1]]); $validation = ($this->records ?? new NotificationRecord())->validateMarkRead($command); if ($user === false || !$validation['valid']) throw new InvalidArgumentException('Invalid mark-read command.'); $stmt = $pdo->prepare('UPDATE notifications SET read_at = COALESCE(read_at, CURRENT_TIMESTAMP) WHERE id = :id AND user_id = :user'); $stmt->execute(['id' => $validation['notification_id'], 'user' => $user]); return $stmt->rowCount() > 0; } /** Mark a notification unread only for the authenticated user's row. */ public function markUnread(PDO $pdo, int|string $userId, array $command): bool { $user = filter_var($userId, FILTER_VALIDATE_INT, ['options' => ['min_range' => 1]]); $validation = ($this->records ?? new NotificationRecord())->validateMarkUnread($command); if ($user === false || !$validation['valid']) throw new InvalidArgumentException('Invalid mark-unread command.'); $stmt = $pdo->prepare('UPDATE notifications SET read_at = NULL WHERE id = :id AND user_id = :user'); $stmt->execute(['id' => $validation['notification_id'], 'user' => $user]); return $stmt->rowCount() > 0; } }