<?php

declare(strict_types=1);

class EntityTranslateProcessor
{
    private const CRON_INTERVAL          = 5;
    private const MAX_TICK_SECONDS       = 50;
    private const POOL_SIZE              = 5;
    private const STALE_SECONDS          = 300;
    private const CONSECUTIVE_FAIL_PAUSE = 3;
    private const COOLDOWN_SECONDS       = 60;
    private const MAX_ITEM_ATTEMPTS      = 5;
    private const BATCH_UNITS            = 100;
    private const MIN_BATCH              = 5;
    private const MAX_CHARGE_SPLITS      = 6;
    private const REFUND_SETTLE_LIMIT    = 25;
    private const REFUND_MAX_ATTEMPTS    = 6;
    private const REFUND_HUB_CEILING     = 2000.0;

    public static function typeMeta(string $entityType): ?array
    {
        return TranslateTypes::meta($entityType);
    }

    public static function autoEnabled(string $entityType): bool
    {
        return Setting::get('entity_autotranslate_' . $entityType, '0') === '1';
    }

    public static function syncAutoBlockedFlag(): void
    {
        foreach (TranslateTypes::keys() as $type) {
            if (self::autoEnabled($type)) return;
        }
        Setting::set('entity_autotranslate_blocked', '0');
        Setting::set('entity_autotranslate_blocked_msg', '');
    }

    public static function registerCron(): void
    {
        CronManager::register('entity_translate', self::CRON_INTERVAL, [self::class, 'tick']);
        CronManager::register('entity_autotranslate', 120, [self::class, 'autoTick']);
    }

    public static function targetLanguages(): array
    {
        return array_values(array_filter(
            Translator::available(),
            fn($l) => !($l['is_default'] ?? false)
        ));
    }

    public static function targetLanguageCodes(): array
    {
        return array_column(self::targetLanguages(), 'code');
    }

    

    public static function createJob(array $spec): int
    {
        $type = (string) $spec['entity_type'];
        $meta = self::typeMeta($type);
        if ($meta === null) {
            throw new \InvalidArgumentException('Unknown entity type: ' . $type);
        }

        $scope = ($spec['scope'] ?? 'ids') === 'all' ? 'all' : 'ids';
        $ids = array_values(array_unique(array_map('intval', $spec['ids'] ?? [])));
        sort($ids);
        $langs = array_values(array_unique(array_map('strval', $spec['languages'] ?? [])));
        $fields = TranslateTypes::fieldKeys($meta, $spec['fields'] ?? null);
        $typeFilter = (string) ($spec['type_filter'] ?? '');

        $total = $scope === 'all'
            ? self::countScope($meta, $typeFilter)
            : count($ids);

        $id = Database::insert('entity_translate_jobs', [
            'entity_type' => $type,
            'scope'       => $scope,
            'type_filter' => $typeFilter,
            'entity_ids'  => json_encode($scope === 'all' ? [] : $ids),
            'languages'   => json_encode($langs),
            'fields_json' => json_encode($fields),
            'total_items' => $total,
            'user_id'     => (int) ($spec['user_id'] ?? 0),
        ]);

        Database::query("UPDATE settings SET setting_value = '0' WHERE setting_key = 'cron_entity_translate_last'");
        self::asyncPing();

        return $id;
    }

    private static function countScope(array $meta, string $typeFilter): int
    {
        $sql = 'SELECT COUNT(*) AS c FROM ' . $meta['main'];
        $params = [];
        if ($typeFilter !== '' && !empty($meta['type_col'])) {
            $sql .= ' WHERE ' . $meta['type_col'] . ' = ?';
            $params[] = $typeFilter;
        }
        return (int) (Database::fetchOne($sql, $params)['c'] ?? 0);
    }

    public static function getJob(int $jobId): ?array
    {
        return Database::fetchOne('SELECT * FROM entity_translate_jobs WHERE id = ?', [$jobId]);
    }

    public static function getActiveJob(string $entityType): ?array
    {
        return Database::fetchOne(
            "SELECT * FROM entity_translate_jobs WHERE entity_type = ? AND status IN ('pending','processing','paused') ORDER BY id DESC LIMIT 1",
            [$entityType]
        );
    }

    

    public static function tick(): void
    {
        self::recoverStalled();
        self::sweepRefunds();

        $job = Database::fetchOne(
            "SELECT * FROM entity_translate_jobs WHERE status IN ('pending','processing') ORDER BY id ASC LIMIT 1"
        );
        if (!$job) return;

        $cooldown = (string) ($job['pause_reason'] ?? '');
        if (str_starts_with($cooldown, 'cooldown:')) {
            $until = (int) substr($cooldown, 9);
            if (time() < $until) return;
            Database::query('UPDATE entity_translate_jobs SET pause_reason = NULL WHERE id = ?', [(int) $job['id']]);
        }

        $lockName = 'etrans_job_' . (int) $job['id'];
        if (!(Database::fetchOne('SELECT GET_LOCK(?, 0) AS l', [$lockName])['l'] ?? 0)) return;

        try {
            if ($job['status'] === 'pending') {
                Database::query(
                    "UPDATE entity_translate_jobs SET status = 'processing', started_at = COALESCE(started_at, NOW()) WHERE id = ?",
                    [(int) $job['id']]
                );
            }
            self::processJob((int) $job['id'], time());
        } catch (\Throwable $e) {
            error_log('EntityTranslateProcessor: Exception on job #' . $job['id'] . ': ' . $e->getMessage());
            Database::query(
                'UPDATE entity_translate_jobs SET error_message = ? WHERE id = ?',
                [mb_substr($e->getMessage(), 0, 1000), (int) $job['id']]
            );
        } finally {
            Database::query('SELECT RELEASE_LOCK(?)', [$lockName]);
        }
    }

    

