<?php

/*
 * Copyright 2026 Google LLC
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 *     * Redistributions of source code must retain the above copyright
 * notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above
 * copyright notice, this list of conditions and the following disclaimer
 * in the documentation and/or other materials provided with the
 * distribution.
 *     * Neither the name of Google Inc. nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
namespace Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\ResumableUpload;

use Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\ApiException;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\ApiStatus;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\Call;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\CredentialsWrapper;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\Middleware\RetryMiddleware;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\RetrySettings;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\ApiCore\ValidationException;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\Protobuf\Internal\Message;
use Matomo\Dependencies\GoogleAnalyticsImporter\Google\Rpc\Code;
use Matomo\Dependencies\GoogleAnalyticsImporter\GuzzleHttp\Exception\RequestException;
use Matomo\Dependencies\GoogleAnalyticsImporter\GuzzleHttp\Promise\Create;
use Matomo\Dependencies\GoogleAnalyticsImporter\GuzzleHttp\Psr7\Request;
use Matomo\Dependencies\GoogleAnalyticsImporter\Psr\Http\Message\RequestInterface;
use Matomo\Dependencies\GoogleAnalyticsImporter\Psr\Http\Message\ResponseInterface;
use Matomo\Dependencies\GoogleAnalyticsImporter\Psr\Http\Message\StreamInterface;
use Throwable;
/**
 * Manages the REST transport and authentication credentials for resumable upload RPCs,
 * and executes the HTTP upload stream loop.
 * Instantiated during GAPIC client initialization when the service has resumable upload methods.
 *
 * @internal
 */
