<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Repositories\StoreReportRepository;
use Modules\EcommerceFinance\Repositories\StoreReportUpstreamRepository;
use RuntimeException;
use Throwable;

class StoreReportMirrorSynchronizer
{
    private readonly ConnectionInterface $connection;

    public function __construct(
        private readonly StoreReportUpstreamRepository $upstream,
        DatabaseManager $databases,
    ) {
        $this->connection = $databases->connection(
            (string) config('upstream_mirrors.connection', 'business_mirror'),
        );
    }

    /** @return array{lifecycles:array<string,mixed>,bills:array<string,mixed>} */
    public function sync(?int $batchSize = null, ?int $maxBatches = null): array
    {
        $this->assertSchemaReady();

        $lifecycleBatch = min(max((int) config('upstream_mirrors.store_report.lifecycle_batch_size', 1000), 100), 5000);
        $lifecycleMax = max((int) config('upstream_mirrors.store_report.lifecycle_max_batches_per_run', 5), 1);
        $billBatch = min(max($batchSize ?? (int) config('upstream_mirrors.store_report.batch_size', 5000), 100), 10000);
        $billMax = max($maxBatches ?? (int) config('upstream_mirrors.store_report.max_batches_per_run', 4), 1);

        return [
            'lifecycles' => $this->syncDataset(
                StoreReportRepository::LIFECYCLES_DATASET,
                'store_report_lifecycle_rows',
                $lifecycleBatch,
                $lifecycleMax,
                fn (int $cursor, int $limit): array => $this->upstream->lifecycleBatch($cursor, $limit),
                fn (array $rows, string $generation): array => $this->mapLifecycles($rows, $generation),
            ),
            'bills' => $this->syncDataset(
                StoreReportRepository::BILLS_DATASET,
                'store_report_bill_rows',
                $billBatch,
                $billMax,
                fn (int $cursor, int $limit): array => $this->upstream->billBatch($cursor, $limit),
                fn (array $rows, string $generation): array => $this->mapBills($rows, $generation),
            ),
        ];
    }

    /**
     * @param callable(int,int):array<int,array<string,mixed>> $fetch
     * @param callable(array<int,array<string,mixed>>,string):array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} $map
     * @return array{status:string,mode:string,batches:int,source_rows:int,stored_rows:int,cursor_id:int}
     */
    private function syncDataset(
        string $dataset,
        string $table,
        int $batchSize,
        int $maxBatches,
        callable $fetch,
        callable $map,
    ): array {
        $state = $this->state($dataset);
        $initial = empty($state->active_generation);
        $mode = $initial ? 'full' : 'incremental';
        $resume = in_array((string) $state->status, ['running', 'failed'], true)
            && (string) ($state->run_mode ?? '') === $mode;
        $generation = $initial
            ? ($resume && ! empty($state->pending_generation)
                ? (string) $state->pending_generation
                : (string) Str::uuid())
            : (string) $state->active_generation;
        $cursor = $resume
            ? (int) ($state->cursor_id ?? 0)
            : ($initial
                ? 0
                : max((int) ($state->cursor_id ?? 0)
                    - max((int) config('upstream_mirrors.store_report.overlap_rows', 50000), 0), 0));
        $storedRows = $initial && ! $resume ? 0 : (int) $state->row_count;
        $sourceRows = 0;
        $batches = 0;

        $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([
            'status' => 'running',
            'run_mode' => $mode,
            'pending_generation' => $initial ? $generation : null,
            'cursor_id' => $cursor,
            'last_attempted_at' => now(),
            'last_error_code' => null,
            'updated_at' => now(),
        ]);

        try {
            while ($batches < $maxBatches) {
                $rows = $fetch($cursor, $batchSize);
                if ($rows === []) {
                    return $this->complete($dataset, $generation, $mode, $cursor, $storedRows, $batches, $sourceRows);
                }

                $nextCursor = max(array_map(static fn (array $row): int => (int) ($row['source_id'] ?? 0), $rows));
                if ($nextCursor <= $cursor) {
                    throw new RuntimeException('store_report_cursor_did_not_advance');
                }

                $mapped = $map($rows, $generation);
                $this->storeBatch($dataset, $table, $generation, $nextCursor, $mapped, $initial, $storedRows);
                $cursor = $nextCursor;
                $sourceRows += count($rows);
                $storedRows += $initial ? count($mapped['upserts']) : 0;
                $batches++;

                if (count($rows) < $batchSize) {
                    return $this->complete($dataset, $generation, $mode, $cursor, $storedRows, $batches, $sourceRows);
                }
            }

            return $this->result('running', $mode, $batches, $sourceRows, $storedRows, $cursor);
        } catch (Throwable $exception) {
            $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([
                'status' => 'failed',
                'last_error_code' => class_basename($exception),
                'updated_at' => now(),
            ]);
            throw $exception;
        }
    }

