<?php
/**
 * @package     com_htprotect
 * @copyright   (C) 2026 Pascal Lohmann / website-bereinigung.de
 * @license     GNU General Public License version 2 or later
 *
 * HtpRegClean - Neuanmeldung: Einschaetzung + sicheres Aufraeumen von Spam-Konten (Connector 2.6.8).
 *
 * ZWEI TEILE:
 *  A) EINSCHAETZUNG (nicht destruktiv): liest den Registrierungs-Zustand von com_users (an/aus) sowie
 *     - gecacht mit TTL, weil sysfactsBundle() JEDEN Heartbeat laeuft - Kennzahlen ueber echte vs.
 *     spam-verdaechtige Anmeldungen (#__users). Die Verdict-Ampel (12-Monats-Fenster) bildet das
 *     DASHBOARD aus diesen Rohzahlen. Gemeldet ueber sysfacts.reg (an/aus, im Hash) + sysfacts.reg_stats.
 *
 *  B) AUFRAEUMEN (destruktiv, streng gegated): loescht NUR beweisbar inerte Bot-Konten in resumierbaren
 *     Batches (skaliert auf 500k-5M Zeilen). Gestaffelte Sicherheit:
 *       - STRENGE KRITERIEN: nie eingeloggt (lastvisitDate) UND alt genug (Karenz) UND KEIN backend-
 *         faehiges Konto (aus der echten ACL ermittelt, NIE per Gruppen-ID hartkodiert) UND kein Inhalts-
 *         Autor (created_by/modified_by der Kern-Tabellen) UND nicht auf der Whitelist des Betreibers.
 *       - FAIL-SAFE: laesst sich die ACL nicht sicher lesen ODER waere die Schutzmenge leer (Admins nicht
 *         bestimmbar) -> Abbruch, NIE loeschen.
 *       - HARTES GATE: ist eine Mitglieder-/Shop-/LMS-/Forum-KOMPONENTE aktiv ODER ein externes Auth-/
 *         Provisionierungs-Plugin (LDAP/SAML/SSO/OAuth), koennen nie-eingeloggte Konten echt (extern
 *         verwaltet / vorprovisioniert) sein -> Bulk-Loeschen wird verweigert.
 *       - VORSCHAU + BESTAETIGUNG: preview liefert Anzahl + maskierte Beispiele + Domain-Konzentration +
 *         Kriterien-Hash; der Lauf verlangt PFLICHTIG den Hash zurueck (confirm) UND die erwartete Anzahl
 *         (expected, Count-Lock gegen Drift) - fehlt eines, wird verweigert.
 *       - UNDO-JOURNAL (default an, abschaltbar): vor dem Loeschen wird JEDE geloeschte Zeile (aus ALLEN
 *         betroffenen Tabellen) in kleine gzip-CHUNK-Dateien je Batch gestreamt -> vollstaendige, O(1)-
 *         resumierbare Wiederherstellung. Ist das Journal gewuenscht, aber nicht schreibbar -> Abbruch
 *         (kein Loeschen ohne das versprochene Netz). 30 Tage Auto-Purge; schlanker Meta-Eintrag (Anzahl/
 *         Zeit/Hash, KEINE PII) bleibt fuers Audit.
 *       - ABBRUCH: reg.cleanup.cancel setzt ein Flag, das die Engine pro Slice prueft.
 *
 * Ausschliesslich ueber die signierten Befehle reg.set / reg.cleanup / reg.cleanup.cancel aufgerufen
 * (HtpLink::dispatch). Der Batch-Lauf reitet auf dem gemeinsamen Job-Motor (HtpLink::advanceJob ->
 * HtpRegClean::advance): Wall-Clock-Budget je Tick, resumierbarer Cursor im $st['job'], flock-Single-Run,
 * FFR-detached, Heartbeat-Fortschritt. KEINE Transaktionen - Robustheit kommt aus dem persistierten
 * Cursor (Loeschen ist idempotent: schon geloescht = matcht nicht erneut).
 */

defined('_JEXEC') or die;

class HtpRegClean
{
    /** Einschaetzungs-Fenster in Monaten (Dashboard-Verdict). 12 = deckt Jahres-Events ab. */
    const WINDOW_MONTHS = 12;

    /** Cache-Lebensdauer der (teureren) #__users-Zaehler. sysfactsBundle laeuft jeden Heartbeat -> die
     *  Zaehler duerfen NICHT jedes Mal ueber eine 560k-Tabelle rechnen. 12h. */
    const STATS_TTL = 43200;

    /** Loesch-Kandidat muss AELTER sein als das (Karenz gegen frische, evtl. echte Anmeldung). */
    const GRACE_DAYS = 90;

    /** Schwelle "nie eingeloggt": lastvisitDate <= diese Marke (Joomla-Null 0000-00-00 bzw. Epoch). */
    const NEVER_LOGIN = '1970-01-02 00:00:00';

    /** Zeilen pro Loesch-Batch (keyset per Primaerschluessel -> indizierter Range-Scan, Sub-Sekunde).
     *  = eine Journal-Chunk-Datei je Batch (klein genug, dass die Wiederherstellung sie je Slice
     *  komplett zuruecklegt -> kein Re-Skip, kein O(n^2)). */
    const DEL_BATCH = 2000;

    /** Wall-Clock je advance()-Scheibe: bewusst KURZ (nicht das volle JOB_BUDGET des Treibers). Der
     *  Treiber speichert den Cursor + heartbeatet den Fortschritt ZWISCHEN den Scheiben - so ist der
     *  Fortschritt sichtbar UND uebersteht einen Prozess-Abschuss (Hoster killt lange Requests): ein Kill
     *  verliert hoechstens diese ~2 s (idempotent nachgeholt), statt den ganzen Lauf. */
    const SLICE_BUDGET = 2.0;

    /** Freier Plattenplatz, unter den das Journal NIE schreibt: lieber fail-closed stoppen (mit intaktem
     *  Cursor -> spaeter fortsetzbar), als die Platte des Hosts vollaufen zu lassen. 200 MB Sicherheits-
     *  Boden. Bei riesigen Aufraeumungen kann der Betreiber ohne Journal weitermachen (dann kein Platz-
     *  bedarf) - die Wahl liegt bei ihm. */
    const MIN_FREE_BYTES = 209715200; // 200 * 1024 * 1024

    /** Beispiel-Konten in der Vorschau (maskiert). Klein halten: das Ergebnis reist im log-Feld
     *  (2000-Zeichen-Deckel bei reportResult). */
    const SAMPLE_MAX = 15;

    /** Mehr geschuetzte/echte Konten als das = definitiv KEINE Spam-Flut -> Bulk-Loeschen verweigern. */
    const MAX_PROTECTED = 50000;

    /** Journal-Aufbewahrung (dann Auto-Purge). 30 Tage. */
    const JOURNAL_TTL = 2592000;

    /** Count-Lock-Toleranz (Vorschau -> Lauf): groesser von Prozent bzw. Absolutwert. */
    const COUNT_TOL_PCT = 5;
    const COUNT_TOL_ABS = 50;