    private static function processJob(int $jobId, int $startTime): void
    {
        while (true) {
            $job = self::getJob($jobId);
            if (!$job || !in_array($job['status'], ['pending', 'processing'], true)) return;

            $meta = self::typeMeta((string) $job['entity_type']);
            if ($meta === null) {
                self::failJob($jobId, 'Unknown entity type: ' . $job['entity_type']);
                return;
            }

            $langs = json_decode((string) $job['languages'], true);
            $fields = json_decode((string) $job['fields_json'], true);
            if (!is_array($langs) || !is_array($fields) || empty($langs) || empty($fields)) {
                self::failJob($jobId, 'Corrupted job definition.');
                return;
            }
            $fields = TranslateTypes::fieldKeys($meta, $fields);

            $rows = self::nextBatch($job, $meta, count($fields));
            if (empty($rows)) {
                self::finishJob($jobId);
                return;
            }

            $lastId = 0;
            foreach ($rows as $r) { $lastId = max($lastId, (int) $r['id']); }

            $outcome = self::processBatch($jobId, $job, $meta, $rows, $fields, $langs, $startTime);
            if ($outcome === 'paused' || $outcome === 'stop') return;

            if ($outcome === 'partial') {
                Database::query("UPDATE settings SET setting_value = '0' WHERE setting_key = 'cron_entity_translate_last'");
                self::asyncPing();
                return;
            }

            Database::query(
                'UPDATE entity_translate_jobs
                    SET cursor_id = ?, batch_seq = batch_seq + 1,
                        processed_items = processed_items + ?, updated_at = NOW()
                  WHERE id = ?',
                [$lastId, count($rows), $jobId]
            );

            if ((time() - $startTime) >= self::MAX_TICK_SECONDS) {
                Database::query("UPDATE settings SET setting_value = '0' WHERE setting_key = 'cron_entity_translate_last'");
                self::asyncPing();
                return;
            }
        }
    }

    private static function nextBatch(array $job, array $meta, int $fieldCount): array
    {
        $batchSize = max(self::MIN_BATCH, (int) floor(self::BATCH_UNITS / max(1, $fieldCount)));
        $cursor = (int) $job['cursor_id'];

        $cols = ['id'];
        if (!empty($meta['type_col'])) { $cols[] = $meta['type_col']; }
        foreach ($meta['fields'] as $f) { $cols[] = $f['src']; }
        $cols = array_values(array_unique($cols));
        $select = implode(', ', $cols);

        if (($job['scope'] ?? 'ids') === 'all') {
            $sql = "SELECT {$select} FROM {$meta['main']} WHERE id > ?";
            $params = [$cursor];
            if (!empty($job['type_filter']) && !empty($meta['type_col'])) {
                $sql .= ' AND ' . $meta['type_col'] . ' = ?';
                $params[] = $job['type_filter'];
            }
            $sql .= ' ORDER BY id ASC LIMIT ' . $batchSize;
            return Database::fetchAll($sql, $params);
        }

        $ids = json_decode((string) $job['entity_ids'], true);
        if (!is_array($ids) || empty($ids)) return [];
        $slice = [];
        foreach ($ids as $id) {
            $id = (int) $id;
            if ($id > $cursor) { $slice[] = $id; }
            if (count($slice) >= $batchSize) break;
        }
        if (empty($slice)) return [];
        $ph = implode(',', array_fill(0, count($slice), '?'));
        return Database::fetchAll("SELECT {$select} FROM {$meta['main']} WHERE id IN ({$ph}) ORDER BY id ASC", $slice);
    }

    

