move_detect/server/src/Repository/Pdo/PdoEventRepository.php

<?php
declare(strict_types=1);
namespace App\Repository\Pdo;

use App\Repository\EventRepository;
use PDO;

final class PdoEventRepository implements EventRepository
{
    private const SELECT_COLUMNS =
        'event_id, event_uuid, device_id, site, camera_id, event_type, severity, track_id, zone_id,
         occurred_at, duration_s, metrics_json, spatial_x, spatial_y, acknowledged_by, acknowledged_at,
         false_positive_by, false_positive_at';

    public function __construct(private PDO $pdo) {}

    public function insert(array $event): int
    {
        $st = $this->pdo->prepare(
            'INSERT INTO move_events
                (event_uuid, device_id, site, camera_id, event_type, severity, track_id, zone_id,
                 occurred_at, duration_s, metrics_json, spatial_x, spatial_y, received_at)
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
        );
        $st->execute([
            $event['event_uuid'] ?? null,   // NULL when the edge sent no idempotency key (legacy)
            (string)$event['device_id'],
            $event['site'] ?? null,          // denormalized from move_devices.site by IngestController
            (string)$event['camera_id'],
            (string)$event['event_type'],
            (string)$event['severity'],
            $event['track_id'] ?? null,
            $event['zone_id'] ?? null,
            (float)$event['occurred_at'],
            (float)($event['duration_s'] ?? 0.0),
            json_encode($event['metrics'] ?? [], JSON_THROW_ON_ERROR),
            $event['spatial_x'] ?? null,
            $event['spatial_y'] ?? null,
            time(),
        ]);
        return (int)$this->pdo->lastInsertId();
    }

    public function idByUuid(string $uuid): ?int
    {
        $st = $this->pdo->prepare('SELECT event_id FROM move_events WHERE event_uuid = ? LIMIT 1');
        $st->execute([$uuid]);
        $id = $st->fetchColumn();
        return $id === false ? null : (int)$id;
    }

    public function recent(int $limit, ?string $severity): array
    {
        if ($severity !== null) {
            $st = $this->pdo->prepare(
                'SELECT ' . self::SELECT_COLUMNS . '
                   FROM move_events WHERE severity = ?
               ORDER BY occurred_at DESC, event_id DESC LIMIT ?'
            );
            $st->bindValue(1, $severity, PDO::PARAM_STR);
            $st->bindValue(2, $limit, PDO::PARAM_INT);
        } else {
            $st = $this->pdo->prepare(
                'SELECT ' . self::SELECT_COLUMNS . '
                   FROM move_events
               ORDER BY occurred_at DESC, event_id DESC LIMIT ?'
            );
            $st->bindValue(1, $limit, PDO::PARAM_INT);
        }
        $st->execute();
        return array_map(self::mapRow(...), $st->fetchAll());
    }

    public function page(int $limit, ?string $severity, ?string $beforeTs, ?int $beforeId, ?string $site = null): array
    {
        // Bind list as [value, PDO type] pairs, positional in SQL order.
        $where = [];
        $binds = [];
        if ($severity !== null) {
            $where[] = 'severity = ?';
            $binds[] = [$severity, PDO::PARAM_STR];
        }
        if ($site !== null) {
            // Site-scope access control: a scoped operator sees only their site's events.
            $where[] = 'site = ?';
            $binds[] = [$site, PDO::PARAM_STR];
        }
        // Keyset predicate: strictly "older" than the client's last-seen (occurred_at, event_id)
        // on the (occurred_at DESC, event_id DESC) order. occurred_at arrives as its exact numeric
        // STRING (never round-tripped through a PHP float) and is bound as a string, so MySQL
        // parses it to the same DOUBLE the column holds — the page-edge boundary stays exact.
        if ($beforeTs !== null && $beforeId !== null) {
            $where[] = '(occurred_at < ? OR (occurred_at = ? AND event_id < ?))';
            $binds[] = [$beforeTs, PDO::PARAM_STR];
            $binds[] = [$beforeTs, PDO::PARAM_STR];
            $binds[] = [$beforeId, PDO::PARAM_INT];
        }
        $sql = 'SELECT ' . self::SELECT_COLUMNS . ' FROM move_events';
        if ($where) $sql .= ' WHERE ' . implode(' AND ', $where);
        $sql .= ' ORDER BY occurred_at DESC, event_id DESC LIMIT ?';
        $binds[] = [$limit, PDO::PARAM_INT];

        $st = $this->pdo->prepare($sql);
        $pos = 1;
        foreach ($binds as [$val, $type]) $st->bindValue($pos++, $val, $type);
        $st->execute();
        return array_map(self::mapRow(...), $st->fetchAll());
    }