    /**
     * Abhaengige KOMPONENTEN: ist EINE davon aktiv (enabled), ist eine offene Registrierung plausibel
     * ECHT genutzt (Mitglieder/Shop/Kurs/Forum) -> HARTES Gate.
     */
    private static function depComponents()
    {
        return array(
            'com_comprofiler', 'com_community', 'com_easysocial', 'com_jomsocial', 'com_socialcommunity',
            'com_membershippro', 'com_osmembership', 'com_payplans', 'com_akeebasubs', 'com_jgive',
            'com_rsmembershippro', 'com_virtuemart', 'com_hikashop', 'com_j2store', 'com_redshop',
            'com_phocacart', 'com_joomlalms', 'com_guru', 'com_kunena', 'com_easydiscuss',
        );
    }

    /** Kern-Authentifizierungs-Plugins, die NICHT auf externe Provisionierung hindeuten. Jedes ANDERE
     *  aktive authentication-Plugin (LDAP/SAML/OAuth/CAS ...) heisst: Konten koennen extern verwaltet
     *  sein und melden sich nie in DIESER Joomla an -> wie eine Abhaengigkeit behandeln (Gate). */
    private static function coreAuthPlugins()
    {
        return array('joomla', 'cookie');
    }

    /** Tabellen, die beim Wiederherstellen zugelassen sind (Journal koennte lokal manipuliert sein). */
    private static function restoreTables()
    {
        return array(
            '#__users' => 1, '#__user_usergroup_map' => 1, '#__user_keys' => 1, '#__user_profiles' => 1,
            '#__user_notes' => 1, '#__user_mfa' => 1, '#__messages' => 1, '#__session' => 1, '#__action_logs' => 1,
        );
    }

    private static function db()
    {
        return function_exists('htp_db') ? htp_db() : null;
    }

    /** Die user-verketteten Kern-Tabellen als [Tabelle, WHERE-Fragment] fuer die gegebene ID-Liste.
     *  Genutzt fuer Journal-LESEN und LOESCHEN (identische Menge -> vollstaendige Undo). Jede wird
     *  einzeln try-gekapselt behandelt; fehlende (versions-/extension-abhaengige) werden uebersprungen. */
    private static function delTargets($db, $inList)
    {
        return array(
            array('#__user_usergroup_map', $db->quoteName('user_id') . ' IN (' . $inList . ')'),
            array('#__user_keys',          $db->quoteName('user_id') . ' IN (' . $inList . ')'),
            array('#__user_profiles',      $db->quoteName('user_id') . ' IN (' . $inList . ')'),
            array('#__user_notes',         $db->quoteName('user_id') . ' IN (' . $inList . ')'),
            array('#__user_mfa',           $db->quoteName('user_id') . ' IN (' . $inList . ')'),
            array('#__messages',           $db->quoteName('user_id_from') . ' IN (' . $inList . ') OR ' . $db->quoteName('user_id_to') . ' IN (' . $inList . ')'),
            array('#__session',            $db->quoteName('userid') . ' IN (' . $inList . ')'),
            array('#__action_logs',        $db->quoteName('user_id') . ' IN (' . $inList . ')'),
        );
    }

    // ─────────────────────────────────────────────────────────────────────────
    //  TEIL A - Zustand + Kennzahlen
    // ─────────────────────────────────────────────────────────────────────────

    /**
     * Registrierung an (1) / aus (0) / nicht ermittelbar (null). Primaer der GESPEICHERTE Param aus
     * #__extensions (autoritativ, kein In-Request-Cache -> spiegelt reg.set sofort); fehlt der Schluessel
     * (nie gespeichert), der EFFEKTIVE Default via ComponentHelper (config.xml-Vorgabe).
     */
    public static function state()
    {
        $raw = self::rawComUsersParams();

        if (is_array($raw) && array_key_exists('allowUserRegistration', $raw)) {
            return ((int) $raw['allowUserRegistration'] === 1) ? 1 : 0;
        }

        try {
            $p = null;
            if (class_exists('Joomla\\CMS\\Component\\ComponentHelper')) {
                $p = \Joomla\CMS\Component\ComponentHelper::getParams('com_users');
            } elseif (class_exists('JComponentHelper')) {
                $p = JComponentHelper::getParams('com_users');
            }
            if (is_object($p) && method_exists($p, 'get')) {
                $v = $p->get('allowUserRegistration', null);
                if ($v !== null) {
                    return ((int) $v === 1) ? 1 : 0;
                }
            }
        } catch (Exception $e) {
        } catch (Throwable $e) {
        }

        if ($raw === null) {
            return null;
        }
        return 0;
    }

    /** Rohe com_users-Params aus #__extensions als Array (null = nicht lesbar). */
    private static function rawComUsersParams()
    {
        try {
            $db = self::db();
            if (!$db) {
                return null;
            }
            $q = $db->getQuery(true)
                ->select($db->quoteName('params'))
                ->from($db->quoteName('#__extensions'))
                ->where($db->quoteName('type') . ' = ' . $db->quote('component'))
                ->where($db->quoteName('element') . ' = ' . $db->quote('com_users'));
            $db->setQuery($q);
            $rawv = $db->loadResult();
            if ($rawv === null) {
                return null;
            }
            if ((string) $rawv === '') {
                return array();
            }
            $p = json_decode((string) $rawv, true);
            return is_array($p) ? $p : array();
        } catch (Exception $e) {
            return null;
        } catch (Throwable $e) {
            return null;
        }
    }

    /**
     * reg.set: allowUserRegistration = 1/0 in com_users setzen (read-modify-write in #__extensions,
     * versionsneutral wie setJoomlaAutoUpdate). true = ok.
     */
    public static function setRegistration($on)
    {
        try {
            $db = self::db();
            if (!$db) {
                return false;
            }
            $q = $db->getQuery(true)
                ->select($db->quoteName('params'))
                ->from($db->quoteName('#__extensions'))
                ->where($db->quoteName('type') . ' = ' . $db->quote('component'))
                ->where($db->quoteName('element') . ' = ' . $db->quote('com_users'));
            $db->setQuery($q);
            $rawv = $db->loadResult();
            if ($rawv === null) {
                return false;
            }
            $p = ((string) $rawv !== '') ? json_decode((string) $rawv, true) : array();
            if (!is_array($p)) {
                $p = array();
            }
            $p['allowUserRegistration'] = $on ? 1 : 0;

            $u = $db->getQuery(true)
                ->update($db->quoteName('#__extensions'))
                ->set($db->quoteName('params') . ' = ' . $db->quote((string) json_encode($p)))
                ->where($db->quoteName('type') . ' = ' . $db->quote('component'))
                ->where($db->quoteName('element') . ' = ' . $db->quote('com_users'));
            $db->setQuery($u);
            $db->execute();
            return true;
        } catch (Exception $e) {
            return false;
        } catch (Throwable $e) {
            return false;
        }
    }

    /**
     * Gecachte Kennzahlen (TTL) fuers Dashboard-Verdict. Liest den Cache aus der Config; ist er abgelaufen
     * (oder fehlt), einmalig neu rechnen + zuruecklegen. Rechnet NIE bei jedem Heartbeat. Gibt bei
     * Rechen-Fehlschlag den alten Cache zurueck (kann null sein).
     *
     * @return array|null
     */
    public static function stats($cfg)
    {
        $c   = (is_array($cfg) && isset($cfg['reg_stats']) && is_array($cfg['reg_stats'])) ? $cfg['reg_stats'] : null;
        $now = time();

        if ($c !== null && isset($c['at']) && ($now - (int) $c['at']) < self::STATS_TTL) {
            return $c;
        }

        $fresh = self::computeStats();
        if ($fresh === null) {
            return $c;
        }
        $fresh['at'] = $now;

        try {
            if (class_exists('HtpConfig')) {
                HtpConfig::saveKeys(array('reg_stats' => $fresh));
            }
        } catch (Exception $e) {
        } catch (Throwable $e) {
        }
        return $fresh;
    }