    private static function processBatch(int $jobId, array $job, array $meta, array $rows, array $fields, array $langs, int $startTime): string
    {
        $ids = array_map(fn($r) => (int) $r['id'], $rows);
        $existing = self::existingTranslations($meta, $ids, $langs);
        $doneBefore = self::countFullyDone($meta, $ids, $fields, $langs, $rows, $existing);

        $work = [];
        foreach ($rows as $row) {
            $id = (int) $row['id'];
            foreach ($fields as $fieldKey) {
                $field = $meta['fields'][$fieldKey];
                $src = trim((string) ($row[$field['src']] ?? ''));
                if ($src === '' || mb_strlen($src) > (int) $field['max']) continue;

                $missing = [];
                foreach ($langs as $lang) {
                    $val = $existing[$id][$lang][$field['dst']] ?? '';
                    if (trim((string) $val) === '') { $missing[] = $lang; }
                }
                if (empty($missing)) continue;

                $work[] = [
                    'id'    => $id,
                    'field' => $fieldKey,
                    'src'   => $src,
                    'langs' => $missing,
                    'type'  => !empty($meta['type_col']) ? (string) ($row[$meta['type_col']] ?? '') : '',
                ];
            }
        }

        if (empty($work)) {
            return 'ok';
        }

        if (!CatalogueClient::isConfigured()) {
            self::pauseJob($jobId, 'Connect your TubePress account to run AI translation.');
            return 'paused';
        }

        $chargedMap = json_decode((string) ($job['charged_map'] ?? '[]'), true);
        if (!is_array($chargedMap)) {
            $chargedMap = [];
        }

        $billable = [];
        foreach ($work as $w) {
            $paid = $chargedMap[$w['id'] . ':' . $w['field']] ?? [];
            $need = is_array($paid) ? array_values(array_diff($w['langs'], $paid)) : $w['langs'];
            if (!empty($need)) {
                $b = $w;
                $b['langs'] = $need;
                $billable[] = $b;
            }
        }

        if (!empty($billable)) {
            $charged = self::chargeBatch($jobId, $job, $meta, $billable, $fields);
            if ($charged === null) {
                return 'paused';
            }
            foreach ($billable as $b) {
                $key = $b['id'] . ':' . $b['field'];
                $chargedMap[$key] = array_values(array_unique(array_merge($chargedMap[$key] ?? [], $b['langs'])));
            }
            Database::query(
                'UPDATE entity_translate_jobs SET charged_map = ?, updated_at = NOW() WHERE id = ?',
                [json_encode($chargedMap), $jobId]
            );
        }

        self::prefillFromCache($meta, $work, $existing);

        $remaining = [];
        foreach ($work as $w) {
            $field = $meta['fields'][$w['field']];
            $still = [];
            foreach ($w['langs'] as $lang) {
                $val = $existing[$w['id']][$lang][$field['dst']] ?? '';
                if (trim((string) $val) === '') { $still[] = $lang; }
            }
            if (!empty($still)) { $w['langs'] = $still; $remaining[] = $w; }
        }
        if (empty($remaining)) {
            self::bumpTranslated($jobId, max(0, self::countFullyDone($meta, $ids, $fields, $langs, $rows) - $doneBefore));
            return 'ok';
        }

        if (!CatalogueClient::pingHealth()) {
            self::pauseJob($jobId, 'cooldown:' . (time() + self::COOLDOWN_SECONDS));
            return 'paused';
        }

        $aiConfig = CatalogueClient::getAiConfig();
        if ($aiConfig === null) {
            self::pauseJob($jobId, 'cooldown:' . (time() + self::COOLDOWN_SECONDS));
            return 'paused';
        }

        $languageMeta = [];
        foreach (self::targetLanguages() as $l) {
            $languageMeta[$l['code']] = $l['name'];
        }

        $attempts = json_decode((string) ($job['attempt_map'] ?? '[]'), true);
        if (!is_array($attempts)) $attempts = [];

        $chunks = self::planCalls($meta, $remaining);

        $requestMap = [];
        $requestIdx = 0;
        $consecutiveFails = 0;
        $apiDown = false;
        $cacheWrites = [];

        $truncated = false;

        $iterator = (function () use (&$requestMap, &$requestIdx, &$apiDown, &$truncated, $chunks, $meta, $aiConfig, $languageMeta, $startTime, &$attempts) {
            foreach ($chunks as $w) {
                if ($apiDown) break;
                if ($requestIdx > 0 && (time() - $startTime) >= self::MAX_TICK_SECONDS) {
                    $truncated = true;
                    break;
                }

                $key = $w['id'] . ':' . $w['field'];
                if ((int) ($attempts[$key] ?? 0) >= self::MAX_ITEM_ATTEMPTS) continue;

                $field = $meta['fields'][$w['field']];
                $prompt = self::buildPrompt($aiConfig, $field, $w['src'], $w['langs'], $languageMeta);

                $idx = $requestIdx++;
                $requestMap[$idx] = $w;
                yield $idx => ['messages' => [['role' => 'user', 'content' => $prompt]], 'temperature' => 0.3];
            }
        })();

        CatalogueClient::aiCompletionPool(
            $iterator,
            function (int $idx, ?array $aiResult) use (&$requestMap, &$attempts, &$consecutiveFails, &$apiDown, &$cacheWrites, $meta) {
                $w = $requestMap[$idx] ?? null;
                if (!$w) return;
                unset($requestMap[$idx]);
                $key = $w['id'] . ':' . $w['field'];

                if ($aiResult === null) {
                    $attempts[$key] = (int) ($attempts[$key] ?? 0) + 1;
                    $consecutiveFails++;
                    if ($consecutiveFails >= self::CONSECUTIVE_FAIL_PAUSE) { $apiDown = true; }
                    return;
                }
                $consecutiveFails = 0;

                $content = $aiResult['choices'][0]['message']['content'] ?? null;
                if ($content === null) {
                    $attempts[$key] = (int) ($attempts[$key] ?? 0) + 1;
                    return;
                }

                $field = $meta['fields'][$w['field']];
                $saved = false;
                foreach ($w['langs'] as $lang) {
                    $tag = strtoupper($lang);
                    if (!preg_match('/<' . preg_quote($tag, '/') . '>(.*?)<\/' . preg_quote($tag, '/') . '>/si', $content, $m)) continue;
                    $text = trim($m[1]);
                    if ($text === '' || mb_strlen($text) > (int) $field['max'] * 2) continue;

                    self::persist($meta, $field, $w['id'], $lang, $text, $w['type'], $w['src']);
                    $saved = true;
                    if (!empty($field['cache'])) {
                        $cacheWrites[] = ['name' => $w['src'], 'video_type' => $w['type'], 'lang' => $lang, 'translated' => $text];
                    }
                }
                if (!$saved) { $attempts[$key] = (int) ($attempts[$key] ?? 0) + 1; }
            },
            self::POOL_SIZE
        );

        if (!empty($cacheWrites) && Setting::get('entity_translation_cache_enabled', '1') === '1') {
            CatalogueClient::translateEntitiesStore($cacheWrites);
        }

        $failed = 0;
        foreach ($attempts as $n) { if ((int) $n >= self::MAX_ITEM_ATTEMPTS) $failed++; }

        Database::query(
            'UPDATE entity_translate_jobs SET attempt_map = ?, failed_count = ?, updated_at = NOW() WHERE id = ?',
            [json_encode($attempts), $failed, $jobId]
        );

        self::bumpTranslated($jobId, max(0, self::countFullyDone($meta, $ids, $fields, $langs, $rows) - $doneBefore));
        self::markSitemapDirty();

        if ($apiDown) {
            self::pauseJob($jobId, 'cooldown:' . (time() + self::COOLDOWN_SECONDS));
            return 'paused';
        }

        if ($truncated) {
            return 'partial';
        }

        return 'ok';
    }

    

    /**
     * One work item becomes one or more AI calls: long fields are split across several calls so
     * a reply never has to carry every language at once. Splitting changes nothing about the
     * price (that is per item, per field, per language) — only how the work is issued.
     *
     * @return array<int, array{id:int,field:string,src:string,langs:array,type:string}>
     */
    private static function planCalls(array $meta, array $work): array
    {
        $calls = [];
        foreach ($work as $w) {
            $kind = $meta['fields'][$w['field']]['kind'];
            $per = TranslateTypes::langsPerCall($kind, mb_strlen($w['src']));
            foreach (array_chunk($w['langs'], $per) as $slice) {
                $calls[] = ['id' => $w['id'], 'field' => $w['field'], 'src' => $w['src'], 'langs' => $slice, 'type' => $w['type']];
            }
        }
        return $calls;
    }

    private static function countPlannedCalls(array $meta, array $work): int
    {
        return count(self::planCalls($meta, $work));
    }

