feat(tahesab): pump pending documents slowly from admin, not schedule
Send leftover orders and charges one-by-one with a gap so Tahesab is not rate-limited, skip already-synced rows, and expose start/stop plus progress on existing status APIs. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Cursor
@@ -0,0 +1,373 @@
|
||||
<?php
|
||||
|
||||
namespace App\Services;
|
||||
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Facades\Log;
|
||||
|
||||
/**
|
||||
* ارسال آرام همهٔ سندهای جاافتاده به ته‌حساب — بدون schedule.
|
||||
* استارت از API؛ حلقه با artisan در پس‌زمینه یا یک تیک روی 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);
|
||||
}
|
||||
|
||||
/**
|
||||
* @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 = [
|
||||
'running' => true,
|
||||
'stop' => false,
|
||||
'stopped' => false,
|
||||
'finished' => false,
|
||||
'total' => $total,
|
||||
'processed' => 0,
|
||||
'success' => 0,
|
||||
'failed' => 0,
|
||||
'current' => '',
|
||||
'errors' => [],
|
||||
'skip' => ['row' => [], 'withdraw' => [], 'order' => []],
|
||||
'started_at' => now()->toDateTimeString(),
|
||||
'finished_at' => null,
|
||||
'beat_at' => now()->toDateTimeString(),
|
||||
];
|
||||
$this->write($job, self::TTL_RUNNING);
|
||||
$this->spawnPump();
|
||||
|
||||
return [
|
||||
'ok' => true,
|
||||
'message' => 'ارسال سندها شروع شد',
|
||||
'doc_sync' => $this->publicSnapshot($job),
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* @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;
|
||||
if ($beat && now()->diffInSeconds(\Illuminate\Support\Carbon::parse($beat)) < self::STALE_SECONDS) {
|
||||
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;
|
||||
}
|
||||
|
||||
$sync = app(TahesabSyncService::class);
|
||||
$next = $sync->peekNextPending($job['skip'] ?? []);
|
||||
if ($next === null) {
|
||||
$this->finish($job, false);
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
$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(),
|
||||
]);
|
||||
}
|
||||
|
||||
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' => $job['current'] ?: ($type.' #'.$id),
|
||||
'error' => $message !== '' ? $message : 'ناموفق',
|
||||
];
|
||||
$job['errors'] = array_slice($errors, -30);
|
||||
$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;
|
||||
}
|
||||
|
||||
$job['beat_at'] = now()->toDateTimeString();
|
||||
$this->write($job, self::TTL_RUNNING);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
private function finish(array $job, bool $stopped): void
|
||||
{
|
||||
$job['running'] = false;
|
||||
$job['finished'] = true;
|
||||
$job['stopped'] = $stopped;
|
||||
$job['stop'] = false;
|
||||
$job['current'] = '';
|
||||
$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, 'تعداد درخواست');
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
return [
|
||||
'running' => $running,
|
||||
'finished' => $finished,
|
||||
'stopped' => (bool) ($job['stopped'] ?? false),
|
||||
'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'] : [],
|
||||
'started_at' => $job['started_at'] ?? null,
|
||||
'finished_at' => $job['finished_at'] ?? null,
|
||||
];
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user