presseportale/app/Services/Archive/ArchivePendingQueue.php

<?php

namespace App\Services\Archive;

use App\Enums\ArchiveEvent;
use App\Enums\ArchiveReasonCategory;
use App\Models\ArchiveEntry;
use App\Models\ArchivePendingJob;
use App\Models\PressReleaseImage;
use Illuminate\Database\Eloquent\Model;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Str;
use Throwable;

/**
 * Aufträge für ausstehende Beweisarchiv-Einträge (Umsetzungsaudit
 * 29.09.2026, A-02/A-04/A-09/A-10).
 *
 * Ablauf für jeden archivpflichtigen Vorgang:
 *
 *     $job = DB::transaction(function () use (…) {
 *         // fachliche Änderung …
 *         return $queue->push($pressReleaseId, ArchiveEvent::LabelCorrected, $correction,
 *             ArchiveReasonCategory::Correction, actorId: $admin->id);
 *     });
 *     $entry = $queue->completeWith($job, fn () => $archiver->labelCorrected(…));
 *
 * push() gehört in die fachliche Transaktion: Auftrag und Marker
 * `press_releases.archive_pending_at` entstehen oder fallen mit dem
 * Vorgang. completeWith() läuft nach dem Commit, wirft nie und löscht den
 * Auftrag nur zusammen mit dem geschriebenen Eintrag. Scheitert die
 * Ablage, bleibt der Auftrag stehen und `archive:retry-pending` holt ihn
 * je Ereignistyp nach (PendingArchiveRetry).
 *
 * Der Marker ist nur noch ein Sammelanzeiger: gesetzt, wenn er NULL ist
 * (ein älterer Wert bleibt), gelöscht, wenn für die Meldung kein Auftrag
 * mehr offen ist — beides per Query Builder, `updated_at` bleibt
 * unberührt. Ein erfolgreicher späterer Vorgang kann einen älteren offenen
 * Auftrag damit weder überschreiben noch mitlöschen.
 */
class ArchivePendingQueue
{
    /**
     * Auftrag anlegen (innerhalb der fachlichen Transaktion aufrufen).
     *
     * @param  Model|null  $subject  Bezug des späteren Eintrags bzw. des Nachholers:
     *                               Meldung, Redaction-Snapshot, Bild oder Korrekturzeile
     * @param  array<string, mixed>  $payload  Was der Nachholer braucht (z. B. `state` bei
     *                                         `pm_depublished`, `before`/`after` bei `gdpr_change`)
     */
    public function push(
        int $pressReleaseId,
        ArchiveEvent $event,
        ?Model $subject = null,
        ?ArchiveReasonCategory $reason = null,
        ?int $legalRequestId = null,
        ?int $actorId = null,
        ?string $note = null,
        array $payload = [],
    ): ArchivePendingJob {
        return $this->create($pressReleaseId, $event->value, $subject, $reason, $legalRequestId, $actorId, $note, $payload);
    }

    /**
     * Sofortlöschung eines nie veröffentlichten Bildes ist gescheitert
     * (A-10): der Nachholer wiederholt sie, der Systemstatus zählt den Rest.
     */
    public function pushImagePurge(PressReleaseImage $image, ?string $error = null): ArchivePendingJob
    {
        $job = $this->create((int) $image->press_release_id, ArchivePendingJob::EVENT_IMAGE_PURGE, $image, null, null, null, 'Sofortlöschung fehlgeschlagen', []);

        if ($error !== null) {
            $this->recordFailure($job, $error);
        }

        return $job;
    }

    /**
     * Archiveintrag schreiben und den Auftrag im selben Schritt abschließen.
     * Wirft nie: ein Fehler oder ein nicht geschriebener Eintrag (fehlender
     * Schlüssel) zählt als Versuch, der Auftrag bleibt offen.
     *
     * @param  callable(): ?ArchiveEntry  $step
     */
    public function completeWith(ArchivePendingJob $job, callable $step): ?ArchiveEntry
    {
        $alreadyCompleted = false;

        try {
            $entry = DB::transaction(function () use ($job, $step, &$alreadyCompleted): ?ArchiveEntry {
                // Erst beanspruchen, dann archivieren (Umsetzungsaudit
                // 02.10.2026, O07): Nachholer und Direktweg dürfen denselben
                // Auftrag nicht beide ablegen.
                if ($this->claim($job) === null) {
                    $alreadyCompleted = true;

                    return null;
                }

                $entry = $step();

                if ($entry !== null) {
                    $this->complete($job);
                }

                return $entry;
            });
        } catch (Throwable $exception) {
            $this->recordFailure($job, $exception->getMessage());
            Log::error('Beweisarchiv: Eintrag konnte nicht geschrieben werden, Auftrag bleibt offen.', [
                'job_id' => $job->getKey(),
                'event' => $job->event,
                'press_release_id' => $job->press_release_id,
                'error' => $exception->getMessage(),
            ]);

            return null;
        }

        if ($entry === null && ! $alreadyCompleted) {
            $this->recordFailure($job, 'Eintrag wurde nicht archiviert (ARCHIVE_KEY fehlt oder ist ungültig).');
        }

        return $entry;
    }

