CaiWuXiTong/modules/EcommerceFinance/Services/DingProcessRefreshQueue.php

<?php

namespace Modules\EcommerceFinance\Services;

use Illuminate\Support\Facades\Storage;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Support\OaProcessTime;
use RuntimeException;
use Throwable;

class DingProcessRefreshQueue
{
    public const DIRECTORY = 'ding-process-refresh';

    private const RESULT_TTL_SECONDS = 3600;

    private const STALE_PROCESSING_SECONDS = 300;

    /**
     * @param  list<string>  $businessIds
     * @return array{queued:bool,token:?string,count:int}
     */
    public function enqueue(array $businessIds, int $companyId, int $userId): array
    {
        $businessIds = $this->normalizeBusinessIds($businessIds);
        if ($businessIds === []) {
            return ['queued' => false, 'token' => null, 'count' => 0];
        }

        Storage::disk('local')->makeDirectory(self::DIRECTORY);
        $token = (string) Str::uuid();
        $createdAt = OaProcessTime::now();
        $this->writeJson($token . '.pending', [
            'token' => $token,
            'status' => 'pending',
            'business_ids' => $businessIds,
            'company_id' => $companyId,
            'user_id' => $userId,
            'created_at' => $createdAt->toIso8601String(),
            'started_at' => null,
            'completed_at' => null,
            'error' => null,
        ]);

        return ['queued' => true, 'token' => $token, 'count' => count($businessIds)];
    }

    /**
     * @return array{token:string,business_ids:list<string>,company_id:int,user_id:int}|null
     */
    public function claimNext(): ?array
    {
        $this->failStaleProcessing();
        $paths = array_values(array_filter(
            Storage::disk('local')->files(self::DIRECTORY),
            static fn (string $path): bool => str_ends_with($path, '.pending'),
        ));
        usort($paths, fn (string $left, string $right): int =>
            Storage::disk('local')->lastModified($left)
            <=> Storage::disk('local')->lastModified($right)
        );

        foreach ($paths as $pendingPath) {
            $token = pathinfo($pendingPath, PATHINFO_FILENAME);
            if (preg_match('/^[0-9a-f-]{36}$/i', $token) !== 1) {
                continue;
            }
            $processingPath = self::DIRECTORY . '/' . $token . '.processing';
            if (! Storage::disk('local')->move($pendingPath, $processingPath)) {
                continue;
            }
            $payload = $this->readJson($processingPath);
            if ($payload === null) {
                continue;
            }
            $payload['status'] = 'processing';
            $payload['started_at'] = OaProcessTime::now()->toIso8601String();
            $this->writeJson($token . '.processing', $payload);

            return [
                'token' => $token,
                'business_ids' => $this->normalizeBusinessIds($payload['business_ids'] ?? []),
                'company_id' => (int) ($payload['company_id'] ?? 0),
                'user_id' => (int) ($payload['user_id'] ?? 0),
            ];
        }

        return null;
    }

    /** @param array<string,mixed> $result */
    public function complete(string $token, array $result): void
    {
        $this->finish($token, 'completed', null, $result);
    }

    public function fail(string $token, string $error): void
    {
        $this->finish($token, 'failed', $error, null);
    }

    /** @return array<string,mixed>|null */
    public function status(string $token, int $companyId, int $userId): ?array
    {
        if (preg_match('/^[0-9a-f-]{36}$/i', $token) !== 1) {
            return null;
        }

        $payload = null;
        foreach (['completed', 'failed', 'processing', 'pending'] as $state) {
            $payload = $this->readJson(self::DIRECTORY . '/' . $token . '.' . $state);
            if ($payload !== null) {
                break;
            }
        }
        if ($payload === null
            || (int) ($payload['company_id'] ?? 0) !== $companyId
            || (int) ($payload['user_id'] ?? 0) !== $userId) {
            return null;
        }

        return [
            'status' => (string) ($payload['status'] ?? 'pending'),
            'count' => count($this->normalizeBusinessIds($payload['business_ids'] ?? [])),
            'created_at' => $payload['created_at'] ?? null,
            'started_at' => $payload['started_at'] ?? null,
            'completed_at' => $payload['completed_at'] ?? null,
            'error' => $payload['error'] ?? null,
        ];
    }

    public function cleanup(): void
    {
        $disk = Storage::disk('local');
        $cutoff = time() - self::RESULT_TTL_SECONDS;
        foreach ($disk->files(self::DIRECTORY) as $path) {
            if (! str_ends_with($path, '.completed')
                && ! str_ends_with($path, '.failed')) {
                continue;
            }
            if ($disk->lastModified($path) < $cutoff) {
                $disk->delete($path);
            }
        }
    }

    private function failStaleProcessing(): void
    {
        $disk = Storage::disk('local');
        $cutoff = time() - self::STALE_PROCESSING_SECONDS;
        foreach ($disk->files(self::DIRECTORY) as $path) {
            if (! str_ends_with($path, '.processing') || $disk->lastModified($path) >= $cutoff) {
                continue;
            }
            $token = pathinfo($path, PATHINFO_FILENAME);
            $this->fail($token, '刷新任务处理超时，请重新打开页面重试。');
        }
    }

    /** @param array<string,mixed>|null $result */
    private function finish(
        string $token,
        string $status,
        ?string $error,
        ?array $result,
    ): void {
        if (preg_match('/^[0-9a-f-]{36}$/i', $token) !== 1) {
            return;
        }
        $processingPath = self::DIRECTORY . '/' . $token . '.processing';
        $payload = $this->readJson($processingPath);
        if ($payload === null) {
            return;
        }
        $payload['status'] = $status;
        $payload['completed_at'] = OaProcessTime::now()->toIso8601String();
        $payload['error'] = $error === null ? null : mb_substr($error, 0, 500);
        $payload['result'] = $result;
        $this->writeJson($token . '.' . $status, $payload);
        Storage::disk('local')->delete($processingPath);
    }

    /** @return list<string> */
    private function normalizeBusinessIds(mixed $value): array
    {
        if (! is_array($value)) {
            return [];
        }

        return array_values(array_unique(array_filter(array_map(
            static function (mixed $businessId): string {
                $businessId = trim((string) $businessId);

                return preg_match('/^[A-Za-z0-9_-]{1,32}$/', $businessId) === 1
                    ? $businessId
                    : '';
            },
            $value,
        ))));
    }

    /** @param array<string,mixed> $payload */
    private function writeJson(string $path, array $payload): void
    {
        $disk = Storage::disk('local');
        $fullPath = self::DIRECTORY . '/' . ltrim($path, '/');
        $tempPath = $fullPath . '.' . Str::random(8) . '.tmp';
        $json = json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR);
        if (! $disk->put($tempPath, $json) || ! $disk->move($tempPath, $fullPath)) {
            throw new RuntimeException('OA流程刷新请求写入失败。');
        }
    }

    /** @return array<string,mixed>|null */
    private function readJson(string $path): ?array
    {
        $disk = Storage::disk('local');
        if (! $disk->exists($path)) {
            return null;
        }
        try {
            $payload = json_decode(
                (string) $disk->get($path),
                true,
                512,
                JSON_THROW_ON_ERROR,
            );
        } catch (Throwable) {
            return null;
        }

        return is_array($payload) ? $payload : null;
    }
}