<?php

namespace App\Services\AutoCount;

use App\Services\AutoCount\Exceptions\AutoCountAuthException;
use App\Services\AutoCount\Exceptions\AutoCountClientException;
use App\Services\AutoCount\Exceptions\AutoCountJobException;
use App\Services\AutoCount\Exceptions\AutoCountServerException;
use App\Services\AutoCount\Exceptions\DuplicateRecordException;
use App\Services\AutoCount\Exceptions\JobTimeoutException;
use App\Services\DataverseAuthService;
use Illuminate\Http\Client\Response;
use Illuminate\Support\Facades\Http;
use RuntimeException;

class AutoCountClient
{
    private string $baseUrl;
    private string $entitySet;
    private int $defaultTimeout;
    private int $pollInterval;

    // Desktop multi-site routing, set per-company via setDesktopRouting() before
    // dispatching desktop jobs. Null = not set (single-site / cloud).
    private ?string $desktopSite = null;
    private ?string $desktopServerIp = null;

    public function __construct(private DataverseAuthService $auth)
    {
        $this->baseUrl = rtrim(config('autocount.environment_url') ?? config('dataverse.environment_url'), '/')
            . '/api/data/' . config('dataverse.api_version');
        $this->entitySet      = config('autocount.job_table', 'cr6f2_autocountjobs');
        $this->defaultTimeout = (int) config('autocount.default_timeout', 60);
        $this->pollInterval   = (int) config('autocount.poll_interval', 1);
    }

    /**
     * Desktop multi-site routing. Call once before dispatching desktop jobs for a
     * company so every job row is stamped with the site whose on-prem worker
     * should pick it up (cr6f2_site — the worker's poll filter) and the AutoCount
     * server IP that worker must connect to (carried inside cr6f2_params, read
     * after the worker claims the job). Resolve values with DesktopRouting::resolve().
     * Empty strings are treated as null; has no effect on cloud jobs.
     */
    public function setDesktopRouting(?string $site, ?string $serverIp): void
    {
        $this->desktopSite     = ($site !== null && trim($site) !== '') ? trim($site) : null;
        $this->desktopServerIp = ($serverIp !== null && trim($serverIp) !== '') ? trim($serverIp) : null;
    }

    // ─── Cloud jobs ─────────────────────────────────────────────────────────

    public function createDepartment(string $company, array $data): array
    {
        return $this->runCloudJob('create_department', [
            'company' => $company,
            'data'    => $data,
        ], 'high');
    }

    /**
     * Reads default to normal priority per cloud-jobs-master-data.md. Pass
     * $highPriority=true when a user is actively waiting on the fetch
     * (e.g. an on-demand sync from the UI).
     */
    public function fetchMasterData(string $company, string|array $endpoint, bool $highPriority = false): array
    {
        $params = is_array($endpoint)
            ? ['company' => $company, 'endpoints' => array_values($endpoint)]
            : ['company' => $company, 'endpoint'  => $endpoint];

        return $this->runCloudJob('fetch_masterdata', $params, $highPriority ? 'high' : 'normal');
    }

    public function captureToken(): array
    {
        return $this->runCloudJob('capture_token', [], 'high');
    }

    // TODO(cloud-jobs-master-data): wrap `create_paymentmethod` cloud job
    // when the spec is finalized. Mirror createDepartment (high priority).
    // Until then, callers use the desktop `payment_method_listing` read path.

    // ─── Desktop jobs (see autocount-listing-api.md) ────────────────────────
    //
    // Each listing job returns { type, accountBook, count, rows }. The methods
    // below return just `rows` (a list of plain arrays) so callers can iterate
    // directly without unpacking the envelope.

    /** AR debtor master — maps to our debtors table. */
    public function fetchDesktopDebtors(string $accountBook, ?array $columns = null): array
    {
        return $this->runDesktopListing('account_listing', $accountBook, $columns);
    }

    /** AP creditor master. */
    public function fetchDesktopCreditors(string $accountBook, ?array $columns = null): array
    {
        return $this->runDesktopListing('creditor_listing', $accountBook, $columns);
    }

    /**
     * GL chart of accounts with full hierarchy + AccType + SpecialAccType.
     * Slower than the basic gl_account_listing (~1 SDK call per account), so
     * expect 5–15s for 300+ accounts.
     */
    public function fetchDesktopGlAccounts(string $accountBook): array
    {
        return $this->runDesktopListing('gl_account_full_listing', $accountBook, null);
    }

    /** Project master. */
    public function fetchDesktopProjects(string $accountBook): array
    {
        return $this->runDesktopListing('project_listing', $accountBook, null);
    }