    /**
     * Auftragszeile unter Sperre neu lesen — innerhalb der Transaktion, die
     * den Eintrag schreibt. Wer zweiter ist, wartet auf die Sperre und findet
     * den Auftrag danach nicht mehr (null): dann ist nichts mehr zu tun.
     */
    public function claim(ArchivePendingJob $job): ?ArchivePendingJob
    {
        return ArchivePendingJob::query()->whereKey($job->getKey())->lockForUpdate()->first();
    }

    /**
     * Auftrag löschen; den Sammelmarker nur leeren, wenn für die Meldung
     * kein weiterer Auftrag offen ist.
     */
    public function complete(ArchivePendingJob $job): void
    {
        ArchivePendingJob::query()->whereKey($job->getKey())->delete();

        DB::table('press_releases')
            ->where('id', $job->press_release_id)
            ->whereNotNull('archive_pending_at')
            ->whereNotExists(function ($query): void {
                $query->select(DB::raw(1))
                    ->from('archive_pending_jobs')
                    ->whereColumn('archive_pending_jobs.press_release_id', 'press_releases.id');
            })
            ->update(['archive_pending_at' => null]);
    }

    public function recordFailure(ArchivePendingJob $job, string $error): void
    {
        ArchivePendingJob::query()->whereKey($job->getKey())->update([
            'attempts' => DB::raw('attempts + 1'),
            'last_error' => Str::limit($error, 990, ''),
            'last_attempted_at' => now(),
        ]);
    }

    /**
     * Marker aus der Zeit vor den Aufträgen (Deploy 28.09.2026) in je einen
     * Auftrag `legacy_marker` überführen, damit sie nicht verloren gehen.
     */
    public function adoptLegacyMarkers(): int
    {
        $adopted = 0;

        DB::table('press_releases')
            ->whereNotNull('archive_pending_at')
            ->whereNotExists(function ($query): void {
                $query->select(DB::raw(1))
                    ->from('archive_pending_jobs')
                    ->whereColumn('archive_pending_jobs.press_release_id', 'press_releases.id');
            })
            ->select(['id'])
            ->lazyById(200)
            ->each(function (object $row) use (&$adopted): void {
                $this->adoptLegacyMarker((int) $row->id);
                $adopted++;
            });

        return $adopted;
    }

    /**
     * Offene Aufträge plus Altmarker ohne Auftrag (Systemstatus).
     *
     * @return array{jobs: int, legacy_markers: int, oldest: ?Carbon}
     */
    public function status(): array
    {
        $oldest = ArchivePendingJob::query()->min('created_at');

        return [
            'jobs' => ArchivePendingJob::query()->count(),
            'legacy_markers' => DB::table('press_releases')
                ->whereNotNull('archive_pending_at')
                ->whereNotExists(function ($query): void {
                    $query->select(DB::raw(1))
                        ->from('archive_pending_jobs')
                        ->whereColumn('archive_pending_jobs.press_release_id', 'press_releases.id');
                })
                ->count(),
            'oldest' => $oldest !== null ? Carbon::parse($oldest) : null,
        ];
    }

    /**
     * @param  array<string, mixed>  $payload
     */
    private function create(int $pressReleaseId, string $event, ?Model $subject, ?ArchiveReasonCategory $reason, ?int $legalRequestId, ?int $actorId, ?string $note, array $payload): ArchivePendingJob
    {
        return DB::transaction(function () use ($pressReleaseId, $event, $subject, $reason, $legalRequestId, $actorId, $note, $payload): ArchivePendingJob {
            $this->adoptLegacyMarker($pressReleaseId);

            $job = ArchivePendingJob::query()->create([
                'press_release_id' => $pressReleaseId,
                'event' => $event,
                'subject_type' => $subject?->getMorphClass(),
                'subject_id' => $subject?->getKey(),
                'reason_category' => $reason,
                'legal_request_id' => $legalRequestId,
                'actor_id' => $actorId,
                'note' => $note !== null ? Str::limit($note, 490, '') : null,
                'payload' => $payload,
                'created_at' => now(),
            ]);

            DB::table('press_releases')
                ->where('id', $pressReleaseId)
                ->whereNull('archive_pending_at')
                ->update(['archive_pending_at' => now()]);

            return $job;
        });
    }

    /**
     * Steht ein Marker ohne Auftrag, stammt er aus der Zeit vor den
     * Aufträgen: als eigener Auftrag festhalten, bevor ein neuer Vorgang
     * ihn „übernimmt".
     */
    private function adoptLegacyMarker(int $pressReleaseId): void
    {
        $marker = DB::table('press_releases')->where('id', $pressReleaseId)->value('archive_pending_at');

        if ($marker === null || ArchivePendingJob::query()->where('press_release_id', $pressReleaseId)->exists()) {
            return;
        }

        ArchivePendingJob::query()->create([
            'press_release_id' => $pressReleaseId,
            'event' => ArchivePendingJob::EVENT_LEGACY_MARKER,
            'payload' => [],
            'created_at' => Carbon::parse($marker),
        ]);
    }
}