    private static function chargeBatch(int $jobId, array $job, array $meta, array $work, array $fields): ?float
    {
        $groups = [];
        foreach ($work as $w) {
            $sig = implode(',', $w['langs']);
            $kind = $meta['fields'][$w['field']]['kind'];
            $groups[$sig]['langs'] = $w['langs'];
            $groups[$sig]['fields'][$kind] = ($groups[$sig]['fields'][$kind] ?? 0) + 1;
            $groups[$sig]['calls'] = ($groups[$sig]['calls'] ?? 0)
                + (int) ceil(count($w['langs']) / TranslateTypes::langsPerCall($kind, mb_strlen($w['src'])));
            foreach ($w['langs'] as $lang) {
                $groups[$sig]['items'][] = $w['id'] . ':' . $w['field'] . ':' . $lang;
            }
        }

        $total = 0.0;
        $seq = (int) $job['batch_seq'];
        $g = 0;
        foreach ($groups as $group) {
            $items = $group['items'] ?? [];
            sort($items);
            $charged = self::chargeGroup(
                $jobId,
                $group,
                'translate:' . $jobId . ':' . $seq . ':' . substr(sha1(implode('|', $items)), 0, 16)
            );
            $g++;
            if ($charged === null) {
                return null;
            }
            $total += $charged;
        }

        if ($total > 0) {
            Database::query(
                'UPDATE entity_translate_jobs SET charged_credits = charged_credits + ? WHERE id = ?',
                [$total, $jobId]
            );
        }
        return $total;
    }

    

    /**
     * Charge one language-signature group, splitting it when the server rejects it as too
     * large for a single call. A high per-language price (a page body, say) can push even a
     * modest batch past the platform's per-charge ceiling, and the operator should never see
     * a run die for that. Each half gets its own stable reference, so a resumed job re-uses
     * them and is never billed twice.
     */
    private static function chargeGroup(int $jobId, array $group, string $ref, int $depth = 0): ?float
    {
        $res = CatalogueClient::translateCharge($group['fields'], $group['langs'], $ref, (int) ($group['calls'] ?? 0));

        if ($res === null) {
            self::pauseJob($jobId, 'cooldown:' . (time() + self::COOLDOWN_SECONDS));
            return null;
        }

        if (empty($res['_error']) && !empty($res['success'])) {
            if (!empty($res['idempotent'])) {
                return 0.0;
            }
            $charged = (float) ($res['charged'] ?? 0);
            self::recordCharge($jobId, $ref, $charged, $group['items'] ?? []);
            return $charged;
        }

        $code = (int) ($res['_http_code'] ?? 0);
        $message = (string) ($res['error'] ?? '');

        if ($code === 400 && stripos($message, 'too large') !== false && $depth < self::MAX_CHARGE_SPLITS) {
            $halves = self::splitGroup($group);
            if ($halves === null) {
                self::pauseJob($jobId, 'One item alone costs more than a single charge allows. Lower the per-language price or pick fewer languages.');
                return null;
            }
            $sum = 0.0;
            foreach ($halves as $i => $half) {
                $part = self::chargeGroup($jobId, $half, $ref . '.' . $i, $depth + 1);
                if ($part === null) {
                    return null;
                }
                $sum += $part;
            }
            return $sum;
        }

        if ($code === 402) {
            $need = $res['required'] ?? '?';
            $have = $res['balance'] ?? '?';
            self::pauseJob($jobId, 'Not enough credits: this batch needs ' . $need . ' Cr and your balance is ' . $have . ' Cr. Top up, then resume.');
            Setting::set('entity_autotranslate_blocked', '1');
            Setting::set('entity_autotranslate_blocked_msg', 'AI translation paused — insufficient credits. Top up to resume.');
        } else {
            self::pauseJob($jobId, 'Billing refused this batch: ' . mb_substr($message !== '' ? $message : 'unknown error', 0, 120));
        }
        return null;
    }

    /** Halve a charge group by item count. Returns null when nothing can be split further. */
    private static function splitGroup(array $group): ?array
    {
        $totalItems = (int) array_sum($group['fields']);
        if ($totalItems < 2) {
            return null;
        }
        $left = ['langs' => $group['langs'], 'fields' => [], 'calls' => 0];
        $right = ['langs' => $group['langs'], 'fields' => [], 'calls' => 0];
        $moved = 0;
        $half = (int) floor($totalItems / 2);
        foreach ($group['fields'] as $kind => $n) {
            $take = max(0, min($n, $half - $moved));
            if ($take > 0) { $left['fields'][$kind] = $take; $moved += $take; }
            if ($n - $take > 0) { $right['fields'][$kind] = $n - $take; }
        }
        $calls = (int) ($group['calls'] ?? 0);
        $left['calls'] = (int) ceil($calls / 2);
        $right['calls'] = $calls - $left['calls'];
        if (empty($left['fields']) || empty($right['fields'])) {
            return null;
        }

        $items = $group['items'] ?? [];
        if (!empty($items)) {
            $cut = (int) round(count($items) * ($moved / max(1, $totalItems)));
            $cut = max(0, min(count($items), $cut));
            $left['items']  = array_slice($items, 0, $cut);
            $right['items'] = array_slice($items, $cut);
        }

        return [$left, $right];
    }

    private static function ensureChargeTable(): void
    {
        static $checked = false;
        if ($checked) {
            return;
        }
        $checked = true;
        try {
            Database::fetchOne('SELECT 1 FROM entity_translate_charges LIMIT 1');
        } catch (\PDOException) {
            $sql = @file_get_contents(APP_ROOT . '/sql/128_entity_translate_charges.sql');
            if (!is_string($sql) || trim($sql) === '') {
                return;
            }
            foreach (array_filter(array_map('trim', explode(';', $sql))) as $statement) {
                try {
                    Database::get()->exec($statement);
                } catch (\PDOException) {
                }
            }
        }
    }

