<?php

/**

* Copyright 2022 Google Inc.

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

namespace Google\Cloud\Firestore;

use Google\ApiCore\ApiException;

use Google\ApiCore\Options\CallOptions;

use Google\Cloud\Core\ApiHelperTrait;

use Google\Cloud\Core\DebugInfoTrait;

use Google\Cloud\Core\Exception\ServiceException;

use Google\Cloud\Core\OptionsValidator;

use Google\Cloud\Core\RequestProcessorTrait;

use Google\Cloud\Core\Timestamp;

use Google\Cloud\Core\TimeTrait;

use Google\Cloud\Core\ValidateTrait;

use Google\Cloud\Firestore\FieldValue\DeleteFieldValue;

use Google\Cloud\Firestore\FieldValue\DocumentTransformInterface;

use Google\Cloud\Firestore\FieldValue\FieldValueInterface;

use Google\Cloud\Firestore\V1\BatchWriteRequest;

use Google\Cloud\Firestore\V1\Client\FirestoreClient;

use Google\Cloud\Firestore\V1\CommitRequest;

use Google\Cloud\Firestore\V1\RollbackRequest;

use Google\Rpc\Code;

/**

* Enqueue and write multiple mutations to Cloud Firestore.

*

* This class may be used directly for multiple non-transactional writes with

* automatic retry on failure. To run changes in a transaction (with automatic

* retry/rollback on failure), use {@see \Google\Cloud\Firestore\Transaction}.

* Single modifications can be made using the various methods on

* {@see \Google\Cloud\Firestore\DocumentReference}.

*

* Example:

* ```

* use Google\Cloud\Firestore\FirestoreClient;

*

* $firestore = new FirestoreClient();

* $batch = $firestore->bulkWriter();

* ```

*/

class BulkWriter

