boursecu_server/app/Services/TahesabDocSyncService.php

<?php

namespace App\Services;

use App\Http\Controllers\Admin\TahesabController;
use App\Models\Customer;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Log;

/**
 * ارسال آرام سندها (و بعد از آن مشتریان) به ته‌حساب — بدون schedule و بدون جاب صف.
 * استارت از API؛ ادامه با همان پمپ موجود یا یک تیک روی pending-summary.
 */
class TahesabDocSyncService
{
    public const CACHE_KEY = 'tahesab_doc_sync_job';

    /** فاصله بین دو آیتم تا ته‌حساب بن نکند (ثانیه) */
    public const GAP_SECONDS = 2.5;

    private const STALE_SECONDS = 8;

    private const TTL_RUNNING = 21600;

    private const TTL_DONE = 7200;

    /** @var resource|null */
    private $lockHandle;

    public function snapshot(): array
    {
        $job = $this->read();

        return $this->publicSnapshot($job);
    }

    public function isRunning(): bool
    {
        return ($this->read()['running'] ?? false) === true;
    }

    public function clearIdleSnapshot(): void
    {
        if ($this->isRunning()) {
            return;
        }
        Cache::forget(self::CACHE_KEY);
    }

    /**
     * @return array{ok:bool,message:string,doc_sync:array}
     */
    public function start(): array
    {
        $sync = app(TahesabSyncService::class);
        if (! $sync->isSyncEnabled()) {
            return [
                'ok' => false,
                'message' => 'همگام‌سازی ته‌حساب فعال نیست یا تنظیمات ناقص است',
                'doc_sync' => $this->snapshot(),
            ];
        }

        $current = $this->read();
        if (($current['running'] ?? false) === true) {
            return [
                'ok' => true,
                'message' => 'ارسال از قبل در جریان است',
                'doc_sync' => $this->publicSnapshot($current),
            ];
        }

        $totals = $sync->pendingTotals();
        $total = (int) ($totals['total'] ?? 0);
        if ($total <= 0) {
            return [
                'ok' => false,
                'message' => 'سند جاافتاده‌ای برای ارسال نیست',
                'doc_sync' => $this->emptySnapshot(),
            ];
        }

        $job = $this->newRunningJob('docs', $total);
        $job['skip'] = ['row' => [], 'withdraw' => [], 'order' => []];
        $this->write($job, self::TTL_RUNNING);
        $this->spawnPump();

        return [
            'ok' => true,
            'message' => 'ارسال سندها شروع شد؛ بعد از سندها مشتریان پنل هم در ته‌حساب ثبت می‌شوند',
            'doc_sync' => $this->publicSnapshot($job),
        ];
    }

    /**
     * ثبت مشتریان پنلِ بدون کد در ته‌حساب — بدون کامند تازه؛ تیک از pending-summary.
     *
     * @return array{ok:bool,message:string,doc_sync:array}
     */
    public function startCustomerOutbound(): array
    {
        $guard = $this->guardCustomerStart();
        if ($guard !== null) {
            return $guard;
        }

        $total = (int) Customer::whereNull('deleted_at')->whereNull('moshtari_code')->count();
        if ($total <= 0) {
            return [
                'ok' => false,
                'message' => 'مشتری پنل بدون کد ته‌حساب نیست',
                'doc_sync' => $this->emptySnapshot(),
            ];
        }

        $job = $this->newRunningJob('customers_out', $total);
        $job['skip_ids'] = [];
        $this->write($job, self::TTL_RUNNING);
        $this->pumpOnce();

        return [
            'ok' => true,
            'message' => 'ثبت مشتریان در ته‌حساب شروع شد',
            'doc_sync' => $this->snapshot(),
        ];
    }