    /** Cache verwerfen -> naechster Heartbeat rechnet neu (nach einem Aufraeumen, das die Zahlen aendert). */
    public static function invalidateStats()
    {
        try {
            if (class_exists('HtpConfig')) {
                HtpConfig::saveKeys(array('reg_stats' => null));
            }
        } catch (Exception $e) {
        } catch (Throwable $e) {
        }
    }

    /** Die #__users-Kennzahlen frisch rechnen (heavy - nur ueber stats() mit TTL). null = Fehler. */
    private static function computeStats()
    {
        $db = self::db();
        if (!$db) {
            return null;
        }
        try {
            $now      = time();
            $winCut   = gmdate('Y-m-d H:i:s', $now - (self::WINDOW_MONTHS * 30 + 5) * 86400);
            $graceCut = gmdate('Y-m-d H:i:s', $now - self::GRACE_DAYS * 86400);
            $qUsers   = $db->quoteName('#__users');
            $lv       = $db->quoteName('lastvisitDate');
            $rd       = $db->quoteName('registerDate');
            $bl       = $db->quoteName('block');
            $ac       = $db->quoteName('activation');
            $never    = $db->quote(self::NEVER_LOGIN);
            $activeSql = $bl . ' = 0 AND (' . $ac . ' = ' . $db->quote('') . ' OR ' . $ac . ' = ' . $db->quote('0') . ')';

            $db->setQuery('SELECT COUNT(*) FROM ' . $qUsers);
            $total = (int) $db->loadResult();

            $db->setQuery(
                'SELECT COUNT(*) AS t,'
                . ' SUM(CASE WHEN ' . $activeSql . ' THEN 1 ELSE 0 END) AS a,'
                . ' SUM(CASE WHEN ' . $lv . ' > ' . $never . ' THEN 1 ELSE 0 END) AS l'
                . ' FROM ' . $qUsers . ' WHERE ' . $rd . ' >= ' . $db->quote($winCut)
            );
            $row     = $db->loadAssoc();
            $rTotal  = is_array($row) ? (int) (isset($row['t']) ? $row['t'] : 0) : 0;
            $rActive = is_array($row) ? (int) (isset($row['a']) ? $row['a'] : 0) : 0;
            $rLogin  = is_array($row) ? (int) (isset($row['l']) ? $row['l'] : 0) : 0;

            $db->setQuery(
                'SELECT MAX(' . $rd . ') FROM ' . $qUsers
                . ' WHERE ' . $activeSql . ' AND ' . $lv . ' > ' . $never
            );
            $lastReal = (string) $db->loadResult();

            $deps    = self::depsActive($db);   // string|null
            $regmenu = self::regMenuCount($db);

            $prot = self::protectedIds($db, array());
            $cand = -1;
            if (is_array($prot)) {
                $where = self::candidateWhereRaw($db, $prot, $graceCut);
                $db->setQuery('SELECT COUNT(*) FROM ' . $qUsers . ' WHERE ' . $where);
                $cand = (int) $db->loadResult();
            }

            self::journalPurge();

            return array(
                'total'     => $total,
                'r_total'   => $rTotal,
                'r_active'  => $rActive,
                'r_login'   => $rLogin,
                'last_real' => ($lastReal !== '' && $lastReal !== '0000-00-00 00:00:00') ? substr($lastReal, 0, 10) : '',
                'deps'      => ($deps === null ? '' : $deps),
                'regmenu'   => $regmenu,
                'cand'      => $cand,
                'win'       => self::WINDOW_MONTHS,
            );
        } catch (Exception $e) {
            return null;
        } catch (Throwable $e) {
            return null;
        }
    }

    /**
     * Aktive Abhaengigkeiten als CSV ('' = keine); null = Query-Fehler (Gate -> fail-closed). Deckt
     * BEIDES ab: gelistete KOMPONENTEN (Mitglieder/Shop/Forum/LMS) UND externe Auth-/Provisionierungs-
     * PLUGINS (jedes aktive authentication-Plugin ausser den Kern-Plugins joomla/cookie -> LDAP/SAML/
     * OAuth/CAS ...). Beide bedeuten: nie-eingeloggte Konten koennen echt sein.
     */
    private static function depsActive($db)
    {
        try {
            $out = array();

            // Komponenten.
            $els = self::depComponents();
            $q   = array();
            foreach ($els as $e) {
                $q[] = $db->quote($e);
            }
            $db->setQuery(
                'SELECT ' . $db->quoteName('element') . ' FROM ' . $db->quoteName('#__extensions')
                . ' WHERE ' . $db->quoteName('type') . ' = ' . $db->quote('component')
                . ' AND ' . $db->quoteName('enabled') . ' = 1'
                . ' AND ' . $db->quoteName('element') . ' IN (' . implode(',', $q) . ')'
            );
            $r = $db->loadColumn();
            if (is_array($r)) {
                foreach ($r as $x) {
                    $out[] = (string) $x;
                }
            }

            // Externe Auth-/Provisionierungs-Plugins (alle aktiven authentication-Plugins ausser Kern).
            $core = self::coreAuthPlugins();
            $cq   = array();
            foreach ($core as $c) {
                $cq[] = $db->quote($c);
            }
            $db->setQuery(
                'SELECT ' . $db->quoteName('element') . ' FROM ' . $db->quoteName('#__extensions')
                . ' WHERE ' . $db->quoteName('type') . ' = ' . $db->quote('plugin')
                . ' AND ' . $db->quoteName('enabled') . ' = 1'
                . ' AND ' . $db->quoteName('folder') . ' = ' . $db->quote('authentication')
                . ' AND ' . $db->quoteName('element') . ' NOT IN (' . implode(',', $cq) . ')'
            );
            $ap = $db->loadColumn();
            if (is_array($ap)) {
                foreach ($ap as $x) {
                    $out[] = 'auth:' . (string) $x;
                }
            }

            return count($out) ? implode(',', $out) : '';
        } catch (Exception $e) {
            return null;
        } catch (Throwable $e) {
            return null;
        }
    }

    /** Anzahl veroeffentlichter Registrierungs-Menuepunkte; -1 = nicht ermittelbar. */
    private static function regMenuCount($db)
    {
        try {
            $db->setQuery(
                'SELECT COUNT(*) FROM ' . $db->quoteName('#__menu')
                . ' WHERE ' . $db->quoteName('published') . ' = 1'
                . ' AND ' . $db->quoteName('link') . ' LIKE ' . $db->quote('%option=com_users%')
                . ' AND ' . $db->quoteName('link') . ' LIKE ' . $db->quote('%view=registration%')
            );
            return (int) $db->loadResult();
        } catch (Exception $e) {
            return -1;
        } catch (Throwable $e) {
            return -1;
        }
    }

    // ─────────────────────────────────────────────────────────────────────────
    //  Sicherheits-Kern - geschuetzte IDs (Admins/Autoren) + Kandidaten-Kriterium
    // ─────────────────────────────────────────────────────────────────────────