    private static function recordCharge(int $jobId, string $ref, float $amount, array $units): void
    {
        if ($amount <= 0) {
            return;
        }
        try {
            self::ensureChargeTable();
            $units = array_values(array_unique(array_map('strval', $units)));
            Database::query(
                'INSERT INTO entity_translate_charges (job_id, reference_id, amount, unit_count, units)
                 VALUES (?, ?, ?, ?, ?)
                 ON DUPLICATE KEY UPDATE
                     amount     = IF(refunded_at IS NULL, VALUES(amount), amount),
                     unit_count = IF(refunded_at IS NULL, VALUES(unit_count), unit_count),
                     units      = IF(refunded_at IS NULL, VALUES(units), units)',
                [$jobId, $ref, $amount, count($units), json_encode($units)]
            );
        } catch (\Throwable $e) {
            error_log('EntityTranslateProcessor: charge ledger write failed for ' . $ref . ': ' . $e->getMessage());
        }
    }

    public static function settleRefunds(int $jobId, int $limit = self::REFUND_SETTLE_LIMIT): float
    {
        $job = self::getJob($jobId);
        if (!$job || !in_array((string) $job['status'], ['completed', 'failed', 'cancelled'], true)) {
            return 0.0;
        }
        if (!CatalogueClient::isConfigured()) {
            return 0.0;
        }

        try {
            self::ensureChargeTable();
            $rows = Database::fetchAll(
                'SELECT * FROM entity_translate_charges
                  WHERE job_id = ? AND refunded_at IS NULL AND refund_attempts < ?
                  ORDER BY id ASC LIMIT ' . max(1, (int) $limit),
                [$jobId, self::REFUND_MAX_ATTEMPTS]
            );
        } catch (\Throwable $e) {
            error_log('EntityTranslateProcessor: refund scan failed on job #' . $jobId . ': ' . $e->getMessage());
            return 0.0;
        }

        $total = 0.0;
        foreach ($rows as $row) {
            $total += self::settleCharge($job, $row);
        }

        $total = round($total, 2);
        if ($total > 0) {
            Database::query(
                'UPDATE entity_translate_jobs SET refund_amount = refund_amount + ?, refunded_at = NOW() WHERE id = ?',
                [$total, $jobId]
            );
        }
        return $total;
    }

    public static function sweepRefunds(): void
    {
        if (!CatalogueClient::isConfigured()) {
            return;
        }
        try {
            self::ensureChargeTable();
            $jobs = Database::fetchAll(
                "SELECT DISTINCT c.job_id AS job_id
                   FROM entity_translate_charges c
                   JOIN entity_translate_jobs j ON j.id = c.job_id
                  WHERE c.refunded_at IS NULL AND c.refund_attempts < ?
                    AND j.status IN ('completed','failed','cancelled')
                  ORDER BY c.job_id ASC LIMIT 3",
                [self::REFUND_MAX_ATTEMPTS]
            );
        } catch (\Throwable) {
            return;
        }
        foreach ($jobs as $j) {
            self::settleRefunds((int) $j['job_id'], 10);
        }
    }

    private static function settleCharge(array $job, array $row): float
    {
        $lock = 'etrans_refund_' . (int) $row['id'];
        if (!(int) (Database::fetchOne('SELECT GET_LOCK(?, 0) AS l', [$lock])['l'] ?? 0)) {
            return 0.0;
        }
        try {
            $fresh = Database::fetchOne('SELECT * FROM entity_translate_charges WHERE id = ?', [(int) $row['id']]);
            if (!$fresh || $fresh['refunded_at'] !== null) {
                return 0.0;
            }
            return self::settleChargeLocked($job, $fresh);
        } finally {
            Database::query('SELECT RELEASE_LOCK(?)', [$lock]);
        }
    }

    private static function settleChargeLocked(array $job, array $row): float
    {
        $id        = (int) $row['id'];
        $ref       = (string) $row['reference_id'];
        $amount    = round((float) $row['amount'], 2);
        $unitCount = (int) $row['unit_count'];
        $units     = json_decode((string) ($row['units'] ?? '[]'), true);
        if (!is_array($units)) {
            $units = [];
        }

        if ($amount <= 0 || $unitCount <= 0 || empty($units)) {
            self::closeCharge($id, 0.0, 'nothing to measure');
            return 0.0;
        }

        $delivered = self::deliveredUnits($job, $units);
        if ($delivered === null) {
            self::bumpChargeAttempt($id, 'could not read what was delivered');
            return 0.0;
        }

        $undelivered = max(0, $unitCount - $delivered);
        if ($undelivered === 0) {
            self::closeCharge($id, 0.0, null);
            return 0.0;
        }

        $refund = round($amount * ($undelivered / $unitCount), 2);
        if ($refund > $amount) {
            $refund = $amount;
        }
        if ($refund < 0.01) {
            self::closeCharge($id, 0.0, null);
            return 0.0;
        }
        $capped = $refund > self::REFUND_HUB_CEILING;
        if ($capped) {
            $refund = self::REFUND_HUB_CEILING;
        }

        $claimed = Database::query(
            'UPDATE entity_translate_charges SET refund_attempts = refund_attempts + 1, refund_amount = ?
              WHERE id = ? AND refunded_at IS NULL',
            [$refund, $id]
        )->rowCount();
        if ($claimed === 0) {
            return 0.0;
        }

        $res = CatalogueClient::refundCredits(
            $refund,
            'AI translation not delivered — job #' . (int) $job['id'] . ' (' . $undelivered . '/' . $unitCount . ' untranslated)',
            $ref,
            1
        );

        if (is_array($res) && empty($res['_error']) && !empty($res['success'])) {
            $given = round((float) ($res['refunded'] ?? $refund), 2);
            self::closeCharge($id, $given, $capped ? 'capped at the platform maximum' : null);
            error_log('EntityTranslateProcessor: refunded ' . $given . ' Cr on ' . $ref . ' (' . $undelivered . '/' . $unitCount . ' undelivered)');
            return $given;
        }

        $code = is_array($res) ? (int) ($res['_http_code'] ?? 0) : 0;
        $msg  = is_array($res) ? (string) ($res['error'] ?? '') : 'no answer from the platform';

        if (in_array($code, [400, 403, 404, 409], true)) {
            $settled = $code === 409 ? $refund : 0.0;
            self::closeCharge($id, $settled, mb_substr($msg !== '' ? $msg : 'refused (HTTP ' . $code . ')', 0, 255));
            return $settled;
        }

        self::noteChargeError($id, mb_substr($msg, 0, 255));
        return 0.0;
    }

