presseportale/app/Console/Commands/AggregatePressReleaseViews.php

<?php

namespace App\Console\Commands;

use App\Enums\PressReleaseEventType;
use App\Models\AppSetting;
use App\Models\PressRelease;
use App\Models\PressReleaseEvent;
use App\Scopes\PortalScope;
use Illuminate\Console\Command;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use RuntimeException;

/**
 * Schreibt die gezählten Aufrufe aus dem Ereignis-Log in den Gesamtzähler
 * `press_releases.hits`.
 *
 * Hintergrund: Früher lief pro Seitenaufruf ein `UPDATE press_releases SET
 * hits = hits + 1` im Request. Auf der über 1 GB großen Tabelle blockiert
 * dieser Schreibzugriff, sobald eine Schemaänderung läuft (FULLTEXT-Indizes
 * erzwingen dort einen COPY-Rebuild) — die wartenden Requests haben dann
 * PHP-FPM-Worker und MySQL-Verbindungen belegt, bis die ganze Seite stand.
 * Der Request schreibt jetzt nur noch das Ereignis in die kleine Log-Tabelle;
 * die Summierung passiert hier gebündelt.
 *
 * Fortschritt läuft über eine Wasserstandsmarke (höchste bereits verarbeitete
 * Event-ID) statt über einen Zwischenspeicher: Das Log ist die dauerhafte
 * Quelle, dadurch geht bei einem Neustart kein Aufruf verloren und ein
 * doppelter Lauf zählt nichts doppelt. Zähler und Marke werden je Durchgang
 * gemeinsam committet — sonst wäre nur der Durchgang selbst, nicht der
 * einzelne Abbruch darin, wiederanlaufsicher.
 */
class AggregatePressReleaseViews extends Command
{
    protected $signature = 'press-releases:aggregate-views
                            {--chunk=5000 : Ereignisse je Durchgang}
                            {--budget=480 : Zeitbudget in Sekunden; der Rest wartet auf den nächsten Lauf}';

    protected $description = 'Überträgt gezählte Aufrufe aus dem Ereignis-Log in press_releases.hits.';

    /**
     * Wasserstandsmarke: höchste bereits in `hits` verbuchte Event-ID.
     */
    public const WatermarkKey = 'press_release_views.aggregated_through_id';

    /**
     * Cache-Lock über alle Einstiege (Scheduler UND manueller Aufruf): Zwei
     * gleichzeitige Läufe lesen dieselbe Marke und verbuchen denselben Chunk
     * doppelt (Audit F19). withoutOverlapping() schützt nur den Scheduler.
     */
    public const LockKey = 'press-releases:aggregate-views';

    /**
     * Der Lock verfällt nach dieser Zeit. Das Zeitbudget (--budget) muss
     * darunter liegen, sonst kann ein zweiter Lauf den Lock übernehmen,
     * während der erste noch verbucht (Audit U13). Als letzte Schranke wird
     * die Marke nur unter Zeilensperre und nur vom erwarteten Stand aus
     * fortgeschrieben – ein fremder Stand bricht den Durchgang ab.
     */
    public const LockSeconds = 600;

    public function handle(): int
    {
        $lock = Cache::lock(self::LockKey, self::LockSeconds);

        if (! $lock->get()) {
            $this->warn('Ein anderer Aggregationslauf ist noch aktiv – dieser Aufruf tut nichts.');

            return self::SUCCESS;
        }

        try {
            return $this->aggregate();
        } finally {
            $lock->release();
        }
    }

