<?php

declare(strict_types=1);

namespace Atdt\Import;

use Atdt\Db;
use Atdt\Messages;
use Atdt\Settings;

/**
 * The import engine (SPEC-IMPORT §15): every pass runs in chunks sized
 * by count, driven by the console polling step(); nothing ever runs
 * unattended. Phases per batch: scan (counts, date ranges, samples for
 * the preview), stage (normalize into import_staging with the final
 * encoding and timezone), write (drain staging into messages in
 * ascending date order per area), link (reply threading from the
 * source id map, cycle-safe), finish (last-read sweep, seed watermark,
 * staging cleanup). A single advisory, expiring lock keeps two imports
 * from running at once; undo deletes by batch id and removes areas the
 * batch created that are now empty.
 */
final class Batch
{
    /** §15.2: chunks are sized by count, one easy-to-lower constant. */
    public const CHUNK = 250;
    /** Link works from an in-memory id map, so its chunks run bigger. */
    public const LINK_CHUNK = 2000;
    /** Past this many staged messages the link map would strain PHP's
     *  memory; the per-row indexed path takes over (slower, correct). */
    public const LINK_MAP_MAX = 150000;
    /** Advisory lock lifetime; a crashed import never wedges the page. */
    public const LOCK_SECONDS = 600;
    /** §6.4: at or over this share of unparseable records, the parse fails. */
    public const BAD_SHARE = 0.05;

    // ---- lifecycle --------------------------------------------------------

    /**
     * Create a batch against a resolved source directory and survey it.
     * @return array{ok: bool, error?: string, id?: int, open?: array}
     */
    public static function open(
        Db $db,
        Settings $settings,
        int $sysopId,
        string $package,
        string $format,
        string $label,
        string $dir
    ): array {
        $src = Sources::make($format);
        if ($src === null) {
            return ['ok' => false, 'error' => "Unknown source format \"$format\"."];
        }
        $held = self::lockHolder($db, $settings);
        if ($held !== null) {
            return ['ok' => false, 'error' => 'Another import (batch #' . $held . ') is running. '
                . 'One import at a time; resume or undo it from Import History.'];
        }
        $open = $src->open($dir);
        if (!$open['ok']) {
            return ['ok' => false, 'error' => (string)$open['error']];
        }
        $encoding = Sources::defaultEncoding($format);
        $tz = (string)($settings->get('timezone', 'UTC') ?? 'UTC');
        $defaults = (array)($open['defaults'] ?? []);
        if (in_array((string)($defaults['encoding'] ?? ''), Normalizer::ENCODINGS, true)) {
            $encoding = (string)$defaults['encoding'];
        }
        if ((string)($defaults['timezone'] ?? '') !== '') {
            $tz = (string)$defaults['timezone'];
        }
        $id = $db->insert('import_batches', [
            'created_at' => now_utc(),
            'sysop_user_id' => $sysopId,
            'source_package' => $package,
            'source_format' => $format,
            'source_label' => $label,
            'source_dir' => $dir,
            'encoding' => $encoding,
            'timezone' => $tz,
            'status' => 'scanning',
            'phase' => 'scan',
            'cursor' => '{}',
            'stats' => (string)json_encode([
                'areas' => $open['areas'],
                'units' => $open['units'],
                'survey' => [],           // area_key => [count, first, last]
                'samples' => [],          // up to 3 of the largest area
                'scanned' => 0,
                'skipped' => 0,
                'host' => parse_url((string)($settings->get('base_url', '') ?? ''), PHP_URL_HOST) ?: 'atdt.local',
            ]),
            'mapping' => '{}',
            'total_messages' => 0,
            'done_messages' => 0,
            'created_areas' => '[]',
        ]);
        self::lock($db, $settings, $id);
        atdt_log_event('admin', 'import', "Import batch #$id opened: $package ($format) from $label");
        return ['ok' => true, 'id' => $id, 'open' => $open];
    }

    public static function row(Db $db, int $id): ?array
    {
        return $db->one('SELECT * FROM import_batches WHERE id = ?', [$id]);
    }

