Appwrite/appwrite
1
0
Fork 0
mirror of https://github.com/appwrite/appwrite.git synced 2026-05-26 13:51:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update DB

Browse Source
This commit is contained in:
Jake Barnby
2025-09-05 01:02:30 +12:00
parent 7eba6582c6
commit d28643bd2b
8 changed files with 433 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Appwrite/Platform/Modules/Databases/Http/Databases/Collections/Documents/Bulk/Upsert.php
+1 -1
View File
@@ -161,7 +161,7 @@ class Upsert extends Action
try {
$modified = $dbForProject->withPreserveDates(function () use ($dbForProject, $database, $collection, $documents, $plan, &$upserted) {
return $dbForProject->createOrUpdateDocuments(
return $dbForProject->upsertDocuments(
'database_' . $database->getSequence() . '_collection_' . $collection->getSequence(),
$documents,
onNext: function (Document $document) use ($plan, &$upserted) {
src/Appwrite/Platform/Modules/Databases/Http/Databases/Collections/Documents/Upsert.php
+1 -1
View File
@@ -299,7 +299,7 @@ class Upsert extends Action
$upserted = [];
try {
$dbForProject->withPreserveDates(function () use (&$upserted, $dbForProject, $database, $collection, $newDocument) {
return $dbForProject->createOrUpdateDocuments(
return $dbForProject->upsertDocuments(
'database_' . $database->getSequence() . '_collection_' . $collection->getSequence(),
[$newDocument],
onNext: function (Document $document) use (&$upserted) {
src/Appwrite/Platform/Modules/Databases/Http/Transactions/Update.php
+390 -149
View File
@@ -12,9 +12,11 @@ use Appwrite\SDK\Response as SDKResponse;
use Appwrite\Utopia\Response as UtopiaResponse;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Exception\Authorization;
use Utopia\Database\Exception\Conflict as ConflictException;
use Utopia\Database\Exception\Duplicate as DuplicateException;
use Utopia\Database\Exception\NotFound as NotFoundException;
use Utopia\Database\Exception\Structure;
use Utopia\Database\Exception\Transaction as TransactionException;
use Utopia\Database\Query;
use Utopia\Database\Validator\UID;
@@ -65,6 +67,23 @@ class Update extends Action
->callback($this->action(...));
}
/**
* @param string $transactionId
* @param bool $commit
* @param bool $rollback
* @param UtopiaResponse $response
* @param Database $dbForProject
* @param Delete $queueForDeletes
* @return void
* @throws ConflictException
* @throws Exception
* @throws \DateMalformedStringException
* @throws \Throwable
* @throws \Utopia\Database\Exception
* @throws Authorization
* @throws Structure
* @throws \Utopia\Exception
*/
public function action(string $transactionId, bool $commit, bool $rollback, UtopiaResponse $response, Database $dbForProject, Delete $queueForDeletes): void
{
if (!$commit && !$rollback) {
@@ -113,149 +132,38 @@ class Update extends Action
$action = $operation['action'];
$data = $operation['data'];
// Check if this operation depends on documents created in same transaction
$dependent = \in_array($action, ['update', 'increment', 'decrement'])
&& isset($state[$collectionId][$documentId]);
if ($dependent) {
// Don't use timestamp wrapper for dependent operations
switch ($action) {
case 'update':
// Update the state document directly
$existing = $state[$collectionId][$documentId];
foreach ($data as $key => $value) {
$existing->setAttribute($key, $value);
}
$state[$collectionId][$documentId] = $dbForProject->updateDocument(
$collectionId,
$documentId,
$existing
);
break;
case 'increment':
$state[$collectionId][$documentId] = $dbForProject->increaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
max: $data['max'] ?? null
);
break;
case 'decrement':
$state[$collectionId][$documentId] = $dbForProject->decreaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
min: $data['min'] ?? null
);
break;
}
} else {
// Use timestamp wrapper for independent operations
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $queueForDeletes, $action, $collectionId, $documentId, $data, &$state) {
switch ($action) {
case 'create':
if ($documentId && !isset($data['$id'])) {
$data['$id'] = $documentId;
}
$state[$collectionId][$documentId] = $dbForProject->createDocument(
$collectionId,
new Document($data),
);
break;
case 'update':
$document = $dbForProject->updateDocument(
$collectionId,
$documentId,
new Document($data),
);
if ($document->isEmpty()) {
throw new ConflictException('');
}
$state[$collectionId][$documentId] = $document;
break;
case 'upsert':
$document = new Document($data);
$dbForProject->createOrUpdateDocuments(
$collectionId,
[$document],
onNext: function (Document $document) use (&$state, $collectionId) {
$state[$collectionId][$document->getId()] = $document;
}
);
break;
case 'delete':
$dbForProject->deleteDocument($collectionId, $documentId);
if (isset($state[$collectionId][$documentId])) {
unset($state[$collectionId][$documentId]);
}
break;
case 'increment':
$dbForProject->increaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
max: $data['max'] ?? null
);
break;
case 'decrement':
$dbForProject->decreaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
min: $data['min'] ?? null
);
break;
case 'bulkCreate':
$dbForProject->createDocuments(
$collectionId,
$data,
onNext: function (Document $document) use (&$state, $collectionId) {
$state[$collectionId][$document->getId()] = $document;
}
);
break;
case 'bulkUpdate':
$dbForProject->updateDocuments(
$collectionId,
new Document($data['data']),
Query::parseQueries($data['queries'] ?? [])
);
break;
case 'bulkUpsert':
$dbForProject->createOrUpdateDocuments(
$collectionId,
$data,
onNext: function (Document $document) use (&$state, $collectionId) {
$state[$collectionId][$document->getId()] = $document;
}
);
break;
case 'bulkDelete':
$dbForProject->deleteDocuments(
$collectionId,
Query::parseQueries($data['queries'] ?? [])
);
break;
}
});
// Execute the operation based on its type
switch ($action) {
case 'create':
$this->handleCreateOperation($dbForProject, $collectionId, $documentId, $data, $createdAt, $state);
break;
case 'update':
$this->handleUpdateOperation($dbForProject, $collectionId, $documentId, $data, $createdAt, $state);
break;
case 'upsert':
$this->handleUpsertOperation($dbForProject, $collectionId, $documentId, $data, $createdAt, $state);
break;
case 'delete':
$this->handleDeleteOperation($dbForProject, $collectionId, $documentId, $createdAt, $state);
break;
case 'increment':
$this->handleIncrementOperation($dbForProject, $collectionId, $documentId, $data, $createdAt, $state);
break;
case 'decrement':
$this->handleDecrementOperation($dbForProject, $collectionId, $documentId, $data, $createdAt, $state);
break;
case 'bulkCreate':
$this->handleBulkCreateOperation($dbForProject, $collectionId, $data, $createdAt, $state);
break;
case 'bulkUpdate':
$this->handleBulkUpdateOperation($dbForProject, $collectionId, $data, $createdAt, $state);
break;
case 'bulkUpsert':
$this->handleBulkUpsertOperation($dbForProject, $collectionId, $data, $createdAt, $state);
break;
case 'bulkDelete':
$this->handleBulkDeleteOperation($dbForProject, $collectionId, $data, $createdAt, $state);
break;
}
}
@@ -269,7 +177,6 @@ class Update extends Action
$queueForDeletes
->setType(DELETE_TYPE_DOCUMENT)
->setDocument($transaction);
} catch (NotFoundException $e) {
$dbForProject->updateDocument('transactions', $transactionId, new Document([
'status' => 'failed',
@@ -290,17 +197,351 @@ class Update extends Action
}
if ($rollback) {
$queueForDeletes
->setType(DELETE_TYPE_DOCUMENT)
->setDocument($transaction);
$transaction = $dbForProject->updateDocument('transactions', $transactionId, new Document([
'status' => 'rolledBack',
]));
$queueForDeletes
->setType(DELETE_TYPE_DOCUMENT)
->setDocument($transaction);
}
$response
->setStatusCode(SwooleResponse::STATUS_CODE_OK)
->dynamic($transaction, UtopiaResponse::MODEL_TRANSACTION);
}
}
/**
* Handle create operation
* @throws \Utopia\Database\Exception
*/
private function handleCreateOperation(
Database $dbForProject,
string $collectionId,
?string $documentId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
if ($documentId && !isset($data['$id'])) {
$data['$id'] = $documentId;
}
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $collectionId, $data, &$state) {
$state[$collectionId][$data['$id']] = $dbForProject->createDocument(
$collectionId,
new Document($data),
);
});
}
/**
* Handle update operation
* @throws ConflictException
* @throws \Utopia\Database\Exception
*/
private function handleUpdateOperation(
Database $dbForProject,
string $collectionId,
string $documentId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
$dependent = isset($state[$collectionId][$documentId]);
if ($dependent) {
// Update the state document directly without timestamp wrapper
$state[$collectionId][$documentId] = $dbForProject->updateDocument(
$collectionId,
$documentId,
new Document($data),
);
return;
}
// Use timestamp wrapper for independent operations
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $collectionId, $documentId, $data, &$state) {
$document = $dbForProject->updateDocument(
$collectionId,
$documentId,
new Document($data),
);
if ($document->isEmpty()) {
throw new ConflictException('');
}
$state[$collectionId][$documentId] = $document;
});
}
/**
* Handle upsert operation
* @throws \Utopia\Database\Exception
*/
private function handleUpsertOperation(
Database $dbForProject,
string $collectionId,
?string $documentId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
$dependent = isset($state[$collectionId][$documentId]);
if ($dependent) {
// Upsert the state document directly without timestamp wrapper
$state[$collectionId][$documentId] = $dbForProject->upsertDocument(
$collectionId,
new Document($data),
);
return;
}
// Use timestamp wrapper for independent operations
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $collectionId, $documentId, $data, &$state) {
$state[$collectionId][$documentId] = $dbForProject->upsertDocument(
$collectionId,
new Document($data),
);
});
}
/**
* Handle delete operation
*/
private function handleDeleteOperation(
Database $dbForProject,
string $collectionId,
string $documentId,
\DateTime $createdAt,
array &$state
): void
{
$dependent = isset($state[$collectionId][$documentId]);
if ($dependent) {
// Delete without timestamp wrapper
$dbForProject->deleteDocument($collectionId, $documentId);
unset($state[$collectionId][$documentId]);
return;
}
// Use timestamp wrapper for independent operations
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $collectionId, $documentId, &$state) {
$dbForProject->deleteDocument($collectionId, $documentId);
if (isset($state[$collectionId][$documentId])) {
unset($state[$collectionId][$documentId]);
}
});
}
/**
* Handle increment operation
*/
private function handleIncrementOperation(
Database $dbForProject,
string $collectionId,
string $documentId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
$dependent = isset($state[$collectionId][$documentId]);
if ($dependent) {
// Increment without timestamp wrapper
$state[$collectionId][$documentId] = $dbForProject->increaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
max: $data['max'] ?? null
);
return;
}
// Use timestamp wrapper for independent operations
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $collectionId, $documentId, $data) {
$dbForProject->increaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
max: $data['max'] ?? null
);
});
}
/**
* Handle decrement operation
*/
private function handleDecrementOperation(
Database $dbForProject,
string $collectionId,
string $documentId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
$dependent = isset($state[$collectionId][$documentId]);
if ($dependent) {
// Decrement without timestamp wrapper
$state[$collectionId][$documentId] = $dbForProject->decreaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
min: $data['min'] ?? null
);
return;
}
// Use timestamp wrapper for independent operations
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $collectionId, $documentId, $data) {
$dbForProject->decreaseDocumentAttribute(
collection: $collectionId,
id: $documentId,
attribute: $data['attribute'],
value: $data['value'] ?? 1,
min: $data['min'] ?? null
);
});
}
/**
* Handle bulk create operation
*/
private function handleBulkCreateOperation(
Database $dbForProject,
string $collectionId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
$dbForProject->withRequestTimestamp($createdAt, function () use ($dbForProject, $collectionId, $data, &$state) {
$dbForProject->createDocuments(
$collectionId,
$data,
onNext: function (Document $document) use (&$state, $collectionId) {
$state[$collectionId][$document->getId()] = $document;
}
);
});
}
/**
* Handle bulk update operation with manual timestamp checking
* @throws \Utopia\Database\Exception
* @throws \Utopia\Database\Exception\Query
* @throws ConflictException
*/
private function handleBulkUpdateOperation(
Database $dbForProject,
string $collectionId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
$queries = Query::parseQueries($data['queries'] ?? []);
$dbForProject->updateDocuments(
$collectionId,
new Document($data['data']),
$queries,
onNext: function (Document $updated, Document $old) use (&$state, $collectionId, $createdAt) {
// Check if this document was created/modified in this transaction
$dependent = isset($state[$collectionId][$updated->getId()]);
// If not in transaction state, check for timestamp conflicts
if (!$dependent) {
$oldUpdatedAt = new \DateTime($old->getUpdatedAt());
if ($oldUpdatedAt > $createdAt) {
throw new ConflictException('Document was updated after the request timestamp');
}
}
$state[$collectionId][$updated->getId()] = $updated;
}
);
}
/**
* Handle bulk upsert operation with manual timestamp checking
* @throws ConflictException
*/
private function handleBulkUpsertOperation(
Database $dbForProject,
string $collectionId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
// Run bulk upsert without timestamp wrapper, checking manually in callback
$dbForProject->upsertDocuments(
$collectionId,
$data,
onNext: function (Document $upserted, ?Document $old) use (&$state, $collectionId, $createdAt) {
if ($old !== null) {
// This is an update - check if document was created/modified in this transaction
$dependent = isset($state[$collectionId][$upserted->getId()]);
// If not in transaction state, check for timestamp conflicts
if (!$dependent) {
$oldUpdatedAt = new \DateTime($old->getUpdatedAt());
if ($oldUpdatedAt > $createdAt) {
throw new ConflictException('Document was updated after the request timestamp');
}
}
}
// If $old is null, this is a create operation - no timestamp check needed
$state[$collectionId][$upserted->getId()] = $upserted;
}
);
}
/**
* Handle bulk delete operation with manual timestamp checking
* @throws \Utopia\Database\Exception\Query
* @throws ConflictException
*/
private function handleBulkDeleteOperation(
Database $dbForProject,
string $collectionId,
array $data,
\DateTime $createdAt,
array &$state
): void
{
$queries = Query::parseQueries($data['queries'] ?? []);
$dbForProject->deleteDocuments(
$collectionId,
$queries,
onNext: function (Document $deleted, Document $old) use (&$state, $collectionId, $createdAt) {
$dependent = isset($state[$collectionId][$deleted->getId()]);
// If not in transaction state, check for timestamp conflicts
if (!$dependent) {
$oldUpdatedAt = new \DateTime($old->getUpdatedAt());
if ($oldUpdatedAt > $createdAt) {
throw new ConflictException('Document was updated after the transaction operation');
}
}
// Remove from state after successful deletion
if (isset($state[$collectionId][$deleted->getId()])) {
unset($state[$collectionId][$deleted->getId()]);
}
}
);
}
}
src/Appwrite/Platform/Workers/StatsResources.php
+1 -1
View File
@@ -432,7 +432,7 @@ class StatsResources extends Action
protected function writeDocuments(Database $dbForLogs, Document $project): void
{
$dbForLogs->createOrUpdateDocuments(
$dbForLogs->upsertDocuments(
'stats',
$this->documents
);
src/Appwrite/Platform/Workers/StatsUsage.php
+2 -2
View File
@@ -424,7 +424,7 @@ class StatsUsage extends Action
try {
$dbForProject = $getProjectDB($projectStats['project']);
Console::log('Processing batch with ' . count($projectStats['stats']) . ' stats');
$dbForProject->createOrUpdateDocumentsWithIncrease('stats', 'value', $projectStats['stats']);
$dbForProject->upsertDocumentsWithIncrease('stats', 'value', $projectStats['stats']);
Console::success('Batch successfully written to DB');
unset($this->projects[$sequence]);
@@ -468,7 +468,7 @@ class StatsUsage extends Action
try {
Console::log('Processing batch with ' . count($this->statDocuments) . ' stats');
$dbForLogs->createOrUpdateDocumentsWithIncrease(
$dbForLogs->upsertDocumentsWithIncrease(
'stats',
'value',
$this->statDocuments