    /**
     * Backend-faehige Gruppen-IDs aus der ECHTEN ACL (nie hartkodiert). Seed = Gruppen mit einer
     * expliziten Erlaubnis (1) fuer core.admin / core.login.admin / core.manage am Wurzel-Asset;
     * plus deren Kind-Gruppen (Erben). null = ACL nicht sicher lesbar -> Aufrufer verweigert jedes Loeschen.
     *
     * @return array|null
     */
    private static function backendGroupIds($db)
    {
        try {
            $db->setQuery(
                'SELECT ' . $db->quoteName('rules') . ' FROM ' . $db->quoteName('#__assets')
                . ' WHERE ' . $db->quoteName('id') . ' = 1'
            );
            $raw = (string) $db->loadResult();
            if ($raw === '') {
                return null;
            }
            $rules = json_decode($raw, true);
            if (!is_array($rules)) {
                return null;
            }

            $seed = array();
            foreach (array('core.admin', 'core.login.admin', 'core.manage') as $act) {
                if (isset($rules[$act]) && is_array($rules[$act])) {
                    foreach ($rules[$act] as $gid => $val) {
                        if ((int) $val === 1) {
                            $seed[(int) $gid] = true;
                        }
                    }
                }
            }
            if (empty($seed)) {
                return null;
            }

            $db->setQuery(
                'SELECT ' . $db->quoteName('id') . ', ' . $db->quoteName('parent_id')
                . ' FROM ' . $db->quoteName('#__usergroups')
            );
            $groups = $db->loadAssocList();
            if (!is_array($groups)) {
                return null;
            }
            $children = array();
            foreach ($groups as $g) {
                $children[(int) $g['parent_id']][] = (int) $g['id'];
            }

            $out   = $seed;
            $stack = array_keys($seed);
            while ($stack) {
                $gid = (int) array_pop($stack);
                if (isset($children[$gid])) {
                    foreach ($children[$gid] as $ch) {
                        if (!isset($out[$ch])) {
                            $out[$ch]  = true;
                            $stack[]   = $ch;
                        }
                    }
                }
            }
            return array_map('intval', array_keys($out));
        } catch (Exception $e) {
            return null;
        } catch (Throwable $e) {
            return null;
        }
    }

    /**
     * Geschuetzte User-IDs, die NIE geloescht werden: Mitglieder backend-faehiger Gruppen (Admins),
     * Inhalts-Autoren und die Whitelist des Betreibers (params.keep). null = ACL/Admin-Ermittlung
     * fehlgeschlagen ODER Ergebnis leer -> Aufrufer verweigert das Loeschen (fail-safe).
     *
     * @return array|null
     */
    private static function protectedIds($db, $params)
    {
        $groups = self::backendGroupIds($db);
        if ($groups === null || count($groups) === 0) {
            return null;
        }

        $prot = array();

        try {
            $db->setQuery(
                'SELECT DISTINCT ' . $db->quoteName('user_id') . ' FROM ' . $db->quoteName('#__user_usergroup_map')
                . ' WHERE ' . $db->quoteName('group_id') . ' IN (' . implode(',', array_map('intval', $groups)) . ')'
            );
            $m = $db->loadColumn();
            if (!is_array($m)) {
                return null;
            }
            foreach ($m as $x) {
                $x = (int) $x;
                if ($x > 0) {
                    $prot[$x] = true;
                }
            }
        } catch (Exception $e) {
            return null;
        } catch (Throwable $e) {
            return null;
        }

        foreach (self::contentAuthorIds($db) as $x) {
            $x = (int) $x;
            if ($x > 0) {
                $prot[$x] = true;
            }
        }

        if (isset($params['keep']) && is_array($params['keep'])) {
            foreach ($params['keep'] as $x) {
                $x = (int) $x;
                if ($x > 0) {
                    $prot[$x] = true;
                }
            }
        }

        // Leere Schutzmenge = Admins nicht bestimmbar (kein Super-User gefunden) -> fail-closed.
        if (empty($prot)) {
            return null;
        }
        return array_map('intval', array_keys($prot));
    }

    /** Inhalts-Autoren der Kern-Tabellen (created_by/modified_by/created_user_id). Best-effort je Spalte. */
    private static function contentAuthorIds($db)
    {
        $checks = array(
            array('#__content', array('created_by', 'modified_by')),
            array('#__categories', array('created_user_id', 'modified_user_id')),
            array('#__tags', array('created_user_id')),
            array('#__contact_details', array('created_by', 'modified_by')),
        );
        $out = array();
        foreach ($checks as $c) {
            foreach ($c[1] as $col) {
                try {
                    $db->setQuery(
                        'SELECT DISTINCT ' . $db->quoteName($col) . ' FROM ' . $db->quoteName($c[0])
                        . ' WHERE ' . $db->quoteName($col) . ' > 0'
                    );
                    $r = $db->loadColumn();
                    if (is_array($r)) {
                        foreach ($r as $x) {
                            $x = (int) $x;
                            if ($x > 0) {
                                $out[$x] = true;
                            }
                        }
                    }
                } catch (Exception $e) {
                } catch (Throwable $e) {
                }
            }
        }
        return array_keys($out);
    }

    /**
     * Das Kandidaten-Kriterium als WHERE-Klausel (gemeinsam fuer Vorschau-COUNT, Kandidaten-Zaehler und
     * den Batch-Delete): nie eingeloggt UND alt genug UND gueltiges registerDate UND nicht geschuetzt.
     * Ist die Schutzmenge leer, matcht die Klausel bewusst NICHTS (fail-closed - der Aufrufer laesst
     * eine leere Schutzmenge aber ohnehin nicht bis hierher kommen).
     */
    private static function candidateWhereRaw($db, $protected, $graceCut)
    {
        $lv = $db->quoteName('lastvisitDate');
        $rd = $db->quoteName('registerDate');

        $w = '(' . $lv . ' IS NULL OR ' . $lv . ' <= ' . $db->quote(self::NEVER_LOGIN) . ')'
            . ' AND ' . $rd . ' < ' . $db->quote($graceCut)
            . ' AND ' . $rd . ' > ' . $db->quote(self::NEVER_LOGIN);

        if (is_array($protected) && count($protected)) {
            $w .= ' AND ' . $db->quoteName('id') . ' NOT IN (' . implode(',', array_map('intval', $protected)) . ')';
        } else {
            $w .= ' AND 0 = 1'; // leere Schutzmenge -> nie loeschen
        }
        return $w;
    }

    /**
     * Bindet die Bestaetigung an die geschuetzte Menge + die Kriterien-KONSTANTEN (Confirm-Lock).
     * WICHTIG: NICHT den sekundengenauen graceCut-Zeitstempel einbeziehen - Vorschau und Lauf laufen
     * Sekunden auseinander, das ergaebe jedes Mal einen anderen Hash. Gebunden wird an GRACE_DAYS.
     */
    private static function criteriaHash($protected)
    {
        $p = is_array($protected) ? array_map('intval', $protected) : array();
        sort($p);
        return substr(hash('sha256', 'rcv1|' . self::GRACE_DAYS . '|' . self::NEVER_LOGIN . '|' . implode(',', $p)), 0, 16);
    }