    /**
     * Advance the batch one chunk, whatever phase it is in. Returns the
     * refreshed row plus 'progress' hints for the polling console.
     */
    public static function step(Db $db, Settings $settings, int $id): array
    {
        $b = self::row($db, $id);
        if ($b === null) {
            return ['ok' => false, 'error' => 'No such batch.'];
        }
        $status = (string)$b['status'];
        if (!in_array($status, ['scanning', 'staging', 'running'], true)) {
            return ['ok' => true, 'batch' => $b];
        }
        $holder = self::lockHolder($db, $settings);
        if ($holder !== null && $holder !== $id) {
            return ['ok' => false, 'error' => "Another import (batch #$holder) holds the lock."];
        }
        self::lock($db, $settings, $id);
        try {
            $b = match ((string)$b['phase']) {
                'scan' => self::stepScan($db, $b),
                'stage' => self::stepStage($db, $b),
                'write' => self::stepWrite($db, $b),
                'link' => self::stepLink($db, $b),
                'finish' => self::stepFinish($db, $settings, $b),
                default => $b,
            };
        } catch (\Throwable $e) {
            $db->q('UPDATE import_batches SET status = ?, error = ? WHERE id = ?',
                ['failed', mb_substr($e->getMessage(), 0, 500), $id]);
            atdt_log_event('error', 'import', "Import batch #$id failed: " . $e->getMessage());
            self::unlock($db, $settings, $id);
            $b = self::row($db, $id);
        }
        return ['ok' => true, 'batch' => $b];
    }

    /** The SysOp closed the page or pressed pause: hold everything. */
    public static function pause(Db $db, Settings $settings, int $id): void
    {
        $db->q(
            "UPDATE import_batches SET status = 'paused'
              WHERE id = ? AND status IN ('scanning', 'staging', 'running')",
            [$id]
        );
        self::unlock($db, $settings, $id);
        atdt_log_event('admin', 'import', "Import batch #$id paused");
    }

    /** Resume a paused (or failed) batch from its chunk boundary. */
    public static function resume(Db $db, Settings $settings, int $id): array
    {
        $b = self::row($db, $id);
        if ($b === null) {
            return ['ok' => false, 'error' => 'No such batch.'];
        }
        $holder = self::lockHolder($db, $settings);
        if ($holder !== null && $holder !== $id) {
            return ['ok' => false, 'error' => "Another import (batch #$holder) holds the lock."];
        }
        $status = in_array((string)$b['phase'], ['scan'], true) ? 'scanning'
            : ((string)$b['phase'] === 'stage' ? 'staging' : 'running');
        $db->q('UPDATE import_batches SET status = ?, error = NULL WHERE id = ?', [$status, $id]);
        self::lock($db, $settings, $id);
        atdt_log_event('admin', 'import', "Import batch #$id resumed");
        return ['ok' => true, 'batch' => self::row($db, $id)];
    }

    // ---- phase: scan ------------------------------------------------------