    /**
     * ورود مشتریان ته‌حساب به پنل — لیست یک‌بار سمت سرور؛ کلاینت فقط وضعیت می‌گیرد.
     *
     * @return array{ok:bool,message:string,doc_sync:array}
     */
    public function startCustomerInbound(): array
    {
        $guard = $this->guardCustomerStart();
        if ($guard !== null) {
            return $guard;
        }

        $rows = app(TahesabController::class)->collectImportableTahesabCustomers();
        if ($rows === []) {
            return [
                'ok' => false,
                'message' => 'مشتری جدیدی با شماره برای ورود به پنل نیست',
                'doc_sync' => $this->emptySnapshot(),
            ];
        }

        $job = $this->newRunningJob('customers_in', count($rows));
        $job['queue'] = $rows;
        $job['skip_codes'] = [];
        $job['current'] = 'لیست از ته‌حساب گرفته شد';
        $this->write($job, self::TTL_RUNNING);

        return [
            'ok' => true,
            'message' => 'ورود مشتریان از ته‌حساب شروع شد',
            'doc_sync' => $this->snapshot(),
        ];
    }

    /**
     * @return array{ok:bool,message:string,doc_sync:array}
     */
    public function stop(): array
    {
        $job = $this->read();
        if (($job['running'] ?? false) !== true) {
            return [
                'ok' => true,
                'message' => 'ارسال در جریان نیست',
                'doc_sync' => $this->publicSnapshot($job),
            ];
        }

        $job['stop'] = true;
        $this->write($job, self::TTL_RUNNING);

        return [
            'ok' => true,
            'message' => 'درخواست توقف ثبت شد',
            'doc_sync' => $this->publicSnapshot($job),
        ];
    }

    public function tickIfStale(): void
    {
        $job = $this->read();
        if (($job['running'] ?? false) !== true) {
            return;
        }

        $beat = $job['beat_at'] ?? null;
        $stage = (string) ($job['stage'] ?? 'docs');
        $staleAfter = ($stage === 'customers_out' || $stage === 'customers_in') ? 2 : self::STALE_SECONDS;
        if ($beat && now()->diffInSeconds(\Illuminate\Support\Carbon::parse($beat)) < $staleAfter) {
            return;
        }

        $this->pumpOnce();
    }

    public function runLoop(): int
    {
        $done = 0;
        while (true) {
            if (! $this->tryLock()) {
                return $done;
            }
            try {
                $cont = $this->pumpOnceUnlocked();
            } finally {
                $this->releaseLock();
            }
            if (! $cont) {
                break;
            }
            $done++;
            usleep((int) (self::GAP_SECONDS * 1_000_000));
        }

        return $done;
    }

    public function pumpOnce(): bool
    {
        if (! $this->tryLock()) {
            return true;
        }
        try {
            return $this->pumpOnceUnlocked();
        } finally {
            $this->releaseLock();
        }
    }

    private function pumpOnceUnlocked(): bool
    {
        $job = $this->read();
        if (($job['running'] ?? false) !== true) {
            return false;
        }

        if (($job['stop'] ?? false) === true) {
            $this->finish($job, true);

            return false;
        }

        $stage = (string) ($job['stage'] ?? 'docs');
        if ($stage === 'customers_out') {
            return $this->pumpOutbound($job);
        }
        if ($stage === 'customers_in') {
            return $this->pumpInbound($job);
        }

        return $this->pumpDocs($job);
    }

    private function pumpDocs(array $job): bool
    {
        $sync = app(TahesabSyncService::class);
        $next = $sync->peekNextPending($job['skip'] ?? []);
        if ($next === null) {
            return $this->advanceToOutboundOrFinish($job);
        }

        $job['current'] = (string) ($next['label'] ?? '');
        $job['beat_at'] = now()->toDateTimeString();
        $this->write($job, self::TTL_RUNNING);

        $type = (string) $next['type'];
        $id = (int) $next['id'];
        $ok = false;
        $message = '';
        try {
            $result = $sync->syncOne($type, $id);
            $ok = (bool) ($result['ok'] ?? false);
            $message = (string) ($result['message'] ?? '');
        } catch (\Throwable $e) {
            $ok = false;
            $message = $e->getMessage();
            Log::warning('Tahesab doc sync item failed', [
                'type' => $type,
                'id' => $id,
                'error' => $e->getMessage(),
            ]);
        }

        return $this->recordItem($ok, $message, $job['current'] ?: ($type.' #'.$id), function (array $job, bool $okNow) use ($type, $id) {
            if ($okNow) {
                return $job;
            }
            $skip = is_array($job['skip'] ?? null) ? $job['skip'] : ['row' => [], 'withdraw' => [], 'order' => []];
            $skip[$type] = array_values(array_unique(array_merge($skip[$type] ?? [], [$id])));
            $job['skip'] = $skip;

            return $job;
        });
    }