    public function find(int $id): ?array
    {
        $st = $this->pdo->prepare('SELECT ' . self::SELECT_COLUMNS . ' FROM move_events WHERE event_id = ?');
        $st->execute([$id]);
        $row = $st->fetch();
        return $row === false ? null : self::mapRow($row);
    }

    public function acknowledge(int $id, string $user): bool
    {
        // Existence is checked separately (rather than trusting UPDATE's
        // rowCount()) because MySQL's PDO driver, by default, reports
        // "changed rows" rather than "matched rows": re-acknowledging an
        // already-acknowledged event by the same user would otherwise report
        // 0 affected rows and this method would wrongly return false.
        $exists = $this->pdo->prepare('SELECT 1 FROM move_events WHERE event_id = ? LIMIT 1');
        $exists->execute([$id]);
        if ($exists->fetchColumn() === false) {
            return false;
        }
        $st = $this->pdo->prepare(
            'UPDATE move_events SET acknowledged_by = ?, acknowledged_at = ? WHERE event_id = ?'
        );
        $st->execute([$user, time(), $id]);
        return true;
    }

    public function heatmapPoints(string $cameraId, float $sinceOccurredAt, ?string $site, int $limit): array
    {
        $sql = 'SELECT spatial_x, spatial_y, severity, event_type, occurred_at
                  FROM move_events
                 WHERE camera_id = ? AND occurred_at >= ?
                   AND spatial_x IS NOT NULL AND spatial_y IS NOT NULL';
        $binds = [[$cameraId, PDO::PARAM_STR], [sprintf('%.6f', $sinceOccurredAt), PDO::PARAM_STR]];
        if ($site !== null) {
            // Same fail-closed site scoping as page(): an empty-string scope matches nothing.
            $sql .= ' AND site = ?';
            $binds[] = [$site, PDO::PARAM_STR];
        }
        $sql .= ' ORDER BY occurred_at DESC LIMIT ?';
        $binds[] = [$limit, PDO::PARAM_INT];
        $st = $this->pdo->prepare($sql);
        $pos = 1;
        foreach ($binds as [$val, $type]) $st->bindValue($pos++, $val, $type);
        $st->execute();
        return array_map(static fn ($r) => [
            'x' => (float)$r['spatial_x'],
            'y' => (float)$r['spatial_y'],
            'severity' => $r['severity'],
            'event_type' => $r['event_type'],
            'occurred_at' => (float)$r['occurred_at'],
        ], $st->fetchAll());
    }

    public function markFalsePositive(int $id, string $user): bool
    {
        // Same matched-vs-changed discipline as acknowledge(): existence checked separately,
        // because a no-op UPDATE reports 0 affected rows on MySQL's default PDO settings.
        $exists = $this->pdo->prepare('SELECT 1 FROM move_events WHERE event_id = ? LIMIT 1');
        $exists->execute([$id]);
        if ($exists->fetchColumn() === false) return false;
        $now = time();
        // COALESCE keeps the FIRST tagger/acknowledger — a second tag must not rewrite who
        // made the original call (it is an audit-relevant judgement).
        $st = $this->pdo->prepare(
            'UPDATE move_events
                SET false_positive_by = COALESCE(false_positive_by, ?),
                    false_positive_at = COALESCE(false_positive_at, ?),
                    acknowledged_by   = COALESCE(acknowledged_by, ?),
                    acknowledged_at   = COALESCE(acknowledged_at, ?)
              WHERE event_id = ?'
        );
        $st->execute([$user, $now, $user, $now, $id]);
        return true;
    }

    public function fpStats(float $sinceOccurredAt, ?string $site): array
    {
        $sql = 'SELECT camera_id, event_type, COUNT(*) AS total,
                       SUM(false_positive_by IS NOT NULL) AS fp
                  FROM move_events
                 WHERE occurred_at >= ?';
        $binds = [[sprintf('%.6f', $sinceOccurredAt), PDO::PARAM_STR]];
        if ($site !== null) {
            $sql .= ' AND site = ?';
            $binds[] = [$site, PDO::PARAM_STR];
        }
        $sql .= ' GROUP BY camera_id, event_type HAVING fp > 0 ORDER BY fp DESC';
        $st = $this->pdo->prepare($sql);
        $pos = 1;
        foreach ($binds as [$val, $type]) $st->bindValue($pos++, $val, $type);
        $st->execute();
        return array_map(static fn ($r) => [
            'camera_id' => $r['camera_id'],
            'event_type' => $r['event_type'],
            'total' => (int)$r['total'],
            'fp' => (int)$r['fp'],
        ], $st->fetchAll());
    }