    /** Count, survey dates per area, and keep the preview samples. */
    private static function stepScan(Db $db, array $b): array
    {
        $stats = json_decode((string)$b['stats'], true) ?: [];
        $src = Sources::make((string)$b['source_format']);
        $n = new Normalizer((string)$b['encoding'], (string)$b['timezone']);
        $survey = (array)($stats['survey'] ?? []);
        $samples = (array)($stats['samples'] ?? []);
        $res = $src->walk(
            (string)$b['source_dir'],
            (array)($stats['units'] ?? []),
            json_decode((string)$b['cursor'], true) ?: [],
            self::CHUNK,
            $n,
            function (array $r) use (&$survey, &$samples): void {
                $key = (string)$r['area_key'];
                $s = (array)($survey[$key] ?? ['count' => 0, 'first' => null, 'last' => null]);
                $s['count']++;
                $d = $r['date'];
                if ($d !== null && ($s['first'] === null || $d < $s['first'])) {
                    $s['first'] = $d;
                }
                if ($d !== null && ($s['last'] === null || $d > $s['last'])) {
                    $s['last'] = $d;
                }
                $survey[$key] = $s;
                // Sample slots: earliest, roughly middle, latest of the
                // area that currently leads the count (§3.2).
                $lead = array_key_first($survey);
                foreach ($survey as $k => $row) {
                    if ((int)$row['count'] > (int)$survey[$lead]['count']) {
                        $lead = $k;
                    }
                }
                if ($key === $lead) {
                    $keep = fn(array $r): array => [
                        'from' => base64_encode((string)$r['from_raw']),
                        'to' => base64_encode((string)$r['to_raw']),
                        'subject' => base64_encode(substr((string)$r['subject_raw'], 0, 256)),
                        'body' => base64_encode(substr((string)$r['body_raw'], 0, 4096)),
                        'date' => $r['date'],
                    ];
                    if (($samples['area'] ?? '') !== $key) {
                        $samples = ['area' => $key, 'first' => $keep($r), 'mid' => null, 'last' => $keep($r), 'n' => 1];
                    } else {
                        $samples['n'] = (int)$samples['n'] + 1;
                        $samples['last'] = $keep($r);
                        if ((int)$samples['n'] === max(2, intdiv((int)$survey[$key]['count'], 2))) {
                            $samples['mid'] = $keep($r);
                        }
                    }
                }
            }
        );
        $stats['survey'] = $survey;
        $stats['samples'] = $samples;
        $scanned = 0;
        foreach ($survey as $s) {
            $scanned += (int)$s['count'];
        }
        $stats['scanned'] = $scanned;
        $stats['skipped'] = (int)($stats['skipped'] ?? 0) + (int)$res['skipped'];
        $done = (bool)$res['done'];
        if ($done) {
            $total = $scanned + (int)$stats['skipped'];
            if ($scanned === 0) {
                $db->q('UPDATE import_batches SET status = ?, error = ?, stats = ?, cursor = ? WHERE id = ?', [
                    'failed', 'Zero messages found. Nothing to import.',
                    (string)json_encode($stats), '{}', (int)$b['id'],
                ]);
                return (array)self::row($db, (int)$b['id']);
            }
            if ($total > 0 && (int)$stats['skipped'] / $total >= self::BAD_SHARE) {
                $db->q('UPDATE import_batches SET status = ?, error = ?, stats = ?, cursor = ? WHERE id = ?', [
                    'failed',
                    number_format((int)$stats['skipped']) . ' of ' . number_format($total)
                    . ' records would not parse (5 percent or more): this file is probably not what it claims to be.',
                    (string)json_encode($stats), '{}', (int)$b['id'],
                ]);
                return (array)self::row($db, (int)$b['id']);
            }
        }
        $db->q('UPDATE import_batches SET stats = ?, cursor = ?, total_messages = ?, status = ?, phase = ? WHERE id = ?', [
            (string)json_encode($stats),
            (string)json_encode($res['cursor']),
            $scanned,
            $done ? 'preview' : 'scanning',
            $done ? 'stage' : 'scan',
            (int)$b['id'],
        ]);
        return (array)self::row($db, (int)$b['id']);
    }

    // ---- confirm: mapping locked in, staging begins -----------------------