    private static function closeCharge(int $id, float $amount, ?string $error): void
    {
        Database::query(
            'UPDATE entity_translate_charges SET refund_amount = ?, refunded_at = NOW(), refund_error = ? WHERE id = ? AND refunded_at IS NULL',
            [round($amount, 2), $error, $id]
        );
    }

    private static function bumpChargeAttempt(int $id, ?string $error): void
    {
        Database::query(
            'UPDATE entity_translate_charges SET refund_attempts = refund_attempts + 1, refund_error = ? WHERE id = ? AND refunded_at IS NULL',
            [$error, $id]
        );
    }

    private static function noteChargeError(int $id, ?string $error): void
    {
        Database::query(
            'UPDATE entity_translate_charges SET refund_error = ? WHERE id = ? AND refunded_at IS NULL',
            [$error, $id]
        );
    }

    private static function deliveredUnits(array $job, array $units): ?int
    {
        $meta = self::typeMeta((string) $job['entity_type']);
        if ($meta === null) {
            return null;
        }

        $wanted = [];
        foreach ($units as $unit) {
            $parts = explode(':', (string) $unit);
            if (count($parts) < 3) continue;
            $entityId = (int) array_shift($parts);
            $lang     = (string) array_pop($parts);
            $field    = implode(':', $parts);
            if ($entityId <= 0 || $lang === '' || !isset($meta['fields'][$field])) continue;
            $wanted[$field][$entityId][$lang] = true;
        }
        if (empty($wanted)) {
            return null;
        }

        $delivered = 0;
        try {
            foreach ($wanted as $field => $byId) {
                $col   = $meta['fields'][$field]['dst'];
                $ids   = array_keys($byId);
                $langs = [];
                foreach ($byId as $set) {
                    foreach (array_keys($set) as $l) { $langs[$l] = true; }
                }
                $langs = array_keys($langs);

                $iph = implode(',', array_fill(0, count($ids), '?'));
                $lph = implode(',', array_fill(0, count($langs), '?'));
                $rows = Database::fetchAll(
                    "SELECT {$meta['fk']} AS entity_id, lang_code, {$col} AS translated
                       FROM {$meta['translations']}
                      WHERE {$meta['fk']} IN ({$iph}) AND lang_code IN ({$lph})",
                    array_merge($ids, $langs)
                );
                foreach ($rows as $r) {
                    $entityId = (int) $r['entity_id'];
                    $lang     = (string) $r['lang_code'];
                    if (!isset($byId[$entityId][$lang])) continue;
                    if (trim((string) ($r['translated'] ?? '')) !== '') $delivered++;
                }
            }
        } catch (\Throwable $e) {
            error_log('EntityTranslateProcessor: delivery check failed on job #' . (int) $job['id'] . ': ' . $e->getMessage());
            return null;
        }

        return $delivered;
    }

    private static function unbillRefunded(int $jobId, array $job): void
    {
        try {
            self::ensureChargeTable();
            $rows = Database::fetchAll(
                'SELECT units FROM entity_translate_charges WHERE job_id = ? AND refunded_at IS NOT NULL AND refund_amount > 0',
                [$jobId]
            );
        } catch (\Throwable) {
            return;
        }
        if (empty($rows)) {
            return;
        }

        $map = json_decode((string) ($job['charged_map'] ?? '[]'), true);
        if (!is_array($map) || empty($map)) {
            return;
        }

        $changed = false;
        foreach ($rows as $row) {
            $units = json_decode((string) ($row['units'] ?? '[]'), true);
            if (!is_array($units)) continue;
            foreach ($units as $unit) {
                $parts = explode(':', (string) $unit);
                if (count($parts) < 3) continue;
                $lang = (string) array_pop($parts);
                $key  = implode(':', $parts);
                if (!isset($map[$key]) || !is_array($map[$key])) continue;
                $left = array_values(array_diff($map[$key], [$lang]));
                if (count($left) === count($map[$key])) continue;
                $changed = true;
                if (empty($left)) { unset($map[$key]); } else { $map[$key] = $left; }
            }
        }

        if ($changed) {
            Database::query('UPDATE entity_translate_jobs SET charged_map = ? WHERE id = ?', [json_encode($map), $jobId]);
        }
    }

    private static function existingTranslations(array $meta, array $ids, array $langs): array
    {
        if (empty($ids) || empty($langs)) return [];
        $cols = [];
        foreach ($meta['fields'] as $f) { $cols[] = $f['dst']; }
        $cols = array_values(array_unique($cols));
        $select = $meta['fk'] . ', lang_code, ' . implode(', ', $cols);

        $iph = implode(',', array_fill(0, count($ids), '?'));
        $lph = implode(',', array_fill(0, count($langs), '?'));
        $rows = Database::fetchAll(
            "SELECT {$select} FROM {$meta['translations']} WHERE {$meta['fk']} IN ({$iph}) AND lang_code IN ({$lph})",
            array_merge($ids, $langs)
        );
        $out = [];
        foreach ($rows as $r) {
            $out[(int) $r[$meta['fk']]][$r['lang_code']] = $r;
        }
        return $out;
    }

    private static function countFullyDone(array $meta, array $ids, array $fields, array $langs, array $rows, ?array $existing = null): int
    {
        $existing ??= self::existingTranslations($meta, $ids, $langs);
        $done = 0;
        foreach ($rows as $row) {
            $id = (int) $row['id'];
            $ok = true;
            foreach ($fields as $fieldKey) {
                $field = $meta['fields'][$fieldKey];
                $src = trim((string) ($row[$field['src']] ?? ''));
                if ($src === '') continue;
                foreach ($langs as $lang) {
                    if (trim((string) ($existing[$id][$lang][$field['dst']] ?? '')) === '') { $ok = false; break 2; }
                }
            }
            if ($ok) $done++;
        }
        return $done;
    }

