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 $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 */ 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, ]; } }