    /** Payment method master. */
    public function fetchDesktopPaymentMethods(string $accountBook): array
    {
        return $this->runDesktopListing('payment_method_listing', $accountBook, null);
    }

    /**
     * Fetch AC payment listing via the desktop job queue.
     * Returns raw rows from the result envelope ({ type, rows: [...] }).
     * Each row matches the AutoCount payment/listing API shape:
     *   master.docKey, master.docNo, master.docType, master.docDate,
     *   master.description, paymentDetails[].paymentMethod,
     *   paymentDetails[].paymentAmt, paymentDetails[].bankCharge
     */
    public function fetchDesktopPaymentListing(
        string $accountBook,
        string $startDate,
        string $endDate,
        array  $paymentMethods,
        ?int   $timeout = null,
    ): array {
        $params = [
            'accountBook'    => $accountBook,
            'startDate'      => $startDate,
            'endDate'        => $endDate,
            'paymentMethods' => array_values($paymentMethods),
        ];
        return $this->paginateDesktopJob('fetch_payment_listing', $params, $timeout ?? 120);
    }

    /**
     * Open documents + outstanding balance for one debtor/creditor — the exact
     * knock-off candidates for an A/R or A/P Payment / Refund. Backed by the SDK's
     * own knock-off lookup (fetch_outstanding_listing, see autocount-listing-api.md #12),
     * so the list matches the AutoCount payment screen.
     *
     * Each row carries `Type` (the knock-off doc-type code: RI/PB/CN/DN — use
     * verbatim as knockOffs[].docType), `DocNo`, `Outstanding` (balance cap), etc.
     *
     * @param string $side 'ar'|'debtor' or 'ap'|'creditor' (omit-safe — worker infers from code prefix)
     */
    public function fetchDesktopOutstanding(
        string $accountBook,
        string $accountCode,
        string $side,
        bool   $outstandingOnly = true,
        ?int   $timeout = null,
    ): array {
        return $this->paginateDesktopJob('fetch_outstanding_listing', [
            'accountBook'     => $accountBook,
            'accountCode'     => $accountCode,
            'side'            => $side,
            'outstandingOnly' => $outstandingOnly,
        ], $timeout ?? 120);
    }

    /**
     * BankTrans rows for a bank GL account — each row carries `dtlKey`, `amount` (signed),
     * `docDate`, `docNo`, `reconciled`, `statementDate`.
     * Pass `dtlKey`s to `reconcileDesktopBankStatement()`.
     *
     * @param string $bankAccount AutoCount GL AccNo (e.g. "310-2000")
     * @param string $reconciled  'all' | 'reconciled' | 'unreconciled'
     */
    public function fetchDesktopBankTransactions(
        string $accountBook,
        string $bankAccount,
        string $startDate,
        string $endDate,
        string $reconciled = 'all',
        ?int   $timeout = null,
    ): array {
        return $this->paginateDesktopJob('get_bank_transactions', [
            'accountBook' => $accountBook,
            'bankAccount' => $bankAccount,
            'startDate'   => $startDate,
            'endDate'     => $endDate,
            'reconciled'  => $reconciled,
        ], $timeout ?? $this->defaultTimeout);
    }

    /**
     * Upsert the BankRecon header for (bankAccount, statementDate) and tick all dtlKeys.
     * Idempotent — safe to call again; re-stamps the same statement date.
     * Set $rollbackOnly = true for a dry-run validation.
     *
     * Returns the raw result envelope:
     *   { type, bankAccount, statementDate, actualBalance, requested, ticked, rolledBack, status }
     */
    public function reconcileDesktopBankStatement(
        string $accountBook,
        string $bankAccount,
        string $statementDate,
        float  $actualBalance,
        array  $dtlKeys,
        bool   $rollbackOnly = false,
        ?int   $timeout = null,
        array  $docKeys = [],
    ): array {
        $params = [
            'accountBook'   => $accountBook,
            'bankAccount'   => $bankAccount,
            'statementDate' => $statementDate,
            'actualBalance' => $actualBalance,
            'rollbackOnly'  => $rollbackOnly,
        ];
        if (!empty($docKeys)) {
            $params['docKeys'] = array_values($docKeys);
        } else {
            $params['dtlKeys'] = array_values($dtlKeys);
        }
        return $this->runJob('desktop', 'normal', 'reconcile_bank_statement', $params, $timeout ?? 120);
    }