    private function advanceToOutboundOrFinish(array $job): bool
    {
        $job['doc_total'] = (int) ($job['total'] ?? 0);
        $job['doc_success'] = (int) ($job['success'] ?? 0);
        $job['doc_failed'] = (int) ($job['failed'] ?? 0);

        $total = (int) Customer::whereNull('deleted_at')->whereNull('moshtari_code')->count();
        if ($total <= 0) {
            $this->finish($job, false);

            return false;
        }

        $job['stage'] = 'customers_out';
        $job['total'] = $total;
        $job['processed'] = 0;
        $job['success'] = 0;
        $job['failed'] = 0;
        $job['errors'] = [];
        $job['skip_ids'] = [];
        $job['current'] = 'شروع ثبت مشتریان پنل در ته‌حساب…';
        $job['beat_at'] = now()->toDateTimeString();
        $this->write($job, self::TTL_RUNNING);

        return true;
    }

    private function pumpOutbound(array $job): bool
    {
        $skip = array_values(array_filter($job['skip_ids'] ?? []));
        $customer = Customer::whereNull('deleted_at')
            ->whereNull('moshtari_code')
            ->when($skip !== [], fn ($q) => $q->whereNotIn('id', $skip))
            ->orderBy('id')
            ->first(['id', 'name', 'mobile']);

        if ($customer === null) {
            $this->finish($job, false);

            return false;
        }

        $label = trim((string) $customer->name.' '.(string) $customer->mobile);
        $job['current'] = $label !== '' ? $label : ('مشتری #'.$customer->id);
        $job['beat_at'] = now()->toDateTimeString();
        $this->write($job, self::TTL_RUNNING);

        $ok = false;
        $message = '';
        try {
            $result = app(TahesabSyncService::class)->registerCustomer($customer);
            $ok = (bool) ($result['ok'] ?? false);
            $message = (string) ($result['message'] ?? '');
        } catch (\Throwable $e) {
            $ok = false;
            $message = $e->getMessage();
            Log::warning('Tahesab customer outbound failed', [
                'id' => $customer->id,
                'error' => $e->getMessage(),
            ]);
        }

        $id = (int) $customer->id;

        return $this->recordItem($ok, $message, $job['current'], function (array $job, bool $okNow) use ($id) {
            if ($okNow) {
                return $job;
            }
            $skip = array_values(array_filter($job['skip_ids'] ?? []));
            $skip[] = $id;
            $job['skip_ids'] = array_values(array_unique($skip));

            return $job;
        });
    }

    private function pumpInbound(array $job): bool
    {
        $queue = is_array($job['queue'] ?? null) ? $job['queue'] : [];
        $skip = [];
        foreach ($job['skip_codes'] ?? [] as $code) {
            $skip[(int) $code] = true;
        }

        $idx = null;
        $row = null;
        foreach ($queue as $i => $item) {
            if (! is_array($item)) {
                continue;
            }
            $code = (int) ($item['code'] ?? 0);
            if ($code > 0 && isset($skip[$code])) {
                continue;
            }
            $idx = $i;
            $row = $item;
            break;
        }

        if ($row === null) {
            $this->finish($job, false);

            return false;
        }

        $label = trim((string) ($row['name'] ?? '').' '.(string) ($row['code'] ?? ''));
        $job['current'] = $label !== '' ? $label : 'مشتری ته‌حساب';
        $job['beat_at'] = now()->toDateTimeString();
        $this->write($job, self::TTL_RUNNING);

        $ok = false;
        $message = '';
        try {
            $result = $this->importInboundRow($row);
            $ok = (bool) ($result['ok'] ?? false);
            $message = (string) ($result['message'] ?? '');
        } catch (\Throwable $e) {
            $ok = false;
            $message = $e->getMessage();
            Log::warning('Tahesab customer inbound failed', [
                'code' => $row['code'] ?? null,
                'error' => $e->getMessage(),
            ]);
        }

        $code = (int) ($row['code'] ?? 0);

        return $this->recordItem($ok, $message, $job['current'], function (array $job, bool $okNow) use ($idx, $code) {
            $queue = is_array($job['queue'] ?? null) ? $job['queue'] : [];
            if ($okNow && $idx !== null) {
                array_splice($queue, (int) $idx, 1);
                $job['queue'] = array_values($queue);
            } elseif (! $okNow) {
                $skip = array_values(array_filter($job['skip_codes'] ?? []));
                if ($code > 0) {
                    $skip[] = $code;
                }
                $job['skip_codes'] = array_values(array_unique($skip));
            }

            return $job;
        });
    }