    private static function bumpTranslated(int $jobId, int $n): void
    {
        if ($n <= 0) return;
        Database::query('UPDATE entity_translate_jobs SET translated_count = translated_count + ? WHERE id = ?', [$n, $jobId]);
    }

    

    private static function buildPrompt(array $aiConfig, array $field, string $source, array $langs, array $languageMeta): string
    {
        $p = TranslateTypes::prompt($field['prompt']);
        $begin = (string) ($aiConfig[$p['begin']] ?? '');
        $end   = (string) ($aiConfig[$p['end']] ?? '');
        if ($begin === '') $begin = $p['dbegin'];
        if ($end === '')   $end   = $p['dend'];

        $begin = str_replace(['\\r\\n', '\\n', '\\t'], ["\n", "\n", "\t"], $begin);
        $end   = str_replace(['\\r\\n', '\\n', '\\t'], ["\n", "\n", "\t"], $end);

        $formatLines = [];
        foreach ($langs as $code) {
            $name = $languageMeta[$code] ?? strtoupper($code);
            $formatLines[] = $name . ': <' . strtoupper($code) . '></' . strtoupper($code) . '>';
        }
        $format = implode("\n", $formatLines);

        $tpl = $begin . ($end !== '' ? "\n\n" . $end : '');
        if (!str_contains($tpl, '{format}')) {
            $tpl .= "\n\nReturn the translation in this EXACT format — one XML tag per language, nothing else:\n{format}";
        }
        $tpl = str_replace('{format}', $format, $tpl);
        $tpl = str_replace('{' . $p['token'] . '}', $source, $tpl);

        foreach (['name', 'title', 'description', 'text'] as $alias) {
            $tpl = str_replace('{' . $alias . '}', $source, $tpl);
        }

        return trim($tpl);
    }

    

    private static function persist(array $meta, array $field, int $entityId, string $lang, string $text, string $videoType, string $sourceText): void
    {
        $tr = $meta['translations'];
        $fk = $meta['fk'];
        $col = $field['dst'];

        $existing = Database::fetchOne("SELECT id FROM {$tr} WHERE {$fk} = ? AND lang_code = ?", [$entityId, $lang]);

        $data = [$col => $text];
        if (!empty($field['slug'])) {
            $data['slug'] = self::makeSlug($meta, $entityId, $lang, $text, $videoType);
        }

        if ($existing) {
            Database::update($tr, $data, "{$fk} = ? AND lang_code = ?", [$entityId, $lang]);
            return;
        }

        $insert = array_merge($data, [$fk => $entityId, 'lang_code' => $lang]);
        if (empty($insert['slug'])) {
            $primary = self::primaryField($meta);
            if ($primary !== null && $primary['dst'] !== $col) {
                $insert['slug'] = self::makeSlug($meta, $entityId, $lang, $sourceText, $videoType);
            }
        }
        Database::insert($tr, $insert);
    }

    private static function primaryField(array $meta): ?array
    {
        foreach ($meta['fields'] as $f) {
            if (!empty($f['slug'])) return $f;
        }
        return null;
    }

    private static function makeSlug(array $meta, int $entityId, string $lang, string $text, string $videoType): string
    {
        switch ($meta['slug_fn']) {
            case 'video':
                return Slug::uniqueVideoTranslation($text, $lang, $entityId);
            case 'page':
                return Slug::uniqueTranslation($text, $lang, $entityId);
            case 'album':
                return Slug::uniqueEntityTranslation($meta['translations'], $meta['fk'], $meta['main'], $text, $lang, $entityId);
            default:
                return Slug::uniqueEntityTranslation($meta['translations'], $meta['fk'], $meta['main'], $text, $lang, $entityId, $videoType);
        }
    }

    

    private static function prefillFromCache(array $meta, array $work, array &$existing): void
    {
        if (Setting::get('entity_translation_cache_enabled', '1') !== '1') return;

        $cacheable = [];
        foreach ($work as $w) {
            $field = $meta['fields'][$w['field']];
            if (empty($field['cache'])) continue;
            $cacheable[] = $w;
        }
        if (empty($cacheable)) return;

        foreach (array_chunk($cacheable, 100) as $chunk) {
            $items = array_map(fn($w) => ['name' => $w['src'], 'video_type' => $w['type']], $chunk);
            $langUnion = [];
            foreach ($chunk as $w) { foreach ($w['langs'] as $l) { $langUnion[$l] = true; } }
            $results = CatalogueClient::translateEntitiesLookup($items, array_keys($langUnion));
            if (!is_array($results)) continue;

            foreach ($chunk as $i => $w) {
                $translations = $results[$i]['translations'] ?? null;
                if (!is_array($translations) || empty($translations)) continue;
                $field = $meta['fields'][$w['field']];
                foreach ($translations as $lang => $text) {
                    $lang = strtolower((string) $lang);
                    $text = trim((string) $text);
                    if ($text === '' || !in_array($lang, $w['langs'], true) || mb_strlen($text) > (int) $field['max']) continue;
                    self::persist($meta, $field, $w['id'], $lang, $text, $w['type'], $w['src']);
                    $existing[$w['id']][$lang][$field['dst']] = $text;
                }
            }
        }
    }

    

    public static function pauseJob(int $jobId, string $reason): void
    {
        $isCooldown = str_starts_with($reason, 'cooldown:');
        Database::query(
            'UPDATE entity_translate_jobs SET status = ?, pause_reason = ?, updated_at = NOW() WHERE id = ? AND status IN (?, ?)',
            [$isCooldown ? 'processing' : 'paused', $reason, $jobId, 'pending', 'processing']
        );
        if (!$isCooldown) {
            error_log('EntityTranslateProcessor: job #' . $jobId . ' paused — ' . $reason);
        }
    }

    public static function pauseByUser(int $jobId): bool
    {
        $stmt = Database::query(
            "UPDATE entity_translate_jobs SET status = 'paused', pause_reason = 'Paused by you.', updated_at = NOW()
              WHERE id = ? AND status IN ('pending','processing')",
            [$jobId]
        );
        return $stmt->rowCount() > 0;
    }