    private static function mask($s)
    {
        $s = (string) $s;
        if ($s === '') {
            return '';
        }
        return substr($s, 0, 1) . '***';
    }

    private static function maskEmail($e)
    {
        $e  = (string) $e;
        $at = strpos($e, '@');
        if ($at === false) {
            return substr(self::mask($e), 0, 24);
        }
        return substr($e, 0, 1) . '***@' . substr(substr($e, $at + 1), 0, 40);
    }

    // ─────────────────────────────────────────────────────────────────────────
    //  TEIL B - Vorschau + Batch-Delete
    // ─────────────────────────────────────────────────────────────────────────

    /**
     * Vorschau (synchron): Anzahl loeschbarer Kandidaten + maskierte Beispiele + Ausschluss-Kennzahlen
     * + Domain-Konzentration (Import-Warnung) + Kriterien-Hash. Fuehrt NICHTS aus. Verweigert bei
     * aktiven Abhaengigkeiten / unlesbarer ACL.
     */
    public static function preview($params)
    {
        $db = self::db();
        if (!$db) {
            return array('ok' => false, 'code' => 'no_db');
        }

        $deps = self::depsActive($db);
        if ($deps === null) {
            return array('ok' => false, 'code' => 'deps_check_failed');
        }
        if ($deps !== '') {
            return array('ok' => false, 'code' => 'deps_present:' . $deps);
        }

        $prot = self::protectedIds($db, is_array($params) ? $params : array());
        if ($prot === null) {
            return array('ok' => false, 'code' => 'acl_unreadable');
        }
        if (count($prot) > self::MAX_PROTECTED) {
            return array('ok' => false, 'code' => 'too_many_real', 'protected' => count($prot));
        }

        try {
            $graceCut = gmdate('Y-m-d H:i:s', time() - self::GRACE_DAYS * 86400);
            $where    = self::candidateWhereRaw($db, $prot, $graceCut);
            $qUsers   = $db->quoteName('#__users');

            $db->setQuery('SELECT COUNT(*) FROM ' . $qUsers . ' WHERE ' . $where);
            $count = (int) $db->loadResult();

            $db->setQuery('SELECT COUNT(*) FROM ' . $qUsers);
            $total = (int) $db->loadResult();

            $db->setQuery('SELECT COUNT(*) FROM ' . $qUsers . ' WHERE ' . $db->quoteName('lastvisitDate') . ' > ' . $db->quote(self::NEVER_LOGIN));
            $exLogin = (int) $db->loadResult();
            $db->setQuery('SELECT COUNT(*) FROM ' . $qUsers . ' WHERE ' . $db->quoteName('registerDate') . ' >= ' . $db->quote($graceCut) . ' AND (' . $db->quoteName('lastvisitDate') . ' IS NULL OR ' . $db->quoteName('lastvisitDate') . ' <= ' . $db->quote(self::NEVER_LOGIN) . ')');
            $exFresh = (int) $db->loadResult();

            // Domain-Konzentration: teilen sich viele Kandidaten EINE Domain, ist das eher eine Import-/
            // Mitgliederliste als eine (diverse) Spam-Flut -> Warnung fuers Dashboard.
            $topDom   = '';
            $topShare = 0.0;
            if ($count > 0) {
                $db->setQuery(
                    'SELECT SUBSTRING_INDEX(' . $db->quoteName('email') . ', ' . $db->quote('@') . ', -1) AS dom, COUNT(*) AS c'
                    . ' FROM ' . $qUsers . ' WHERE ' . $where . ' GROUP BY dom ORDER BY c DESC LIMIT 1'
                );
                $td = $db->loadAssoc();
                if (is_array($td) && isset($td['dom'])) {
                    $topDom   = substr((string) $td['dom'], 0, 40);
                    $topShare = round(((int) $td['c']) / $count, 3);
                }
            }

            $db->setQuery(
                'SELECT ' . implode(',', $db->quoteName(array('id', 'username', 'email', 'registerDate')))
                . ' FROM ' . $qUsers . ' WHERE ' . $where . ' ORDER BY ' . $db->quoteName('id') . ' ASC LIMIT ' . (int) self::SAMPLE_MAX
            );
            $rows   = $db->loadAssocList();
            $sample = array();
            if (is_array($rows)) {
                foreach ($rows as $r) {
                    $sample[] = array(
                        'u' => self::mask(isset($r['username']) ? $r['username'] : ''),
                        'e' => self::maskEmail(isset($r['email']) ? $r['email'] : ''),
                        'd' => substr((string) (isset($r['registerDate']) ? $r['registerDate'] : ''), 0, 10),
                    );
                }
            }

            return array(
                'ok'         => true,
                'code'       => 'reg_preview',
                'count'      => $count,
                'total'      => $total,
                'deps'       => $deps,
                'protected'  => count($prot),
                'excluded'   => array('protected' => count($prot), 'loggedin' => $exLogin, 'fresh' => $exFresh),
                'sample'     => $sample,
                'top_domain' => $topDom,
                'top_share'  => $topShare,
                'hash'       => self::criteriaHash($prot),
                'grace'      => self::GRACE_DAYS,
            );
        } catch (Exception $e) {
            return array('ok' => false, 'code' => 'preview_error');
        } catch (Throwable $e) {
            return array('ok' => false, 'code' => 'preview_error');
        }
    }

    /**
     * Batch-Delete anstossen (async). Prueft: busy-Guard, Deps-Gate (hart), ACL-Fail-Safe, Zu-viele-
     * echte, Journal-Verfuegbarkeit (falls gewuenscht), Confirm-Hash + Count-Lock (BEIDE PFLICHT).
     * Seedet dann den Job (Cursor + geschuetzte Menge + Journal-Flag).
     */
    public static function start(&$st, $cid, $params)
    {
        if (!empty($st['job'])) {
            return array('ok' => false, 'code' => 'busy');
        }
        $db = self::db();
        if (!$db) {
            return array('ok' => false, 'code' => 'no_db');
        }

        // Confirm-Hash + erwartete Anzahl sind PFLICHT (die menschliche Doppelpruefung darf nicht
        // umgangen werden). Fehlt eines -> verweigern.
        if (!isset($params['confirm']) || (string) $params['confirm'] === '') {
            return array('ok' => false, 'code' => 'confirm_required');
        }
        if (!isset($params['expected'])) {
            return array('ok' => false, 'code' => 'expected_required');
        }

        $deps = self::depsActive($db);
        if ($deps === null) {
            return array('ok' => false, 'code' => 'deps_check_failed');
        }
        if ($deps !== '') {
            return array('ok' => false, 'code' => 'deps_present:' . $deps);
        }

        $prot = self::protectedIds($db, is_array($params) ? $params : array());
        if ($prot === null) {
            return array('ok' => false, 'code' => 'acl_unreadable');
        }
        if (count($prot) > self::MAX_PROTECTED) {
            return array('ok' => false, 'code' => 'too_many_real:' . count($prot));
        }

        // Journal gewuenscht, aber nicht schreibbar (kein zlib / Ordner nicht beschreibbar) -> NICHT
        // loeschen (das Undo-Netz wurde versprochen).
        $journal = array_key_exists('journal', $params) ? (!empty($params['journal'])) : true;
        if ($journal && !self::journalReady()) {
            return array('ok' => false, 'code' => 'journal_unavailable');
        }

        $graceCut = gmdate('Y-m-d H:i:s', time() - self::GRACE_DAYS * 86400);
        $where    = self::candidateWhereRaw($db, $prot, $graceCut);

        try {
            $db->setQuery('SELECT COUNT(*) FROM ' . $db->quoteName('#__users') . ' WHERE ' . $where);
            $count = (int) $db->loadResult();
        } catch (Exception $e) {
            return array('ok' => false, 'code' => 'count_error');
        } catch (Throwable $e) {
            return array('ok' => false, 'code' => 'count_error');
        }

        $hash = self::criteriaHash($prot);
        if ((string) $params['confirm'] !== $hash) {
            return array('ok' => false, 'code' => 'confirm_mismatch');
        }

        $expected = (int) $params['expected'];
        $tol      = max(self::COUNT_TOL_ABS, (int) ceil($expected * self::COUNT_TOL_PCT / 100));
        if (abs($count - $expected) > $tol) {
            return array('ok' => false, 'code' => 'count_drift:' . $count . '/' . $expected);
        }

        if ($count === 0) {
            return array('ok' => true, 'code' => 'reg_cleanup_empty');
        }

        $ref = self::newRef();
        self::journalMeta($ref, $count, $hash, $journal);

        $st['job'] = array(
            'type'      => 'reg_cleanup',
            'op'        => '',
            'cmd_id'    => (string) $cid,
            'phase'     => 'delete',
            'started'   => time(),
            'advanced'  => time(),
            'progress'  => 0.0,
            'cursor'    => 0,
            'deleted'   => 0,
            'total'     => $count,
            'protected' => array_values($prot),
            'grace_cut' => $graceCut,
            'journal'   => $journal ? 1 : 0,
            'ref'       => $ref,
            'cancel'    => 0,
        );

        return array('ok' => true, 'code' => 'reg_cleanup_started', 'async' => true);
    }