class ResumableUploadClient
{
    private const PHASE_STARTING = 'STARTING';
    private const PHASE_TRANSMITTING = 'TRANSMITTING';
    private const PHASE_FINALIZING = 'FINALIZING';
    private const PHASE_RECOVERY = 'RECOVERY';
    private const PHASE_DONE = 'DONE';
    private const DEFAULT_CHUNK_SIZE = 8388608;
    private const DEFAULT_TOTAL_TIMEOUT_MILLIS = 600000;
    private const MAX_RECOVERY_ATTEMPTS = 3;
    private ?ResponseInterface $finalResponse = null;
    /**
     * @param ResumableUploadTransportInterface $transport Transport implementing buildRequest and sendRawRequest.
     * @param CredentialsWrapper $credentialsWrapper The credentials wrapper from GAPIC client.
     * @param array $headers Custom headers to include with the initial upload request.
     * @param string $uploadPrefix Resumable upload path prefix (default: '/resumable/upload').
     */
    public function __construct(private ResumableUploadTransportInterface $transport, private CredentialsWrapper $credentialsWrapper, private array $headers = [], private string $uploadPrefix = '/resumable/upload')
    {
    }
    /**
     * Starts the resumable upload exchange using the provided data stream.
     *
     * @param ResumableUpload $upload
     * @param StreamInterface $dataStream
     * @param Call $call
     * @param array $callOptions {
     *     Optional.
     *
     *     @type array $headers Optional. Key-value array of custom HTTP headers to
     *           include with upload requests.
     *     @type int $timeoutMillis Optional. The timeout in milliseconds for the
     *           initial start call.
     *     @type RetrySettings|array $retrySettings Optional. Retry settings to use for the
     *           initial start call.
     * }
     * @param array $resumableUploadOptions {
     *     Optional.
     *
     *     @type int $chunkSize Optional. The size of each chunk to upload in bytes.
     *           Must be a multiple of 262144 (256 KB). Values smaller than the server's chunk
     *           granularity (typically 256 KB) will be rounded up to match the granularity.
     *           Defaults to 8388608 (8 MB).
     *     @type callable $progressCallback Optional. A callback function executed after
     *           every chunk upload or query. The callback should accept two arguments:
     *           (int $bytesUploaded, ResumableUpload $upload).
     *     @type int $totalTimeoutMillis Optional. The total timeout in milliseconds for the
     *           entire resumable upload operation. Defaults to 600000 (10 minutes).
     *     @type string $uploadUrl Optional. An existing resumable upload session URL
     *           to resume an upload across process restarts or interruptions.
     * }
     * @return Message
     * @throws ApiException
     */
    public function startUpload(ResumableUpload $upload, StreamInterface $dataStream, Call $call, array $callOptions = [], array $resumableUploadOptions = []) : Message
    {
        $this->finalResponse = null;
        $uploadUrl = $upload->getUploadUrl() ?? $resumableUploadOptions['uploadUrl'] ?? null;
        $totalTimeoutMillis = (float) ($resumableUploadOptions['totalTimeoutMillis'] ?? self::DEFAULT_TOTAL_TIMEOUT_MILLIS);
        $deadlineMs = microtime(\true) * 1000 + $totalTimeoutMillis;
        $state = new ResumableUploadState($resumableUploadOptions['chunkSize'] ?? self::DEFAULT_CHUNK_SIZE, $resumableUploadOptions['progressCallback'] ?? null, $uploadUrl, $uploadUrl !== null ? self::PHASE_RECOVERY : self::PHASE_STARTING);
        while ($state->phase !== self::PHASE_DONE) {
            $this->checkDeadline($deadlineMs);
            try {
                $state->phase = match ($state->phase) {
                    self::PHASE_STARTING => $call->getMessage() !== null ? $this->phaseStarting($state, $upload, $dataStream, $call, $callOptions) : throw new ValidationException('A Call with request message is required when starting a new resumable upload.'),
                    self::PHASE_TRANSMITTING, self::PHASE_FINALIZING => $this->phaseUploading($state, $upload, $dataStream),
                    self::PHASE_RECOVERY => $this->phaseRecovery($state, $upload, $dataStream),
                    default => throw new ApiException("Unexpected phase: {$state->phase}", 0, ApiStatus::INTERNAL),
                };
            } catch (Throwable $e) {
                $state->phase = $this->handleException($e, $state, $deadlineMs);
            }
        }
        $decodeType = $call->getDecodeType();
        if ($decodeType === null || !class_exists($decodeType)) {
            throw new ValidationException('A valid decodeType is required on the Call object.');
        }
        if ($this->finalResponse === null) {
            throw new ApiException('No final response received from server.', 0, ApiStatus::INTERNAL);
        }
        $body = (string) $this->finalResponse->getBody();
        if ($body === '') {
            throw new ApiException('Final response body was empty.', 0, ApiStatus::INTERNAL);
        }
        /** @var Message $responseMessage */
        $responseMessage = new $decodeType();
        $responseMessage->mergeFromJsonString($body, \true);
        return $responseMessage;
    }
    private function phaseStarting(ResumableUploadState $state, ResumableUpload $upload, StreamInterface $dataStream, Call $call, array $callOptions = []) : string
    {
        $headers = array_merge($this->headers, $callOptions['headers'] ?? []);
        $headers['X-Goog-Upload-Protocol'] = 'resumable';
        $headers['X-Goog-Upload-Command'] = 'start';
        if ($dataStream->getSize() !== null) {
            $headers['X-Goog-Upload-Header-Content-Length'] = (string) $dataStream->getSize();
        }
        $request = $this->transport->buildRequest($call->getMethod(), $call->getMessage(), $headers);
        // Add upload prefix
        $uri = $request->getUri();
        $request = $request->withUri($uri->withPath($this->uploadPrefix . $uri->getPath()));
        // Add retry settings
        $retrySettings = $callOptions['retrySettings'] ?? null;
        if ($retrySettings !== null && !$retrySettings instanceof RetrySettings) {
            $retrySettings = RetrySettings::constructDefault()->with($retrySettings);
        }
        // Make the request
        $response = $this->sendRequest($request, $callOptions['timeoutMillis'] ?? null, $retrySettings);
        if ($response->getStatusCode() !== 200) {
            $this->handleErrorResponse($response);
        }
        $urlHeader = $response->getHeaderLine('X-Goog-Upload-URL');
        if (!empty($urlHeader)) {
            if ($request->getUri()->getScheme() === 'https' && str_starts_with($urlHeader, 'http://')) {
                $urlHeader = 'https://' . substr($urlHeader, 7);
            }
            $state->uploadUrl = $urlHeader;
        }
        if ($state->uploadUrl !== null) {
            $upload->setUploadUrl($state->uploadUrl);
        }
        $granularityHeader = $response->getHeaderLine('X-Goog-Upload-Chunk-Granularity');
        $state->chunkGranularity = !empty($granularityHeader) ? (int) $granularityHeader : 1;
        $statusHeader = $response->getHeaderLine('X-Goog-Upload-Status');
        if ($statusHeader === 'final') {
            $this->finalResponse = $response;
            return self::PHASE_DONE;
        }
        return self::PHASE_TRANSMITTING;
    }
    private function phaseUploading(ResumableUploadState $state, ResumableUpload $upload, StreamInterface $dataStream) : string
    {
        $state->prepareBuffer($dataStream);
        $headers = [];
        $headers['X-Goog-Upload-Offset'] = (string) $state->committedOffset;
        $body = (string) $state->buffer;
        if ($state->isEof) {
            $phase = self::PHASE_FINALIZING;
            $headers['X-Goog-Upload-Command'] = strlen($body) > 0 ? 'upload, finalize' : 'finalize';
        } else {
            $phase = self::PHASE_TRANSMITTING;
            $headers['X-Goog-Upload-Command'] = 'upload';
        }
        $response = $this->sendRequest(new Request('POST', (string) $state->uploadUrl, $headers, $body));
        if ($response->getStatusCode() !== 200) {
            $this->handleErrorResponse($response);
        }
        if ($state->progressCallback && $headers['X-Goog-Upload-Command'] !== 'finalize') {
            ($state->progressCallback)($state->committedOffset + strlen($body), $upload);
        }
        if ($response->getHeaderLine('X-Goog-Upload-Status') === 'final') {
            $this->finalResponse = $response;
            return self::PHASE_DONE;
        }
        $state->commitBuffer();
        return self::PHASE_TRANSMITTING;
    }
    private function phaseRecovery(ResumableUploadState $state, ResumableUpload $upload, StreamInterface $dataStream) : string
    {
        if (empty($state->uploadUrl)) {
            throw new ValidationException('Cannot recover resumable upload: uploadUrl is not set.');
        }
        $headers = ['X-Goog-Upload-Command' => 'query'];
        $response = $this->sendRequest(new Request('POST', (string) $state->uploadUrl, $headers, ''));
        $statusCode = $response->getStatusCode();
        if ($statusCode === 200) {
            $serverOffsetStr = $response->getHeaderLine('X-Goog-Upload-Size-Received');
            $serverOffset = !empty($serverOffsetStr) || $serverOffsetStr === '0' ? (int) $serverOffsetStr : $state->committedOffset;
            $state->reconcileRecoveryOffset($serverOffset, $dataStream, self::MAX_RECOVERY_ATTEMPTS);
            $statusHeader = $response->getHeaderLine('X-Goog-Upload-Status');
            if ($statusHeader === 'final') {
                $this->finalResponse = $response;
                return self::PHASE_DONE;
            }
            if ($state->progressCallback) {
                ($state->progressCallback)($state->committedOffset, $upload);
            }
            return self::PHASE_TRANSMITTING;
        }
        $this->handleErrorResponse($response);
    }
    private function sendRequest(RequestInterface $request, ?int $timeoutMillis = null, ?RetrySettings $retrySettings = null) : ResponseInterface
    {
        $reqHeaders = $request->getHeaders();
        if ($authCallback = $this->credentialsWrapper->getAuthorizationHeaderCallback()) {
            $reqHeaders = array_merge($reqHeaders, $authCallback());
        }
        foreach ($reqHeaders as $k => $v) {
            $request = $request->withHeader($k, $v);
        }
        $callOptions = [];
        if ($timeoutMillis !== null && $timeoutMillis > 0) {
            $callOptions['timeout'] = $timeoutMillis / 1000;
        }
        if ($retrySettings !== null) {
            $callOptions['retrySettings'] = $retrySettings;
            $middleware = new RetryMiddleware(fn(Call $unusedCall, array $options) => Create::promiseFor($this->transport->sendRawRequest($request, $options)), $retrySettings);
            $response = $middleware(
                new Call(''),
                // unused
                $callOptions
            );
        } else {
            $response = $this->transport->sendRawRequest($request, $callOptions);
        }
        if (is_object($response) && method_exists($response, 'wait')) {
            $response = $response->wait();
        }
        return $response;
    }
    private function checkDeadline(float $deadlineMs, ?\Throwable $previous = null) : void
    {
        if (microtime(\true) * 1000 >= $deadlineMs) {
            throw new ApiException('Resumable upload total timeout exceeded.', Code::DEADLINE_EXCEEDED, ApiStatus::DEADLINE_EXCEEDED, $previous ? ['previous' => $previous] : []);
        }
    }
    private function handleException(\Throwable $e, ResumableUploadState $state, float $deadlineMs) : string
    {
        $this->checkDeadline($deadlineMs, $e);
        $code = (int) $e->getCode();
        if ($e instanceof RequestException) {
            $response = method_exists($e, 'getResponse') ? $e->getResponse() : null;
            if ($response) {
                $code = $response->getStatusCode();
            }
        }
        // For transient HTTP errors, return the current phase unchanged so that the match loop
        // re-runs the phase and retries the request until the total deadline is exceeded.
        if (in_array($code, [429, 500, 502, 503, 504])) {
            return $state->phase;
        }
        // For range mismatch or bad request errors, transition to the query/recovery phase to
        // verify the server's received offset and resume transmitting from there, provided
        // an uploadUrl session has already been established.
        if ($state->uploadUrl !== null && in_array($code, [308, 400, 412, 416])) {
            return self::PHASE_RECOVERY;
        }
        if ($e instanceof ApiException || $e instanceof ValidationException) {
            throw $e;
        }
        throw new ApiException($e->getMessage(), $code, ApiStatus::INTERNAL, ['previous' => $e]);
    }
    private function handleErrorResponse(ResponseInterface $response) : never
    {
        $statusCode = $response->getStatusCode();
        $body = (string) $response->getBody();
        if ($response->getHeaderLine('X-Goog-Upload-Status') === 'final') {
            throw new ApiException($body ?: 'Upload rejected by server', $statusCode, ApiStatus::INVALID_ARGUMENT);
        }
        throw new ApiException("HTTP error {$statusCode}: {$body}", $statusCode, ApiStatus::statusFromRpcCode(ApiStatus::rpcCodeFromHttpStatusCode($statusCode)));
    }
}