    /**
     * @param  callable(array, bool): array  $mutate
     */
    private function recordItem(bool $ok, string $message, string $label, callable $mutate): bool
    {
        if (! $ok && $this->looksRateLimited($message)) {
            $job = $this->read();
            if (($job['running'] ?? false) !== true) {
                return false;
            }
            $job['current'] = 'محدودیت نرخ ته‌حساب — کمی صبر…';
            $job['beat_at'] = now()->toDateTimeString();
            $this->write($job, self::TTL_RUNNING);
            usleep(5_000_000);

            return true;
        }

        $job = $this->read();
        if (($job['running'] ?? false) !== true) {
            return false;
        }

        $job['processed'] = (int) ($job['processed'] ?? 0) + 1;
        if ($ok) {
            $job['success'] = (int) ($job['success'] ?? 0) + 1;
        } else {
            $job['failed'] = (int) ($job['failed'] ?? 0) + 1;
            $errors = is_array($job['errors'] ?? null) ? $job['errors'] : [];
            $errors[] = [
                'label' => $label !== '' ? $label : 'آیتم',
                'error' => $message !== '' ? $message : 'ناموفق',
            ];
            $job['errors'] = array_slice($errors, -30);
        }
        $job = $mutate($job, $ok);
        $job['beat_at'] = now()->toDateTimeString();
        $this->write($job, self::TTL_RUNNING);

        return true;
    }

    /**
     * @param  array<string, mixed>  $row
     * @return array{ok:bool,message:string}
     */
    private function importInboundRow(array $row): array
    {
        $request = Request::create('/', 'POST', [
            'customers' => [[
                'code' => (int) ($row['code'] ?? 0),
                'name' => (string) ($row['name'] ?? ''),
                'tel' => (string) ($row['tel'] ?? ''),
            ]],
        ]);
        $response = app(TahesabController::class)->importFromTahesab($request);
        $data = method_exists($response, 'getData') ? $response->getData(true) : [];
        $ok = ((int) ($data['success'] ?? 0)) > 0;
        $err = '';
        if (! $ok) {
            $errors = is_array($data['errors'] ?? null) ? $data['errors'] : [];
            $err = (string) ($errors[0]['error'] ?? 'وارد نشد');
        }

        return ['ok' => $ok, 'message' => $err];
    }

    private function finish(array $job, bool $stopped): void
    {
        $job['running'] = false;
        $job['finished'] = true;
        $job['stopped'] = $stopped;
        $job['stop'] = false;
        $job['current'] = '';
        $job['queue'] = [];
        $job['finished_at'] = now()->toDateTimeString();
        $job['beat_at'] = now()->toDateTimeString();
        $this->write($job, self::TTL_DONE);
    }

    private function spawnPump(): void
    {
        if (\PHP_OS_FAMILY === 'Windows') {
            return;
        }

        $php = PHP_BINARY;
        if (stripos((string) $php, 'fpm') !== false || ! is_executable((string) $php)) {
            $php = 'php';
        }
        $artisan = base_path('artisan');
        $cmd = 'nohup '.escapeshellarg($php).' '.escapeshellarg($artisan)
            .' tahesab:pump-doc-sync >/dev/null 2>&1 &';
        if (! function_exists('exec')) {
            Log::warning('Tahesab doc sync spawn skipped: exec disabled');

            return;
        }
        try {
            exec($cmd);
        } catch (\Throwable $e) {
            Log::warning('Tahesab doc sync spawn failed', ['error' => $e->getMessage()]);
        }
    }