    private function state(string $dataset): object
    {
        $now = now();
        $this->connection->table('store_report_import_states')->insertOrIgnore([
            'dataset_key' => $dataset,
            'status' => 'pending',
            'row_count' => 0,
            'source_files_json' => json_encode(['jiansun.dbo.jst_originalBillInfo', 'jiansun.dbo.fys_shopInfo_flow']),
            'created_at' => $now,
            'updated_at' => $now,
        ]);

        return $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->first();
    }

    /** @param array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} $mapped */
    private function storeBatch(
        string $dataset,
        string $table,
        string $generation,
        int $cursor,
        array $mapped,
        bool $initial,
        int $storedRows,
    ): void {
        $this->connection->transaction(function () use (
            $dataset, $table, $generation, $cursor, $mapped, $initial, $storedRows,
        ): void {
            if ($mapped['deletes'] !== []) {
                $this->connection->table($table)
                    ->where('generation', $generation)
                    ->whereIn('source_id', $mapped['deletes'])
                    ->delete();
            }
            if ($mapped['upserts'] !== []) {
                $updates = array_values(array_diff(array_keys($mapped['upserts'][0]), ['generation', 'source_id']));
                // MySQL/MariaDB 对单条预处理语句的占位符数量有限，不能把
                // 数千行、十余列一次塞进同一条 upsert。
                $columnsPerRow = max(count($mapped['upserts'][0]), 1);
                $chunkSize = max(1, min(2000, intdiv(50000, $columnsPerRow)));
                foreach (array_chunk($mapped['upserts'], $chunkSize) as $chunk) {
                    $this->connection->table($table)->upsert(
                        $chunk, ['generation', 'source_id'], $updates,
                    );
                }
            }
            $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([
                'cursor_id' => $cursor,
                'row_count' => $initial ? $storedRows + count($mapped['upserts']) : $storedRows,
                'updated_at' => now(),
            ]);
        });
    }

    /** @return array{status:string,mode:string,batches:int,source_rows:int,stored_rows:int,cursor_id:int} */
    private function complete(
        string $dataset,
        string $generation,
        string $mode,
        int $cursor,
        int $storedRows,
        int $batches,
        int $sourceRows,
    ): array {
        $now = now();
        $actualRows = (int) $this->connection->table(
            $dataset === StoreReportRepository::BILLS_DATASET
                ? 'store_report_bill_rows'
                : 'store_report_lifecycle_rows',
        )->where('generation', $generation)->count();
        $values = [
            'status' => 'succeeded',
            'run_mode' => null,
            'cursor_id' => $cursor,
            'last_succeeded_at' => $now,
            'last_error_code' => null,
            'row_count' => $actualRows,
            'updated_at' => $now,
        ];
        if ($mode === 'full') {
            $values += [
                'active_generation' => $generation,
                'pending_generation' => null,
                'initial_completed_at' => $now,
            ];
        }
        $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update($values);

        return $this->result('succeeded', $mode, $batches, $sourceRows, $actualRows, $cursor);
    }

    /** @return array{status:string,mode:string,batches:int,source_rows:int,stored_rows:int,cursor_id:int} */
    private function result(string $status, string $mode, int $batches, int $sourceRows, int $storedRows, int $cursor): array
    {
        return [
            'status' => $status,
            'mode' => $mode,
            'batches' => $batches,
            'source_rows' => $sourceRows,
            'stored_rows' => $storedRows,
            'cursor_id' => $cursor,
        ];
    }