{

use ApiHelperTrait;

use DebugInfoTrait;

use TimeTrait;

use ValidateTrait;

use RequestProcessorTrait;

const TYPE_UPDATE = 'update';

const TYPE_SET = 'set';

const TYPE_CREATE = 'create';

const TYPE_DELETE = 'delete';

const TYPE_TRANSFORM = 'transform';

/**

* @var array Holds default configurations for Bulkwriter.

* These values are experimentally derived after performance evaluations.

* The underlying constants may change in backwards-incompatible ways.

* Please use with caution, and test thoroughly when upgrading.

*/

private static $defaultOptions = [

/** The maximum number of writes that can be in a single batch. */

'MAX_BATCH_SIZE' => 20,

/** The maximum number of writes that can be in a batch containing retries. */

'RETRY_MAX_BATCH_SIZE' => 20,

/**

* The maximum number of retries that will be attempted with backoff before stopping all retry

* attempts.

*/

'MAX_RETRY_ATTEMPTS' => 10,

/**

* The default initial backoff time in milliseconds after an error. Set to 1s according to

* https://cloud.google.com/apis/design/errors.

*/

'DEFAULT_BACKOFF_INITIAL_DELAY_MS' => 500,

/** The default maximum backoff time in milliseconds when retrying an operation. */

'DEFAULT_BACKOFF_MAX_DELAY_MS' => 60 * 1000,

/** The default factor to increase the backup by after each failed attempt. */

'DEFAULT_BACKOFF_FACTOR' => 1.5,

/**

* The default jitter to apply to the exponential backoff used in retries.

* For example, a factor of 0.3 means a 30% jitter is applied.

*/

'DEFAULT_JITTER_FACTOR' => 0.3,

/**

* The starting maximum number of operations per second as allowed by the 500/50/5 rule.

*

* @see [Ramping up traffic](https://cloud.google.com/firestore/docs/best-practices#ramping_up_traffic)

*/

'DEFAULT_STARTING_MAXIMUM_OPS_PER_SECOND' => 20,

/**

* The maximum number of operations per second as allowed by the 500/50/5 rule.

*

* @see [Ramping up traffic](https://cloud.google.com/firestore/docs/best-practices#ramping_up_traffic)

*/

'DEFAULT_MAXIMUM_OPS_PER_SECOND_LIMIT' => 500,

/**

* The rate by which to increase the capacity as specified by the 500/50/5 rule.

*

* @see [Ramping up traffic](https://cloud.google.com/firestore/docs/best-practices#ramping_up_traffic)

*/

'RATE_LIMITER_MULTIPLIER' => 1.5,

/**

* How often the operations per second capacity should increase in milliseconds as specified by

* the 500/50/5 rule.

*

* @see [Ramping up traffic](https://cloud.google.com/firestore/docs/best-practices#ramping_up_traffic)

*/

'RATE_LIMITER_MULTIPLIER_MILLIS' => 1000,

];

private FirestoreClient $gapicClient;

private ValueMapper $valueMapper;

private string $database;

private string|null $transaction = null;

private bool $isLegacyWriteBatch;

private int $maxBatchSize;

private array $writes = [];

private array $finalResponse = [];

private RateLimiter $rateLimiter;

private Serializer $serializer;

private OptionsValidator $optionsValidator;

/**

* @var array {

* Failed mutations scheduled for retry. Each retry has following fields:

*

* @type int $num_failed_attempts Number of past failures.

* @type int $scheduled_in_millis Latest timestamp in millis when retry

* can be attempted.

* @type int $backoff_in_millis Backoff time in millis included in

* `scheduled_in_millis`.

* }

*/

private $retryScheduledWrites = [];

/**

* @var callable Sets the conditions for whether or not a write should

* be attempted to retry.

*/

private $isRetryable;

private array $uniqueDocuments = [];

private bool $closed;

/**

* @var bool Whether BulkWriter greedily sends operations via

* [https://cloud.google.com/firestore/docs/reference/rpc/google.firestore.v1beta1#batchwriterequest](BatchWriteRequest)

* as soon as sufficient number of operations are enqueued.

*/

private bool $greedilySend;

private int $maxDelayTime;

/**

* @param FirestoreClient $firestoreClient A FirestoreClient instance.

* @param ValueMapper $valueMapper A Value Mapper instance

* @param string $database The current database

* @param array $options [optional] {

* Configuration options is an array.

*

* Please note that the default values are experiementally derived after

* performance evaluations. The underlying constants may change in backwards-

* incompatible ways. Please use with caution, and test thoroughly when

* upgrading.

*

* For legacy reasons if provided as `string` or `null`, its assumed

* to be transaction id for Google\Cloud\Firestore\WriteBatch.

*

* @type int $maxBatchSize Maximum number of requests per BulkWriter batch.

* **Defaults to** `20`.

* @type bool $greedilySend Flag to indicate whether BulkWriter greedily

* sends batches. **Defaults to** `true`.

* @type bool $isThrottlingEnabled Flag to indicate whether rate of

* sending writes can be throttled. **Defaults to** `true`.

* @type int $initialOpsPerSecond Initial number of operations per second.

* **Defaults to** `20`.

* @type int $maxOpsPerSecond Maximum number of operations per second.

* **Defaults to** `500`.

* @type callable $isRetryable Default retry handler for individial writes

* status code to be retried. Should accept error code and return

* true if retryable.

* }

*/

public function __construct(FirestoreClient $firestoreClient, $valueMapper, $database, $options = null)

{

$this->gapicClient = $firestoreClient;

$this->valueMapper = $valueMapper;

$this->database = $database;

$this->closed = false;

$this->isLegacyWriteBatch = false;

$this->maxDelayTime = self::$defaultOptions['DEFAULT_BACKOFF_MAX_DELAY_MS'];

if (is_null($options) || !is_array($options)) {

// convert to transaction id for legacy WriteBatch

$this->transaction = $options;

$this->isLegacyWriteBatch = true;

$options = [];

}

$this->finalResponse = [

'writeResults' => [],

'status' => [],

];

$options += [

'maxBatchSize' => self::$defaultOptions['MAX_BATCH_SIZE'],

'greedilySend' => true,

'isThrottlingEnabled' => true,

'initialOpsPerSecond' => self::$defaultOptions['DEFAULT_STARTING_MAXIMUM_OPS_PER_SECOND'],

'maxOpsPerSecond' => self::$defaultOptions['DEFAULT_MAXIMUM_OPS_PER_SECOND_LIMIT'],

'isRetryable' => $this->defaultWriteErrorHandler(),

];

$this->isRetryable = $this->pluck('isRetryable', $options);

$this->maxBatchSize = $this->pluck('maxBatchSize', $options);

$this->greedilySend = $this->pluck('greedilySend', $options);

if ($options['initialOpsPerSecond'] < 1) {

throw new \InvalidArgumentException(

sprintf(

'Value for argument "initialOpsPerSecond" must be greater than 1, but was: %s',

$options['initialOpsPerSecond']

)

);

}

if ($options['maxOpsPerSecond'] < 1) {

throw new \InvalidArgumentException(

sprintf(

'Value for argument "maxOpsPerSecond" must be greater than 1, but was: %s',

$options['maxOpsPerSecond']

)

);

}

if ($options['initialOpsPerSecond'] > $options['maxOpsPerSecond']) {

throw new \InvalidArgumentException(

"'maxOpsPerSecond' cannot be less than 'initialOpsPerSecond'."

);

}

if ($options['isThrottlingEnabled'] === false) {

$this->rateLimiter = new RateLimiter(

PHP_INT_MAX,

PHP_INT_MAX,

PHP_INT_MAX,

PHP_INT_MAX

);

} else {

$startingRate = $options['initialOpsPerSecond'];

$maxRate = $options['maxOpsPerSecond'];

// The initial validation step ensures that the maxOpsPerSecond is

// greater than initialOpsPerSecond. If this inequality is true, that

// means initialOpsPerSecond was not set and maxOpsPerSecond is less

// than the default starting rate.

if ($maxRate < $startingRate) {

$startingRate = $maxRate;

}

// Ensure that the batch size is not larger than the number of allowed

// operations per second.

if ($startingRate < $this->maxBatchSize) {

$this->maxBatchSize = $startingRate;

}

$this->rateLimiter = new RateLimiter(

$startingRate,

self::$defaultOptions['RATE_LIMITER_MULTIPLIER'],

self::$defaultOptions['RATE_LIMITER_MULTIPLIER_MILLIS'],

$maxRate

);

}

$this->serializer = new Serializer();

$this->optionsValidator = new OptionsValidator($this->serializer);

}

/**

* Enqueue a document creation.

*

* This operation will fail (when committed) if the document already exists.

*

* Example:

* ```

* $batch->create($documentName, [

* 'name' => 'John'

* ]);

* ```

*

* @param DocumentReference|string $document The document to target, either

* as a string document name, or DocumentReference object. Please

* note that DocumentReferences will be used only for the document

* name. Field data must be provided in the `$fields` argument.

* @param array $fields An array containing fields, where keys are the field

* names, and values are field values. Nested arrays are allowed.

* Note that unlike {@see \Google\Cloud\Firestore\DocumentReference::update()},

* field paths are NOT supported by this method.

* @param array $options Configuration options

* @return BulkWriter

* @throws \InvalidArgumentException If `FieldValue::deleteField()` is found in the fields list.

* @throws \InvalidArgumentException If `FieldValue::serverTimestamp()` is found in an array value.

*/

public function create($document, array $fields, array $options = [])

{

$this->checkWriterConditions($document);

// Record whether the document is empty before any filtering.

$emptyDocument = count($fields) === 0;

list($fields, $sentinels, $metadata) = $this->filterFields($fields);

if ($metadata['hasDelete']) {

throw new \InvalidArgumentException('Cannot delete fields when creating a document.');

}

// Cannot create a document that already exists!

$precondition = ['exists' => false];

// Enqueue an update operation if an empty document was provided,

// or if there are still fields after filtering.

$transformOptions = [];

if (!empty($fields) || $emptyDocument) {

$this->writes[] = $this->createDatabaseWrite(self::TYPE_CREATE, $document, [

'fields' => $this->valueMapper->encodeValues($fields),

'precondition' => $precondition,

] + $options);

} else {

// If no UPDATE mutation is enqueued, we need the precondition applied

// to the transform mutation.

$transformOptions = [

'precondition' => $precondition,

];

}

// document transform operations are enqueued as a separate mutation.

$this->enqueueTransforms($document, $sentinels, $transformOptions);

return $this;

}

/**

* Enqueue a set mutation.

*

* Unless `$options['merge']` is set to `true, this method replaces all

* fields in a Firestore document.

*

* Example:

* ```

* $batch->set($documentName, [

* 'name' => 'John'

* ]);

*

* @codingStandardsIgnoreStart

* @param DocumentReference|string $document The document to target, either

* as a string document name, or DocumentReference object. Please

* note that DocumentReferences will be used only for the document

* name. Field data must be provided in the `$fields` argument.

* @param array $fields An array containing fields, where keys are the field

* names, and values are field values. Nested arrays are allowed.

* Note that unlike {@see \Google\Cloud\Firestore\BulkWriter::update()},

* field paths are NOT supported by this method.

* @param array $options {

* Configuration Options

*

* @type bool $merge If true, unwritten fields will be preserved.

* Otherwise, they will be overwritten (removed). **Defaults to**

* `false`.

* }

* @return BulkWriter

* @codingStandardsIgnoreEnd

* @throws \InvalidArgumentException If `FieldValue::deleteField()` is found in the document when `$options.merge`

* is not `true`.

* @throws \InvalidArgumentException If `FieldValue::serverTimestamp()` is found in an array value.

*/

public function set($document, array $fields, array $options = [])

{

$this->checkWriterConditions($document);

$merge = $this->pluck('merge', $options, false) ?: false;

// Record whether the document is empty before any filtering.

$emptyDocument = count($fields) === 0;

list($fields, $sentinels, $metadata) = $this->filterFields($fields);

if (!$merge && $metadata['hasDelete']) {

throw new \InvalidArgumentException('Delete cannot appear in data unless `$options[\'merge\']` is set.');

}

// Enqueue a write if any of the following conditions are met

// - if there are still fields remaining after sentinels were removed

// - if the user provided an empty set to begin with

// - if the user provided only transform sentinel values AND did not specify merge behavior

// - if the user provided only delete sentinel field values.

$updateNotRequired = count($fields) === 0

&& !$emptyDocument

&& !$metadata['hasUpdateMask']

&& $metadata['hasTransform'];

$shouldEnqueueUpdate = $fields

|| $emptyDocument

|| ($updateNotRequired && !$merge)

|| $metadata['hasUpdateMask'];

if ($shouldEnqueueUpdate) {

$write = [];

$write['fields'] = $this->valueMapper->encodeValues($fields);

if ($merge) {

$write['updateMask'] = $this->pathsToStrings($this->encodeFieldPaths($fields), $sentinels);

}

$this->writes[] = $this->createDatabaseWrite(self::TYPE_SET, $document, array_merge($options, $write));

}

// document transform operations are enqueued as a separate mutation.

$this->enqueueTransforms($document, $sentinels, $options);

return $this;

}

/**

* Enqueue an update with field paths and values.

*

* Merges provided data with data stored in Firestore.

*

* Calling this method on a non-existent document will raise an exception.

*

* This method supports various sentinel values, to perform special operations

* on fields. Available sentinel values are provided as methods, found in

* {@see \Google\Cloud\Firestore\FieldValue}.

*

* Note that field names must be provided using field paths, encoded either

* as a dot-delimited string (i.e. `foo.bar`), or an instance of

* {@see \Google\Cloud\Firestore\FieldPath}. Nested arrays are not allowed.

*

* Please note that conflicting paths will result in an exception. Paths

* conflict when one path indicates a location nested within another path.

* For instance, path `a.b` cannot be set directly if path `a` is also

* provided.

*

* Example:

* ```

* $batch->update($documentName, [

* ['path' => 'name', 'value' => 'John'],

* ['path' => 'country', 'value' => 'USA'],

* ['path' => 'cryptoCurrencies.bitcoin', 'value' => 0.5],

* ['path' => 'cryptoCurrencies.ethereum', 'value' => 10],

* ['path' => 'cryptoCurrencies.litecoin', 'value' => 5.51]

* ]);

* ```

*

* ```

* // Google Cloud PHP provides special field values to enable operations such

* // as deleting fields or setting the value to the current server timestamp.

* use Google\Cloud\Firestore\FieldValue;

*

* $batch->update($documentName, [

* ['path' => 'country', 'value' => FieldValue::deleteField()],

* ['path' => 'lastLogin', 'value' => FieldValue::serverTimestamp()]

* ]);

* ```

*

* ```

* // If your field names contain special characters (such as `.`, or symbols),

* // using {@see \Google\Cloud\Firestore\FieldPath} will properly escape each element.

*

* use Google\Cloud\Firestore\FieldPath;

*

* $batch->update($documentName, [

* ['path' => new FieldPath(['cryptoCurrencies', 'big$$$coin']), 'value' => 5.51]

* ]);

* ```

*

* @param DocumentReference|string $document The document to target, either

* as a string document name, or DocumentReference object. Please

* note that DocumentReferences will be used only for the document

* name. Field data must be provided in the `$data` argument.

* @param array[] $data A list of arrays of form

* `[FieldPath|string $path, mixed $value]`.

* @param array $options Configuration options

* @return BulkWriter

* @throws \InvalidArgumentException If data is given in an invalid format

* or is empty.

* @throws \InvalidArgumentException If any field paths are empty.

* @throws \InvalidArgumentException If field paths conflict.

*/

public function update($document, array $data, array $options = [])

{

$this->checkWriterConditions($document);

if (!$data || $this->isAssoc($data)) {

throw new \InvalidArgumentException(

'Field data must be provided as a list of arrays of form `[string|FieldPath $path, mixed $value]`.'

);

}

$paths = [];

$fields = [];

foreach ($data as $field) {

$this->arrayHasKeys($field, ['path', 'value']);

$path = ($field['path'] instanceof FieldPath)

? $field['path']

: FieldPath::fromString($field['path']);

if (!$path->path()) {

throw new \InvalidArgumentException('Field Path cannot be empty.');

}

$paths[] = $path;

$keys = $path->path();

$num = count($keys);

// Construct a nested array to represent a nested field path.

// For instance, `a.b.c` = 'foo' will become

// `['a' => ['b' => ['c' => 'foo']]]`

$val = $field['value'];

foreach (array_reverse($keys) as $index => $key) {

if ($num >= $index + 1) {

$val = [

$key => $val,

];

}

}

$fields = $this->arrayMergeRecursive($fields, $val);

}

if (count(array_unique($paths)) !== count($paths)) {

throw new \InvalidArgumentException('Duplicate field paths are not allowed.');

}

// Record whether the document is empty before any filtering.

$emptyDocument = count($fields) === 0;

list($fields, $sentinels, $metadata) = $this->filterFields($fields, $paths);

// to conform to specification.

if (isset($options['precondition']['exists'])) {

throw new \InvalidArgumentException('Exists preconditions are not supported by this method.');

}

// We only want to enqueue an update write if there are non-sentinel fields

// OR no transform sentinels are found.

// We MUST always enqueue at least one write, so if there are no fields

// and no transform sentinels, we can assume an empty write is intended

// and enqueue an empty UPDATE operation.

$shouldEnqueueUpdate = $fields

|| $emptyDocument

|| $metadata['hasUpdateMask'];

if ($shouldEnqueueUpdate) {

$write = [

'fields' => $this->valueMapper->encodeValues($fields),

'updateMask' => $this->pathsToStrings($paths, $sentinels, true),

];

$this->writes[] = $this->createDatabaseWrite(

self::TYPE_UPDATE,

$document,

$write + $this->formatPrecondition($options, true)

);

} else {

// If no update write is enqueued, preconditions must be applied to

// a transform.

$options = $this->formatPrecondition($options, true);

}

// document transform operations are enqueued as a separate mutation.

$this->enqueueTransforms($document, $sentinels, $options);

return $this;

}

/**

* Delete a Firestore document.

*

* Example:

* ```

* $batch->delete($documentName);

* ```

*

* @codingStandardsIgnoreStart

* @param DocumentReference|string $document The document to target, either

* as a string document name, or DocumentReference object.

* @param array $options Configuration Options

* @return BulkWriter

* @codingStandardsIgnoreEnd

*/

public function delete($document, array $options = [])

{

$this->checkWriterConditions($document);

$options = $this->formatPrecondition($options);

$this->writes[] = $this->createDatabaseWrite(self::TYPE_DELETE, $document, $options);

return $this;

}

/**

* Flushes the enqueued writes in batches with auto-retries. Please note:

* - This method is blocking and may execute many sequential batch write requests.

* - Gradually ramps up writes as specified by the 500/50/5 rule.

* - Does not guarantee the order of writes.

* - Accepts unique document references only.

* Read more: [Ramping up traffic](https://cloud.google.com/firestore/docs/best-practices#ramping_up_traffic)

*

* Example:

* ```

* $batch->flush();

* ```

*

* @param bool $waitForRetryableFailures Flag to indicate whether to wait for

* retryable failures. **Defaults to** `false`.

* @return array [https://firebase.google.com/docs/firestore/reference/rpc/google.firestore.v1beta1#BatchWriteResponse](BatchWriteResponse)

*/

public function flush($waitForRetryableFailures = false)

{

while (true) {

$batchIds = $this->createWritesBatchIds($waitForRetryableFailures);

$writesBatch = [];

foreach ($batchIds as $batchId) {

$writesBatch[] = $this->writes[$batchId];

}

$batchSize = count($writesBatch);

if ($batchSize <= 0) {

// no more writes to process

break;

}

$response = $this->sendBatch($writesBatch);

for ($i = 0; $i < $batchSize; $i++) {

$writeResult = $response['writeResults'][$i];

$status = $response['status'][$i];

$this->finalResponse['writeResults'][$batchIds[$i]] = $writeResult;

$this->finalResponse['status'][$batchIds[$i]] = $status;

if ($status['code'] !== Code::OK) {

$this->handleSendBatchFailure($batchIds[$i], $status['code']);

} else {

// try delete from retries

unset($this->retryScheduledWrites[$batchIds[$i]]);

}

}

}

ksort($this->finalResponse['writeResults']);

ksort($this->finalResponse['status']);

return $this->finalResponse;

}

/**

* Commit writes to the database.

* Not to be used together with flush / close.

*

* Example:

* ```

* $batch->commit();

* ```

*

* @codingStandardsIgnoreStart

* @see https://firebase.google.com/docs/firestore/reference/rpc/google.firestore.v1beta1#google.firestore.v1beta1.Firestore.Commit Commit

*

* @internal Only supposed to be used internally in Transaction class.

* @access private

* @param array $options Configuration Options

* @return array [https://firebase.google.com/docs/firestore/reference/rpc/google.firestore.v1beta1#commitresponse](CommitResponse)

* @codingStandardsIgnoreEnd

* @throws ServiceException

*/

public function commit(array $options = []): array

{

unset($options['merge'], $options['precondition']);

/**

* @var CommitRequest $request,

* @var CallOptions $callOptions

*/

[$request, $callOptions] = $this->validateOptions(

[

'database' => $this->database,

'writes' => $this->writes,

'transaction' => $this->transaction

] + $options,

new CommitRequest(),

CallOptions::class

);

try {

$response = $this->gapicClient->commit($request, $callOptions);

} catch (ApiException $ex) {

throw $this->convertToGoogleException($ex);

}

$responseArray = $this->serializer->encodeMessage($response);

if (!is_null($response->getCommitTime())) {

$time = $this->parseTimeString($responseArray['commitTime']);

$responseArray['commitTime'] = new Timestamp($time[0], $time[1]);

}

if (!is_null($response->getWriteResults())) {

foreach ($responseArray['writeResults'] as &$result) {

if (isset($result['updateTime'])) {

$time = $this->parseTimeString($result['updateTime']);

$result['updateTime'] = new Timestamp($time[0], $time[1]);

}

}

}

return $responseArray;

}

/**

* Rollback a transaction.

*

* If the class was created without a Transaction ID, this method will fail.

* Not to be used together with flush / close.

*

* This method is intended for use internally and should not be considered

* part of the public API.

*

* @internal Only supposed to be used internally in Transaction class.

* @access private

* @param array $options Configuration Options

* @return void

* @throws \RuntimeException If no transaction ID is provided at class construction.

* @throws ServiceException

*/

public function rollback(array $options = []): void

{

if (!$this->transaction) {

throw new \RuntimeException('Cannot rollback because no transaction id was provided.');

}

/**

* @var RollbackRequest $request

* @var CallOptions $callOptions

*/

[$request, $callOptions] = $this->validateOptions(

[

'database' => $this->database,

'transaction' => $this->transaction

] + $options,

new RollbackRequest(),

CallOptions::class

);

try {

$this->gapicClient->rollback($request, $callOptions);

} catch (ApiException $ex) {

throw $this->convertToGoogleException($ex);

}

}

/**

* Check if the BulkWriter has any writes enqueued.

*

* @access private

* @return bool

*/

public function isEmpty()

{

return !(bool) $this->writes;

}

/**

* Close the bulk writer instance for further writes.

* Also, flushes all retries and pending writes.

*

* @return array [https://firebase.google.com/docs/firestore/reference/rpc/google.firestore.v1beta1#BatchWriteResponse](BatchWriteResponse)

*/

public function close()

{

$this->closed = true;

return $this->flush(true);

}

/**

* Gets updated backoff duration provided last status code and backoff duration.

*

* @internal

* @access private

* @param int $lastStatus Previous status code of batchWrite

* @param int $backoffDurationInMillis Previous backoff duration in milliseconds

* @return int

*/

public function getBackoffDuration($lastStatus, $backoffDurationInMillis = 0)

{

if ($lastStatus === Code::RESOURCE_EXHAUSTED) {

$backoffDurationInMillis = $this->maxDelayTime;

} elseif ($backoffDurationInMillis <= 0) {

$backoffDurationInMillis = self::$defaultOptions['DEFAULT_BACKOFF_INITIAL_DELAY_MS'];

} else {

$backoffDurationInMillis *= self::$defaultOptions['DEFAULT_BACKOFF_FACTOR'];

}

return min($this->maxDelayTime, $backoffDurationInMillis);

}

/**

* Change the maximum delay time for rescheduling a failed mutation or

* awaiting a batch creation.

*

* @internal

* @access private

* @param int $maxTime The maximum delay time in millis for rescheduling

* a failed mutation or awaiting a batch creation.

* @return void

*/

public function setMaxRetryTimeInMs($maxTime)

{

$this->maxDelayTime = max(0, $maxTime);

}

/**

* Reschedule failed mutations if retryable.

*

* @param int $writesId Sequence of mutation among all enqueued writes

* @param int $lastRunStatusCode Previous status code of batchWrite

* @return void

*/

private function handleSendBatchFailure($writesId, $lastRunStatusCode)

{

if (!call_user_func_array($this->isRetryable, [$lastRunStatusCode])) {

return;

}

$numFailedAttempts = 1;

$backoffDurationInMillis = 0;

if (array_key_exists($writesId, $this->retryScheduledWrites)) {

$numFailedAttempts += $this->retryScheduledWrites[$writesId]['num_failed_attempts'];

$backoffDurationInMillis = $this->retryScheduledWrites[$writesId]['backoff_in_millis'];

}

// calculate the new backoff time

$backoffDurationInMillis = $this->getBackoffDuration($lastRunStatusCode, $backoffDurationInMillis);

$rescheduleTime = floor(microtime(true) * 1000) + $backoffDurationInMillis;

// upsert to reschedule array

$this->retryScheduledWrites[$writesId] = [

'scheduled_in_millis' => $rescheduleTime,

'backoff_in_millis' => $backoffDurationInMillis,

'num_failed_attempts' => $numFailedAttempts,

];

}

/**

* Get whether individual writes with provided status code shall be retried.

* The default error handler retries for UNAVAILABLE and ABORTED errors.

*

* @return callable Accepts error code and returns true if retryable.

*/

private function defaultWriteErrorHandler()

{

return function ($lastRunStatusCode) {

return in_array($lastRunStatusCode, [

Code::UNAVAILABLE,

Code::ABORTED,

]);

};

}

/**

* Creates writes array indices which form a batch

*

* @param bool $waitForRetryableFailures Flag to indicate whether to wait for

* retryable failures. **Defaults to** `false`.

* @return array

*/

private function createWritesBatchIds($waitForRetryableFailures = false)

{

$writesBatchIds = [];

$curTimeInMillis = floor(microtime(true) * 1000);

$maxScheduledDelayInMillis = 0;

$batchSize = $this->maxBatchSize;

// check for scheduling fresh writes

foreach (array_keys($this->writes) as $writeId) {

if (count($writesBatchIds) >= $batchSize) {

break;

}

// ignore if write has a result and result is already successful

if (array_key_exists($writeId, $this->finalResponse['status']) &&

$this->finalResponse['status'][$writeId]['code'] === Code::OK) {

continue;

}

// ignore if write is yet to reach rescheduled time OR retried more than enough

if (array_key_exists($writeId, $this->retryScheduledWrites)) {

if ($this->retryScheduledWrites[$writeId]['num_failed_attempts'] >=

self::$defaultOptions['MAX_RETRY_ATTEMPTS']

) {

// let the failures eventually trickle down to finalResponse

continue;

}

if (!$waitForRetryableFailures &&

$this->retryScheduledWrites[$writeId]['scheduled_in_millis'] > $curTimeInMillis) {

// cannot wait for writes that are yet to be scheduled

continue;

}

// Delay greater than 0 implies that this batch is a retry.

// Retries are sent with RETRY_MAX_BATCH_SIZE in order to guarantee

// that the batch is under the 10MiB limit.

$batchSize = self::$defaultOptions['RETRY_MAX_BATCH_SIZE'];

$maxScheduledDelayInMillis = max(

$maxScheduledDelayInMillis,

$this->retryScheduledWrites[$writeId]['scheduled_in_millis']

);

}

$writesBatchIds[] = $writeId;

}

// sleep needs delay, not the actual timestamp

$maxScheduledDelayInMillis -= $curTimeInMillis;

if ($maxScheduledDelayInMillis > 0) {

$actualDelay = $this->applyJitter($maxScheduledDelayInMillis);

// sleep for max possible delay to form batches for all writes

usleep($actualDelay * 1000);

}

return $writesBatchIds;

}

/**

* Sends a batch write request to the database.

*

* @param array $writes The writes to send in a batch. Please note:

* - Writes do not apply atomically

* - Writes do not guarantee ordering.

* - Each write can succeed or fail independently.

* @param array $options [optional] {

* Configuration options

*

* @type array $labels

* Labels associated with this batch write.

* }

* @return array [https://firebase.google.com/docs/firestore/reference/rpc/google.firestore.v1beta1#BatchWriteResponse](BatchWriteResponse)

* @throws ServiceException

*/

private function sendBatch(array $writes, array $options = [])

{

$rateLimiterDelayMs = $this->rateLimiter->getNextRequestDelayMs(count($writes));

// avoid very long sleep

$rateLimiterDelayMs = min(

$rateLimiterDelayMs,

$this->maxDelayTime

);

if ($rateLimiterDelayMs > 0) {

usleep($rateLimiterDelayMs * 1000);

}

$this->rateLimiter->tryMakeRequest(count($writes));

unset($options['merge'], $options['precondition']);

$options += ['labels' => []];

$options += [

'database' => $this->database,

'writes' => $writes

];

/**

* @var BatchWriteRequest $request

* @var CallOptions $callOptions

*/

[$request, $callOptions] = $this->validateOptions(

$options,

new BatchWriteRequest(),

CallOptions::class,

);

try {

$response = $this->gapicClient->batchWrite($request, $callOptions);

} catch (ApiException $ex) {

throw $this->convertToGoogleException($ex);

}

$responseArray = $this->serializer->encodeMessage($response);

if (!empty($response->getWriteResults())) {

foreach ($responseArray['writeResults'] as &$result) {

if (isset($result['updateTime'])) {

$time = $this->parseTimeString($result['updateTime']);

$result['updateTime'] = new Timestamp($time[0], $time[1]);

}

}

}

return $responseArray;

}

/**

* Enqueue transforms for CREATE, UPDATE, and SET calls.

*

* @param DocumentReference|string $document The document to target, either

* as a string document name, or DocumentReference object.

* @param DocumentTransformInterface[] $transforms

* @param array $options

* @return void

*/

private function enqueueTransforms($document, array $transforms, array $options = [])

{

$operations = [];

foreach ($transforms as $transform) {

if (!($transform instanceof DocumentTransformInterface)) {