    public function reportSummary(float $sinceOccurredAt, ?string $site, int $topCameras,
                                  ?float $untilOccurredAt = null): array
    {
        $sinceStr = sprintf('%.6f', $sinceOccurredAt);
        // Optional upper bound so the SAME aggregation can run over the preceding window
        // (period-over-period trend) without duplicating any of these queries.
        $until = $untilOccurredAt !== null ? ' AND occurred_at < ?' : '';
        $untilBind = $untilOccurredAt !== null ? [sprintf('%.6f', $untilOccurredAt)] : [];
        $scope = $site !== null ? ' AND site = ?' : '';
        $scopeBind = $site !== null ? [$site] : [];

        $byType = $this->pdo->prepare(
            "SELECT event_type, severity, COUNT(*) AS count,
                    SUM(acknowledged_by IS NULL) AS unacked,
                    SUM(false_positive_by IS NOT NULL) AS fp
               FROM move_events WHERE occurred_at >= ?$until$scope
           GROUP BY event_type, severity"
        );
        $byType->execute([$sinceStr, ...$untilBind, ...$scopeBind]);

        $byCamera = $this->pdo->prepare(
            "SELECT camera_id, COUNT(*) AS count,
                    SUM(severity IN ('critical','high')) AS critical_high
               FROM move_events WHERE occurred_at >= ?$until$scope
           GROUP BY camera_id ORDER BY critical_high DESC, count DESC LIMIT ?"
        );
        $pos = 1;
        $byCamera->bindValue($pos++, $sinceStr, PDO::PARAM_STR);
        foreach ($untilBind as $v) $byCamera->bindValue($pos++, $v, PDO::PARAM_STR);
        foreach ($scopeBind as $v) $byCamera->bindValue($pos++, $v, PDO::PARAM_STR);
        $byCamera->bindValue($pos, $topCameras, PDO::PARAM_INT);
        $byCamera->execute();

        // Response time counts only real emergencies someone actually confirmed: critical/high,
        // acknowledged, not FP-tagged (dismissing a false alarm is not an emergency response).
        // acknowledged_at >= occurred_at guards clock-skewed rows from producing negative TTA.
        $tta = $this->pdo->prepare(
            "SELECT COUNT(*) AS acked,
                    AVG(acknowledged_at - occurred_at) AS avg_s,
                    MAX(acknowledged_at - occurred_at) AS max_s
               FROM move_events
              WHERE occurred_at >= ?$until$scope AND severity IN ('critical','high')
                AND acknowledged_at IS NOT NULL AND false_positive_by IS NULL
                AND acknowledged_at >= occurred_at"
        );
        $tta->execute([$sinceStr, ...$untilBind, ...$scopeBind]);
        $ttaRow = $tta->fetch() ?: ['acked' => 0, 'avg_s' => null, 'max_s' => null];

        $ergo = $this->pdo->prepare(
            "SELECT metrics_json FROM move_events
              WHERE occurred_at >= ?$until$scope AND event_type = 'ergo_report'
           ORDER BY occurred_at DESC LIMIT 2000"
        );
        $ergo->execute([$sinceStr, ...$untilBind, ...$scopeBind]);

        return [
            'by_type' => array_map(static fn ($r) => [
                'event_type' => $r['event_type'], 'severity' => $r['severity'],
                'count' => (int)$r['count'], 'unacked' => (int)$r['unacked'], 'fp' => (int)$r['fp'],
            ], $byType->fetchAll()),
            'by_camera' => array_map(static fn ($r) => [
                'camera_id' => $r['camera_id'], 'count' => (int)$r['count'],
                'critical_high' => (int)$r['critical_high'],
            ], $byCamera->fetchAll()),
            'tta' => [
                'acked' => (int)$ttaRow['acked'],
                'avg_s' => $ttaRow['avg_s'] !== null ? (float)$ttaRow['avg_s'] : null,
                'max_s' => $ttaRow['max_s'] !== null ? (float)$ttaRow['max_s'] : null,
            ],
            'ergo_metrics' => array_map(
                static fn ($r) => json_decode((string)$r['metrics_json'], true) ?: [],
                $ergo->fetchAll()
            ),
        ];
    }