    /** Department master — DeptNo, ParentDeptNo, Description, Desc2, IsActive. */
    public function fetchDesktopDepartments(string $accountBook): array
    {
        return $this->runDesktopListing('department_listing', $accountBook, null);
    }

    /** Stock / inventory item master — flat rows, no variants. */
    public function fetchDesktopStockItems(string $accountBook): array
    {
        return $this->runDesktopListing('stock_item_listing', $accountBook, null);
    }

    // ─── Generic ────────────────────────────────────────────────────────────

    public function getJob(string $id): array
    {
        $row = $this->readRow($id);
        return $this->shapeJob($row);
    }

    // ─── Job management (surface failed/pending write-backs for retry/dismiss) ──

    /**
     * List recent cloud jobs for one company in the given statuses. The job
     * table has no company FK — the company name lives inside cr6f2_params —
     * so we match on a substring of the params JSON. Cloud-only by category.
     *
     * @param  string[] $statuses
     * @return array<int, array{id,status,type,error,params,result,created_on,completed_on}>
     */
    public function listCompanyJobs(string $companyName, array $statuses = ['error', 'pending'], int $top = 50): array
    {
        $name = str_replace("'", "''", trim($companyName));
        $statusClause = '(' . implode(' or ', array_map(fn($s) => "cr6f2_status eq '{$s}'", $statuses)) . ')';
        $filter = "cr6f2_category eq 'cloud' and contains(cr6f2_params,'{$name}') and {$statusClause}";

        $url = "{$this->baseUrl}/{$this->entitySet}"
            . '?$select=cr6f2_autocountjobid,cr6f2_status,cr6f2_type,cr6f2_params,cr6f2_result,cr6f2_error,createdon,cr6f2_completedon'
            . '&$filter=' . rawurlencode($filter)
            . '&$orderby=createdon desc'
            . '&$top=' . max(1, $top);

        $rows = $this->dataverse('GET', $url)->json('value') ?? [];
        return array_map(fn($r) => [
            'id'           => $r['cr6f2_autocountjobid'] ?? null,
            'status'       => $r['cr6f2_status']         ?? null,
            'type'         => $r['cr6f2_type']           ?? null,
            'error'        => $r['cr6f2_error']          ?? null,
            'params'       => $this->safeJson($r['cr6f2_params'] ?? null),
            'created_on'   => $r['createdon']            ?? null,
            'completed_on' => $r['cr6f2_completedon']    ?? null,
        ], $rows);
    }

    /** Reset a job to pending so the worker re-runs it. Clears the prior error. */
    public function retryJob(string $id): void
    {
        $url = "{$this->baseUrl}/{$this->entitySet}({$id})";
        $this->dataverse('PATCH', $url, [
            'cr6f2_status' => 'pending',
            'cr6f2_error'  => null,
            'cr6f2_result' => null,
        ]);
    }

    /** Delete a job row (dismiss a failed/stuck write-back). */
    public function deleteJob(string $id): void
    {
        $url = "{$this->baseUrl}/{$this->entitySet}({$id})";
        $this->dataverse('DELETE', $url);
    }

    // ─── Bulk Desktop listings (parallel fetch) ─────────────────────────────
    //
    // The serial pattern (fetchDesktopX → fetchDesktopY → …) inserts a job row,
    // waits 1s+, reads, repeats — even though the worker handles jobs one by
    // one anyway. The bulk pattern below dispatches all rows up front and then
    // shares one poll loop across them, which trims the inter-section sleep
    // overhead and lets the worker pipeline its own work.

    /**
     * @param  array<string, array<string, mixed>> $specs  type => params (excluding accountBook)
     * @return array<string, string>                       type => job id
     */
    public function dispatchDesktopListings(string $accountBook, array $specs): array
    {
        $ids = [];
        foreach ($specs as $type => $extraParams) {
            $params = array_merge(['accountBook' => $accountBook], $extraParams);
            $ids[$type] = $this->insertJobRow('desktop', 'normal', $type, $params);
        }
        return $ids;
    }