    /**
     * Store the mapping and create the new areas (§3.5 confirm). The
     * mapping: source_area_key => ['action' => skip|existing|create,
     * 'area_id' =>, 'tag' =>, 'name' =>, 'descr' =>, 'acs_read' =>,
     * 'acs_write' =>, 'category_id' =>].
     */
    public static function confirm(Db $db, Settings $settings, int $id, array $mapping): array
    {
        $b = self::row($db, $id);
        if ($b === null || (string)$b['status'] !== 'preview') {
            return ['ok' => false, 'error' => 'This batch is not waiting at the preview.'];
        }
        $stats = json_decode((string)$b['stats'], true) ?: [];
        $areas = (array)($stats['areas'] ?? []);
        $created = [];
        $resolved = [];
        foreach ((array)($stats['survey'] ?? []) as $key => $s) {
            $plan = (array)($mapping[$key] ?? ['action' => 'skip']);
            $action = (string)($plan['action'] ?? 'skip');
            if ($action === 'existing') {
                $area = $db->one('SELECT * FROM message_areas WHERE id = ?', [(int)($plan['area_id'] ?? 0)]);
                if ($area === null || (string)$area['kind'] === 'mail') {
                    return ['ok' => false, 'error' => "Area for \"$key\" is missing or is the private mail spool."];
                }
                $resolved[$key] = ['action' => 'existing', 'area_id' => (int)$area['id']];
                continue;
            }
            if ($action === 'create') {
                $tag = strtoupper(trim((string)($plan['tag'] ?? '')));
                if (!preg_match('/^[A-Z0-9_]{2,12}$/', $tag)) {
                    return ['ok' => false, 'error' => "\"$key\": the tag \"$tag\" is not 2-12 of A-Z 0-9 _."];
                }
                if ($db->one('SELECT id FROM message_areas WHERE tag = ?', [$tag]) !== null) {
                    return ['ok' => false, 'error' => "\"$key\": an area already wears the tag $tag."];
                }
                $sort = 10 + (int)$db->val('SELECT COALESCE(MAX(sort), 0) FROM message_areas');
                $aid = $db->insert('message_areas', [
                    'tag' => $tag,
                    'name' => mb_strimwidth(trim((string)($plan['name'] ?? $tag)), 0, 40),
                    'descr' => mb_strimwidth(trim((string)($plan['descr'] ?? '')), 0, 120),
                    'sort' => $sort,
                    'category_id' => (int)($plan['category_id'] ?? 0),
                    'acs_read' => trim((string)($plan['acs_read'] ?? '')),
                    'acs_write' => trim((string)($plan['acs_write'] ?? '')),
                    'kind' => 'local',
                    'echo_tag' => ($areas[$key]['echo_tag'] ?? null) !== null
                        ? (string)$areas[$key]['echo_tag'] : null,
                    'network_id' => null,
                    'listed' => 1,
                ]);
                $created[] = $aid;
                $resolved[$key] = ['action' => 'existing', 'area_id' => $aid, 'created' => true];
                continue;
            }
            $resolved[$key] = ['action' => 'skip'];
        }
        if (array_filter($resolved, fn($p) => $p['action'] !== 'skip') === []) {
            return ['ok' => false, 'error' => 'Every area is set to Skip; there is nothing to import.'];
        }
        $db->q('UPDATE import_batches SET mapping = ?, created_areas = ?, status = ?, phase = ?, cursor = ? WHERE id = ?', [
            (string)json_encode($resolved),
            (string)json_encode($created),
            'staging', 'stage', '{}', $id,
        ]);
        atdt_log_event('admin', 'import', "Import batch #$id confirmed: "
            . count($created) . ' area(s) created, staging begins');
        return ['ok' => true];
    }

    /** Update encoding/timezone while the batch waits at the preview. */
    public static function setRendering(Db $db, int $id, string $encoding, string $tz): void
    {
        if (!in_array($encoding, Normalizer::ENCODINGS, true)) {
            return;
        }
        try {
            new \DateTimeZone($tz === '' ? 'UTC' : $tz);
        } catch (\Throwable $e) {
            return;
        }
        $db->q("UPDATE import_batches SET encoding = ?, timezone = ? WHERE id = ? AND status = 'preview'",
            [$encoding, $tz, $id]);
    }

    // ---- phase: stage -----------------------------------------------------