    /** Laufenden Cleanup/Restore abbrechen (Flag; die Engine prueft es pro Slice). */
    public static function cancel(&$st)
    {
        if (empty($st['job']) || !is_array($st['job']) || (isset($st['job']['type']) ? $st['job']['type'] : '') !== 'reg_cleanup') {
            return array('ok' => false, 'code' => 'no_cleanup_running');
        }
        $st['job']['cancel'] = 1;
        return array('ok' => true, 'code' => 'reg_cleanup_canceling');
    }

    /**
     * Ein budgetierter Slice des Batch-Deletes (Kontrakt wie HtpBackup::advance): keyset-Kandidaten oberhalb
     * des Cursors holen -> optional als Journal-Chunk streamen -> Satelliten + #__users loeschen -> Cursor
     * vorruecken. Keine Transaktion (idempotent). Ist das Journal gewuenscht, aber eine Chunk-Datei nicht
     * schreibbar -> Abbruch VOR dem Loeschen dieses Batches (kein Loeschen ohne Netz).
     *
     * @return array{job:array,done:bool,ok:bool,code:string,progress?:float}
     */
    public static function advance($job, $deadline, $cfg)
    {
        $db = self::db();
        if (!$db) {
            return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'no_db');
        }

        if (!empty($job['cancel'])) {
            return array('job' => $job, 'done' => true, 'ok' => true, 'code' => 'reg_cleanup_canceled:' . (int) (isset($job['deleted']) ? $job['deleted'] : 0));
        }