    public static function resumeJob(int $jobId): bool
    {
        $stmt = Database::query(
            "UPDATE entity_translate_jobs SET status = 'pending', pause_reason = NULL, error_message = NULL, updated_at = NOW()
              WHERE id = ? AND status = 'paused'",
            [$jobId]
        );
        if ($stmt->rowCount() > 0) {
            Database::query("UPDATE settings SET setting_value = '0' WHERE setting_key = 'cron_entity_translate_last'");
            self::asyncPing();
            return true;
        }
        return false;
    }

    public static function retryFailed(int $jobId): bool
    {
        $job = self::getJob($jobId);
        if (!$job) return false;
        self::unbillRefunded($jobId, $job);
        Database::query(
            "UPDATE entity_translate_jobs
                SET attempt_map = '[]', failed_count = 0, cursor_id = 0,
                    processed_items = 0, translated_count = 0,
                    status = 'pending', pause_reason = NULL, error_message = NULL,
                    completed_at = NULL, updated_at = NOW()
              WHERE id = ?",
            [$jobId]
        );
        Database::query("UPDATE settings SET setting_value = '0' WHERE setting_key = 'cron_entity_translate_last'");
        self::asyncPing();
        return true;
    }

    public static function cancelJob(int $jobId, int $requestingUserId = 0): array
    {
        $job = self::getJob($jobId);
        if (!$job || !in_array($job['status'], ['pending', 'processing', 'paused'], true)) {
            return ['cancelled' => false];
        }
        if ($requestingUserId > 0 && (int) $job['user_id'] !== $requestingUserId && (int) $job['user_id'] !== 0) {
            return ['cancelled' => false, 'error' => 'Not your job'];
        }
        Database::query(
            "UPDATE entity_translate_jobs SET status = 'cancelled', pause_reason = NULL, completed_at = NOW(), updated_at = NOW() WHERE id = ?",
            [$jobId]
        );
        $refunded = self::settleRefunds($jobId, 5);
        return [
            'cancelled' => true,
            'charged'   => round((float) $job['charged_credits'], 2),
            'refunded'  => $refunded,
        ];
    }

    private static function markSitemapDirty(): void
    {
        if (!class_exists('SitemapController', false)) {
            require_once APP_ROOT . '/src/Front/SitemapController.php';
        }
        SitemapController::markDirty();
    }

    private static function finishJob(int $jobId): void
    {
        Database::query(
            "UPDATE entity_translate_jobs SET status = 'completed', pause_reason = NULL, completed_at = NOW(), updated_at = NOW() WHERE id = ?",
            [$jobId]
        );
        self::markSitemapDirty();
        self::settleRefunds($jobId);
    }

    private static function failJob(int $jobId, string $error): void
    {
        Database::query(
            "UPDATE entity_translate_jobs SET status = 'failed', pause_reason = NULL, error_message = ?, completed_at = NOW() WHERE id = ?",
            [mb_substr($error, 0, 1000), $jobId]
        );
        self::settleRefunds($jobId);
    }

    private static function recoverStalled(): void
    {
        Database::query(
            "UPDATE entity_translate_jobs SET status = 'pending'
              WHERE status = 'processing' AND updated_at < DATE_SUB(NOW(), INTERVAL ? SECOND)",
            [self::STALE_SECONDS]
        );
    }

    

    public static function autoTick(): void
    {
        foreach (TranslateTypes::keys() as $type) {
            if (!self::autoEnabled($type)) continue;
            if (self::getActiveJob($type) !== null) continue;

            $meta = self::typeMeta($type);
            $langs = self::targetLanguageCodes();
            if (empty($langs) || $meta === null) continue;

            $pending = self::countUntranslated($meta, $langs);
            if ($pending === 0) continue;

            if (!CatalogueClient::isConfigured()) continue;

            self::createJob([
                'entity_type' => $type,
                'scope'       => 'all',
                'languages'   => $langs,
                'user_id'     => (int) (Database::fetchOne('SELECT id FROM users ORDER BY id ASC LIMIT 1')['id'] ?? 0),
            ]);
        }
    }

    /** Rows that still miss at least one language on at least one non-empty translatable field. */
    public static function countUntranslated(array $meta, array $langs, string $typeFilter = '', int $limit = 200): int
    {
        if (empty($langs)) return 0;
        $conds = [];
        foreach ($meta['fields'] as $f) {
            $conds[] = "(m.{$f['src']} IS NULL OR m.{$f['src']} = '' OR (t.{$f['dst']} IS NOT NULL AND t.{$f['dst']} <> ''))";
        }
        $lph = implode(',', array_fill(0, count($langs), '?'));
        $params = $langs;
        $where = '';
        if ($typeFilter !== '' && !empty($meta['type_col'])) {
            $where = ' WHERE m.' . $meta['type_col'] . ' = ?';
            $params[] = $typeFilter;
        }
        $params[] = count($langs);
        $params[] = $limit;

        $sql = "SELECT COUNT(*) AS c FROM (
                    SELECT m.id
                      FROM {$meta['main']} m
                      LEFT JOIN {$meta['translations']} t
                        ON t.{$meta['fk']} = m.id AND t.lang_code IN ({$lph}) AND " . implode(' AND ', $conds) . "
                    {$where}
                     GROUP BY m.id
                    HAVING COUNT(DISTINCT t.lang_code) < ?
                     LIMIT ?
                ) sub";
        return (int) (Database::fetchOne($sql, $params)['c'] ?? 0);
    }

    

    private static function asyncPing(): void
    {
        $url = rtrim(Setting::get('site_url', ''), '/') . '/';
        if ($url === '/') return;
        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER    => true,
            CURLOPT_TIMEOUT_MS        => 500,
            CURLOPT_CONNECTTIMEOUT_MS => 500,
            CURLOPT_SSL_VERIFYPEER    => true,
            CURLOPT_USERAGENT         => 'TubePress-Cron/1.0',
        ]);
        @curl_exec($ch);
        curl_close($ch);
    }
}