    private function tryLock(): bool
    {
        $path = storage_path('framework/cache/tahesab-doc-sync.lock');
        $dir = dirname($path);
        if (! is_dir($dir)) {
            @mkdir($dir, 0775, true);
        }
        $handle = fopen($path, 'c');
        if ($handle === false) {
            return false;
        }
        if (! flock($handle, LOCK_EX | LOCK_NB)) {
            fclose($handle);

            return false;
        }
        $this->lockHandle = $handle;

        return true;
    }

    private function releaseLock(): void
    {
        if (! is_resource($this->lockHandle)) {
            $this->lockHandle = null;

            return;
        }
        flock($this->lockHandle, LOCK_UN);
        fclose($this->lockHandle);
        $this->lockHandle = null;
    }

    private function looksRateLimited(string $message): bool
    {
        $m = mb_strtolower($message);

        return str_contains($m, '429')
            || str_contains($m, 'too many')
            || str_contains($m, 'rate limit')
            || str_contains($m, 'تعداد درخواست');
    }

    /**
     * @return array{ok:bool,message:string,doc_sync:array}|null
     */
    private function guardCustomerStart(): ?array
    {
        $sync = app(TahesabSyncService::class);
        if (! $sync->isSyncEnabled()) {
            return [
                'ok' => false,
                'message' => 'همگام‌سازی ته‌حساب فعال نیست یا تنظیمات ناقص است',
                'doc_sync' => $this->snapshot(),
            ];
        }

        $current = $this->read();
        if (($current['running'] ?? false) === true) {
            return [
                'ok' => true,
                'message' => 'انتقال از قبل در جریان است',
                'doc_sync' => $this->publicSnapshot($current),
            ];
        }

        return null;
    }

    /**
     * @return array<string, mixed>
     */
    private function newRunningJob(string $stage, int $total): array
    {
        return [
            'running' => true,
            'stop' => false,
            'stopped' => false,
            'finished' => false,
            'stage' => $stage,
            'total' => $total,
            'processed' => 0,
            'success' => 0,
            'failed' => 0,
            'current' => '',
            'errors' => [],
            'started_at' => now()->toDateTimeString(),
            'finished_at' => null,
            'beat_at' => now()->toDateTimeString(),
        ];
    }

    private function read(): array
    {
        $job = Cache::get(self::CACHE_KEY);

        return is_array($job) ? $job : [];
    }

    private function write(array $job, int $ttl): void
    {
        Cache::put(self::CACHE_KEY, $job, $ttl);
    }

    private function emptySnapshot(): array
    {
        return $this->publicSnapshot([]);
    }

    private function publicSnapshot(array $job): array
    {
        $running = (bool) ($job['running'] ?? false);
        $finished = (bool) ($job['finished'] ?? false);
        $total = (int) ($job['total'] ?? 0);
        $processed = (int) ($job['processed'] ?? 0);
        $percent = $total > 0 ? min(100, (int) round(($processed / $total) * 100)) : 0;
        if ($finished && ! ($job['stopped'] ?? false)) {
            $percent = 100;
        }

        $stage = (string) ($job['stage'] ?? 'docs');
        if ($stage === '') {
            $stage = 'docs';
        }

        return [
            'running' => $running,
            'finished' => $finished,
            'stopped' => (bool) ($job['stopped'] ?? false),
            'stage' => $stage,
            'percent' => $percent,
            'total' => $total,
            'processed' => $processed,
            'success' => (int) ($job['success'] ?? 0),
            'failed' => (int) ($job['failed'] ?? 0),
            'current' => (string) ($job['current'] ?? ''),
            'errors' => is_array($job['errors'] ?? null) ? $job['errors'] : [],
            'doc_total' => (int) ($job['doc_total'] ?? 0),
            'doc_success' => (int) ($job['doc_success'] ?? 0),
            'doc_failed' => (int) ($job['doc_failed'] ?? 0),
            'started_at' => $job['started_at'] ?? null,
            'finished_at' => $job['finished_at'] ?? null,
        ];
    }
}