    /**
     * Poll a batch of dispatched job rows concurrently. Returns one entry per
     * input type: either { 'rows' => [...] } on success or { 'error' => string }
     * on failure / timeout. A failure in one job never aborts the others.
     *
     * @param  array<string, string> $jobIds  type => job id (from dispatchDesktopListings)
     * @return array<string, array{rows?: array, error?: string, job_id: string}>
     */
    /**
     * @param  array<string, array> $jobParams  type => full params (accountBook + extras).
     *   When provided and a result has hasMore: true, the next page is dispatched
     *   automatically and rows are accumulated until hasMore is false.
     */
    public function awaitDesktopListings(array $jobIds, ?int $timeout = null, array $jobParams = []): array
    {
        $timeout     = $timeout ?? $this->defaultTimeout;
        // 0 = no deadline (nightly sync). Any positive value sets a hard cutoff.
        $deadline    = $timeout > 0 ? microtime(true) + $timeout : null;
        $results     = [];
        $accumulated = []; // type => all rows accumulated across pages

        // pending: key => [type, jobId]. For page 1, key === type.
        // For subsequent pages, key === "{type}__p{N}" (so the same type can be
        // re-added without colliding with the original key).
        $pending = [];
        foreach ($jobIds as $type => $id) {
            $pending[$type] = [$type, $id];
        }

        while ($pending !== [] && ($deadline === null || microtime(true) < $deadline)) {
            sleep(max(1, $this->pollInterval));
            foreach ($pending as $key => [$type, $jobId]) {
                $row    = $this->readRow($jobId);
                $status = $row['cr6f2_status'] ?? null;
                if ($status === 'done') {
                    $resultRaw = $row['cr6f2_result'] ?? '';
                    $decoded   = $resultRaw === '' ? [] : json_decode($resultRaw, true);
                    $rows = is_array($decoded) ? ($decoded['rows'] ?? []) : [];
                    $accumulated[$type] = array_merge($accumulated[$type] ?? [], $rows);
                    unset($pending[$key]);

                    if (!empty($decoded['hasMore']) && !empty($jobParams[$type])) {
                        $nextPage   = (int) ($decoded['page'] ?? 1) + 1;
                        $nextParams = array_merge($jobParams[$type], ['page' => $nextPage]);
                        $nextId     = $this->insertJobRow('desktop', 'normal', $type, $nextParams);
                        $pending["{$type}__p{$nextPage}"] = [$type, $nextId];
                    } else {
                        $results[$type] = ['job_id' => $jobId, 'rows' => $accumulated[$type]];
                    }
                } elseif ($status === 'error') {
                    $results[$type] = [
                        'job_id' => $jobId,
                        'error'  => (string) ($row['cr6f2_error'] ?? 'unknown error'),
                    ];
                    unset($pending[$key]);
                }
            }
        }

        // Any jobs still pending past the deadline get marked as timeouts —
        // we don't throw, so the caller can salvage the sections that did finish.
        foreach ($pending as $key => [$type, $jobId]) {
            if (!isset($results[$type])) {
                $results[$type] = [
                    'job_id' => $jobId,
                    'error'  => "Job {$jobId} did not complete within {$timeout}s.",
                ];
            }
        }

        return $results;
    }

    // ─── Desktop writes (see autocount-write-api.md) ─────────────────────────
    //
    // Write jobs create documents in AutoCount Desktop (AR/AP invoice, CN/DN,
    // journal entry, cash book). They run high priority so the worker picks
    // them up ahead of listing reads. On success cr6f2_result holds
    // { type, status:"created", docNo, docKey }.

    /**
     * Dispatch a single high-priority desktop write job and block until
     * done/error/timeout. Returns the decoded cr6f2_result ({type,status,docNo,docKey}).
     * $params must already include accountBook + master + details (+ payments / cashBookType).
     */
    public function createDesktopDocument(string $type, array $params, ?int $timeout = null): array
    {
        return $this->runJob('desktop', 'high', $type, $params, $timeout);
    }

    /**
     * Insert a batch of desktop write jobs up front (non-blocking). Twin of
     * dispatchDesktopListings — lets a batch be polled in one shared loop instead
     * of N serial blocking waits.
     *
     * @param  array<string, array{type: string, params: array}> $specs  localKey => spec
     * @return array<string, string>                                     localKey => job id
     */
    public function dispatchDesktopWrites(array $specs): array
    {
        $ids = [];
        foreach ($specs as $key => $spec) {
            $ids[$key] = $this->insertJobRow('desktop', 'high', $spec['type'], $spec['params']);
        }
        return $ids;
    }