    private static function stepStage(Db $db, array $b): array
    {
        $stats = json_decode((string)$b['stats'], true) ?: [];
        $mapping = json_decode((string)$b['mapping'], true) ?: [];
        $src = Sources::make((string)$b['source_format']);
        $n = new Normalizer((string)$b['encoding'], (string)$b['timezone']);
        $rows = [];
        $res = $src->walk(
            (string)$b['source_dir'],
            (array)($stats['units'] ?? []),
            json_decode((string)$b['cursor'], true) ?: [],
            self::CHUNK,
            $n,
            function (array $r) use (&$rows, $mapping, $b, $n): void {
                $plan = (array)($mapping[(string)$r['area_key']] ?? ['action' => 'skip']);
                if (($plan['action'] ?? 'skip') !== 'existing') {
                    return;
                }
                $rows[] = [
                    'batch_id' => (int)$b['id'],
                    'source_area_key' => (string)$r['area_key'],
                    'target_area_id' => (int)$plan['area_id'],
                    'date_utc' => (string)($r['date'] ?? ''),
                    'from_handle' => $n->handle((string)$r['from_raw']),
                    'to_handle' => $n->toHandle((string)$r['to_raw']),
                    'subject' => $n->subject((string)$r['subject_raw']),
                    'body' => $n->body((string)$r['body_raw']),
                    'source_msg_id' => mb_strimwidth((string)$r['id'], 0, 60),
                    'source_reply_to' => $r['reply_to'] === null ? null : mb_strimwidth((string)$r['reply_to'], 0, 60),
                    'new_msg_id' => 0,
                    'new_num' => 0,
                    'written' => 0,
                ];
            }
        );
        $db->tx(function (Db $db) use ($rows): void {
            foreach ($rows as $row) {
                $db->insert('import_staging', $row);
            }
        });
        $stats['substitutions'] = (int)($stats['substitutions'] ?? 0) + $n->substitutions;
        $staged = (int)$db->val('SELECT COUNT(*) FROM import_staging WHERE batch_id = ?', [(int)$b['id']]);
        $done = (bool)$res['done'];
        if ($done) {
            // Undated messages sort to the front of their area (§14.2):
            // the earliest dated stamp in the same area, minus a second.
            $stats['undated'] = (int)$db->val(
                "SELECT COUNT(*) FROM import_staging WHERE batch_id = ? AND date_utc = ''", [(int)$b['id']]
            );
            foreach ($db->all(
                "SELECT target_area_id, MIN(date_utc) AS lo FROM import_staging
                  WHERE batch_id = ? AND date_utc <> '' GROUP BY target_area_id", [(int)$b['id']]
            ) as $r) {
                $lo = (string)$r['lo'];
                $stamp = $lo === '' ? (string)$b['created_at']
                    : gmdate('Y-m-d\TH:i:s\Z', max(0, (int)strtotime($lo) - 1));
                $db->q("UPDATE import_staging SET date_utc = ? WHERE batch_id = ? AND target_area_id = ? AND date_utc = ''",
                    [$stamp, (int)$b['id'], (int)$r['target_area_id']]);
            }
            $db->q("UPDATE import_staging SET date_utc = ? WHERE batch_id = ? AND date_utc = ''",
                [(string)$b['created_at'], (int)$b['id']]);
        }
        $db->q('UPDATE import_batches SET stats = ?, cursor = ?, status = ?, phase = ?, total_messages = ? WHERE id = ?', [
            (string)json_encode($stats),
            (string)json_encode($res['cursor']),
            $done ? 'running' : 'staging',
            $done ? 'write' : 'stage',
            $done ? $staged : (int)$b['total_messages'],
            (int)$b['id'],
        ]);
        return (array)self::row($db, (int)$b['id']);
    }

    // ---- phase: write -----------------------------------------------------