    /** @return array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} */
    private function mapBills(array $rows, string $generation): array
    {
        $upserts = [];
        $deletes = [];
        foreach ($rows as $row) {
            $sourceId = (int) ($row['source_id'] ?? 0);
            $storeName = $this->text($row['shop_name'] ?? null, 191);
            $occurredAt = $this->dateTime($row['fee_time'] ?? null);
            $direction = (string) ($row['inout_type'] ?? '');
            if ($sourceId < 1 || (int) ($row['del_flag'] ?? 0) !== 0 || $storeName === null
                || $occurredAt === null || ! in_array($direction, ['1', '-1'], true)) {
                if ($sourceId > 0) {
                    $deletes[] = $sourceId;
                }
                continue;
            }
            $amount = abs((float) ($row['amount'] ?? 0));
            $platform = $this->text($row['channel'] ?? null, 32) ?? 'unknown';
            $business = [$sourceId, $platform, $storeName, $occurredAt, $direction, $amount];
            $upserts[] = [
                'generation' => $generation,
                'source_id' => $sourceId,
                'shop_id' => (int) ($row['shop_id'] ?? 0) ?: null,
                'platform' => $platform,
                'platform_label' => $this->platformLabel($platform),
                'store_name' => $storeName,
                'occurred_at' => $occurredAt,
                'income' => $direction === '1' ? $amount : 0,
                'expense' => $direction === '-1' ? $amount : 0,
                'bill_item' => $this->text($row['fee_name'] ?? null, 255),
                'record_code' => (string) $sourceId,
                'source_file' => 'jiansun.dbo.jst_originalBillInfo',
                'row_hash' => hash('sha256', json_encode($business, JSON_UNESCAPED_UNICODE)),
                'source_updated_at' => $this->dateTime($row['source_updated_at'] ?? null),
                'imported_at' => now(),
            ];
        }
        return ['upserts' => $upserts, 'deletes' => $deletes];
    }

    /** @return array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} */
    private function mapLifecycles(array $rows, string $generation): array
    {
        $upserts = [];
        $deletes = [];
        foreach ($rows as $row) {
            $sourceId = (int) ($row['source_id'] ?? 0);
            $storeName = $this->text($row['erp_store_name'] ?? null, 191);
            $creditCode = $this->text($row['unified_social_credit_code'] ?? null, 64);
            $effectiveAt = $this->dateTime($row['effective_at'] ?? null);
            if ($sourceId < 1 || $storeName === null || $creditCode === null || $effectiveAt === null) {
                if ($sourceId > 0) {
                    $deletes[] = $sourceId;
                }
                continue;
            }
            $business = [$sourceId, $storeName, $creditCode, $effectiveAt];
            $upserts[] = [
                'generation' => $generation,
                'source_id' => $sourceId,
                'business_key' => hash('sha256', (string) $sourceId),
                'erp_store_name' => $storeName,
                'company_name' => $this->text($row['company_name'] ?? null, 191),
                'unified_social_credit_code' => $creditCode,
                'effective_at' => $effectiveAt,
                'expires_at' => null,
                'source_file' => 'jiansun.dbo.fys_shopInfo_flow',
                'row_hash' => hash('sha256', json_encode($business, JSON_UNESCAPED_UNICODE)),
                'imported_at' => now(),
            ];
        }
        return ['upserts' => $upserts, 'deletes' => $deletes];
    }

    private function platformLabel(string $channel): string
    {
        return match ($channel) {
            'alipayCSV' => '淘系',
            'TouTiaoFXGZJ' => '抖音',
            'PinduoduoImport' => '拼多多',
            'Jos' => '京东',
            default => $channel,
        };
    }

    private function text(mixed $value, int $length): ?string
    {
        $value = trim((string) ($value ?? ''));
        if ($value === '') {
            return null;
        }
        if (! mb_check_encoding($value, 'UTF-8')) {
            $value = mb_convert_encoding($value, 'UTF-8', 'GB18030');
        }
        return mb_substr($value, 0, $length);
    }

    private function dateTime(mixed $value): ?string
    {
        $value = trim((string) ($value ?? ''));
        return $value === '' ? null : str_replace('T', ' ', substr($value, 0, 26));
    }

    private function assertSchemaReady(): void
    {
        $schema = Schema::connection($this->connection->getName());
        foreach (['store_report_import_states', 'store_report_bill_rows', 'store_report_lifecycle_rows'] as $table) {
            if (! $schema->hasTable($table)) {
                throw new RuntimeException('store_report_mirror_schema_missing:' . $table);
            }
        }
        foreach (['run_mode', 'cursor_id', 'initial_completed_at'] as $column) {
            if (! $schema->hasColumn('store_report_import_states', $column)) {
                throw new RuntimeException('store_report_mirror_upgrade_missing:' . $column);
            }
        }
    }
}