<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Carbon;
use Illuminate\Support\Str;
use RuntimeException;
use Throwable;

/**
 * 将钉钉应付申请流程的固定流程编码同步为按流程聚合的共享镜像。
 *
 * 钉钉表单字段存放在 JSON 中，且历史版本字段不完全一致；镜像保留原始
 * 主表、表单和审批任务 payload，页面只在镜像库中解析和筛选。
 */
class DingPayableWorkflowMirrorSynchronizer
{
    private readonly ConnectionInterface $connection;

    public function __construct(private readonly UpstreamReadOnlyQueryFactory $queries)
    {
        $this->connection = app('db')->connection(
            (string) config('upstream_mirrors.connection', 'business_mirror')
        );
    }

    /** @return array{status:string,mode:string,batches:int,rows:int,dataset:string} */
    public function sync(bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array
    {
        $config = (array) config('upstream_mirrors.ding_payable_workflows', []);
        $dataset = (string) ($config['dataset_key'] ?? 'operations_replica.rpt.ding_payable_workflows');
        $target = (string) ($config['target_table'] ?? 'ding_payable_workflow_mirror');
        $source = (string) ($config['source_key'] ?? 'operations_replica');
        $processCode = (string) ($config['process_code'] ?? '');

        if ($processCode === '') {
            throw new RuntimeException('ding_payable_workflow_mirror_not_initialized');
        }

        $this->ensureMirrorTable($target);

        UpstreamOdbcConnection::reset($source);
        $state = $this->state($dataset);

        if (! $full && ($state['initial_completed_at'] ?? null) === null) {
            return ['status' => 'requires_full_sync', 'mode' => 'incremental', 'batches' => 0, 'rows' => 0, 'dataset' => $dataset];
        }

        $mode = $full ? 'full' : 'incremental';
        $stateCursor = $this->decodeJson($state['cursor'] ?? null);
        $highWatermark = $this->cursorFromState($this->decodeJson($state['high_watermark'] ?? null));
        $resume = is_array($state)
            && ($state['run_mode'] ?? null) === $mode
            && in_array($state['status'] ?? null, ['running', 'failed'], true)
            && is_string($state['generation'] ?? null)
            && $this->isCompatibleCursor($stateCursor);
        $generation = $resume ? (string) $state['generation'] : (string) Str::uuid();
        [$windowStart, $windowEnd] = $this->window($state, $mode, $resume);
        $cursor = $resume ? $this->cursorFromState($stateCursor) : null;
        $rows = $resume ? (int) ($state['last_upserted_rows'] ?? 0) : 0;
        $batchSize = min(max($batchSize ?? (int) ($config['batch_size'] ?? 100), 1), 500);
        $maxBatches = max($maxBatches ?? (int) ($config['max_batches_per_run'] ?? 2), 0);

        $this->start($dataset, $target, $source, $mode, $generation, $resume, $windowStart, $windowEnd);
        $batches = 0;

        try {
            while (true) {
                $workflows = $this->readWorkflows($source, $processCode, $cursor, $windowStart, $windowEnd, $batchSize);
                if ($workflows === []) {
                    $this->complete($dataset, $target, $mode, $generation, $rows, $cursor ?? $highWatermark);

                    return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset];
                }

                $ids = array_values(array_filter(array_map(
                    static fn (array $row): string => (string) ($row['processInstanceId'] ?? ''),
                    $workflows,
                )));
                $originatorIds = array_values(array_unique(array_filter(array_map(
                    static fn (array $row): string => trim((string) ($row['originatorUserId'] ?? '')),
                    $workflows,
                ))));
                $schemes = $this->readRelated($source, 'ding_WorkflowScheme', ['id', 'processInstanceId', 'componentType', 'name', 'value', 'extValue'], $ids, 'id');
                $tasks = $this->readRelated($source, 'ding_WorkflowTask', ['processInstanceId', 'result', 'activityId', 'finishTime', 'createTime', 'userId', 'taskId', 'status', 'id'], $ids, 'id');
                $originatorNames = $this->readOriginatorNames($source, $originatorIds);

                $schemeById = $this->groupBy($schemes, 'processInstanceId');
                $taskById = $this->groupBy($tasks, 'processInstanceId');
                $now = now();
                $payloads = [];

                foreach ($workflows as $workflow) {
                    $id = (string) ($workflow['processInstanceId'] ?? '');
                    if ($id === '') {
                        continue;
                    }
                    $originatorId = trim((string) ($workflow['originatorUserId'] ?? ''));
                    $originatorName = trim((string) ($originatorNames[$originatorId] ?? ''));
                    if ($originatorName !== '') {
                        // 保留钉钉原始用户 ID，同时把稳定的中文姓名写入镜像，
                        // 页面无需在请求期间再次直连上游或猜测标题内容。
                        $workflow['originatorUserName'] = $originatorName;
                    }
                    $payloads[] = [
                        'process_instance_id' => $id,
                        'workflow_payload' => json_encode($workflow, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR),
                        'scheme_payload' => json_encode($schemeById[$id] ?? [], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR),
                        'task_payload' => json_encode($taskById[$id] ?? [], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR),
                        'workflow_created_at' => isset($workflow['createAt']) ? (string) $workflow['createAt'] : null,
                        'sync_generation' => $generation,
                        'synced_at' => $now,
                    ];
                }

                $nextCursor = $this->cursorFromWorkflow($workflows[array_key_last($workflows)]);
                if ($nextCursor === null) {
                    throw new RuntimeException('ding_payable_workflow_missing_watermark');
                }
                $cursor = $nextCursor;
                $rows += count($payloads);
                $batches++;
                $this->connection->transaction(function () use ($dataset, $target, $payloads, $cursor, $generation, $rows, $now): void {
                    if ($payloads !== []) {
                        $this->connection->table($target)->upsert(
                            $payloads,
                            ['process_instance_id'],
                            ['workflow_payload', 'scheme_payload', 'task_payload', 'workflow_created_at', 'sync_generation', 'synced_at'],
                        );
                    }
                    $this->connection->table('business_mirror_table_states')
                        ->where('dataset_key', $dataset)
                        ->update([
                            'cursor' => $cursor === null
                                ? null
                                : json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
                            'last_upserted_rows' => $rows,
                            'updated_at' => $now,
                        ]);
                });

                if (count($workflows) < $batchSize) {
                    $this->complete($dataset, $target, $mode, $generation, $rows, $cursor);

                    return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset];
                }

                if ($maxBatches > 0 && $batches >= $maxBatches) {
                    return ['status' => 'running', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset];
                }
            }
        } catch (Throwable $exception) {
            $this->connection->table('business_mirror_table_states')
                ->where('dataset_key', $dataset)
                ->update(['status' => 'failed', 'last_error_code' => mb_substr($exception::class, 0, 120), 'updated_at' => now()]);
            throw $exception;
        }
    }

    private function ensureMirrorTable(string $target): void
    {
        $schema = Schema::connection($this->connection->getName());
        if ($schema->hasTable($target)) {
            return;
        }
        $schema->create($target, function (Blueprint $table): void {
            $table->string('process_instance_id', 100)->primary();
            $table->json('workflow_payload');
            $table->json('scheme_payload');
            $table->json('task_payload')->nullable();
            $table->string('workflow_created_at', 32)->nullable()->index();
            $table->uuid('sync_generation')->nullable()->index();
            $table->dateTime('synced_at', 6);
        });
    }

    /** @return array<string,mixed>|null */
    private function state(string $dataset): ?array
    {
        $row = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset)->first();
        if ($row === null) {
            return null;
        }
        $state = (array) $row;
        $state['cursor'] = is_string($state['cursor'] ?? null) ? (json_decode($state['cursor'], true) ?: []) : ($state['cursor'] ?? []);

        return $state;
    }

    private function start(
        string $dataset,
        string $target,
        string $source,
        string $mode,
        string $generation,
        bool $resume,
        ?Carbon $windowStart,
        ?Carbon $windowEnd,
    ): void
    {
        $now = now();
        $values = [
            'source_key' => $source,
            'source_object' => 'rpt.ding_Workflow + rpt.ding_WorkflowScheme + rpt.ding_WorkflowTask',
            'target_table' => $target,
            'status' => 'running',
            'run_mode' => $mode,
            'generation' => $generation,
            'last_attempted_at' => $now,
            'last_error_code' => null,
            'updated_at' => $now,
        ];
        if (! $resume) {
            $values += [
                'cursor' => null,
                'window_started_at' => $windowStart,
                'window_ends_at' => $windowEnd,
                'last_upserted_rows' => 0,
            ];
        }
        $query = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset);
        $query->exists()
            ? $query->update($values)
            : $this->connection->table('business_mirror_table_states')->insert(['dataset_key' => $dataset, 'created_at' => $now] + $values);
    }

    private function complete(string $dataset, string $target, string $mode, string $generation, int $rows, ?array $cursor): void
    {
        $now = now();
        $this->connection->transaction(function () use ($dataset, $target, $mode, $generation, $rows, $cursor, $now): void {
            if ($mode === 'full') {
                $this->connection->table($target)->where(function ($query) use ($generation): void {
                    $query->whereNull('sync_generation')->orWhere('sync_generation', '!=', $generation);
                })->delete();
            }
            $values = [
                'status' => 'succeeded',
                'cursor' => null,
                'high_watermark' => $cursor === null
                    ? null
                    : json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
                'last_succeeded_at' => $now,
                'last_upserted_rows' => $rows,
                'mirror_row_count' => $this->connection->table($target)->count(),
                'last_error_code' => null,
                'updated_at' => $now,
            ];
            if ($mode === 'full') {
                $values['initial_completed_at'] = $now;
                $values['last_full_succeeded_at'] = $now;
            }
            $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset)->update($values);
        });
    }

    /** @return array<string,mixed> */
    private function decodeJson(mixed $value): array
    {
        if (is_array($value)) {
            return $value;
        }
        if (! is_string($value) || $value === '') {
            return [];
        }
        $decoded = json_decode($value, true);
        return is_array($decoded) ? $decoded : [];
    }

    /** @return list<array<string,mixed>> */
    private function readWorkflows(
        string $source,
        string $processCode,
        ?array $cursor,
        ?Carbon $windowStart,
        ?Carbon $windowEnd,
        int $limit,
    ): array
    {
        $watermark = 'COALESCE([updateAt], [createAt])';
        $where = '[processCode] = ?';
        $bindings = [$processCode];
        if ($windowStart !== null && $windowEnd !== null) {
            $where .= ' AND ' . $watermark . ' >= ? AND ' . $watermark . ' < ?';
            $bindings[] = $this->sqlServerDateTime($windowStart);
            $bindings[] = $this->sqlServerDateTime($windowEnd);
        }
        if ($cursor !== null) {
            $where .= ' AND (' . $watermark . ' > ? OR (' . $watermark . ' = ? AND [processInstanceId] > ?))';
            $bindings[] = (string) ($cursor['updateAt'] ?? '');
            $bindings[] = (string) ($cursor['updateAt'] ?? '');
            $bindings[] = (string) ($cursor['processInstanceId'] ?? '');
        }
        return $this->queries->forSource($source)->select(
            'SELECT TOP (' . $limit . ') [finishTime],[businessId],[title],[result],[originatorUserId],[originatorDeptId],[originatorDeptName],[status],[bizAction],[processInstanceId], [joinAt], COALESCE([updateAt], [createAt]) AS [updateAt], [createAt], [processCode], [flowType] FROM [rpt].[ding_Workflow] WHERE ' . $where . ' ORDER BY ' . $watermark . ', [processInstanceId]',
            $bindings,
        );
    }

    /** @return array{0:?Carbon,1:?Carbon} */
    private function window(?array $state, string $mode, bool $resume): array
    {
        if ($mode === 'full') {
            return [null, null];
        }

        if ($resume && ($state['window_started_at'] ?? null) !== null && ($state['window_ends_at'] ?? null) !== null) {
            return [
                Carbon::parse((string) $state['window_started_at']),
                Carbon::parse((string) $state['window_ends_at']),
            ];
        }

        $overlap = max((int) config('upstream_mirrors.table_sync.overlap_hours', 48), 1);
        $lastSucceeded = ($state['last_succeeded_at'] ?? null) !== null
            ? Carbon::parse((string) $state['last_succeeded_at'])
            : now();

        return [$lastSucceeded->copy()->subHours($overlap), now()->addSecond()];
    }

    /** @return array{updateAt:string,processInstanceId:string}|null */
    private function cursorFromState(array $cursor): ?array
    {
        if (($cursor['updateAt'] ?? '') === '' && ($cursor['processInstanceId'] ?? '') === '') {
            return null;
        }

        if (($cursor['updateAt'] ?? '') === '') {
            return null;
        }

        return [
            'updateAt' => (string) $cursor['updateAt'],
            'processInstanceId' => (string) ($cursor['processInstanceId'] ?? ''),
        ];
    }

    /** @param array<string,mixed> $workflow @return array{updateAt:string,processInstanceId:string}|null */
    private function cursorFromWorkflow(array $workflow): ?array
    {
        $id = trim((string) ($workflow['processInstanceId'] ?? ''));
        $updateAt = trim((string) ($workflow['updateAt'] ?? $workflow['createAt'] ?? ''));

        if ($id === '' || $updateAt === '') {
            return null;
        }

        return ['updateAt' => $updateAt, 'processInstanceId' => $id];
    }

    private function isCompatibleCursor(array $cursor): bool
    {
        return $cursor === [] || array_key_exists('updateAt', $cursor);
    }

    private function sqlServerDateTime(Carbon $value): string
    {
        return $value->format('Y-m-d\TH:i:s.v');
    }

    /** @param list<string> $ids @param list<string> $columns @return list<array<string,mixed>> */
    private function readRelated(string $source, string $table, array $columns, array $ids, string $order): array
    {
        if ($ids === []) {
            return [];
        }
        $placeholders = implode(',', array_fill(0, count($ids), '?'));
        return $this->queries->forSource($source)->select(
            'SELECT ' . implode(', ', array_map(static fn (string $column): string => '[' . $column . ']', $columns))
                . ' FROM [rpt].[' . $table . '] WHERE [processInstanceId] IN (' . $placeholders . ') ORDER BY [' . $order . ']',
            $ids,
        );
    }

    /** @param list<string> $userIds @return array<string,string> */
    private function readOriginatorNames(string $source, array $userIds): array
    {
        if ($userIds === []) {
            return [];
        }

        $names = [];
        try {
            $names = $this->readUserNameRows($source, 'hr_employee', ['userid', 'name'], $userIds);
        } catch (Throwable) {
            // 姓名目录短暂不可用时不应阻断应付流程同步；页面会回退显示原始用户 ID。
        }

        $missing = array_values(array_diff($userIds, array_keys($names)));
        if ($missing !== []) {
            try {
                // 已离职申请人不一定还存在于 hr_employee，使用同一只读来源的
                // 离职视图补齐姓名；两个查询均为固定字段和参数化 IN 条件。
                $retired = $this->readUserNameRows($source, 'hr_employee_dimission', ['userid', 'realName', 'name'], $missing);
                $names += $retired;
            } catch (Throwable) {
                // 离职目录不可用时保留已解析的在职姓名。
            }
        }

        return $names;
    }

    /** @param list<string> $columns @param list<string> $userIds @return array<string,string> */
    private function readUserNameRows(string $source, string $table, array $columns, array $userIds): array
    {
        $placeholders = implode(',', array_fill(0, count($userIds), '?'));
        $rows = $this->queries->forSource($source)->select(
            'SELECT ' . implode(', ', array_map(static fn (string $column): string => '[' . $column . ']', $columns))
                . ' FROM [rpt].[' . $table . '] WHERE [userid] IN (' . $placeholders . ')',
            $userIds,
        );
        $names = [];

        foreach ($rows as $row) {
            $id = trim((string) ($row['userid'] ?? $row['userId'] ?? ''));
            $name = trim((string) ($row['name'] ?? $row['realName'] ?? ''));
            if ($id !== '' && $name !== '') {
                $names[$id] = $name;
            }
        }

        return $names;
    }

    /** @param list<array<string,mixed>> $rows @return array<string,list<array<string,mixed>>> */
    private function groupBy(array $rows, string $column): array
    {
        $grouped = [];
        foreach ($rows as $row) {
            $key = (string) ($row[$column] ?? '');
            if ($key !== '') {
                $grouped[$key][] = $row;
            }
        }
        return $grouped;
    }
}