    private static function stepWrite(Db $db, array $b): array
    {
        $stats = json_decode((string)$b['stats'], true) ?: [];
        $next = (array)($stats['nextnum'] ?? []);
        $host = (string)($stats['host'] ?? '') !== '' ? (string)$stats['host'] : 'atdt.local';
        $tags = [];
        $rows = $db->all(
            'SELECT * FROM import_staging WHERE batch_id = ? AND written = 0
              ORDER BY target_area_id, date_utc, id LIMIT ' . self::CHUNK,
            [(int)$b['id']]
        );
        if ($rows === []) {
            $db->q('UPDATE import_batches SET phase = ? WHERE id = ?', ['link', (int)$b['id']]);
            return (array)self::row($db, (int)$b['id']);
        }
        $db->tx(function (Db $db) use ($rows, &$next, &$tags, $host, $b): void {
            foreach ($rows as $r) {
                $aid = (int)$r['target_area_id'];
                if (!isset($next[$aid])) {
                    $next[$aid] = 1 + (int)$db->val(
                        'SELECT COALESCE(MAX(num), 0) FROM messages WHERE area_id = ?', [$aid]
                    );
                }
                $num = (int)$next[$aid];
                $next[$aid] = $num + 1;
                $tags[$aid] = $tags[$aid] ?? (string)($db->val('SELECT tag FROM message_areas WHERE id = ?', [$aid]) ?? 'AREA');
                $tag = $tags[$aid];
                $mid = $db->insert('messages', [
                    'area_id' => $aid,
                    'num' => $num,
                    'from_user_id' => null,
                    'from_name' => (string)$r['from_handle'],
                    'to_user_id' => null,
                    'to_name' => $r['to_handle'] === null || (string)$r['to_handle'] === ''
                        ? 'All' : (string)$r['to_handle'],
                    'subject' => (string)$r['subject'],
                    'body' => (string)$r['body'],
                    'posted_at' => (string)$r['date_utc'],
                    'reply_to_id' => null,
                    'msgid' => '<' . $tag . '.' . $num . '@' . $host . '>',
                    'reply_msgid' => null,
                    'origin' => null,
                    'kludges' => '{}',
                    'is_read' => 0,
                    'deleted' => 0,
                    'import_batch_id' => (int)$b['id'],
                ]);
                $db->q('UPDATE import_staging SET written = 1, new_msg_id = ?, new_num = ? WHERE id = ?',
                    [$mid, $num, (int)$r['id']]);
            }
        });
        $stats['nextnum'] = $next;
        $done = (int)$db->val(
            'SELECT COUNT(*) FROM import_staging WHERE batch_id = ? AND written > 0', [(int)$b['id']]
        );
        $db->q('UPDATE import_batches SET stats = ?, done_messages = ? WHERE id = ?', [
            (string)json_encode($stats), $done, (int)$b['id'],
        ]);
        return (array)self::row($db, (int)$b['id']);
    }

    // ---- phase: link ------------------------------------------------------

    private static function stepLink(Db $db, array $b): array
    {
        $useMap = (int)$b['total_messages'] <= self::LINK_MAP_MAX;
        $rows = $db->all(
            'SELECT * FROM import_staging WHERE batch_id = ? AND written = 1 LIMIT '
            . ($useMap ? self::LINK_CHUNK : self::CHUNK),
            [(int)$b['id']]
        );
        if ($rows === []) {
            $db->q('UPDATE import_batches SET phase = ? WHERE id = ?', ['finish', (int)$b['id']]);
            return (array)self::row($db, (int)$b['id']);
        }
        $stats = json_decode((string)$b['stats'], true) ?: [];
        $orphans = (int)($stats['orphans'] ?? 0);
        $cycles = (int)($stats['cycles'] ?? 0);
        if ($useMap) {
            // The whole batch's id map in memory (a few dozen bytes per
            // message): every parent lookup and cycle walk is then a
            // hash hit instead of its own query. The per-row road did
            // one query per message plus one per ancestor in the chain
            // walk, thousands of them a chunk, and the SysOp watched a
            // frozen progress line while they ran.
            $map = [];
            foreach ($db->all(
                'SELECT source_area_key, source_msg_id, source_reply_to, new_msg_id
                   FROM import_staging WHERE batch_id = ? AND written > 0', [(int)$b['id']]
            ) as $m) {
                $map[(string)$m['source_area_key'] . "\0" . (string)$m['source_msg_id']] = [
                    'new_msg_id' => (int)$m['new_msg_id'],
                    'source_reply_to' => $m['source_reply_to'],
                ];
            }
            $lookup = function (int $batch, string $areaKey, string $sid) use ($map): ?array {
                return $map[$areaKey . "\0" . $sid] ?? null;
            };
        } else {
            $lookup = function (int $batch, string $areaKey, string $sid) use ($db): ?array {
                return $db->one(
                    'SELECT id, new_msg_id, source_reply_to FROM import_staging
                      WHERE batch_id = ? AND source_area_key = ? AND source_msg_id = ? AND written > 0',
                    [$batch, $areaKey, $sid]
                );
            };
        }
        $db->tx(function (Db $db) use ($rows, $lookup, $b, &$orphans, &$cycles): void {
            foreach ($rows as $r) {
                $reply = $r['source_reply_to'];
                if ($reply !== null && (string)$reply !== '') {
                    $parent = $lookup((int)$b['id'], (string)$r['source_area_key'], (string)$reply);
                    if ($parent === null || (int)$parent['new_msg_id'] === 0) {
                        $orphans++;                       // §14.7: becomes a root
                    } else {
                        // Cycle check: walk the source chain with a cap.
                        $seen = [(string)$r['source_msg_id'] => true];
                        $cur = $parent;
                        $bad = false;
                        for ($i = 0; $i < 64 && $cur !== null; $i++) {
                            $sid = (string)($cur['source_reply_to'] ?? '');
                            if ($sid === '') {
                                break;
                            }
                            if (isset($seen[$sid])) {
                                $bad = true;
                                break;
                            }
                            $seen[$sid] = true;
                            $cur = $lookup((int)$b['id'], (string)$r['source_area_key'], $sid);
                        }
                        if ($bad) {
                            $cycles++;                    // break the loop: this one is a root
                        } else {
                            $pm = $db->one('SELECT msgid FROM messages WHERE id = ?', [(int)$parent['new_msg_id']]);
                            $db->q('UPDATE messages SET reply_to_id = ?, reply_msgid = ? WHERE id = ?', [
                                (int)$parent['new_msg_id'],
                                $pm === null ? null : (string)$pm['msgid'],
                                (int)$r['new_msg_id'],
                            ]);
                        }
                    }
                }
                $db->q('UPDATE import_staging SET written = 2 WHERE id = ?', [(int)$r['id']]);
            }
        });
        $stats['orphans'] = $orphans;
        $stats['cycles'] = $cycles;
        $db->q('UPDATE import_batches SET stats = ? WHERE id = ?', [(string)json_encode($stats), (int)$b['id']]);
        return (array)self::row($db, (int)$b['id']);
    }