    /**
     * Batch-read job rows by id — one query per ~15 ids instead of one GET per
     * job. Returns id => raw row (status/result/error/completedon). Used by the
     * async export reconciler so a 50-doc batch polls cheaply (1–4 reads/pass
     * instead of 50), which also avoids Dataverse throttling on big batches.
     *
     * @param  string[] $ids
     * @return array<string, array>  job id => raw row
     */
    public function readRowsBatch(array $ids): array
    {
        $out = [];
        $ids = array_values(array_filter(array_unique($ids)));
        foreach (array_chunk($ids, 15) as $chunk) {
            // GUID equality in OData is unquoted: `cr6f2_autocountjobid eq <guid>`.
            $clause = '(' . implode(' or ', array_map(fn ($id) => "cr6f2_autocountjobid eq {$id}", $chunk)) . ')';
            $url = "{$this->baseUrl}/{$this->entitySet}"
                . '?$select=cr6f2_autocountjobid,cr6f2_status,cr6f2_result,cr6f2_error,cr6f2_completedon'
                . '&$filter=' . rawurlencode($clause)
                . '&$top=' . count($chunk);
            $rows = $this->dataverse('GET', $url)->json('value') ?? [];
            foreach ($rows as $r) {
                if (!empty($r['cr6f2_autocountjobid'])) {
                    $out[$r['cr6f2_autocountjobid']] = $r;
                }
            }
        }
        return $out;
    }

    /**
     * Poll a batch of dispatched write jobs concurrently. Returns one entry per
     * input key: { 'result' => {type,status,docNo,docKey} } on success or
     * { 'error' => string } on failure / timeout. A failure in one job never
     * aborts the others. Mirrors awaitDesktopListings but keeps the full result.
     *
     * @param  array<string, string> $jobIds  localKey => job id (from dispatchDesktopWrites)
     * @return array<string, array{result?: array, error?: string, job_id: string}>
     */
    public function awaitDesktopWrites(array $jobIds, ?int $timeout = null): array
    {
        $timeout  = $timeout ?? $this->defaultTimeout;
        $deadline = microtime(true) + $timeout;
        $results  = [];
        $pending  = $jobIds; // key => id

        while ($pending !== [] && microtime(true) < $deadline) {
            sleep(max(1, $this->pollInterval));
            foreach ($pending as $key => $jobId) {
                $row    = $this->readRow($jobId);
                $status = $row['cr6f2_status'] ?? null;
                if ($status === 'done') {
                    $resultRaw = $row['cr6f2_result'] ?? '';
                    $decoded   = $resultRaw === '' ? [] : json_decode($resultRaw, true);
                    $results[$key] = [
                        'job_id' => $jobId,
                        'result' => is_array($decoded) ? $decoded : [],
                    ];
                    unset($pending[$key]);
                } elseif ($status === 'error') {
                    $results[$key] = [
                        'job_id' => $jobId,
                        'error'  => (string) ($row['cr6f2_error'] ?? 'unknown error'),
                    ];
                    unset($pending[$key]);
                }
            }
        }

        // Any rows still pending past the deadline get marked as timeouts —
        // we don't throw, so the caller can salvage the docs that did finish.
        foreach ($pending as $key => $jobId) {
            $results[$key] = [
                'job_id' => $jobId,
                'error'  => "Job {$jobId} did not complete within {$timeout}s.",
            ];
        }

        return $results;
    }

    // ─── Cloud REST document writes (edit / delete) ─────────────────────────
    //
    // The AC Cloud create path POSTs directly to the REST API (UploadToAutocountJob::
    // handleCloud) rather than the cr6f2_autocountjobs queue. Edit/delete mirror that:
    // PUT/DELETE {acApiEndpoint}{route}?docNo=… with the per-company API-Key / Key-ID.
    // Single-document, synchronous — the caller is a user waiting on one action.

    /**
     * PUT (replace) an existing AC Cloud document. Body is the FULL document (same
     * shape as create), since Cloud PUT replaces the record. Returns a normalized
     * outcome so the caller never has to guess HTTP semantics.
     *
     * @return array{ok: bool, status: int, notFound: bool, error: ?string, body: string}
     */
    public function updateCloudDocument(string $endpoint, string $apiKey, string $keyId, string $route, string $docNo, array $body): array
    {
        $response = Http::withHeaders([
                'API-Key' => $apiKey,
                'Key-ID'  => $keyId,
                'Accept'  => 'application/json',
            ])
            ->timeout(30)
            ->put($this->cloudUrl($endpoint, $route, $docNo), $body);

        return $this->normalizeCloudResponse($response);
    }

    /**
     * DELETE an AC Cloud document by docNo.
     *
     * @return array{ok: bool, status: int, notFound: bool, error: ?string, body: string}
     */
    public function deleteCloudDocument(string $endpoint, string $apiKey, string $keyId, string $route, string $docNo): array
    {
        $response = Http::withHeaders([
                'API-Key' => $apiKey,
                'Key-ID'  => $keyId,
                'Accept'  => 'application/json',
            ])
            ->timeout(30)
            ->delete($this->cloudUrl($endpoint, $route, $docNo));

        return $this->normalizeCloudResponse($response);
    }

