<?php

declare(strict_types=1);

namespace app\common\service\prescriptionai;

use DomainException;
use think\facade\Db;

/** Database-backed outbox, immutable result storage and fenced work leases. */
final class PrescriptionAiStore
{
    /** Cached per connection for rolling deployment before the additive migration arrives. */
    public static function supportsProgress(): bool
    {
        static $columns = [];
        $query = Db::name('prescription_ai_task');
        $connection = $query->getConnection();
        $key = spl_object_id($connection) . ':' . $query->getConfig('database');
        if (!array_key_exists($key, $columns)) {
            $columns[$key] = isset($query->getFields()['progress_json']);
        }
        return $columns[$key];
    }

    public static function enabled(): bool
    {
        return (bool) config('prescription_analysis.enabled', false);
    }

    public static function recordSaved(array $rx, int $actorId, ?array $info = null, array $options = []): ?int
    {
        if (!self::enabled() || !PrescriptionAiPolicy::isManual($rx)) {
            return null;
        }
        $info = $info ?? PrescriptionAiAccess::actor($actorId);
        if (!$info || !PrescriptionAiAccess::canGenerate($actorId, $info)) {
            return null;
        }
        // Caller holds the prescription mutation transaction; failure must propagate.
        return Db::transaction(static function () use ($rx, $actorId, $options): ?int {
            $id = (int) $rx['id'];
            $subject = Db::name('prescription_ai_subject')->where('prescription_id', $id)->lock(true)->find();
            $hash = PrescriptionAiPolicy::fingerprint($rx);
            if ($subject && hash_equals((string) $subject['clinical_hash'], $hash)
                && empty($options['restored'])) {
                return (int) $subject['latest_batch_id'];
            }
            $revision = $subject ? (int) $subject['revision'] + 1 : 1;
            $now = time();
            if (!$subject) {
                Db::name('prescription_ai_subject')->insert([
                    'prescription_id' => $id, 'revision' => $revision, 'clinical_hash' => $hash,
                    'first_batch_id' => 0, 'latest_batch_id' => 0, 'updated_at' => $now,
                ]);
            } else {
                self::invalidate($id, 'prescription_changed');
                Db::name('prescription_ai_subject')->where('prescription_id', $id)->update([
                    'revision' => $revision, 'clinical_hash' => $hash, 'updated_at' => $now,
                ]);
            }
            $trigger = $subject ? 'clinical_change' : (string) ($options['trigger'] ?? 'first_manual');
            if ($subject && !config('prescription_analysis.auto_refresh_prescription', true)) {
                return null;
            }
            return self::insertBatch($rx, $revision, $actorId, $trigger,
                hash('sha256', "saved:{$id}:{$revision}"), $options);
        });
    }