    public function occurrenceTimes(float $sinceOccurredAt, ?string $site, int $limit): array
    {
        $sql = "SELECT occurred_at, event_type FROM move_events
                 WHERE occurred_at >= ? AND event_type <> 'ergo_report'";
        $binds = [[sprintf('%.6f', $sinceOccurredAt), PDO::PARAM_STR]];
        if ($site !== null) {
            $sql .= ' AND site = ?';
            $binds[] = [$site, PDO::PARAM_STR];
        }
        $sql .= ' ORDER BY occurred_at DESC LIMIT ?';
        $binds[] = [$limit, PDO::PARAM_INT];
        $st = $this->pdo->prepare($sql);
        $pos = 1;
        foreach ($binds as [$val, $type]) $st->bindValue($pos++, $val, $type);
        $st->execute();
        return array_map(static fn ($r) => [
            'occurred_at' => (float)$r['occurred_at'],
            'event_type' => $r['event_type'],
        ], $st->fetchAll());
    }

    public function exportRows(float $sinceOccurredAt, ?string $site, int $limit): array
    {
        $sql = 'SELECT ' . self::SELECT_COLUMNS . ' FROM move_events WHERE occurred_at >= ?';
        $binds = [[sprintf('%.6f', $sinceOccurredAt), PDO::PARAM_STR]];
        if ($site !== null) {
            $sql .= ' AND site = ?';
            $binds[] = [$site, PDO::PARAM_STR];
        }
        $sql .= ' ORDER BY occurred_at DESC, event_id DESC LIMIT ?';
        $binds[] = [$limit, PDO::PARAM_INT];
        $st = $this->pdo->prepare($sql);
        $pos = 1;
        foreach ($binds as [$val, $type]) $st->bindValue($pos++, $val, $type);
        $st->execute();
        return array_map(self::mapRow(...), $st->fetchAll());
    }

    public function forNotification(float $sinceOccurredAt, int $limit): array
    {
        // No site filter: the dispatcher is a server-side job, not a logged-in operator, and it
        // must be able to notify every site the installation serves. Channel rows carry their
        // own site, which is where the scoping actually belongs.
        $sql = 'SELECT ' . self::SELECT_COLUMNS . ' FROM move_events
                 WHERE occurred_at >= ? ORDER BY occurred_at ASC, event_id ASC LIMIT ?';
        $st = $this->pdo->prepare($sql);
        $st->bindValue(1, sprintf('%.6f', $sinceOccurredAt), PDO::PARAM_STR);
        $st->bindValue(2, $limit, PDO::PARAM_INT);
        $st->execute();
        return array_map(self::mapRow(...), $st->fetchAll());
    }

    public function siteSummary(float $sinceOccurredAt, ?float $untilOccurredAt = null): array
    {
        // COUNT(DISTINCT camera_id) alongside the totals: "three sites, one of which is watching
        // with a single camera" is a different picture from "three comparable sites", and the
        // head-office view is exactly where that distinction gets lost.
        $sql = "SELECT site,
                       COUNT(*) AS count,
                       SUM(CASE WHEN severity IN ('critical','high') THEN 1 ELSE 0 END) AS critical_high,
                       SUM(CASE WHEN acknowledged_by IS NULL THEN 1 ELSE 0 END) AS unacked,
                       COUNT(DISTINCT camera_id) AS cameras
                  FROM move_events
                 WHERE occurred_at >= ? AND event_type <> 'ergo_report'";
        $binds = [sprintf('%.6f', $sinceOccurredAt)];
        if ($untilOccurredAt !== null) {
            $sql .= ' AND occurred_at < ?';
            $binds[] = sprintf('%.6f', $untilOccurredAt);
        }
        $sql .= ' GROUP BY site ORDER BY critical_high DESC, count DESC';
        $st = $this->pdo->prepare($sql);
        $st->execute($binds);
        return array_map(static fn (array $r): array => [
            'site' => $r['site'] !== null ? (string)$r['site'] : null,
            'count' => (int)$r['count'],
            'critical_high' => (int)$r['critical_high'],
            'unacked' => (int)$r['unacked'],
            'cameras' => (int)$r['cameras'],
        ], $st->fetchAll());
    }