    /** Build `{endpoint}{route}?docNo=…`. $endpoint already ends in `{accountBookId}/`. */
    private function cloudUrl(string $endpoint, string $route, string $docNo): string
    {
        return $endpoint . $route . '?docNo=' . rawurlencode($docNo);
    }

    /** Normalize an AC Cloud PUT/DELETE response; flags 404 so the caller won't fake success. */
    private function normalizeCloudResponse(Response $response): array
    {
        $status = $response->status();
        $body   = $response->body();

        if ($status >= 200 && $status < 300) {
            return ['ok' => true, 'status' => $status, 'notFound' => false, 'error' => null, 'body' => $body];
        }

        $notFound = $status === 404;
        $message  = json_decode($body, true)['message'] ?? trim($body);

        return [
            'ok'       => false,
            'status'   => $status,
            'notFound' => $notFound,
            'error'    => $message !== '' ? $message : "HTTP {$status}",
            'body'     => $body,
        ];
    }

    /**
     * Dispatch a `cloud` job and block until done/error/timeout. Priority is
     * caller-controlled: writes (createDepartment) and on-demand user-waiting
     * calls use 'high'; background reads use 'normal' (cloud-jobs-master-data.md).
     */
    private function runCloudJob(string $type, array $params, string $priority = 'normal', ?int $timeout = null): array
    {
        return $this->runJob('cloud', $priority, $type, $params, $timeout);
    }

    /**
     * Blocking master-data write-back (create/update/delete) for either worker.
     * Inserts a high-priority job and polls until the worker writes done/error
     * (or times out), then returns the decoded cr6f2_result. On a worker/AC
     * failure it throws a typed AutoCount* exception (via throwForWorkerError),
     * preserving AutoCount's own wording — e.g. "Can't change Bank Account
     * because Cash Book Entry is using this Payment Method." Callers use this so
     * the local Dataverse write can be gated on the AutoCount write succeeding.
     *
     * @param string $category 'cloud' or 'desktop'
     */
    public function runMasterDataJob(string $category, string $type, array $params, ?int $timeout = null): array
    {
        return $this->runJob($category, 'high', $type, $params, $timeout);
    }

    /**
     * Fire-and-forget cloud job dispatch. Returns the new cr6f2_autocountjobs
     * row id immediately — does NOT poll for completion. Use when:
     *   - the user shouldn't wait (sub-second response required)
     *   - the local Dataverse write is the source of truth and the cloud push
     *     is eventual-consistency (e.g. master-data write-back: edit lands in
     *     Dataverse first, AutoCount Cloud gets the change asynchronously).
     * If the worker fails, the row sits in 'error' status — surface that via
     * cr6f2_systemlog / the existing sync-status UI rather than re-running here.
     */
    public function enqueueCloudJob(string $type, array $params, string $priority = 'normal'): string
    {
        return $this->insertJobRow('cloud', $priority, $type, $params);
    }

    /**
     * Fire-and-forget desktop job dispatch — twin of enqueueCloudJob for the
     * AutoCountDesktop C# worker. Returns the new job-row id immediately; does
     * NOT poll. Used for master-data CRUD write-back on Desktop-synced
     * companies (the C# worker handles category=desktop rows via the SDK).
     */
    public function enqueueDesktopJob(string $type, array $params, string $priority = 'high'): string
    {
        return $this->insertJobRow('desktop', $priority, $type, $params);
    }

    /**
     * Dispatch a normal-priority `desktop` listing job and block until done.
     * Returns just the `rows` array from the result envelope.
     */
    private function runDesktopListing(string $type, string $accountBook, ?array $columns, ?int $timeout = null): array
    {
        $params = ['accountBook' => $accountBook];
        if ($columns !== null && $columns !== []) {
            $params['columns'] = array_values($columns);
        }
        return $this->paginateDesktopJob($type, $params, $timeout);
    }

    /**
     * Run a desktop job and follow hasMore pagination until all rows are fetched.
     * The desktop worker caps each page at 100 rows; this loops until done.
     */
    private function paginateDesktopJob(string $type, array $baseParams, ?int $timeout = null): array
    {
        $allRows = [];
        $page    = 1;
        do {
            $result  = $this->runJob('desktop', 'normal', $type, array_merge($baseParams, ['page' => $page]), $timeout);
            $allRows = array_merge($allRows, $result['rows'] ?? []);
            $hasMore = !empty($result['hasMore']);
            $page++;
        } while ($hasMore);
        return $allRows;
    }