    // ---- phase: finish ----------------------------------------------------

    private static function stepFinish(Db $db, Settings $settings, array $b): array
    {
        // §14.6: current callers never face the archive as unread mail
        // from 1987; the seed watermark protects accounts created later.
        $areas = $db->all(
            'SELECT DISTINCT target_area_id AS aid FROM import_staging WHERE batch_id = ?', [(int)$b['id']]
        );
        $users = $db->all('SELECT * FROM users');
        $seed = json_decode((string)($settings->get('import_seed', '{}') ?? '{}'), true) ?: [];
        foreach ($areas as $a) {
            $aid = (int)$a['aid'];
            $top = (int)$db->val('SELECT COALESCE(MAX(num), 0) FROM messages WHERE area_id = ?', [$aid]);
            foreach ($users as $u) {
                Messages::markRead($db, $u, $aid, $top);
            }
            $seed[(string)$aid] = max((int)($seed[(string)$aid] ?? 0), $top);
        }
        $settings->set('import_seed', (string)json_encode($seed));
        $stats = json_decode((string)$b['stats'], true) ?: [];
        $stats['touched'] = array_map(fn($a) => (int)$a['aid'], $areas);
        $db->q('DELETE FROM import_staging WHERE batch_id = ?', [(int)$b['id']]);
        $db->q('UPDATE import_batches SET status = ?, phase = ?, error = NULL, stats = ? WHERE id = ?',
            ['complete', 'finish', (string)json_encode($stats), (int)$b['id']]);
        self::unlock($db, $settings, (int)$b['id']);
        atdt_log_event('admin', 'import', sprintf(
            'Import batch #%d complete: %s message(s) into %d area(s); %d skipped, %d substitutions, '
            . '%d undated, %d orphan replies, %d cycles broken',
            (int)$b['id'],
            number_format((int)$b['done_messages']),
            count($areas),
            (int)($stats['skipped'] ?? 0),
            (int)($stats['substitutions'] ?? 0),
            (int)($stats['undated'] ?? 0),
            (int)($stats['orphans'] ?? 0),
            (int)($stats['cycles'] ?? 0)
        ));
        return (array)self::row($db, (int)$b['id']);
    }

    // ---- undo (§3.7) ------------------------------------------------------