        $protected = (isset($job['protected']) && is_array($job['protected'])) ? $job['protected'] : array();
        if (empty($protected)) {
            // Sollte nie vorkommen (start() verlangt eine nicht-leere Schutzmenge) - fail-closed.
            return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'no_protected');
        }

        $cursor   = (int) (isset($job['cursor']) ? $job['cursor'] : 0);
        $deleted  = (int) (isset($job['deleted']) ? $job['deleted'] : 0);
        $total    = (int) (isset($job['total']) ? $job['total'] : 0);
        $journal  = !empty($job['journal']);
        $ref      = (string) (isset($job['ref']) ? $job['ref'] : '');
        $graceCut = (string) (isset($job['grace_cut']) ? $job['grace_cut'] : gmdate('Y-m-d H:i:s', time() - self::GRACE_DAYS * 86400));

        $where  = self::candidateWhereRaw($db, $protected, $graceCut);
        $qUsers = $db->quoteName('#__users');
        $qId    = $db->quoteName('id');
        $jdir   = $journal ? self::journalDir() : ''; // fuer den Platten-Boden-Check je Batch
        $done   = false;

        // Kurze Scheibe: nach ~SLICE_BUDGET zurueck an den Treiber, damit er Cursor speichert + Fortschritt
        // heartbeatet (nicht das volle $deadline in EINEM Rutsch abarbeiten). $deadline bleibt harte Obergrenze.
        $sliceEnd = microtime(true) + self::SLICE_BUDGET;
        if ($sliceEnd > $deadline) {
            $sliceEnd = $deadline;
        }

        try {
            while (microtime(true) < $sliceEnd) {
                $db->setQuery(
                    'SELECT ' . $qId . ' FROM ' . $qUsers . ' WHERE ' . $where
                    . ' AND ' . $qId . ' > ' . (int) $cursor
                    . ' ORDER BY ' . $qId . ' ASC LIMIT ' . (int) self::DEL_BATCH
                );
                $ids = $db->loadColumn();
                if (!is_array($ids) || count($ids) === 0) {
                    $done = true;
                    break;
                }
                $ids     = array_map('intval', $ids);
                $inList  = implode(',', $ids);
                $maxId   = max($ids); // streng monoton -> kollisionsfreier Chunk-Name (crash-robust)
                $targets = self::delTargets($db, $inList);

                if ($journal) {
                    // Platten-Boden: bleibt zu wenig frei, NICHT weiterschreiben (Host-Platte nie vollaufen
                    // lassen). Cursor bleibt intakt -> spaeter fortsetzbar, sobald wieder Platz da ist.
                    $freeB = ($jdir !== '') ? @disk_free_space($jdir) : false;
                    if ($freeB !== false && (float) $freeB < self::MIN_FREE_BYTES) {
                        $job['cursor']  = $cursor;
                        $job['deleted'] = $deleted;
                        return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'reg_cleanup_lowdisk:' . $deleted);
                    }
                    // Chunk VOR dem Loeschen, benannt nach der Batch-Max-ID: stirbt der Prozess mitten im
                    // Slice (Cursor noch nicht gespeichert), ueberschreibt kein Resume diesen Chunk -> die
                    // hier geloeschten Konten bleiben rueckholbar.
                    if (!self::journalChunk($db, $ref, $maxId, $inList, $targets)) {
                        // Netz versprochen, aber nicht schreibbar -> diesen Batch NICHT loeschen, Job beenden.
                        $job['cursor']  = $cursor;
                        $job['deleted'] = $deleted;
                        return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'reg_cleanup_journal_fail:' . $deleted);
                    }
                }

                foreach ($targets as $t) {
                    try {
                        $db->setQuery('DELETE FROM ' . $db->quoteName($t[0]) . ' WHERE ' . $t[1]);
                        $db->execute();
                    } catch (Exception $e) {
                    } catch (Throwable $e) {
                    }
                }
                $db->setQuery('DELETE FROM ' . $qUsers . ' WHERE ' . $qId . ' IN (' . $inList . ')');
                $db->execute();

                $cursor  = (int) $maxId;
                $deleted += count($ids);
                $job['cursor']   = $cursor;
                $job['deleted']  = $deleted;
                $job['progress'] = ($total > 0) ? min(0.999, $deleted / $total) : 0.5;

                if (count($ids) < self::DEL_BATCH) {
                    $done = true;
                    break;
                }
            }
        } catch (Exception $e) {
            $job['cursor'] = $cursor;
            $job['deleted'] = $deleted;
            return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'reg_cleanup_db_error:' . $deleted);
        } catch (Throwable $e) {
            $job['cursor'] = $cursor;
            $job['deleted'] = $deleted;
            return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'reg_cleanup_db_error:' . $deleted);
        }

        if ($done) {
            $job['progress'] = 1.0;
            self::invalidateStats();
            return array('job' => $job, 'done' => true, 'ok' => true, 'code' => 'reg_cleanup_done:' . $deleted);
        }
        return array('job' => $job, 'done' => false, 'ok' => true, 'code' => 'reg_cleanup_run', 'progress' => (float) $job['progress']);
    }

    // ─────────────────────────────────────────────────────────────────────────
    //  Undo-Journal (gzip-Chunk je Batch unter data/regclean; 30 Tage Auto-Purge)
    // ─────────────────────────────────────────────────────────────────────────

    private static function newRef()
    {
        $rnd = '';
        try {
            if (function_exists('random_bytes')) {
                $rnd = bin2hex(random_bytes(4));
            }
        } catch (Exception $e) {
        } catch (Throwable $e) {
        }
        if ($rnd === '') {
            $rnd = substr(md5(uniqid('rc', true)), 0, 8);
        }
        return 'rc-' . gmdate('Ymd-His') . '-' . $rnd;
    }

    private static function journalDir()
    {
        $base = function_exists('htp_state_dir') ? (string) htp_state_dir() : '';
        if ($base === '') {
            return '';
        }
        $d = $base . '/regclean';
        if (!is_dir($d)) {
            @mkdir($d, 0700, true);
        }
        return is_dir($d) ? $d : '';
    }

    /** Ist das Journal ueberhaupt schreibbar (zlib vorhanden + Ordner beschreibbar)? */
    private static function journalReady()
    {
        if (!function_exists('gzopen')) {
            return false;
        }
        $d = self::journalDir();
        return $d !== '' && @is_writable($d);
    }

    /** Schlanker Audit-Eintrag (Anzahl/Zeit/Hash - KEINE PII), bleibt bis zum Purge. */
    private static function journalMeta($ref, $count, $hash, $journal)
    {
        $d = self::journalDir();
        if ($d === '') {
            return;
        }
        $meta = array(
            'ref'     => $ref,
            'at'      => time(),
            'count'   => (int) $count,
            'hash'    => (string) $hash,
            'journal' => $journal ? 1 : 0,
            'win'     => self::WINDOW_MONTHS,
            'grace'   => self::GRACE_DAYS,
        );
        @file_put_contents($d . '/' . $ref . '.meta.json', (string) json_encode($meta));
        @chmod($d . '/' . $ref . '.meta.json', 0600);
    }

    /** Einen Batch als eigene gzip-Chunk-Datei wegschreiben (ALLE betroffenen Tabellen). true = ok. */
    private static function journalChunk($db, $ref, $seq, $inList, $targets)
    {
        if (!function_exists('gzopen')) {
            return false;
        }
        $dir = self::journalDir();
        if ($dir === '' || $ref === '') {
            return false;
        }
        $path = $dir . '/' . $ref . '.' . sprintf('%012d', (int) $seq) . '.jsonl.gz';
        $fh   = @gzopen($path, 'wb6');
        if (!$fh) {
            return false;
        }
        @chmod($path, 0600);

        // #__users zuerst, dann die Satelliten - jede in der gleichen Menge, die geloescht wird.
        // Scheitert EIN Schreibvorgang (volle Platte), brechen wir ab und behandeln den Chunk als kaputt.
        $ok = self::journalTable($db, $fh, '#__users', $db->quoteName('id') . ' IN (' . $inList . ')');
        foreach ($targets as $t) {
            if (!self::journalTable($db, $fh, $t[0], $t[1])) {
                $ok = false;
            }
        }

        $closed = @gzclose($fh); // gepufferte Bytes werden erst hier geflusht -> Platte-voll faellt hier auf
        if (!$ok || $closed === false) {
            @unlink($path); // Teil-Chunk nicht liegen lassen (kein Muell, kein falsches Netz)
            return false;
        }
        // Sanity: eine echte Menge muss echte Bytes ergeben (0-Byte = harter Abbruch/volle Platte).
        $sz = @filesize($path);
        if ($sz === false || (int) $sz < 20) {
            @unlink($path);
            return false;
        }
        return true;
    }

    /** Alle Zeilen einer Tabelle (per WHERE-Fragment) als JSON-Zeilen ins offene gzip schreiben.
     *  true = alles geschrieben; false = Schreibfehler (z. B. volle Platte) ODER DB-Fehler -> der
     *  Aufrufer bricht den Batch VOR dem Loeschen ab (kein Loeschen ohne vollstaendiges Netz). */
    private static function journalTable($db, $fh, $table, $where)
    {
        try {
            $db->setQuery('SELECT * FROM ' . $db->quoteName($table) . ' WHERE ' . $where);
            $rows = $db->loadAssocList();
            if (is_array($rows)) {
                foreach ($rows as $r) {
                    $line = (string) json_encode(array('t' => $table, 'r' => $r)) . "\n";
                    $w    = @gzwrite($fh, $line);
                    // gzwrite gibt die Zahl der UNKOMPRIMIERTEN Bytes zurueck (0/false = Platte voll).
                    if ($w === false || (int) $w < strlen($line)) {
                        return false;
                    }
                }
            }
            return true;
        } catch (Exception $e) {
            return false;
        } catch (Throwable $e) {
            return false;
        }
    }

    /** Journale (Chunks + Meta) aelter als die Aufbewahrung loeschen. */
    public static function journalPurge()
    {
        $d = self::journalDir();
        if ($d === '') {
            return;
        }
        $now  = time();
        $list = @glob($d . '/*');
        if (!is_array($list)) {
            return;
        }
        foreach ($list as $f) {
            if (@is_file($f) && ($now - (int) @filemtime($f)) > self::JOURNAL_TTL) {
                @unlink($f);
            }
        }
    }

    // ─────────────────────────────────────────────────────────────────────────
    //  Wiederherstellung (Break-Glass): Journal-Chunks zurueckspielen (O(1)-resumierbar)
    // ─────────────────────────────────────────────────────────────────────────

    /** Restore anstossen (async). Spielt die gzip-Chunk-Dateien eines Refs der Reihe nach zurueck. */
    public static function startRestore(&$st, $cid, $params)
    {
        if (!empty($st['job'])) {
            return array('ok' => false, 'code' => 'busy');
        }
        $ref = isset($params['ref']) ? (string) $params['ref'] : '';
        if ($ref === '' || !preg_match('~^rc-[0-9]{8}-[0-9]{6}-[0-9a-f]{8}$~', $ref)) {
            return array('ok' => false, 'code' => 'bad_ref');
        }
        $files = self::chunkFiles($ref);
        if (empty($files)) {
            return array('ok' => false, 'code' => 'journal_gone');
        }

        $st['job'] = array(
            'type'     => 'reg_cleanup',
            'op'       => 'restore',
            'cmd_id'   => (string) $cid,
            'phase'    => 'restore',
            'started'  => time(),
            'advanced' => time(),
            'progress' => 0.0,
            'ref'      => $ref,
            'rseq'     => -1,
            'rfiles'   => 0,
            'rtotal'   => count($files),
            'restored' => 0,
            'cancel'   => 0,
        );
        return array('ok' => true, 'code' => 'reg_restore_started', 'async' => true);
    }

    /** Chunk-Dateien eines Refs als [seq => path], nach seq aufsteigend sortiert. */
    private static function chunkFiles($ref)
    {
        $d = self::journalDir();
        if ($d === '') {
            return array();
        }
        $glob = @glob($d . '/' . $ref . '.*.jsonl.gz');
        if (!is_array($glob)) {
            return array();
        }
        $out = array();
        foreach ($glob as $f) {
            if (preg_match('~\.([0-9]{6,})\.jsonl\.gz$~', $f, $m)) {
                $out[(int) $m[1]] = $f;
            }
        }
        ksort($out);
        return $out;
    }

    /**
     * Ein budgetierter Slice der Wiederherstellung: die naechsten Chunk-Dateien (seq > rseq) der Reihe
     * nach KOMPLETT zurueckspielen (INSERT IGNORE). Weil jede Chunk-Datei ~ ein Batch (klein) ist, wird sie
     * je Slice vollstaendig verarbeitet -> kein Re-Skip, echter datei-basierter Fortschritt (Fingerprint
     * bewegt sich -> kein Stall-Fehlalarm).
     *
     * @return array{job:array,done:bool,ok:bool,code:string,progress?:float}
     */
    public static function advanceRestore($job, $deadline, $cfg)
    {
        $db = self::db();
        if (!$db) {
            return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'no_db');
        }
        if (!empty($job['cancel'])) {
            return array('job' => $job, 'done' => true, 'ok' => true, 'code' => 'reg_restore_canceled:' . (int) (isset($job['restored']) ? $job['restored'] : 0));
        }
        if (!function_exists('gzopen')) {
            return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'no_gzip');
        }

        $ref   = (string) (isset($job['ref']) ? $job['ref'] : '');
        $files = self::chunkFiles($ref);
        if (empty($files)) {
            return array('job' => $job, 'done' => true, 'ok' => false, 'code' => 'journal_gone');
        }

        $rseq     = (int) (isset($job['rseq']) ? $job['rseq'] : -1);
        $rfiles   = (int) (isset($job['rfiles']) ? $job['rfiles'] : 0);
        $rtotal   = (int) (isset($job['rtotal']) ? $job['rtotal'] : count($files));
        $restored = (int) (isset($job['restored']) ? $job['restored'] : 0);
        $allow    = self::restoreTables();

        // Kurze Scheibe wie advance(): nach ~SLICE_BUDGET zurueck -> Treiber speichert rseq + heartbeatet.
        $sliceEnd = microtime(true) + self::SLICE_BUDGET;
        if ($sliceEnd > $deadline) {
            $sliceEnd = $deadline;
        }

        foreach ($files as $seq => $path) {
            if ($seq <= $rseq) {
                continue; // schon erledigt
            }
            if (microtime(true) >= $sliceEnd) {
                break; // sauberer Wiederaufsatz beim naechsten seq
            }

            $fh = @gzopen($path, 'rb');
            if ($fh) {
                while (!gzeof($fh)) {
                    $raw = @gzgets($fh);
                    if ($raw === false) {
                        break;
                    }
                    $raw = trim($raw);
                    if ($raw === '') {
                        continue;
                    }
                    $rec = json_decode($raw, true);
                    if (is_array($rec) && isset($rec['t'], $rec['r']) && is_array($rec['r']) && isset($allow[$rec['t']])) {
                        self::insertRow($db, (string) $rec['t'], $rec['r']);
                        $restored++;
                    }
                }
                @gzclose($fh);
            }

            $rseq   = (int) $seq;
            $rfiles++;
            $job['rseq']     = $rseq;
            $job['rfiles']   = $rfiles;
            $job['restored'] = $restored;
            $job['progress'] = ($rtotal > 0) ? min(0.999, $rfiles / $rtotal) : 0.5;
        }

        // Fertig, wenn keine Chunk-Datei mit seq > rseq mehr aussteht.
        $remaining = false;
        foreach ($files as $seq => $path) {
            if ((int) $seq > $rseq) {
                $remaining = true;
                break;
            }
        }

        if (!$remaining) {
            $job['progress'] = 1.0;
            self::invalidateStats();
            return array('job' => $job, 'done' => true, 'ok' => true, 'code' => 'reg_restore_done:' . $restored);
        }
        return array('job' => $job, 'done' => false, 'ok' => true, 'code' => 'reg_restore_run', 'progress' => (float) $job['progress']);
    }

    /**
     * Liste der letzten Aufraeum-Aktionen (aus den schlanken Meta-Dateien): {ref, at, count, restorable}.
     * restorable = es liegen noch Journal-Chunks (nicht gepurgt) -> rueckholbar. Neueste zuerst, max 20.
     * Fuer den Rueckgaengig-/Verlauf-Dialog.
     */
    public static function journalList()
    {
        $d = self::journalDir();
        if ($d === '') {
            return array('ok' => true, 'code' => 'reg_list', 'items' => array());
        }
        $items = array();
        $metas = @glob($d . '/*.meta.json');
        if (is_array($metas)) {
            foreach ($metas as $f) {
                $j = json_decode((string) @file_get_contents($f), true);
                if (is_array($j) && isset($j['ref'])) {
                    $items[] = array(
                        'ref'        => (string) $j['ref'],
                        'at'         => (int) (isset($j['at']) ? $j['at'] : 0),
                        'count'      => (int) (isset($j['count']) ? $j['count'] : 0),
                        'journal'    => (int) (isset($j['journal']) ? $j['journal'] : 0),
                        'restorable' => (!empty($j['journal']) && count(self::chunkFiles((string) $j['ref'])) > 0) ? 1 : 0,
                    );
                }
            }
        }
        usort($items, function ($a, $b) {
            return $b['at'] - $a['at'];
        });
        if (count($items) > 20) {
            $items = array_slice($items, 0, 20);
        }
        return array('ok' => true, 'code' => 'reg_list', 'items' => $items);
    }

    /** Eine Journal-Zeile dublettensicher zurueckspielen (INSERT IGNORE). */
    private static function insertRow($db, $tbl, $row)
    {
        try {
            $cols = array();
            $vals = array();
            foreach ($row as $k => $v) {
                $cols[] = $db->quoteName((string) $k);
                $vals[] = ($v === null) ? 'NULL' : $db->quote((string) $v);
            }
            if (empty($cols)) {
                return;
            }
            $db->setQuery(
                'INSERT IGNORE INTO ' . $db->quoteName($tbl) . ' (' . implode(',', $cols) . ') VALUES (' . implode(',', $vals) . ')'
            );
            $db->execute();
        } catch (Exception $e) {
        } catch (Throwable $e) {
        }
    }
}