    private function runJob(string $category, string $priority, string $type, array $params, ?int $timeout): array
    {
        $id = $this->insertJobRow($category, $priority, $type, $params);
        return $this->pollJob($id, $timeout);
    }

    /**
     * Block on an existing job row until done/error/timeout. Returns the decoded
     * cr6f2_result on success, maps a worker error to the right exception, or throws
     * JobTimeoutException (carrying the row id, since the worker may still finish).
     */
    private function pollJob(string $id, ?int $timeout): array
    {
        $timeout  = $timeout ?? $this->defaultTimeout;
        $deadline = microtime(true) + $timeout;
        while (microtime(true) < $deadline) {
            // 1 Hz cap per the vault doc — Dataverse rate limits are real and shared.
            sleep(max(1, $this->pollInterval));

            $row    = $this->readRow($id);
            $status = $row['cr6f2_status'] ?? null;

            if ($status === 'done') {
                $resultRaw = $row['cr6f2_result'] ?? '';
                $decoded   = $resultRaw === '' ? [] : json_decode($resultRaw, true);
                if (!is_array($decoded)) {
                    throw new AutoCountJobException("AutoCount job {$id} returned a malformed result payload.");
                }
                return $decoded;
            }

            if ($status === 'error') {
                $this->throwForWorkerError((string) ($row['cr6f2_error'] ?? 'unknown error'));
            }
        }

        // Don't delete the row — the worker may still finish. Hand the caller the id.
        throw new JobTimeoutException($id, "AutoCount job {$id} did not complete within {$timeout}s.");
    }

    /**
     * Idempotent desktop single-write for edit/delete: if a job with the same
     * (category|type|params) idempotency key already exists, REUSE it instead of
     * dispatching a conflicting duplicate — so a retry after a PHP timeout converges
     * to the worker's single result rather than editing/deleting twice.
     * Returns the same decoded result envelope as createDesktopDocument.
     */
    public function createDesktopDocumentIdempotent(string $type, array $params, ?int $timeout = null): array
    {
        $key      = $this->idempotencyKey('desktop', $type, $params);
        $existing = $this->findJobByIdempotencyKey($key);
        if ($existing !== null) {
            $status = $existing['cr6f2_status'] ?? null;
            $jobId  = (string) ($existing['cr6f2_autocountjobid'] ?? '');
            if ($status === 'done') {
                $decoded = $this->safeJson($existing['cr6f2_result'] ?? null);
                if (is_array($decoded)) return $decoded;
            } elseif ($status === 'error') {
                $this->throwForWorkerError((string) ($existing['cr6f2_error'] ?? 'unknown error'));
            } elseif ($jobId !== '' && in_array($status, ['pending', 'running'], true)) {
                // An identical job is still in flight — wait on it, don't create a second.
                return $this->pollJob($jobId, $timeout);
            }
        }
        return $this->createDesktopDocument($type, $params, $timeout);
    }

    /** Most recent job row for an idempotency key, or null. */
    private function findJobByIdempotencyKey(string $key): ?array
    {
        $url = "{$this->baseUrl}/{$this->entitySet}"
            . '?$select=cr6f2_autocountjobid,cr6f2_status,cr6f2_result,cr6f2_error'
            . '&$filter=' . rawurlencode("cr6f2_idempotencykey eq '{$key}'")
            . '&$orderby=createdon desc&$top=1';
        try {
            $rows = $this->dataverse('GET', $url)->json('value') ?? [];
        } catch (\Throwable) {
            return null; // non-fatal — fall back to a fresh dispatch
        }
        return $rows[0] ?? null;
    }

    /**
     * Map a worker-side error string to the right PHP exception class.
     * Always preserves the full original message so the caller can surface
     * AutoCount's own wording ("DeptNo is empty.", etc.).
     */
    private function throwForWorkerError(string $err): never
    {
        if (str_starts_with($err, 'ValueError:')) {
            throw new \InvalidArgumentException($err);
        }

        if (preg_match('/^RuntimeError: AutoCount Cloud returned HTTP (\d{3}):/', $err, $m)) {
            $code = (int) $m[1];
            if ($code === 409) {
                throw new DuplicateRecordException($err);
            }
            if ($code >= 400 && $code < 500) {
                throw new AutoCountClientException($err);
            }
            if ($code >= 500) {
                throw new AutoCountServerException($err);
            }
        }

        if (str_starts_with($err, 'RuntimeError: Token expired and could not be refreshed')) {
            throw new AutoCountAuthException($err);
        }

        throw new AutoCountJobException($err);
    }

