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\Facades\Cache;
use Illuminate\Support\Facades\DB;

/**
 * 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}';

    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';

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

        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'));
        $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) {
            // 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.
            DB::transaction(function () use ($counts, $upperBound, &$totalEvents, &$totalReleases): void {
                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);
            });

            $watermark = $upperBound;
        }

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

        return self::SUCCESS;
    }
}