    public static function enqueue(array $rx, int $actorId, string $trigger, string $eventKey, array $options = []): int
    {
        return Db::transaction(static function () use ($rx, $actorId, $trigger, $eventKey, $options): int {
            $rx = Db::name('tcm_prescription')->where('id', (int) $rx['id'])->lock(true)->find();
            if (!self::enabled() || !$rx || !PrescriptionAiPolicy::isManual($rx)) {
                throw new DomainException('该处方不能安排分析');
            }
            $subject = Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->lock(true)->find();
            if (!$subject || !hash_equals($subject['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
                $info = PrescriptionAiAccess::actor($actorId);
                $id = self::recordSaved($rx, $actorId, $info, ['trigger' => $trigger] + $options);
                if (!$id) {
                    throw new DomainException('无法为该处方安排分析');
                }
                return $id;
            }
            $existing = Db::name('prescription_ai_batch')->where('event_key', $eventKey)->value('id');
            if ($existing) {
                return (int) $existing;
            }
            $latest = Db::name('prescription_ai_batch')->where('id', (int) $subject['latest_batch_id'])->find();
            if ($trigger === 'manual_refresh' && $latest && $latest['validity'] === 'current'
                && in_array($latest['status'], ['preparing', 'waiting_sources', 'queued', 'running'], true)) {
                return (int) $latest['id'];
            }
            self::invalidate((int) $rx['id'], 'source_updated');
            return self::insertBatch($rx, (int) $subject['revision'], $actorId, $trigger, $eventKey, $options);
        });
    }

    private static function insertBatch(array $rx, int $revision, int $actorId, string $trigger, string $key, array $options): int
    {
        $now = time();
        $diagnosisId = (int) ($rx['diagnosis_id'] ?? 0);
        $patientId = $diagnosisId > 0 ? (int) Db::name('tcm_diagnosis')->where('id', $diagnosisId)
            ->whereNull('delete_time')->value('patient_id') : 0;
        $bindingValid = $patientId > 0 && ((int) ($rx['patient_id'] ?? 0) === 0 || (int) $rx['patient_id'] === $patientId);
        $aiAssisted = array_key_exists('ai_assisted', $options)
            ? (in_array($options['ai_assisted'], [true, 1, '1'], true) ? 'yes'
                : (in_array($options['ai_assisted'], [false, 0, '0'], true) ? 'no' : 'unknown')) : 'unknown';
        $id = (int) Db::name('prescription_ai_batch')->insertGetId([
            'event_key' => $key, 'prescription_id' => (int) $rx['id'], 'prescription_revision' => $revision,
            'clinical_hash' => PrescriptionAiPolicy::fingerprint($rx), 'diagnosis_id' => $diagnosisId,
            'patient_id' => $patientId, 'doctor_id' => (int) ($rx['creator_id'] ?? 0), 'actor_id' => $actorId,
            'trigger_type' => $trigger, 'reason' => mb_substr((string) ($options['reason'] ?? ''), 0, 500),
            'ai_assisted' => $aiAssisted, 'status' => $bindingValid ? 'preparing' : 'blocked',
            'validity' => 'current', 'comparison_type' => 'latest_context', 'baseline_eligible' => 0,
            'prescription_cipher' => (new PrescriptionAiCipher())->encrypt($rx, 'prescription'),
            'decision_at' => $now, 'wait_until' => $now + (int) config('prescription_analysis.transcript_wait_seconds', 300),
            'next_run_at' => $now + ($trigger === 'clinical_change' ? (int) config('prescription_analysis.debounce_seconds', 60) : 0),
            'error_code' => $bindingValid ? '' : 'PATIENT_BINDING_REQUIRED', 'created_at' => $now, 'updated_at' => $now,
        ]);
        $subject = Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->find();
        Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->update([
            'latest_batch_id' => $id, 'first_batch_id' => (int) ($subject['first_batch_id'] ?? 0) ?: $id, 'updated_at' => $now,
        ]);
        return $id;
    }

    public static function invalidate(int $rxId, string $validity): void
    {
        if (!self::enabled()) {
            return;
        }
        $ids = Db::name('prescription_ai_batch')->where('prescription_id', $rxId)->where('validity', 'current')->column('id');
        if ($ids === []) {
            return;
        }
        $now = time();
        Db::name('prescription_ai_batch')->whereIn('id', $ids)->update([
            'validity' => $validity, 'lock_token' => '', 'lock_until' => 0, 'updated_at' => $now,
        ]);
        Db::name('prescription_ai_batch')->whereIn('id', $ids)
            ->whereIn('status', ['preparing', 'waiting_sources', 'retry_wait', 'queued', 'running'])->update(['status' => 'cancelled']);
        Db::name('prescription_ai_task')->whereIn('batch_id', $ids)->whereIn('status', PrescriptionAiPolicy::ACTIVE_TASKS)
            ->update(['status' => 'cancelled', 'lock_token' => '', 'lock_until' => 0, 'updated_at' => $now]);
        $taskIds = Db::name('prescription_ai_task')->whereIn('batch_id', $ids)->column('id');
        if ($taskIds !== []) {
            Db::name('prescription_ai_attempt')->whereIn('task_id', $taskIds)->where('status', 'running')
                ->update(['status' => 'cancelled', 'finished_at' => $now]);
        }
    }

    public static function claimBatch(): ?array
    {
        $now = time();
        $ids = Db::name('prescription_ai_batch')->where('validity', 'current')
            ->whereIn('status', ['preparing', 'waiting_sources', 'retry_wait'])->where('next_run_at', '<=', $now)
            ->where('lock_until', '<=', $now)->order('id')->limit(20)->column('id');
        foreach ($ids as $id) {
            $row = Db::transaction(static function () use ($id, $now): ?array {
                $row = Db::name('prescription_ai_batch')->where('id', $id)->lock(true)->find();
                if (!$row || $row['validity'] !== 'current' || (int) $row['lock_until'] > $now
                    || !in_array($row['status'], ['preparing', 'waiting_sources', 'retry_wait'], true)
                    || (int) $row['next_run_at'] > $now) {
                    return null;
                }
                $row['lock_token'] = bin2hex(random_bytes(16));
                $row['lock_until'] = $now + (int) config('prescription_analysis.lease_seconds', 600);
                Db::name('prescription_ai_batch')->where('id', $id)->update([
                    'lock_token' => $row['lock_token'], 'lock_until' => $row['lock_until'], 'updated_at' => $now,
                ]);
                return $row;
            });
            if ($row) {
                return $row;
            }
        }
        return null;
    }

    public static function finishPreparation(array $batch, array $context): bool
    {
        return Db::transaction(static function () use ($batch, $context): bool {
            $current = Db::name('prescription_ai_batch')->where('id', $batch['id'])->lock(true)->find();
            if (!self::owns($current, $batch)) {
                return false;
            }
            $now = time();
            if (!empty($context['wait_for_transcript']) && $now < (int) $current['wait_until']) {
                self::releasePreparation($batch, ['status' => 'waiting_sources', 'next_run_at' => $now + 15]);
                return true;
            }
            $exclusions = (array) ($context['baseline_exclusion_reasons'] ?? []);
            if ($current['ai_assisted'] !== 'no') {
                $exclusions[] = $current['ai_assisted'] === 'yes' ? 'ai_assisted' : 'assistance_unknown';
            }
            if ($current['trigger_type'] !== 'first_manual' && $current['trigger_type'] !== 'blank_to_manual') {
                $exclusions[] = 'not_original_submission';
            }
            $eligible = !empty($context['baseline_eligible']) && $exclusions === [];
            Db::name('prescription_ai_batch')->where('id', $batch['id'])->update([
                'context_cipher' => (new PrescriptionAiCipher())->encrypt($context, 'context'),
                'access_cipher' => (new PrescriptionAiCipher())->encrypt(
                    ['source_access_manifest' => $context['source_access_manifest'] ?? []], 'access:' . $batch['id']),
                'source_hash' => (string) ($context['source_hash'] ?? ''),
                'source_diagnosis_ids_json' => PrescriptionAiPolicy::canonical($context['source_diagnosis_ids'] ?? []),
                'source_summary_json' => PrescriptionAiPolicy::canonical($context['source_summary'] ?? []),
                'missing_json' => PrescriptionAiPolicy::canonical($context['missing'] ?? []),
                'comparison_type' => (string) ($context['comparison_type'] ?? 'latest_context'),
                'baseline_eligible' => (int) $eligible,
                'baseline_exclusions_json' => PrescriptionAiPolicy::canonical(array_values(array_unique($exclusions))),
                'cutoff_at' => (int) ($context['cutoff_at'] ?? $now), 'status' => 'queued',
                'coverage_status' => empty($context['missing']) ? 'pending' : 'partial',
                'lock_token' => '', 'lock_until' => 0, 'error_code' => '', 'updated_at' => $now,
            ]);
            foreach (PrescriptionAiPolicy::MODELS as $model) {
                Db::name('prescription_ai_task')->insert([
                    'batch_id' => $batch['id'], 'model_key' => $model, 'status' => 'queued',
                    'next_run_at' => $now, 'updated_at' => $now,
                ]);
            }
            return true;
        });
    }

    public static function releasePreparation(array $batch, array $values): bool
    {
        return (bool) Db::name('prescription_ai_batch')->where('id', $batch['id'])
            ->where('lock_token', $batch['lock_token'])->where('validity', 'current')->update($values + [
                'lock_token' => '', 'lock_until' => 0, 'updated_at' => time(),
            ]);
    }

    public static function claimTask(string $model): ?array
    {
        if (!in_array($model, PrescriptionAiPolicy::MODELS, true)) {
            throw new DomainException('未知模型');
        }
        $now = time();
        $ids = Db::name('prescription_ai_task')->where('model_key', $model)->whereIn('status', PrescriptionAiPolicy::ACTIVE_TASKS)
            ->where('next_run_at', '<=', $now)->where('lock_until', '<=', $now)->order('id')->limit(30)->column('id');
        foreach ($ids as $id) {
            $result = Db::transaction(static function () use ($id, $model, $now): ?array {
                // One locked daily model counter serializes concurrency and budget reservations.
                $key = date('Y-m-d', $now) . ':' . $model;
                Db::name('prescription_ai_limit')->extra('IGNORE')->insert(['limit_key' => $key, 'used_count' => 0, 'updated_at' => $now]);
                $budget = Db::name('prescription_ai_limit')->where('limit_key', $key)->lock(true)->find();
                $batchId = Db::name('prescription_ai_task')->where('id', $id)->value('batch_id');
                // All mutations lock batch before task, matching invalidation and completion.
                $batch = $batchId ? Db::name('prescription_ai_batch')->where('id', $batchId)->lock(true)->find() : null;
                $row = Db::name('prescription_ai_task')->where('id', $id)->lock(true)->find();
                if (!$row || !in_array($row['status'], PrescriptionAiPolicy::ACTIVE_TASKS, true)
                    || (int) $row['lock_until'] > $now || (int) $row['next_run_at'] > $now) {
                    return null;
                }
                if (!$batch || $batch['validity'] !== 'current') {
                    Db::name('prescription_ai_task')->where('id', $id)->update(['status' => 'cancelled', 'updated_at' => $now]);
                    return null;
                }
                $limit = max(1, (int) config('prescription_analysis.max_attempts', 3));
                if ($row['status'] === 'running') {
                    Db::name('prescription_ai_attempt')->where('task_id', $id)->where('status', 'running')
                        ->update(['status' => 'expired', 'error_code' => 'LEASE_EXPIRED', 'finished_at' => $now]);
                }
                if ((int) $row['attempts'] >= $limit) {
                    Db::name('prescription_ai_task')->where('id', $id)->update([
                        'status' => 'failed', 'error_code' => 'LEASE_EXPIRED', 'finished_at' => $now, 'updated_at' => $now,
                    ]);
                    self::refreshBatch((int) $row['batch_id']);
                    return null;
                }
                if ((int) $budget['used_count'] >= max(1, (int) config('prescription_analysis.daily_model_tasks', 200))) {
                    Db::name('prescription_ai_task')->where('id', $id)->update([
                        'status' => 'retry_wait', 'error_code' => 'BUDGET_PAUSED',
                        'next_run_at' => strtotime('tomorrow', $now), 'updated_at' => $now,
                    ]);
                    return null;
                }
                $running = Db::name('prescription_ai_task')->where('model_key', $model)->where('status', 'running')
                    ->where('lock_until', '>', $now)->count();
                if ($running >= max(1, (int) config('prescription_analysis.max_parallel_per_model', 1))) {
                    return null;
                }
                $row['attempts'] = (int) $row['attempts'] + 1;
                $row['total_attempts'] = (int) $row['total_attempts'] + 1;
                $row['status'] = 'running';
                $row['lock_token'] = bin2hex(random_bytes(16));
                $row['lock_until'] = $now + (int) config('prescription_analysis.lease_seconds', 600);
                $row['started_at'] = $now;
                $row['finished_at'] = 0;
                $public = self::supportsProgress() ? ['progress_json' => PrescriptionAiPolicy::canonical(
                    PrescriptionAiProgress::advance([], 'preparing', 'running', null, null, $now))] : [];
                Db::name('prescription_ai_task')->where('id', $id)->update(array_intersect_key($row, array_flip([
                    'attempts', 'total_attempts', 'status', 'lock_token', 'lock_until', 'started_at', 'finished_at',
                ])) + $public + ['error_code' => '', 'updated_at' => $now]);
                Db::name('prescription_ai_attempt')->insert([
                    'task_id' => $id, 'attempt_no' => $row['total_attempts'], 'status' => 'running', 'started_at' => $now,
                ]);
                Db::name('prescription_ai_limit')->where('limit_key', $key)->inc('used_count')->update(['updated_at' => $now]);
                Db::name('prescription_ai_batch')->where('id', $row['batch_id'])->update(['status' => 'running', 'updated_at' => $now]);
                return $row;
            });
            if ($result) {
                return $result;
            }
        }
        return null;
    }

    public static function checkpoint(array $task, array $progress, bool $persistCache = true): bool
    {
        $now = time();
        $values = ['lock_until' => $now + (int) config('prescription_analysis.lease_seconds', 600), 'updated_at' => $now];
        if (self::supportsProgress()) {
            $values['progress_json'] = PrescriptionAiPolicy::canonical(PrescriptionAiProgress::sanitize($progress['public'] ?? null));
        }
        // Stage notifications do not rewrite the growing encrypted model cache.
        if ($persistCache) {
            $values['progress_cipher'] = (new PrescriptionAiCipher())->encrypt($progress, 'progress:' . $task['id']);
        }
        $owned = static fn () => Db::name('prescription_ai_task')->where('id', $task['id'])->where('status', 'running')
            ->where('lock_token', $task['lock_token'])->where('lock_until', '>', time());
        $changed = $owned()->update($values);
        // Identical metadata twice in one second (also old-schema heartbeats) can be a no-op.
        return $changed > 0 || $owned()->count() > 0;
    }

    public static function complete(array $task, array $output, array $comparison): bool
    {
        return Db::transaction(static function () use ($task, $output, $comparison): bool {
            $rxId = Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->value('prescription_id');
            $rx = Db::name('tcm_prescription')->where('id', $rxId)->lock(true)->find();
            $batch = Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->lock(true)->find();
            $row = Db::name('prescription_ai_task')->where('id', $task['id'])->lock(true)->find();
            if (!self::owns($row, $task) || $row['status'] !== 'running') {
                return false;
            }
            if (!$batch || $batch['validity'] !== 'current' || !$rx || !PrescriptionAiPolicy::isManual($rx)
                || !hash_equals($batch['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
                return false;
            }
            $coverage = $output['coverage'] ?? [];
            $coverageStatus = is_array($coverage) ? (string) ($coverage['status'] ?? 'partial') : 'partial';
            $coverageStatus = in_array($coverageStatus, ['complete', 'full'], true) ? 'complete' : 'partial';
            $body = ['report' => $output['report'] ?? [], 'candidate' => $output['candidate'] ?? null,
                'comparison' => $comparison, 'coverage' => $coverage, 'usage' => $output['usage'] ?? []];
            $now = time();
            $resultId = (int) Db::name('prescription_ai_result')->insertGetId([
                'batch_id' => $task['batch_id'], 'model_key' => $task['model_key'],
                'body_cipher' => (new PrescriptionAiCipher())->encrypt($body, 'result:' . $task['batch_id'] . ':' . $task['model_key']),
                'score' => $comparison['score'] ?? null, 'herb_score' => $comparison['herb_score'] ?? null,
                'comparison_status' => (string) ($comparison['status'] ?? 'not_comparable'),
                'comparison_reason_code' => (string) ($comparison['reason_code'] ?? ''), 'coverage_status' => $coverageStatus,
                'model_name' => mb_substr((string) ($output['model_name'] ?? ''), 0, 100),
                'prompt_version' => mb_substr((string) ($output['prompt_version'] ?? ''), 0, 100),
                'algorithm_version' => mb_substr((string) ($comparison['algorithm_version'] ?? ''), 0, 100), 'generated_at' => $now,
                'dictionary_version' => mb_substr((string) ($comparison['dictionary_version'] ?? ''), 0, 100),
            ]);
            $public = self::supportsProgress() ? ['progress_json' => PrescriptionAiPolicy::canonical(
                PrescriptionAiProgress::advance([], 'completed', 'completed', null, null, $now))] : [];
            Db::name('prescription_ai_task')->where('id', $task['id'])->update($public + [
                'status' => 'success', 'result_id' => $resultId, 'finished_at' => $now, 'updated_at' => $now,
                'lock_token' => '', 'lock_until' => 0, 'error_code' => '', 'progress_cipher' => null,
            ]);
            Db::name('prescription_ai_attempt')->where('task_id', $task['id'])->where('attempt_no', $row['total_attempts'])
                ->update(['status' => 'success', 'finished_at' => $now]);
            self::refreshBatch((int) $task['batch_id']);
            return true;
        });
    }

    public static function fail(array $task, string $code, bool $retryable): void
    {
        Db::transaction(static function () use ($task, $code, $retryable): void {
            Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->lock(true)->find();
            $current = Db::name('prescription_ai_task')->where('id', $task['id'])->lock(true)->find();
            if (!self::owns($current, $task) || $current['status'] !== 'running') {
                return;
            }
            $now = time();
            $code = preg_match('/^[A-Z0-9_]{1,64}$/D', $code) ? $code : 'INTERNAL_ERROR';
            $next = PrescriptionAiPolicy::retryAt((int) $task['attempts'], $now, $retryable,
                max(1, (int) config('prescription_analysis.max_attempts', 3)));
            Db::name('prescription_ai_task')->where('id', $task['id'])->where('lock_token', $task['lock_token'])
                ->where('status', 'running')->update([
                    'status' => $next === null ? 'failed' : 'retry_wait', 'next_run_at' => $next ?? $now,
                    'error_code' => $code, 'lock_token' => '', 'lock_until' => 0, 'finished_at' => $now, 'updated_at' => $now,
                ]);
            self::refreshBatch((int) $task['batch_id']);
            Db::name('prescription_ai_attempt')->where('task_id', $task['id'])->where('attempt_no', $current['total_attempts'])
                ->update(['status' => 'failed', 'error_code' => $code, 'finished_at' => $now]);
        });
    }

    public static function refreshBatch(int $batchId): void
    {
        $states = Db::name('prescription_ai_task')->where('batch_id', $batchId)->column('status');
        $coverage = Db::name('prescription_ai_result')->where('batch_id', $batchId)->column('coverage_status');
        Db::name('prescription_ai_batch')->where('id', $batchId)->where('validity', 'current')->update([
            'status' => PrescriptionAiPolicy::aggregate($states),
            'coverage_status' => count($coverage) === 2 && $coverage === ['complete', 'complete'] ? 'complete' : 'partial',
            'updated_at' => time(),
        ]);
    }

    private static function owns(?array $current, array $claim): bool
    {
        return $current && !empty($claim['lock_token'])
            && hash_equals((string) $current['lock_token'], (string) $claim['lock_token'])
            && (int) $current['lock_until'] > time()
            && (!isset($current['validity']) || $current['validity'] === 'current');
    }
}