    // ─── Dataverse transport ────────────────────────────────────────────────

    /**
     * Build the standard job-row body and POST it. Returns the new row's GUID.
     * Shared by runJob (blocking) and dispatchDesktopListings (non-blocking).
     */
    private function insertJobRow(string $category, string $priority, string $type, array $params): string
    {
        // Desktop multi-site routing (see setDesktopRouting): the server IP rides
        // inside params under the "server" key (worker reads it after claiming);
        // site is a top-level column so each worker's OData poll can filter
        // cr6f2_site eq '<its site>'.
        if ($category === 'desktop' && $this->desktopServerIp !== null && !isset($params['server'])) {
            $params['server'] = $this->desktopServerIp;
        }
        $body = [
            'cr6f2_category'       => $category,
            'cr6f2_type'           => $type,
            'cr6f2_priority'       => $priority,
            'cr6f2_status'         => 'pending',
            'cr6f2_params'         => json_encode($params, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE),
            'cr6f2_idempotencykey' => $this->idempotencyKey($category, $type, $params),
        ];
        if ($category === 'desktop' && $this->desktopSite !== null) {
            $body['cr6f2_site'] = $this->desktopSite;
        }
        return $this->insertRow($body);
    }

    /** Stable dedupe key for a job — same (category, type, params) → same key. */
    private function idempotencyKey(string $category, string $type, array $params): string
    {
        return hash('sha256', $category . '|' . $type . '|' . json_encode($params, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE));
    }

    private function insertRow(array $body): string
    {
        $url = "{$this->baseUrl}/{$this->entitySet}";

        $res = $this->dataverse('POST', $url, $body);

        // POST with Prefer: return=representation returns the new row inline.
        $rowBody = $res->json();
        if (is_array($rowBody) && !empty($rowBody['cr6f2_autocountjobid'])) {
            return (string) $rowBody['cr6f2_autocountjobid'];
        }

        $entityId = $res->header('OData-EntityId');
        if ($entityId && preg_match('/\(([0-9a-f\-]{36})\)$/i', $entityId, $m)) {
            return $m[1];
        }

        throw new RuntimeException('Dataverse insert succeeded but no job id was returned.');
    }

    private function readRow(string $id): array
    {
        $url = "{$this->baseUrl}/{$this->entitySet}({$id})";
        return $this->dataverse('GET', $url)->json() ?? [];
    }

    private function shapeJob(array $row): array
    {
        return [
            'id'           => $row['cr6f2_autocountjobid'] ?? null,
            'status'       => $row['cr6f2_status']         ?? null,
            'type'         => $row['cr6f2_type']           ?? null,
            'priority'     => $row['cr6f2_priority']       ?? null,
            'params'       => $this->safeJson($row['cr6f2_params'] ?? null),
            'result'       => $this->safeJson($row['cr6f2_result'] ?? null),
            'error'        => $row['cr6f2_error']          ?? null,
            'completed_on' => $row['cr6f2_completedon']    ?? null,
        ];
    }

    private function safeJson(?string $s): mixed
    {
        if ($s === null || $s === '') return null;
        $decoded = json_decode($s, true);
        return $decoded ?? $s; // hand back the raw string if it isn't valid JSON
    }

    /**
     * Authenticated Dataverse request with one 401-retry-after-refresh, mirroring
     * the pattern used in DataverseService. Kept in this client (not delegated to
     * DataverseService) so the polling loop owns its own timeouts and headers.
     */
    private function dataverse(string $method, string $url, array $body = []): Response
    {
        $send = function (string $token) use ($method, $url, $body): Response {
            $http = Http::withToken($token)
                ->withHeaders([
                    'Accept'           => 'application/json',
                    'OData-MaxVersion' => '4.0',
                    'OData-Version'    => '4.0',
                    'Prefer'           => 'return=representation',
                ])
                ->timeout(config('dataverse.timeout'));

            return match (strtoupper($method)) {
                'GET'    => $http->get($url),
                'POST'   => $http->post($url, $body),
                'PATCH'  => $http->patch($url, $body),
                'DELETE' => $http->delete($url),
                default  => throw new RuntimeException("Unsupported method {$method}"),
            };
        };

        $res = $send($this->auth->getAccessToken());

        if ($res->status() === 401) {
            $res = $send($this->auth->refreshAccessToken());
        }

        if ($res->failed()) {
            $msg = $res->json('error.message') ?? $res->body();
            throw new RuntimeException("Dataverse {$method} {$url} failed [{$res->status()}]: {$msg}");
        }

        return $res;
    }
}