    private function aggregate(): int
    {
        $chunk = max(100, (int) $this->option('chunk'));
        $budget = min(self::LockSeconds - 60, max(0, (int) $this->option('budget')));
        $startedAt = microtime(true);
        $watermark = AppSetting::getValue(self::WatermarkKey);

        // Fehlende Marke heißt: noch nie initialisiert. Sie jetzt als 0 zu
        // lesen wäre fatal — `press_releases.hits` enthält den Bestand bereits,
        // ein Lauf ab 0 würde jedes Alt-Ereignis ein zweites Mal aufaddieren.
        // Das kann im Deploy-Fenster passieren, wenn der Scheduler den neuen
        // Task startet, bevor die Migration gelaufen ist. Also setzen wir die
        // Marke auf den Ist-Stand und zählen erst ab dem nächsten Aufruf.
        if ($watermark === null) {
            $latest = (int) PressReleaseEvent::query()->max('id');
            AppSetting::setValue(self::WatermarkKey, $latest);

            $this->info("Wasserstandsmarke erstmalig auf Event-ID {$latest} gesetzt — der Bestand steckt bereits in `hits`.");

            return self::SUCCESS;
        }

        $watermark = (int) $watermark;

        $totalEvents = 0;
        $totalReleases = 0;

        while (true) {
            if (microtime(true) - $startedAt > $budget) {
                $this->warn(sprintf('Zeitbudget von %d s erreicht – der Rest folgt im nächsten Lauf.', $budget));

                break;
            }

            // Obergrenze des Durchgangs zuerst festlegen, damit gleichzeitig
            // eintreffende Aufrufe die Summe nicht gegen die Marke verschieben.
            // Bewusst über die IDs statt über max(): eine Aggregatfunktion
            // ignoriert das limit und lieferte die Marke des GESAMTEN Rests.
            $ids = PressReleaseEvent::query()
                ->where('id', '>', $watermark)
                ->orderBy('id')
                ->limit($chunk)
                ->pluck('id');

            if ($ids->isEmpty()) {
                break;
            }

            $upperBound = (int) $ids->last();

            $counts = PressReleaseEvent::query()
                ->where('id', '>', $watermark)
                ->where('id', '<=', $upperBound)
                ->where('type', PressReleaseEventType::View->value)
                ->selectRaw('press_release_id, count(*) as views')
                ->groupBy('press_release_id')
                ->pluck('views', 'press_release_id');

            // Zähler und Marke gehören in EINE Transaktion. Ohne sie wäre der
            // Lauf nur zwischen den Durchgängen wiederanlaufsicher: Bricht der
            // Prozess mitten im Chunk ab (OOM, Deploy, kill), sind ein Teil der
            // Zähler schon erhöht, die Marke steht aber noch auf dem alten
            // Stand — der nächste Lauf verbuchte genau diese Aufrufe ein
            // zweites Mal.
            try {
                $this->bookChunk($counts, $watermark, $upperBound, $totalEvents, $totalReleases);
            } catch (RuntimeException $exception) {
                report($exception);
                $this->error($exception->getMessage());

                return self::FAILURE;
            }

            $watermark = $upperBound;
        }

        $this->info("Aufrufe verbucht: {$totalEvents} auf {$totalReleases} Pressemitteilungen (Stand Event-ID {$watermark}).");

        return self::SUCCESS;
    }

    /**
     * Zähler und Marke in EINER Transaktion; siehe Kommentar im Aufrufer.
     *
     * @param  Collection<int|string, int|string>  $counts
     */
    private function bookChunk(Collection $counts, int $watermark, int $upperBound, int &$totalEvents, int &$totalReleases): void
    {
        DB::transaction(function () use ($counts, $watermark, $upperBound, &$totalEvents, &$totalReleases): void {
            // Marke unter Zeilensperre gegen den erwarteten Stand prüfen:
            // hat ein anderer Lauf sie inzwischen verschoben (Lock
            // verfallen), würde dieser Durchgang dieselben Ereignisse ein
            // zweites Mal verbuchen (Audit U13).
            $current = AppSetting::query()->whereKey(self::WatermarkKey)->lockForUpdate()->first();

            if ($current === null || (int) $current->value !== $watermark) {
                throw new RuntimeException(sprintf(
                    'Wasserstandsmarke steht auf %s statt %d – ein anderer Lauf hat übernommen, Durchgang verworfen.',
                    $current === null ? 'null' : (string) $current->value,
                    $watermark,
                ));
            }

            foreach ($counts as $pressReleaseId => $views) {
                PressRelease::query()
                    ->withoutGlobalScope(PortalScope::class)
                    ->whereKey((int) $pressReleaseId)
                    ->increment('hits', (int) $views);

                $totalReleases++;
                $totalEvents += (int) $views;
            }

            AppSetting::setValue(self::WatermarkKey, $upperBound);
        });
    }
}