mirror of
https://github.com/appwrite/appwrite.git
synced 2026-05-26 13:51:13 +00:00
feat: add afterQuery hook to list-documents/rows action
Wrap each database call (find, count, transaction list/count) with a measuring closure so the actual DB duration is known — cache hits report near-zero, cache misses report only the DB time, not cache save / response serialization. After the response is sent, invoke a protected afterQuery() hook with the measured duration, the database/collection documents, and both parsed + raw query arrays. CE impl is a no-op; downstreams (e.g., cloud) can override it to log slow queries without relying on HTTP shutdown hooks or route-path matching. Exceptions from afterQuery are swallowed so observability never breaks the response.
This commit is contained in:
Damodar Lohani
@@ -2,15 +2,27 @@
|
||||
|
||||
namespace Appwrite\Platform\Modules\Databases\Http\TablesDB\Tables\Rows;
|
||||
|
||||
use Appwrite\Databases\TransactionState;
|
||||
use Appwrite\Extend\Exception;
|
||||
use Appwrite\Platform\Modules\Databases\Http\Databases\Collections\Documents\XList as DocumentXList;
|
||||
use Appwrite\SDK\AuthType;
|
||||
use Appwrite\SDK\ContentType;
|
||||
use Appwrite\SDK\Method;
|
||||
use Appwrite\SDK\Response as SDKResponse;
|
||||
use Appwrite\Usage\Context;
|
||||
use Appwrite\Utopia\Database\Documents\User;
|
||||
use Appwrite\Utopia\Response as UtopiaResponse;
|
||||
use Utopia\Database\Database;
|
||||
use Utopia\Database\Document;
|
||||
use Utopia\Database\Exception\Order as OrderException;
|
||||
use Utopia\Database\Exception\Query as QueryException;
|
||||
use Utopia\Database\Exception\Timeout;
|
||||
use Utopia\Database\Query;
|
||||
use Utopia\Database\Validator\Authorization;
|
||||
use Utopia\Database\Validator\Query\Cursor;
|
||||
use Utopia\Database\Validator\UID;
|
||||
use Utopia\Http\Adapter\Swoole\Response as SwooleResponse;
|
||||
use Utopia\Http\Http;
|
||||
use Utopia\Validator\ArrayList;
|
||||
use Utopia\Validator\Boolean;
|
||||
use Utopia\Validator\Nullable;
|
||||
@@ -65,6 +77,200 @@ class XList extends DocumentXList
|
||||
->inject('usage')
|
||||
->inject('transactionState')
|
||||
->inject('authorization')
|
||||
->inject('utopia')
|
||||
->callback($this->action(...));
|
||||
}
|
||||
|
||||
/**
|
||||
* List rows with actual database duration measurement and a post-query
|
||||
* observability hook. Mirrors the parent listDocuments action body but
|
||||
* wraps each DB call with a timer so subclasses can observe just the DB
|
||||
* portion of the request via afterQuery().
|
||||
*
|
||||
* @param array<string> $queries
|
||||
*/
|
||||
public function action(string $databaseId, string $collectionId, array $queries, ?string $transactionId, bool $includeTotal, int $ttl, UtopiaResponse $response, Database $dbForProject, User $user, callable $getDatabasesDB, Context $usage, TransactionState $transactionState, Authorization $authorization, ?Http $utopia = null): void
|
||||
{
|
||||
$isAPIKey = $user->isApp($authorization->getRoles());
|
||||
$isPrivilegedUser = $user->isPrivileged($authorization->getRoles());
|
||||
|
||||
$database = $authorization->skip(fn () => $dbForProject->getDocument('databases', $databaseId));
|
||||
if ($database->isEmpty() || (!$database->getAttribute('enabled', false) && !$isAPIKey && !$isPrivilegedUser)) {
|
||||
throw new Exception(Exception::DATABASE_NOT_FOUND, params: [$databaseId]);
|
||||
}
|
||||
|
||||
$collection = $authorization->skip(fn () => $dbForProject->getDocument('database_' . $database->getSequence(), $collectionId));
|
||||
if ($collection->isEmpty() || (!$collection->getAttribute('enabled', false) && !$isAPIKey && !$isPrivilegedUser)) {
|
||||
throw new Exception($this->getParentNotFoundException(), params: [$collectionId]);
|
||||
}
|
||||
|
||||
try {
|
||||
$queries = Query::parseQueries($queries);
|
||||
} catch (QueryException $e) {
|
||||
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
|
||||
}
|
||||
|
||||
$dbForDatabases = $getDatabasesDB($database);
|
||||
$cursor = Query::getCursorQueries($queries, false);
|
||||
$cursor = \reset($cursor);
|
||||
|
||||
if ($cursor !== false) {
|
||||
$validator = new Cursor();
|
||||
if (!$validator->isValid($cursor)) {
|
||||
throw new Exception(Exception::GENERAL_QUERY_INVALID, $validator->getDescription());
|
||||
}
|
||||
|
||||
$documentId = $cursor->getValue();
|
||||
|
||||
$cursorDocument = $authorization->skip(fn () => $dbForDatabases->getDocument('database_' . $database->getSequence() . '_collection_' . $collection->getSequence(), $documentId));
|
||||
|
||||
if ($cursorDocument->isEmpty()) {
|
||||
$type = ucfirst($this->getContext());
|
||||
throw new Exception(Exception::GENERAL_CURSOR_NOT_FOUND, "$type '{$documentId}' for the 'cursor' value not found.");
|
||||
}
|
||||
|
||||
$cursor->setValue($cursorDocument);
|
||||
}
|
||||
|
||||
$dbDurationMs = 0.0;
|
||||
$measure = function (callable $fn) use (&$dbDurationMs) {
|
||||
$start = \microtime(true);
|
||||
try {
|
||||
return $fn();
|
||||
} finally {
|
||||
$dbDurationMs += (\microtime(true) - $start) * 1000;
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
$hasSelects = ! empty(Query::groupByType($queries)['selections'] ?? []);
|
||||
$collectionTableId = 'database_' . $database->getSequence() . '_collection_' . $collection->getSequence();
|
||||
$find = $hasSelects
|
||||
? fn () => $measure(fn () => $dbForDatabases->find($collectionTableId, $queries))
|
||||
: fn () => $measure(fn () => $dbForDatabases->skipRelationships(fn () => $dbForDatabases->find($collectionTableId, $queries)));
|
||||
|
||||
if ($transactionId !== null) {
|
||||
$documents = $measure(fn () => $transactionState->listDocuments($database, $collectionTableId, $transactionId, $queries));
|
||||
$total = $includeTotal ? $measure(fn () => $transactionState->countDocuments($database, $collectionTableId, $transactionId, $queries)) : 0;
|
||||
} elseif ((int)$ttl > 0) {
|
||||
$cacheKey = $this->getListCacheKey($dbForProject, $collectionId);
|
||||
$roles = $dbForProject->getAuthorization()->getRoles();
|
||||
$documentsField = $this->getListCacheField($collection, $roles, $queries, self::LIST_CACHE_FIELD_DOCUMENTS);
|
||||
|
||||
$documentsCacheHit = false;
|
||||
try {
|
||||
$cachedDocuments = $dbForProject->getCache()->load($cacheKey, $ttl, $documentsField);
|
||||
} catch (\Throwable) {
|
||||
$cachedDocuments = null;
|
||||
}
|
||||
|
||||
if ($cachedDocuments !== null &&
|
||||
$cachedDocuments !== false &&
|
||||
\is_array($cachedDocuments)) {
|
||||
$documents = \array_map(function ($doc) {
|
||||
return new Document($doc);
|
||||
}, $cachedDocuments);
|
||||
$documentsCacheHit = true;
|
||||
} else {
|
||||
$documents = $find();
|
||||
|
||||
$documentsArray = \array_map(function ($doc) {
|
||||
return $doc->getArrayCopy();
|
||||
}, $documents);
|
||||
try {
|
||||
$dbForProject->getCache()->save($cacheKey, $documentsArray, $documentsField);
|
||||
} catch (\Throwable) {
|
||||
}
|
||||
}
|
||||
|
||||
if ($includeTotal) {
|
||||
$totalField = $this->getListCacheField($collection, $roles, $queries, self::LIST_CACHE_FIELD_TOTAL);
|
||||
try {
|
||||
$cachedTotal = $dbForProject->getCache()->load($cacheKey, $ttl, $totalField);
|
||||
} catch (\Throwable) {
|
||||
$cachedTotal = null;
|
||||
}
|
||||
if ($cachedTotal !== null && $cachedTotal !== false) {
|
||||
$total = $cachedTotal;
|
||||
} else {
|
||||
$total = $measure(fn () => $dbForDatabases->count($collectionTableId, $queries, APP_LIMIT_COUNT));
|
||||
try {
|
||||
$dbForProject->getCache()->save($cacheKey, $total, $totalField);
|
||||
} catch (\Throwable) {
|
||||
}
|
||||
}
|
||||
} else {
|
||||
$total = 0;
|
||||
}
|
||||
|
||||
$response->addHeader('X-Appwrite-Cache', $documentsCacheHit ? 'hit' : 'miss');
|
||||
} else {
|
||||
$documents = $find();
|
||||
$total = $includeTotal ? $measure(fn () => $dbForDatabases->count($collectionTableId, $queries, APP_LIMIT_COUNT)) : 0;
|
||||
}
|
||||
} catch (OrderException $e) {
|
||||
$documents = $this->isCollectionsAPI() ? 'documents' : 'rows';
|
||||
$attribute = $this->isCollectionsAPI() ? 'attribute' : 'column';
|
||||
$message = "The order $attribute '{$e->getAttribute()}' had a null value. Cursor pagination requires all $documents order $attribute values are non-null.";
|
||||
throw new Exception(Exception::DATABASE_QUERY_ORDER_NULL, $message);
|
||||
} catch (QueryException $e) {
|
||||
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
|
||||
} catch (Timeout) {
|
||||
throw new Exception(Exception::DATABASE_TIMEOUT);
|
||||
}
|
||||
|
||||
$operations = 0;
|
||||
$collectionsCache = [];
|
||||
foreach ($documents as $document) {
|
||||
$this->processDocument(
|
||||
database: $database,
|
||||
collection: $collection,
|
||||
document: $document,
|
||||
dbForProject: $dbForProject,
|
||||
collectionsCache: $collectionsCache,
|
||||
authorization: $authorization,
|
||||
operations: $operations
|
||||
);
|
||||
}
|
||||
|
||||
$usage
|
||||
->addMetric($this->getDatabasesOperationReadMetric(), max($operations, 1))
|
||||
->addMetric(str_replace('{databaseInternalId}', $database->getSequence(), $this->getDatabasesIdOperationReadMetric()), $operations);
|
||||
|
||||
$response->dynamic(new Document([
|
||||
'total' => $total,
|
||||
$this->getSDKGroup() => $documents,
|
||||
]), $this->getResponseModel());
|
||||
|
||||
try {
|
||||
$this->afterQuery($dbDurationMs, $database, $collection, $queries, $utopia);
|
||||
} catch (\Throwable) {
|
||||
// Observers must never break the response.
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hook invoked after listRows completes the response. Under Swoole (the
|
||||
* default transport) the client connection has already been closed by
|
||||
* `$response->dynamic()`, so observers here do not delay the client.
|
||||
* Under synchronous transports observers would run before the bytes
|
||||
* reach the client — keep work here cheap regardless.
|
||||
*
|
||||
* Runs with the actual measured database duration (cache hits report
|
||||
* near-zero). Intended to be overridden for observability (e.g., slow-
|
||||
* query logging in downstream distributions). CE implementation is a
|
||||
* no-op.
|
||||
*
|
||||
* The `$utopia` Http instance is passed so overrides can resolve
|
||||
* additional resources (e.g., a downstream-specific logger) via
|
||||
* `$utopia->getResource(...)` without needing to inject them here.
|
||||
*
|
||||
* @param array<Query> $queries parsed Query objects (pass directly to
|
||||
* `Query::fingerprint()` if you need a
|
||||
* shape hash)
|
||||
*/
|
||||
protected function afterQuery(float $dbDurationMs, Document $database, Document $collection, array $queries, ?Http $utopia): void
|
||||
{
|
||||
// no-op in CE
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user