    public function windowForCamera(string $cameraId, float $from, float $to, ?string $site, int $limit): array
    {
        $sql = 'SELECT ' . self::SELECT_COLUMNS . ' FROM move_events
                 WHERE camera_id = ? AND occurred_at >= ? AND occurred_at < ?';
        $binds = [[$cameraId, PDO::PARAM_STR], [sprintf('%.6f', $from), PDO::PARAM_STR],
                  [sprintf('%.6f', $to), PDO::PARAM_STR]];
        if ($site !== null) {
            $sql .= ' AND site = ?';
            $binds[] = [$site, PDO::PARAM_STR];
        }
        $sql .= ' ORDER BY occurred_at ASC, event_id ASC LIMIT ?';
        $binds[] = [$limit, PDO::PARAM_INT];
        $st = $this->pdo->prepare($sql);
        $pos = 1;
        foreach ($binds as [$val, $type]) $st->bindValue($pos++, $val, $type);
        $st->execute();
        return array_map(self::mapRow(...), $st->fetchAll());
    }

    public function countForCamera(string $cameraId, float $from, float $to, ?string $site): int
    {
        $sql = "SELECT COUNT(*) FROM move_events
                 WHERE camera_id = ? AND occurred_at >= ? AND occurred_at < ?
                   AND event_type <> 'ergo_report'";
        $binds = [$cameraId, sprintf('%.6f', $from), sprintf('%.6f', $to)];
        if ($site !== null) {
            $sql .= ' AND site = ?';
            $binds[] = $site;
        }
        $st = $this->pdo->prepare($sql);
        $st->execute($binds);
        return (int)$st->fetchColumn();
    }

    public function purgeOlderThan(float $cutoff, int $limit = 500): int
    {
        // Bounded batch, same shape as the snapshot purge: a shared-hosting request must not
        // stall deleting a year of rows. ON DELETE CASCADE on move_event_snapshots would drop
        // snapshot ROWS with the event, so purge only what the snapshot purge has already
        // emptied of files — cutoff is expected to be far older than the image retention window.
        $sel = $this->pdo->prepare(
            'SELECT event_id FROM move_events WHERE occurred_at < ? ORDER BY event_id LIMIT ?');
        $sel->bindValue(1, sprintf('%.6f', $cutoff), PDO::PARAM_STR);
        $sel->bindValue(2, $limit, PDO::PARAM_INT);
        $sel->execute();
        $ids = array_map('intval', $sel->fetchAll(PDO::FETCH_COLUMN));
        if (!$ids) {
            return 0;
        }
        $in = implode(',', array_fill(0, count($ids), '?'));
        $this->pdo->prepare("DELETE FROM move_events WHERE event_id IN ($in)")->execute($ids);
        return count($ids);
    }

    public function cameraExists(string $cameraId): bool
    {
        $st = $this->pdo->prepare('SELECT 1 FROM move_events WHERE camera_id = ? LIMIT 1');
        $st->execute([$cameraId]);
        return $st->fetchColumn() !== false;
    }

    public function siteForCamera(string $cameraId): ?string
    {
        // Newest event wins: if a camera is re-homed to another site, later actions belong to
        // the site it reports to NOW. event_id DESC (not occurred_at) because ingest order is
        // the authoritative "most recently seen" — occurred_at comes from the edge clock.
        $st = $this->pdo->prepare(
            'SELECT site FROM move_events WHERE camera_id = ? AND site IS NOT NULL
              ORDER BY event_id DESC LIMIT 1'
        );
        $st->execute([$cameraId]);
        $site = $st->fetchColumn();
        return $site === false ? null : (string)$site;
    }

    private static function mapRow(array $row): array
    {
        return [
            'event_id' => (int)$row['event_id'],
            'event_uuid' => $row['event_uuid'] ?? null,
            'device_id' => $row['device_id'],
            'site' => $row['site'] ?? null,
            'camera_id' => $row['camera_id'],
            'event_type' => $row['event_type'],
            'severity' => $row['severity'],
            'track_id' => $row['track_id'] !== null ? (int)$row['track_id'] : null,
            'zone_id' => $row['zone_id'],
            'occurred_at' => (float)$row['occurred_at'],
            'duration_s' => (float)$row['duration_s'],
            'metrics' => json_decode((string)($row['metrics_json'] ?? '[]'), true) ?? [],
            'spatial_x' => $row['spatial_x'] !== null ? (float)$row['spatial_x'] : null,
            'spatial_y' => $row['spatial_y'] !== null ? (float)$row['spatial_y'] : null,
            'false_positive_by' => $row['false_positive_by'] ?? null,
            'false_positive_at' => ($row['false_positive_at'] ?? null) !== null ? (int)$row['false_positive_at'] : null,
            'acknowledged_by' => $row['acknowledged_by'],
            'acknowledged_at' => $row['acknowledged_at'] !== null ? (int)$row['acknowledged_at'] : null,
        ];
    }
}