    /** Chunked: call until it reports done. */
    public static function undo(Db $db, Settings $settings, int $id): array
    {
        $b = self::row($db, $id);
        if ($b === null) {
            return ['ok' => false, 'error' => 'No such batch.'];
        }
        $ids = $db->all('SELECT id, area_id FROM messages WHERE import_batch_id = ? LIMIT ' . self::CHUNK, [$id]);
        if ($ids !== []) {
            $db->tx(function (Db $db) use ($ids): void {
                foreach ($ids as $r) {
                    $db->q('DELETE FROM messages WHERE id = ?', [(int)$r['id']]);
                }
            });
            return ['ok' => true, 'done' => false, 'deleted' => count($ids)];
        }
        // Messages gone: drop areas this batch created that are now empty,
        // clamp the new-account seed, drain any leftover staging rows.
        $created = json_decode((string)$b['created_areas'], true) ?: [];
        $seed = json_decode((string)($settings->get('import_seed', '{}') ?? '{}'), true) ?: [];
        foreach ($created as $aid) {
            $aid = (int)$aid;
            $left = (int)$db->val('SELECT COUNT(*) FROM messages WHERE area_id = ?', [$aid]);
            if ($left === 0) {
                $db->q('DELETE FROM message_areas WHERE id = ?', [$aid]);
                $db->q('DELETE FROM lastread WHERE area_id = ?', [$aid]);
                $db->q('DELETE FROM read_marks WHERE area_id = ?', [$aid]);
                unset($seed[(string)$aid]);
            }
        }
        // A seed above the area's new top would hide the next organic
        // posts from new accounts; clamp it back to reality.
        foreach ($seed as $aid => $num) {
            $top = (int)$db->val('SELECT COALESCE(MAX(num), 0) FROM messages WHERE area_id = ?', [(int)$aid]);
            if ($top === 0) {
                unset($seed[$aid]);
            } elseif ((int)$num > $top) {
                $seed[$aid] = $top;
            }
        }
        $settings->set('import_seed', (string)json_encode($seed));
        $db->q('DELETE FROM import_staging WHERE batch_id = ?', [$id]);
        $db->q('UPDATE import_batches SET status = ? WHERE id = ?', ['undone', $id]);
        self::unlock($db, $settings, $id);
        atdt_log_event('admin', 'import', "Import batch #$id undone");
        return ['ok' => true, 'done' => true];
    }

    // ---- the mapping screen's tag prefill (§3.4) --------------------------

    /**
     * Suggest an ATDT tag from the source tag or name. Deliberately
     * never invents a GENERAL2: a collision or a too-short result
     * returns '' and the row is marked as needing the SysOp's call.
     * @param array<string, bool> $taken tags already used in this grid
     */
    public static function suggestTag(Db $db, string $sourceTag, string $sourceName, array $taken): string
    {
        $base = trim($sourceTag) !== '' ? $sourceTag : $sourceName;
        $tag = strtoupper($base);
        $tag = preg_replace('/[^A-Z0-9_]/', '_', $tag) ?? '';
        $tag = preg_replace('/_+/', '_', $tag) ?? '';
        $tag = trim($tag, '_');
        $tag = substr($tag, 0, 12);
        if (strlen($tag) < 2
            || isset($taken[$tag])
            || $db->one('SELECT id FROM message_areas WHERE tag = ?', [$tag]) !== null) {
            return '';
        }
        return $tag;
    }

    // ---- the advisory lock (§15.2) ----------------------------------------

    public static function lockHolder(Db $db, Settings $settings): ?int
    {
        $raw = (string)($settings->get('import_lock', '') ?? '');
        if ($raw === '' || !str_contains($raw, ':')) {
            return null;
        }
        [$id, $at] = explode(':', $raw, 2);
        if ((int)$at < time() - self::LOCK_SECONDS) {
            return null;                                 // expired: a crash never wedges
        }
        return (int)$id;
    }

    private static function lock(Db $db, Settings $settings, int $id): void
    {
        $settings->set('import_lock', $id . ':' . time());
    }

    private static function unlock(Db $db, Settings $settings, int $id): void
    {
        if (self::lockHolder($db, $settings) === $id) {
            $settings->set('import_lock', '');
        }
    }
}
