feat(notifications): scheduled campaigns are actually sent

Campaigns saved as 'scheduled' were never looked at again: nothing in either
app processed them. CampaignService::processDue() sends every due campaign
into the game's bell (notifications with title_ar/body_ar, as the game's
notifications.php reads them) and records recipients, claiming each one with
a conditional scheduled -> sending update so overlapping runs never send
twice. It runs from GET/POST /api/cron/notifications (X-Service-Key, or
X-Cron-Secret / ?secret= when CRON_SECRET is set, the same credentials as
the game's api/cron.php) for a scheduler, and as a fallback at most once a
minute after any staff page load.

Scheduling now stores the typed Cairo time as UTC (the raw datetime-local
value was saved as UTC, 2-3 hours late) and refuses past times; filters
are stored as jsonb, not a JSON string; cancelling only cancels a campaign
that has not started; send-now shares the same delivery code.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

api/cron-notifications.php

+<?php
+
+/**
+ * Send scheduled notification campaigns that are due.
+ *
+ *   GET or POST /api/cron/notifications
+ *
+ * For a scheduler (a crontab line, an external pinger). Authenticated by a
+ * shared secret, the same two the player app's api/cron.php accepts:
+ *   X-Service-Key: <SUPABASE_SERVICE_KEY>   (both apps already have it), or
+ *   X-Cron-Secret: <CRON_SECRET> / ?secret=  (when CRON_SECRET is set here).
+ * Refuses when neither matches. Idempotent: a campaign is claimed before it is
+ * sent, so overlapping runs never send one twice.
+ */
+
+$cronSecret = (string)(getenv('CRON_SECRET') ?: '');
+$given = (string)($_SERVER['HTTP_X_CRON_SECRET'] ?? ($_GET['secret'] ?? ''));
+$serviceKey = (string)($_SERVER['HTTP_X_SERVICE_KEY'] ?? '');
+
+$authorised =
+    ($cronSecret !== '' && $given !== '' && hash_equals($cronSecret, $given)) ||
+    ($serviceKey !== '' && hash_equals(SUPABASE_SERVICE_KEY, $serviceKey));
+
+if (!$authorised) {
+    Response::json(['error' => 'forbidden'], 403);
+}
+
+require_once MODULES_PATH . '/notifications/services/CampaignService.php';
+Response::json(CampaignService::processDue(25));

config/routes.php

@@ -227,6 +227,7 @@ return [
 
     // API
     'api/health' => ['module' => 'api', 'action' => 'health'],
+    'api/cron/notifications' => ['module' => 'api', 'action' => 'cron-notifications'],
     'api/players/search' => ['module' => 'players', 'action' => 'apiSearch'],
 
     // Player Avatar & Frame

core/Router.php

@@ -128,6 +128,9 @@ class Router
         if ($action === 'health') {
             require_once BASE_PATH . '/api/health.php';
         }
+        if ($action === 'cron-notifications') {
+            require_once BASE_PATH . '/api/cron-notifications.php';
+        }
     }
 
     private function moduleToClassName(string $module): string

index.php

@@ -25,5 +25,20 @@ if (str_starts_with($route, 'public/')) {
     return false;
 }
 
+// Scheduled notifications are sent by /api/cron/notifications when a scheduler
+// calls it. As a fallback, staff traffic also sends anything due, at most once
+// a minute and after the page has been produced. Never on public pages.
+register_shutdown_function(function () use ($route) {
+    if (str_starts_with($route, 'live/') || str_starts_with($route, 'api/')) return;
+    if (empty($_SESSION['user'])) return;
+    try {
+        if (function_exists('fastcgi_finish_request')) fastcgi_finish_request();
+        require_once MODULES_PATH . '/notifications/services/CampaignService.php';
+        CampaignService::tick(60);
+    } catch (\Throwable $e) {
+        error_log('[campaigns] tick failed: ' . $e->getMessage());
+    }
+});
+
 $router = new Router();
 $router->dispatch($route, $method);

modules/notifications/services/CampaignService.php

+<?php
+
+/**
+ * Notification campaigns: who they reach, and delivering them into the game's
+ * bell (public.notifications, which the game's api/notifications.php reads:
+ * title_ar/body_ar, type, data).
+ *
+ * Scheduled campaigns used to be saved with status 'scheduled' and then never
+ * looked at again: nothing in either app processed them. processDue() is that
+ * processor. It runs from the protected endpoint /api/cron/notifications (for a
+ * scheduler) and, as a fallback, at most once a minute at the end of any staff
+ * page load in the manager. Each campaign is claimed with a conditional update
+ * (scheduled -> sending) before it is sent, so two runs at once can never send
+ * the same campaign twice.
+ */
+class CampaignService
+{
+    /** Campaigns handled per run, so one run stays short. */
+    public const BATCH = 5;
+
+    /** Players for a broadcast or a demographic campaign: id and display name only. */
+    public static function targetPlayers(string $targetType, array $filters): array
+    {
+        $db = Database::getInstance();
+        $queryParams = ['select' => 'id,display_name', 'order' => 'created_at.desc'];
+        if ($targetType !== 'broadcast') self::applyFilters($queryParams, $filters);
+
+        $all = [];
+        $offset = 0;
+        $limit = 1000;
+        while (true) {
+            $queryParams['limit'] = $limit;
+            $queryParams['offset'] = $offset;
+            $batch = $db->select('profiles', $queryParams);
+            if (empty($batch)) break;
+            $all = array_merge($all, $batch);
+            if (count($batch) < $limit) break;
+            $offset += $limit;
+        }
+        return $all;
+    }
+
+    public static function countPlayers(array $filters): int
+    {
+        $queryParams = [];
+        self::applyFilters($queryParams, $filters);
+        return Database::getInstance()->count('profiles', $queryParams);
+    }
+
+    /**
+     * Write one notification per player (and the recipient log), in chunks.
+     *
+     * @return array{sent:int, failed:int}
+     */
+    public static function deliver(string $campaignId, string $titleAr, string $bodyAr, string $type, array $players): array
+    {
+        $db = Database::getInstance();
+        $sent = 0;
+        $failed = 0;
+        $sentAt = date('c');
+
+        foreach (array_chunk($players, 100) as $chunk) {
+            $rows = [];
+            $recipients = [];
+            foreach ($chunk as $player) {
+                $rows[] = [
+                    'user_id' => $player['id'],
+                    'type' => $type,
+                    'title' => $titleAr,
+                    'title_ar' => $titleAr,
+                    'body' => $bodyAr,
+                    'body_ar' => $bodyAr,
+                    'data' => ['campaign_id' => $campaignId],
+                ];
+            }
+            $ok = $db->insert('notifications', $rows) !== null;
+            if ($ok) { $sent += count($chunk); } else { $failed += count($chunk); }
+
+            foreach ($chunk as $player) {
+                $recipients[] = [
+                    'campaign_id' => $campaignId,
+                    'user_id' => $player['id'],
+                    'display_name' => $player['display_name'] ?? null,
+                    'status' => $ok ? 'sent' : 'failed',
+                    'sent_at' => $ok ? $sentAt : null,
+                    'error_message' => $ok ? null : mb_substr($db->lastErrorMessage(), 0, 500),
+                ];
+            }
+            // The recipient log is bookkeeping; it must not undo a delivery.
+            $db->soft(fn(Database $d) => $d->insert('push_campaign_recipients', $recipients));
+        }
+
+        return ['sent' => $sent, 'failed' => $failed];
+    }
+
+    /**
+     * Send every scheduled campaign whose time has come.
+     *
+     * @return array{processed: array, errors: array}
+     */
+    public static function processDue(int $limit = self::BATCH): array
+    {
+        $db = Database::getInstance();
+        $report = ['processed' => [], 'errors' => []];
+
+        $due = $db->select('push_campaigns', [
+            'select' => '*',
+            'status' => 'eq.scheduled',
+            'scheduled_at' => 'lte.' . gmdate('Y-m-d\TH:i:s\Z'),
+            'order' => 'scheduled_at.asc',
+            'limit' => $limit,
+        ]);
+
+        foreach ($due as $c) {
+            // Claim it. Only the run whose update changes the row sends it.
+            $claimed = $db->update('push_campaigns', ['id' => "eq.{$c['id']}", 'status' => 'eq.scheduled'], ['status' => 'sending']);
+            if (empty($claimed)) continue;
+
+            $filters = $c['filters_json'] ?? [];
+            if (is_string($filters)) $filters = json_decode($filters, true) ?: [];
+            $players = self::targetPlayers((string)($c['type'] ?? 'broadcast'), is_array($filters) ? $filters : []);
+
+            $result = ['sent' => 0, 'failed' => 0];
+            if ($players) {
+                $result = self::deliver($c['id'], (string)$c['title_ar'], (string)$c['body_ar'], (string)($c['notification_type'] ?: 'announcement'), $players);
+            }
+
+            $status = ($result['sent'] === 0 && ($result['failed'] > 0 || !$players)) ? 'failed' : 'sent';
+            $db->update('push_campaigns', ['id' => "eq.{$c['id']}"], [
+                'status' => $status,
+                'target_count' => count($players),
+                'sent_count' => $result['sent'],
+                'failed_count' => $result['failed'],
+                'sent_at' => date('c'),
+            ]);
+
+            $entry = ['id' => $c['id'], 'title' => $c['title_ar'], 'status' => $status] + $result;
+            if ($status === 'failed') $report['errors'][] = $entry; else $report['processed'][] = $entry;
+        }
+
+        // A run that sends notifications is not the place to fail the caller
+        // over one bad row; problems are in the report and the PHP error log.
+        $db->clearWriteErrors();
+        return $report;
+    }
+
+    /**
+     * At most once every $seconds across all PHP workers (a lock file), so a
+     * busy manager does not query for due campaigns on every page.
+     */
+    public static function tick(int $seconds = 60): ?array
+    {
+        $file = STORAGE_PATH . '/cache/campaign_tick';
+        $fh = @fopen($file, 'c+');
+        if (!$fh) return null;
+        try {
+            if (!flock($fh, LOCK_EX | LOCK_NB)) return null;
+            $last = (int)stream_get_contents($fh);
+            if (time() - $last < $seconds) return null;
+            ftruncate($fh, 0);
+            rewind($fh);
+            fwrite($fh, (string)time());
+            fflush($fh);
+            return self::processDue();
+        } finally {
+            flock($fh, LOCK_UN);
+            fclose($fh);
+        }
+    }
+
+    public static function applyFilters(array &$queryParams, array $filters): void
+    {
+        $andConditions = [];
+
+        if (!empty($filters['level_min'])) $andConditions[] = 'level.gte.' . (int)$filters['level_min'];
+        if (!empty($filters['level_max'])) $andConditions[] = 'level.lte.' . (int)$filters['level_max'];
+        if (!empty($filters['games_min'])) $queryParams['games_played'] = 'gte.' . (int)$filters['games_min'];
+        if (!empty($filters['coins_min'])) $andConditions[] = 'coins.gte.' . (int)$filters['coins_min'];
+        if (!empty($filters['coins_max'])) $andConditions[] = 'coins.lte.' . (int)$filters['coins_max'];
+        if (!empty($filters['country'])) $queryParams['country_code'] = 'eq.' . strtoupper(substr((string)$filters['country'], 0, 2));
+
+        if (!empty($filters['activity'])) {
+            $now = new DateTimeImmutable('now', new DateTimeZone('UTC'));
+            $since = match ($filters['activity']) {
+                'active_today' => Time::toUtc((new DateTimeImmutable('today', Time::zone()))->format('Y-m-d H:i')),
+                'active_7d', 'inactive_7d' => $now->modify('-7 days')->format('Y-m-d\TH:i:s\Z'),
+                'active_30d', 'inactive_30d' => $now->modify('-30 days')->format('Y-m-d\TH:i:s\Z'),
+                'inactive_90d' => $now->modify('-90 days')->format('Y-m-d\TH:i:s\Z'),
+                default => null,
+            };
+            if ($since) {
+                $op = str_starts_with($filters['activity'], 'active') ? 'gte' : 'lte';
+                $queryParams['last_seen_at'] = "{$op}.{$since}";
+            }
+        }
+
+        // Registration dates are whole days in Cairo.
+        if (!empty($filters['registered_after'])) {
+            $t = Time::toUtc($filters['registered_after'] . ' 00:00');
+            if ($t) $andConditions[] = "created_at.gte.{$t}";
+        }
+        if (!empty($filters['registered_before'])) {
+            $t = Time::toUtc($filters['registered_before'] . ' 23:59');
+            if ($t) $andConditions[] = "created_at.lte.{$t}";
+        }
+
+        if (!empty($andConditions)) {
+            $queryParams['and'] = '(' . implode(',', $andConditions) . ')';
+        }
+    }
+}