Elgato_dark/appwrite
1
0
Fork 0
mirror of https://github.com/appwrite/appwrite synced 2026-05-18 14:38:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge branch '1.6.x' of https://github.com/appwrite/appwrite into multi-region-support

Browse source 
# Conflicts:
#	composer.json
#	composer.lock
This commit is contained in:
shimon 2024-11-07 13:02:45 +02:00
commit 2cc73151d1
23 changed files with 432 additions and 343 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
app/controllers/api/account.php
View file

@ -274,7 +274,6 @@ $createSession = function (string $userId, string $secret, Request $request, Res
App::post('/v1/account')
->desc('Create account')
->groups(['api', 'account', 'auth'])
->label('event', 'users.[userId].create')
->label('scope', 'sessions.write')
->label('auth.type', 'emailPassword')
->label('audits.event', 'user.create')
@ -297,9 +296,8 @@ App::post('/v1/account')
->inject('user')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $name, Request $request, Response $response, Document $user, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
->action(function (string $userId, string $email, string $password, string $name, Request $request, Response $response, Document $user, Document $project, Database $dbForProject, Hooks $hooks) {
$email = \strtolower($email);
if ('console' === $project->getId()) {
@ -409,8 +407,6 @@ App::post('/v1/account')
Authorization::setRole(Role::user($user->getId())->toString());
Authorization::setRole(Role::users()->toString());
$queueForEvents->setParam('userId', $user->getId());
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
->dynamic($user, Response::MODEL_ACCOUNT);
@ -442,7 +438,6 @@ App::get('/v1/account')
App::delete('/v1/account')
->desc('Delete account')
->groups(['api', 'account'])
->label('event', 'users.[userId].delete')
->label('scope', 'account')
->label('audits.event', 'user.delete')
->label('audits.resource', 'user/{response.$id}')
@ -1499,6 +1494,7 @@ App::get('/v1/account/sessions/oauth2/:provider/redirect')
'providerType' => MESSAGE_TYPE_EMAIL,
'identifier' => $email,
]));
} catch (Duplicate) {
$failureRedirect(Exception::USER_ALREADY_EXISTS);
}

185
app/controllers/api/databases.php
View file

@ -3,7 +3,6 @@
use Appwrite\Auth\Auth;
use Appwrite\Detector\Detector;
use Appwrite\Event\Database as EventDatabase;
use Appwrite\Event\Delete;
use Appwrite\Event\Event;
use Appwrite\Event\Usage;
use Appwrite\Extend\Exception;
@ -26,6 +25,7 @@ use Utopia\Database\Exception\Authorization as AuthorizationException;
use Utopia\Database\Exception\Conflict as ConflictException;
use Utopia\Database\Exception\Duplicate as DuplicateException;
use Utopia\Database\Exception\Limit as LimitException;
use Utopia\Database\Exception\NotFound as NotFoundException;
use Utopia\Database\Exception\Query as QueryException;
use Utopia\Database\Exception\Restricted as RestrictedException;
use Utopia\Database\Exception\Structure as StructureException;
@ -352,13 +352,16 @@ function updateAttribute(
if ($type === Database::VAR_RELATIONSHIP) {
$primaryDocumentOptions = \array_merge($attribute->getAttribute('options', []), $options);
$attribute->setAttribute('options', $primaryDocumentOptions);
$dbForProject->updateRelationship(
collection: $collectionId,
id: $key,
newKey: $newKey,
onDelete: $primaryDocumentOptions['onDelete'],
);
try {
$dbForProject->updateRelationship(
collection: $collectionId,
id: $key,
newKey: $newKey,
onDelete: $primaryDocumentOptions['onDelete'],
);
} catch (NotFoundException) {
throw new Exception(Exception::ATTRIBUTE_NOT_FOUND);
}
if ($primaryDocumentOptions['twoWay']) {
$relatedCollection = $dbForProject->getDocument('database_' . $db->getInternalId(), $primaryDocumentOptions['relatedCollection']);
@ -388,6 +391,8 @@ function updateAttribute(
);
} catch (TruncateException) {
throw new Exception(Exception::ATTRIBUTE_INVALID_RESIZE);
} catch (NotFoundException) {
throw new Exception(Exception::ATTRIBUTE_NOT_FOUND);
}
}
@ -530,12 +535,7 @@ App::get('/v1/databases')
->inject('response')
->inject('dbForProject')
->action(function (array $queries, string $search, Response $response, Database $dbForProject) {
try {
$queries = Query::parseQueries($queries);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$queries = Query::parseQueries($queries);
if (!empty($search)) {
$queries[] = Query::search('search', $search);
@ -816,22 +816,21 @@ App::post('/v1/databases/:databaseId/collections')
$collectionId = $collectionId == 'unique()' ? ID::unique() : $collectionId;
// Map aggregate permissions into the multiple permissions they represent.
$permissions = Permission::aggregate($permissions);
$permissions = Permission::aggregate($permissions) ?? [];
try {
$dbForProject->createDocument('database_' . $database->getInternalId(), new Document([
$collection = $dbForProject->createDocument('database_' . $database->getInternalId(), new Document([
'$id' => $collectionId,
'databaseInternalId' => $database->getInternalId(),
'databaseId' => $databaseId,
'$permissions' => $permissions ?? [],
'$permissions' => $permissions,
'documentSecurity' => $documentSecurity,
'enabled' => $enabled,
'name' => $name,
'search' => implode(' ', [$collectionId, $name]),
]));
$collection = $dbForProject->getDocument('database_' . $database->getInternalId(), $collectionId);
$dbForProject->createCollection('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), permissions: $permissions ?? [], documentSecurity: $documentSecurity);
$dbForProject->createCollection('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), permissions: $permissions, documentSecurity: $documentSecurity);
} catch (DuplicateException) {
throw new Exception(Exception::COLLECTION_ALREADY_EXISTS);
} catch (LimitException) {
@ -849,7 +848,7 @@ App::post('/v1/databases/:databaseId/collections')
});
App::get('/v1/databases/:databaseId/collections')
->alias('/v1/database/collections', ['databaseId' => 'default'])
->alias('/v1/database/collections')
->desc('List collections')
->groups(['api', 'database'])
->label('scope', 'collections.read')
@ -875,11 +874,7 @@ App::get('/v1/databases/:databaseId/collections')
throw new Exception(Exception::DATABASE_NOT_FOUND);
}
try {
$queries = Query::parseQueries($queries);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$queries = Query::parseQueries($queries);
if (!empty($search)) {
$queries[] = Query::search('search', $search);
@ -919,7 +914,7 @@ App::get('/v1/databases/:databaseId/collections')
});
App::get('/v1/databases/:databaseId/collections/:collectionId')
->alias('/v1/database/collections/:collectionId', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId')
->desc('Get collection')
->groups(['api', 'database'])
->label('scope', 'collections.read')
@ -954,7 +949,7 @@ App::get('/v1/databases/:databaseId/collections/:collectionId')
});
App::get('/v1/databases/:databaseId/collections/:collectionId/logs')
->alias('/v1/database/collections/:collectionId/logs', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/logs')
->desc('List collection logs')
->groups(['api', 'database'])
->label('scope', 'collections.read')
@ -988,12 +983,7 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/logs')
throw new Exception(Exception::COLLECTION_NOT_FOUND);
}
try {
$queries = Query::parseQueries($queries);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$queries = Query::parseQueries($queries);
$grouped = Query::groupByType($queries);
$limit = $grouped['limit'] ?? APP_LIMIT_COUNT;
$offset = $grouped['offset'] ?? 0;
@ -1055,7 +1045,7 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/logs')
App::put('/v1/databases/:databaseId/collections/:collectionId')
->alias('/v1/database/collections/:collectionId', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId')
->desc('Update collection')
->groups(['api', 'database', 'schema'])
->label('scope', 'collections.write')
@ -1101,12 +1091,16 @@ App::put('/v1/databases/:databaseId/collections/:collectionId')
$enabled ??= $collection->getAttribute('enabled', true);
$collection = $dbForProject->updateDocument('database_' . $database->getInternalId(), $collectionId, $collection
$collection = $dbForProject->updateDocument(
'database_' . $database->getInternalId(),
$collectionId,
$collection
->setAttribute('name', $name)
->setAttribute('$permissions', $permissions)
->setAttribute('documentSecurity', $documentSecurity)
->setAttribute('enabled', $enabled)
->setAttribute('search', implode(' ', [$collectionId, $name])));
->setAttribute('search', \implode(' ', [$collectionId, $name]))
);
$dbForProject->updateCollection('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $permissions, $documentSecurity);
@ -1119,7 +1113,7 @@ App::put('/v1/databases/:databaseId/collections/:collectionId')
});
App::delete('/v1/databases/:databaseId/collections/:collectionId')
->alias('/v1/database/collections/:collectionId', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId')
->desc('Delete collection')
->groups(['api', 'database', 'schema'])
->label('scope', 'collections.write')
@ -1175,7 +1169,7 @@ App::delete('/v1/databases/:databaseId/collections/:collectionId')
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/string')
->alias('/v1/database/collections/:collectionId/attributes/string', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/string')
->desc('Create string attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1226,14 +1220,13 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/string
'filters' => $filters,
]), $response, $dbForProject, $queueForDatabase, $queueForEvents);
$response
->setStatusCode(Response::STATUS_CODE_ACCEPTED)
->dynamic($attribute, Response::MODEL_ATTRIBUTE_STRING);
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/email')
->alias('/v1/database/collections/:collectionId/attributes/email', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/email')
->desc('Create email attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1276,7 +1269,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/email'
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/enum')
->alias('/v1/database/collections/:collectionId/attributes/enum', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/enum')
->desc('Create enum attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1324,7 +1317,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/enum')
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/ip')
->alias('/v1/database/collections/:collectionId/attributes/ip', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/ip')
->desc('Create IP address attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1367,7 +1360,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/ip')
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/url')
->alias('/v1/database/collections/:collectionId/attributes/url', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/url')
->desc('Create URL attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1410,7 +1403,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/url')
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/integer')
->alias('/v1/database/collections/:collectionId/attributes/integer', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/integer')
->desc('Create integer attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1440,8 +1433,8 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/intege
->action(function (string $databaseId, string $collectionId, string $key, ?bool $required, ?int $min, ?int $max, ?int $default, bool $array, Response $response, Database $dbForProject, EventDatabase $queueForDatabase, Event $queueForEvents) {
// Ensure attribute default is within range
$min = (is_null($min)) ? PHP_INT_MIN : \intval($min);
$max = (is_null($max)) ? PHP_INT_MAX : \intval($max);
$min = \is_null($min) ? PHP_INT_MIN : $min;
$max = \is_null($max) ? PHP_INT_MAX : $max;
if ($min > $max) {
throw new Exception(Exception::ATTRIBUTE_VALUE_INVALID, 'Minimum value must be lesser than maximum value');
@ -1482,7 +1475,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/intege
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/float')
->alias('/v1/database/collections/:collectionId/attributes/float', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/float')
->desc('Create float attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1512,21 +1505,16 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/float'
->action(function (string $databaseId, string $collectionId, string $key, ?bool $required, ?float $min, ?float $max, ?float $default, bool $array, Response $response, Database $dbForProject, EventDatabase $queueForDatabase, Event $queueForEvents) {
// Ensure attribute default is within range
$min = (is_null($min)) ? -PHP_FLOAT_MAX : \floatval($min);
$max = (is_null($max)) ? PHP_FLOAT_MAX : \floatval($max);
$min = \is_null($min) ? -PHP_FLOAT_MAX : $min;
$max = \is_null($max) ? PHP_FLOAT_MAX : $max;
if ($min > $max) {
throw new Exception(Exception::ATTRIBUTE_VALUE_INVALID, 'Minimum value must be lesser than maximum value');
}
// Ensure default value is a float
if (!is_null($default)) {
$default = \floatval($default);
}
$validator = new Range($min, $max, Database::VAR_FLOAT);
if (!is_null($default) && !$validator->isValid($default)) {
if (!\is_null($default) && !$validator->isValid($default)) {
throw new Exception(Exception::ATTRIBUTE_VALUE_INVALID, $validator->getDescription());
}
@ -1557,7 +1545,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/float'
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/boolean')
->alias('/v1/database/collections/:collectionId/attributes/boolean', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/boolean')
->desc('Create boolean attribute')
->groups(['api', 'database', 'schema'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1599,7 +1587,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/boolea
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/datetime')
->alias('/v1/database/collections/:collectionId/attributes/datetime', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/datetime')
->desc('Create datetime attribute')
->groups(['api', 'database'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1618,7 +1606,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/dateti
->param('collectionId', '', new UID(), 'Collection ID. You can create a new collection using the Database service [server integration](https://appwrite.io/docs/server/databases#databasesCreateCollection).')
->param('key', '', new Key(), 'Attribute Key.')
->param('required', null, new Boolean(), 'Is attribute required?')
->param('default', null, new DatetimeValidator(), 'Default value for the attribute in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. Cannot be set when attribute is required.', true)
->param('default', null, fn (Database $dbForProject) => new DatetimeValidator($dbForProject->getAdapter()->getMinDateTime(), $dbForProject->getAdapter()->getMaxDateTime()), 'Default value for the attribute in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. Cannot be set when attribute is required.', true, ['dbForProject'])
->param('array', false, new Boolean(), 'Is attribute an array?', true)
->inject('response')
->inject('dbForProject')
@ -1644,7 +1632,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/dateti
});
App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/relationship')
->alias('/v1/database/collections/:collectionId/attributes/relationship', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/relationship')
->desc('Create relationship attribute')
->groups(['api', 'database'])
->label('event', 'databases.[databaseId].collections.[collectionId].attributes.[attributeId].create')
@ -1773,7 +1761,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/attributes/relati
});
App::get('/v1/databases/:databaseId/collections/:collectionId/attributes')
->alias('/v1/database/collections/:collectionId/attributes', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes')
->desc('List attributes')
->groups(['api', 'database'])
->label('scope', 'collections.read')
@ -1804,16 +1792,12 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/attributes')
throw new Exception(Exception::COLLECTION_NOT_FOUND);
}
try {
$queries = Query::parseQueries($queries);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$queries = Query::parseQueries($queries);
\array_push(
$queries,
Query::equal('databaseInternalId', [$database->getInternalId()]),
Query::equal('collectionInternalId', [$collection->getInternalId()]),
Query::equal('databaseInternalId', [$database->getInternalId()])
);
/**
@ -1822,6 +1806,7 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/attributes')
$cursor = \array_filter($queries, function ($query) {
return \in_array($query->getMethod(), [Query::TYPE_CURSOR_AFTER, Query::TYPE_CURSOR_BEFORE]);
});
$cursor = \reset($cursor);
if ($cursor) {
@ -1832,8 +1817,8 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/attributes')
$attributeId = $cursor->getValue();
$cursorDocument = Authorization::skip(fn () => $dbForProject->find('attributes', [
Query::equal('collectionInternalId', [$collection->getInternalId()]),
Query::equal('databaseInternalId', [$database->getInternalId()]),
Query::equal('collectionInternalId', [$collection->getInternalId()]),
Query::equal('key', [$attributeId]),
Query::limit(1),
]));
@ -1857,7 +1842,7 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/attributes')
});
App::get('/v1/databases/:databaseId/collections/:collectionId/attributes/:key')
->alias('/v1/database/collections/:collectionId/attributes/:key', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/:key')
->desc('Get attribute')
->groups(['api', 'database'])
->label('scope', 'collections.read')
@ -1951,7 +1936,7 @@ App::patch('/v1/databases/:databaseId/collections/:collectionId/attributes/strin
->param('key', '', new Key(), 'Attribute Key.')
->param('required', null, new Boolean(), 'Is attribute required?')
->param('default', null, new Nullable(new Text(0, 0)), 'Default value for attribute when not provided. Cannot be set when attribute is required.')
->param('size', null, new Integer(), 'Maximum size of the string attribute.', true)
->param('size', null, new Range(1, APP_DATABASE_ATTRIBUTE_STRING_MAX_LENGTH, Range::TYPE_INTEGER), 'Maximum size of the string attribute.', true)
->param('newKey', null, new Key(), 'New attribute key.', true)
->inject('response')
->inject('dbForProject')
@ -2309,7 +2294,7 @@ App::patch('/v1/databases/:databaseId/collections/:collectionId/attributes/datet
->param('collectionId', '', new UID(), 'Collection ID. You can create a new collection using the Database service [server integration](https://appwrite.io/docs/server/databases#databasesCreateCollection).')
->param('key', '', new Key(), 'Attribute Key.')
->param('required', null, new Boolean(), 'Is attribute required?')
->param('default', null, new Nullable(new DatetimeValidator()), 'Default value for attribute when not provided. Cannot be set when attribute is required.')
->param('default', null, fn (Database $dbForProject) => new Nullable(new DatetimeValidator($dbForProject->getAdapter()->getMinDateTime(), $dbForProject->getAdapter()->getMaxDateTime())), 'Default value for attribute when not provided. Cannot be set when attribute is required.', injections: ['dbForProject'])
->param('newKey', null, new Key(), 'New attribute key.', true)
->inject('response')
->inject('dbForProject')
@ -2390,7 +2375,7 @@ App::patch('/v1/databases/:databaseId/collections/:collectionId/attributes/:key/
});
App::delete('/v1/databases/:databaseId/collections/:collectionId/attributes/:key')
->alias('/v1/database/collections/:collectionId/attributes/:key', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/attributes/:key')
->desc('Delete attribute')
->groups(['api', 'database', 'schema'])
->label('scope', 'collections.write')
@ -2504,7 +2489,7 @@ App::delete('/v1/databases/:databaseId/collections/:collectionId/attributes/:key
});
App::post('/v1/databases/:databaseId/collections/:collectionId/indexes')
->alias('/v1/database/collections/:collectionId/indexes', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/indexes')
->desc('Create index')
->groups(['api', 'database'])
->label('event', 'databases.[databaseId].collections.[collectionId].indexes.[indexId].create')
@ -2675,7 +2660,7 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/indexes')
});
App::get('/v1/databases/:databaseId/collections/:collectionId/indexes')
->alias('/v1/database/collections/:collectionId/indexes', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/indexes')
->desc('List indexes')
->groups(['api', 'database'])
->label('scope', 'collections.read')
@ -2706,13 +2691,13 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/indexes')
throw new Exception(Exception::COLLECTION_NOT_FOUND);
}
try {
$queries = Query::parseQueries($queries);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$queries = Query::parseQueries($queries);
\array_push($queries, Query::equal('collectionId', [$collectionId]), Query::equal('databaseId', [$databaseId]));
\array_push(
$queries,
Query::equal('databaseId', [$databaseId]),
Query::equal('collectionId', [$collectionId]),
);
/**
* Get cursor document if there was a cursor query, we use array_filter and reset for reference $cursor to $queries
@ -3042,10 +3027,12 @@ App::post('/v1/databases/:databaseId/collections/:collectionId/documents')
try {
$document = $dbForProject->createDocument('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $document);
} catch (StructureException $exception) {
throw new Exception(Exception::DOCUMENT_INVALID_STRUCTURE, $exception->getMessage());
} catch (DuplicateException $exception) {
} catch (StructureException $e) {
throw new Exception(Exception::DOCUMENT_INVALID_STRUCTURE, $e->getMessage());
} catch (DuplicateException $e) {
throw new Exception(Exception::DOCUMENT_ALREADY_EXISTS);
} catch (NotFoundException $e) {
throw new Exception(Exception::COLLECTION_NOT_FOUND);
}
// Add $collectionId and $databaseId for all documents
@ -3176,14 +3163,8 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/documents')
$cursor->setValue($cursorDocument);
}
try {
$documents = $dbForProject->find('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $queries);
$total = $dbForProject->count('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $queries, APP_LIMIT_COUNT);
} catch (AuthorizationException) {
throw new Exception(Exception::USER_UNAUTHORIZED);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$documents = $dbForProject->find('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $queries);
$total = $dbForProject->count('database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(), $queries, APP_LIMIT_COUNT);
// Add $collectionId and $databaseId for all documents
$processDocument = (function (Document $collection, Document $document) use (&$processDocument, $dbForProject, $database): bool {
@ -3469,7 +3450,7 @@ App::get('/v1/databases/:databaseId/collections/:collectionId/documents/:documen
});
App::patch('/v1/databases/:databaseId/collections/:collectionId/documents/:documentId')
->alias('/v1/database/collections/:collectionId/documents/:documentId', ['databaseId' => 'default'])
->alias('/v1/database/collections/:collectionId/documents/:documentId')
->desc('Update document')
->groups(['api', 'database'])
->label('event', 'databases.[databaseId].collections.[collectionId].documents.[documentId].update')
@ -3647,8 +3628,10 @@ App::patch('/v1/databases/:databaseId/collections/:collectionId/documents/:docum
throw new Exception(Exception::USER_UNAUTHORIZED);
} catch (DuplicateException) {
throw new Exception(Exception::DOCUMENT_ALREADY_EXISTS);
} catch (StructureException $exception) {
throw new Exception(Exception::DOCUMENT_INVALID_STRUCTURE, $exception->getMessage());
} catch (StructureException $e) {
throw new Exception(Exception::DOCUMENT_INVALID_STRUCTURE, $e->getMessage());
} catch (NotFoundException $e) {
throw new Exception(Exception::COLLECTION_NOT_FOUND);
}
// Add $collectionId and $databaseId for all documents
@ -3757,12 +3740,16 @@ App::delete('/v1/databases/:databaseId/collections/:collectionId/documents/:docu
throw new Exception(Exception::DOCUMENT_NOT_FOUND);
}
$dbForProject->withRequestTimestamp($requestTimestamp, function () use ($dbForProject, $database, $collection, $documentId) {
$dbForProject->deleteDocument(
'database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(),
$documentId
);
});
try {
$dbForProject->withRequestTimestamp($requestTimestamp, function () use ($dbForProject, $database, $collection, $documentId) {
$dbForProject->deleteDocument(
'database_' . $database->getInternalId() . '_collection_' . $collection->getInternalId(),
$documentId
);
});
} catch (NotFoundException $e) {
throw new Exception(Exception::COLLECTION_NOT_FOUND);
}
// Add $collectionId and $databaseId for all documents
$processDocument = function (Document $collection, Document $document) use (&$processDocument, $dbForProject, $database) {

8
app/controllers/api/functions.php
View file

@ -1753,7 +1753,7 @@ App::post('/v1/functions/:functionId/executions')
->param('path', '/', new Text(2048), 'HTTP path of execution. Path can include query params. Default value is /', true)
->param('method', 'POST', new Whitelist(['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'], true), 'HTTP method of execution. Default value is GET.', true)
->param('headers', [], new AnyOf([new Assoc(), new Text(65535)], AnyOf::TYPE_MIXED), 'HTTP headers of execution. Defaults to empty.', true)
->param('scheduledAt', null, new DatetimeValidator(true, DateTimeValidator::PRECISION_MINUTES, 60), 'Scheduled execution time in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future with precision in minutes.', true)
->param('scheduledAt', null, new DatetimeValidator(requireDateInFuture: true, precision: DateTimeValidator::PRECISION_MINUTES, offset: 60), 'Scheduled execution time in [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) format. DateTime value must be in future with precision in minutes.', true)
->inject('response')
->inject('request')
->inject('project')
@ -2642,9 +2642,11 @@ App::get('/v1/functions/templates/:templateId')
->action(function (string $templateId, Response $response) {
$templates = Config::getParam('function-templates', []);
$template = array_shift(\array_filter($templates, function ($template) use ($templateId) {
$filtered = \array_filter($templates, function ($template) use ($templateId) {
return $template['id'] === $templateId;
}));
});
$template = array_shift($filtered);
if (empty($template)) {
throw new Exception(Exception::FUNCTION_TEMPLATE_NOT_FOUND);

6
app/controllers/api/health.php
View file

@ -135,6 +135,7 @@ App::get('/v1/health/cache')
foreach ($configs as $key => $config) {
foreach ($config as $database) {
try {
/** @var \Utopia\Cache\Adapter $adapter */
$adapter = $pools->get($database)->pop()->getResource();
$checkStart = \microtime(true);
@ -191,11 +192,11 @@ App::get('/v1/health/queue')
foreach ($configs as $key => $config) {
foreach ($config as $database) {
$checkStart = \microtime(true);
try {
/** @var Connection $adapter */
$adapter = $pools->get($database)->pop()->getResource();
$checkStart = \microtime(true);
if ($adapter->ping()) {
$output[] = new Document([
'name' => $key . " ($database)",
@ -249,6 +250,7 @@ App::get('/v1/health/pubsub')
foreach ($configs as $key => $config) {
foreach ($config as $database) {
try {
/** @var \Appwrite\PubSub\Adapter $adapter */
$adapter = $pools->get($database)->pop()->getResource();
$checkStart = \microtime(true);

89
app/controllers/api/storage.php
View file

@ -16,7 +16,8 @@ use Utopia\App;
use Utopia\Config\Config;
use Utopia\Database\Database;
use Utopia\Database\Document;
use Utopia\Database\Exception\Duplicate;
use Utopia\Database\Exception\Duplicate as DuplicateException;
use Utopia\Database\Exception\NotFound as NotFoundException;
use Utopia\Database\Exception\Query as QueryException;
use Utopia\Database\Helpers\ID;
use Utopia\Database\Helpers\Permission;
@ -131,7 +132,7 @@ App::post('/v1/storage/buckets')
$bucket = $dbForProject->getDocument('buckets', $bucketId);
$dbForProject->createCollection('bucket_' . $bucket->getInternalId(), $attributes, $indexes, permissions: $permissions ?? [], documentSecurity: $fileSecurity);
} catch (Duplicate) {
} catch (DuplicateException) {
throw new Exception(Exception::STORAGE_BUCKET_ALREADY_EXISTS);
}
@ -273,10 +274,6 @@ App::put('/v1/storage/buckets/:bucketId')
$encryption ??= $bucket->getAttribute('encryption', true);
$antivirus ??= $bucket->getAttribute('antivirus', true);
/**
* Map aggregate permissions into the multiple permissions they represent,
* accounting for the resource type given that some types not allowed specific permissions.
*/
// Map aggregate permissions into the multiple permissions they represent.
$permissions = Permission::aggregate($permissions);
@ -290,11 +287,11 @@ App::put('/v1/storage/buckets/:bucketId')
->setAttribute('encryption', $encryption)
->setAttribute('compression', $compression)
->setAttribute('antivirus', $antivirus));
$dbForProject->updateCollection('bucket_' . $bucket->getInternalId(), $permissions, $fileSecurity);
$queueForEvents
->setParam('bucketId', $bucket->getId())
;
->setParam('bucketId', $bucket->getId());
$response->dynamic($bucket, Response::MODEL_BUCKET);
});
@ -342,7 +339,7 @@ App::delete('/v1/storage/buckets/:bucketId')
});
App::post('/v1/storage/buckets/:bucketId/files')
->alias('/v1/storage/files', ['bucketId' => 'default'])
->alias('/v1/storage/files')
->desc('Create file')
->groups(['api', 'storage'])
->label('scope', 'files.write')
@ -670,7 +667,11 @@ App::post('/v1/storage/buckets/:bucketId/files')
'metadata' => $metadata,
]);
$file = $dbForProject->createDocument('bucket_' . $bucket->getInternalId(), $doc);
try {
$file = $dbForProject->createDocument('bucket_' . $bucket->getInternalId(), $doc);
} catch (NotFoundException) {
throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND);
}
} else {
$file = $file
->setAttribute('chunksUploaded', $chunksUploaded)
@ -686,15 +687,19 @@ App::post('/v1/storage/buckets/:bucketId/files')
if (!$validator->isValid($bucket->getCreate())) {
throw new Exception(Exception::USER_UNAUTHORIZED);
}
$file = Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $bucket->getInternalId(), $fileId, $file));
try {
$file = Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $bucket->getInternalId(), $fileId, $file));
} catch (NotFoundException) {
throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND);
}
}
}
$queueForEvents
->setParam('bucketId', $bucket->getId())
->setParam('fileId', $file->getId())
->setContext('bucket', $bucket)
;
->setContext('bucket', $bucket);
$metadata = null; // was causing leaks as it was passed by reference
@ -704,7 +709,7 @@ App::post('/v1/storage/buckets/:bucketId/files')
});
App::get('/v1/storage/buckets/:bucketId/files')
->alias('/v1/storage/files', ['bucketId' => 'default'])
->alias('/v1/storage/files')
->desc('List files')
->groups(['api', 'storage'])
->label('scope', 'files.read')
@ -739,11 +744,7 @@ App::get('/v1/storage/buckets/:bucketId/files')
throw new Exception(Exception::USER_UNAUTHORIZED);
}
try {
$queries = Query::parseQueries($queries);
} catch (QueryException $e) {
throw new Exception(Exception::GENERAL_QUERY_INVALID, $e->getMessage());
}
$queries = Query::parseQueries($queries);
if (!empty($search)) {
$queries[] = Query::search('search', $search);
@ -781,12 +782,16 @@ App::get('/v1/storage/buckets/:bucketId/files')
$filterQueries = Query::groupByType($queries)['filters'];
if ($fileSecurity && !$valid) {
$files = $dbForProject->find('bucket_' . $bucket->getInternalId(), $queries);
$total = $dbForProject->count('bucket_' . $bucket->getInternalId(), $filterQueries, APP_LIMIT_COUNT);
} else {
$files = Authorization::skip(fn () => $dbForProject->find('bucket_' . $bucket->getInternalId(), $queries));
$total = Authorization::skip(fn () => $dbForProject->count('bucket_' . $bucket->getInternalId(), $filterQueries, APP_LIMIT_COUNT));
try {
if ($fileSecurity && !$valid) {
$files = $dbForProject->find('bucket_' . $bucket->getInternalId(), $queries);
$total = $dbForProject->count('bucket_' . $bucket->getInternalId(), $filterQueries, APP_LIMIT_COUNT);
} else {
$files = Authorization::skip(fn () => $dbForProject->find('bucket_' . $bucket->getInternalId(), $queries));
$total = Authorization::skip(fn () => $dbForProject->count('bucket_' . $bucket->getInternalId(), $filterQueries, APP_LIMIT_COUNT));
}
} catch (NotFoundException) {
throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND);
}
$response->dynamic(new Document([
@ -796,7 +801,7 @@ App::get('/v1/storage/buckets/:bucketId/files')
});
App::get('/v1/storage/buckets/:bucketId/files/:fileId')
->alias('/v1/storage/files/:fileId', ['bucketId' => 'default'])
->alias('/v1/storage/files/:fileId')
->desc('Get file')
->groups(['api', 'storage'])
->label('scope', 'files.read')
@ -844,7 +849,7 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId')
});
App::get('/v1/storage/buckets/:bucketId/files/:fileId/preview')
->alias('/v1/storage/files/:fileId/preview', ['bucketId' => 'default'])
->alias('/v1/storage/files/:fileId/preview')
->desc('Get file preview')
->groups(['api', 'storage'])
->label('scope', 'files.read')
@ -1017,7 +1022,7 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId/preview')
});
App::get('/v1/storage/buckets/:bucketId/files/:fileId/download')
->alias('/v1/storage/files/:fileId/download', ['bucketId' => 'default'])
->alias('/v1/storage/files/:fileId/download')
->desc('Get file for download')
->groups(['api', 'storage'])
->label('scope', 'files.read')
@ -1158,7 +1163,7 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId/download')
});
App::get('/v1/storage/buckets/:bucketId/files/:fileId/view')
->alias('/v1/storage/files/:fileId/view', ['bucketId' => 'default'])
->alias('/v1/storage/files/:fileId/view')
->desc('Get file for view')
->groups(['api', 'storage'])
->label('scope', 'files.read')
@ -1465,7 +1470,7 @@ App::get('/v1/storage/buckets/:bucketId/files/:fileId/push')
});
App::put('/v1/storage/buckets/:bucketId/files/:fileId')
->alias('/v1/storage/files/:fileId', ['bucketId' => 'default'])
->alias('/v1/storage/files/:fileId')
->desc('Update file')
->groups(['api', 'storage'])
->label('scope', 'files.write')
@ -1555,10 +1560,14 @@ App::put('/v1/storage/buckets/:bucketId/files/:fileId')
$file->setAttribute('name', $name);
}
if ($fileSecurity && !$valid) {
$file = $dbForProject->updateDocument('bucket_' . $bucket->getInternalId(), $fileId, $file);
} else {
$file = Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $bucket->getInternalId(), $fileId, $file));
try {
if ($fileSecurity && !$valid) {
$file = $dbForProject->updateDocument('bucket_' . $bucket->getInternalId(), $fileId, $file);
} else {
$file = Authorization::skip(fn () => $dbForProject->updateDocument('bucket_' . $bucket->getInternalId(), $fileId, $file));
}
} catch (NotFoundException) {
throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND);
}
$queueForEvents
@ -1641,10 +1650,14 @@ App::delete('/v1/storage/buckets/:bucketId/files/:fileId')
->setResource('file/' . $fileId)
;
if ($fileSecurity && !$valid) {
$deleted = $dbForProject->deleteDocument('bucket_' . $bucket->getInternalId(), $fileId);
} else {
$deleted = Authorization::skip(fn () => $dbForProject->deleteDocument('bucket_' . $bucket->getInternalId(), $fileId));
try {
if ($fileSecurity && !$valid) {
$deleted = $dbForProject->deleteDocument('bucket_' . $bucket->getInternalId(), $fileId);
} else {
$deleted = Authorization::skip(fn () => $dbForProject->deleteDocument('bucket_' . $bucket->getInternalId(), $fileId));
}
} catch (NotFoundException) {
throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND);
}
if (!$deleted) {

52
app/controllers/api/users.php
View file

@ -51,7 +51,7 @@ use Utopia\Validator\Text;
use Utopia\Validator\WhiteList;
/** TODO: Remove function when we move to using utopia/platform */
function createUser(string $hash, mixed $hashOptions, string $userId, ?string $email, ?string $password, ?string $phone, string $name, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks): Document
function createUser(string $hash, mixed $hashOptions, string $userId, ?string $email, ?string $password, ?string $phone, string $name, Document $project, Database $dbForProject, Hooks $hooks): Document
{
$plaintextPassword = $password;
$hashOptionsObject = (\is_string($hashOptions)) ? \json_decode($hashOptions, true) : $hashOptions; // Cast to JSON array
@ -176,15 +176,12 @@ function createUser(string $hash, mixed $hashOptions, string $userId, ?string $e
throw new Exception(Exception::USER_ALREADY_EXISTS);
}
$queueForEvents->setParam('userId', $user->getId());
return $user;
}
App::post('/v1/users')
->desc('Create user')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -203,10 +200,9 @@ App::post('/v1/users')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, ?string $email, ?string $phone, ?string $password, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
$user = createUser('plaintext', '{}', $userId, $email, $password, $phone, $name, $project, $dbForProject, $queueForEvents, $hooks);
->action(function (string $userId, ?string $email, ?string $phone, ?string $password, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$user = createUser('plaintext', '{}', $userId, $email, $password, $phone, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
->dynamic($user, Response::MODEL_USER);
@ -215,7 +211,6 @@ App::post('/v1/users')
App::post('/v1/users/bcrypt')
->desc('Create user with bcrypt password')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -233,10 +228,9 @@ App::post('/v1/users/bcrypt')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
$user = createUser('bcrypt', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $queueForEvents, $hooks);
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$user = createUser('bcrypt', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
@ -246,7 +240,6 @@ App::post('/v1/users/bcrypt')
App::post('/v1/users/md5')
->desc('Create user with MD5 password')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -264,10 +257,9 @@ App::post('/v1/users/md5')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
$user = createUser('md5', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $queueForEvents, $hooks);
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$user = createUser('md5', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
@ -277,7 +269,6 @@ App::post('/v1/users/md5')
App::post('/v1/users/argon2')
->desc('Create user with Argon2 password')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -295,10 +286,9 @@ App::post('/v1/users/argon2')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
$user = createUser('argon2', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $queueForEvents, $hooks);
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$user = createUser('argon2', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
@ -308,7 +298,6 @@ App::post('/v1/users/argon2')
App::post('/v1/users/sha')
->desc('Create user with SHA password')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -327,16 +316,15 @@ App::post('/v1/users/sha')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $passwordVersion, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
->action(function (string $userId, string $email, string $password, string $passwordVersion, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$options = '{}';
if (!empty($passwordVersion)) {
$options = '{"version":"' . $passwordVersion . '"}';
}
$user = createUser('sha', $options, $userId, $email, $password, null, $name, $project, $dbForProject, $queueForEvents, $hooks);
$user = createUser('sha', $options, $userId, $email, $password, null, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
@ -346,7 +334,6 @@ App::post('/v1/users/sha')
App::post('/v1/users/phpass')
->desc('Create user with PHPass password')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -364,10 +351,9 @@ App::post('/v1/users/phpass')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
$user = createUser('phpass', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $queueForEvents, $hooks);
->action(function (string $userId, string $email, string $password, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$user = createUser('phpass', '{}', $userId, $email, $password, null, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
@ -377,7 +363,6 @@ App::post('/v1/users/phpass')
App::post('/v1/users/scrypt')
->desc('Create user with Scrypt password')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -400,9 +385,8 @@ App::post('/v1/users/scrypt')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $passwordSalt, int $passwordCpu, int $passwordMemory, int $passwordParallel, int $passwordLength, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
->action(function (string $userId, string $email, string $password, string $passwordSalt, int $passwordCpu, int $passwordMemory, int $passwordParallel, int $passwordLength, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$options = [
'salt' => $passwordSalt,
'costCpu' => $passwordCpu,
@ -411,7 +395,7 @@ App::post('/v1/users/scrypt')
'length' => $passwordLength
];
$user = createUser('scrypt', \json_encode($options), $userId, $email, $password, null, $name, $project, $dbForProject, $queueForEvents, $hooks);
$user = createUser('scrypt', \json_encode($options), $userId, $email, $password, null, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
@ -421,7 +405,6 @@ App::post('/v1/users/scrypt')
App::post('/v1/users/scrypt-modified')
->desc('Create user with Scrypt modified password')
->groups(['api', 'users'])
->label('event', 'users.[userId].create')
->label('scope', 'users.write')
->label('audits.event', 'user.create')
->label('audits.resource', 'user/{response.$id}')
@ -442,10 +425,9 @@ App::post('/v1/users/scrypt-modified')
->inject('response')
->inject('project')
->inject('dbForProject')
->inject('queueForEvents')
->inject('hooks')
->action(function (string $userId, string $email, string $password, string $passwordSalt, string $passwordSaltSeparator, string $passwordSignerKey, string $name, Response $response, Document $project, Database $dbForProject, Event $queueForEvents, Hooks $hooks) {
$user = createUser('scryptMod', '{"signerKey":"' . $passwordSignerKey . '","saltSeparator":"' . $passwordSaltSeparator . '","salt":"' . $passwordSalt . '"}', $userId, $email, $password, null, $name, $project, $dbForProject, $queueForEvents, $hooks);
->action(function (string $userId, string $email, string $password, string $passwordSalt, string $passwordSaltSeparator, string $passwordSignerKey, string $name, Response $response, Document $project, Database $dbForProject, Hooks $hooks) {
$user = createUser('scryptMod', '{"signerKey":"' . $passwordSignerKey . '","saltSeparator":"' . $passwordSaltSeparator . '","salt":"' . $passwordSalt . '"}', $userId, $email, $password, null, $name, $project, $dbForProject, $hooks);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)

2
app/controllers/general.php
View file

@ -775,6 +775,8 @@ App::error()
case 'Utopia\Database\Exception\Relationship':
$error = new AppwriteException(AppwriteException::RELATIONSHIP_VALUE_INVALID, $error->getMessage(), previous: $error);
break;
case 'Utopia\Database\Exception\NotFound':
$error = new AppwriteException(AppwriteException::COLLECTION_NOT_FOUND, $error->getMessage(), previous: $error);
}
$code = $error->getCode();

116
app/controllers/shared/api.php
View file

@ -11,10 +11,11 @@ use Appwrite\Event\Delete;
use Appwrite\Event\Event;
use Appwrite\Event\Func;
use Appwrite\Event\Messaging;
use Appwrite\Event\Realtime;
use Appwrite\Event\Usage;
use Appwrite\Event\Webhook;
use Appwrite\Extend\Exception;
use Appwrite\Extend\Exception as AppwriteException;
use Appwrite\Messaging\Adapter\Realtime;
use Appwrite\Utopia\Request;
use Appwrite\Utopia\Response;
use Utopia\Abuse\Abuse;
@ -28,6 +29,7 @@ use Utopia\Database\DateTime;
use Utopia\Database\Document;
use Utopia\Database\Helpers\Role;
use Utopia\Database\Validator\Authorization;
use Utopia\Queue\Connection;
use Utopia\System\System;
use Utopia\Validator\WhiteList;
@ -57,8 +59,36 @@ $parseLabel = function (string $label, array $responsePayload, array $requestPar
return $label;
};
$databaseListener = function (string $event, Document $document, Document $project, Usage $queueForUsage, Database $dbForProject) {
$eventDatabaseListener = function (Document $document, Response $response, Event $queueForEvents, Func $queueForFunctions, Webhook $queueForWebhooks, Realtime $queueForRealtime) {
// Only trigger events for user creation with the database listener.
if ($document->getCollection() !== 'users') {
return;
}
$queueForEvents
->setEvent('users.[userId].create')
->setParam('userId', $document->getId())
->setPayload($response->output($document, Response::MODEL_USER));
// Trigger functions, webhooks, and realtime events
$queueForFunctions
->from($queueForEvents)
->trigger();
$queueForWebhooks
->from($queueForEvents)
->trigger();
if ($queueForEvents->getProject()->getId() === 'console') {
return;
}
$queueForRealtime
->from($queueForEvents)
->trigger();
};
$usageDatabaseListener = function (string $event, Document $document, Usage $queueForUsage) {
$value = 1;
if ($event === Database::EVENT_DOCUMENT_DELETE) {
$value = -1;
@ -353,6 +383,7 @@ App::init()
->inject('response')
->inject('project')
->inject('user')
->inject('queue')
->inject('queueForEvents')
->inject('queueForMessaging')
->inject('queueForAudits')
@ -362,7 +393,7 @@ App::init()
->inject('queueForUsage')
->inject('dbForProject')
->inject('mode')
->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Event $queueForEvents, Messaging $queueForMessaging, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Usage $queueForUsage, Database $dbForProject, string $mode) use ($databaseListener) {
->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Connection $queue, Event $queueForEvents, Messaging $queueForMessaging, Audit $queueForAudits, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Usage $queueForUsage, Database $dbForProject, string $mode) use ($usageDatabaseListener, $eventDatabaseListener) {
$route = $utopia->getRoute();
@ -456,9 +487,24 @@ App::init()
$queueForBuilds->setProject($project);
$queueForMessaging->setProject($project);
// Clone the queues, to prevent events triggered by the database listener
// from overwriting the events that are supposed to be triggered in the shutdown hook.
$queueForEventsClone = new Event($queue);
$queueForFunctions = new Func($queue);
$queueForWebhooks = new Webhook($queue);
$queueForRealtime = new Realtime();
$dbForProject
->on(Database::EVENT_DOCUMENT_CREATE, 'calculate-usage', fn ($event, $document) => $databaseListener($event, $document, $project, $queueForUsage, $dbForProject))
->on(Database::EVENT_DOCUMENT_DELETE, 'calculate-usage', fn ($event, $document) => $databaseListener($event, $document, $project, $queueForUsage, $dbForProject));
->on(Database::EVENT_DOCUMENT_CREATE, 'calculate-usage', fn ($event, $document) => $usageDatabaseListener($event, $document, $queueForUsage))
->on(Database::EVENT_DOCUMENT_DELETE, 'calculate-usage', fn ($event, $document) => $usageDatabaseListener($event, $document, $queueForUsage))
->on(Database::EVENT_DOCUMENT_CREATE, 'create-trigger-events', fn ($event, $document) => $eventDatabaseListener(
$document,
$response,
$queueForEventsClone->from($queueForEvents),
$queueForFunctions->from($queueForEvents),
$queueForWebhooks->from($queueForEvents),
$queueForRealtime->from($queueForEvents)
));
$useCache = $route->getLabel('cache', false);
if ($useCache) {
@ -591,11 +637,13 @@ App::shutdown()
->inject('queueForDatabase')
->inject('queueForBuilds')
->inject('queueForMessaging')
->inject('dbForProject')
->inject('queueForFunctions')
->inject('queueForWebhooks')
->inject('queueForRealtime')
->inject('dbForProject')
->inject('mode')
->inject('dbForConsole')
->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Event $queueForEvents, Audit $queueForAudits, Usage $queueForUsage, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Messaging $queueForMessaging, Database $dbForProject, Func $queueForFunctions, string $mode, Database $dbForConsole) use ($parseLabel) {
->action(function (App $utopia, Request $request, Response $response, Document $project, Document $user, Event $queueForEvents, Audit $queueForAudits, Usage $queueForUsage, Delete $queueForDeletes, EventDatabase $queueForDatabase, Build $queueForBuilds, Messaging $queueForMessaging, Func $queueForFunctions, Event $queueForWebhooks, Realtime $queueForRealtime, Database $dbForProject, string $mode, Database $dbForConsole) use ($parseLabel) {
$responsePayload = $response->getPayload();
@ -604,54 +652,18 @@ App::shutdown()
$queueForEvents->setPayload($responsePayload);
}
/**
* Trigger functions.
*/
if (!$queueForEvents->isPaused()) {
$queueForFunctions
->from($queueForEvents)
->trigger();
}
/**
* Trigger webhooks.
*/
$queueForEvents
->setClass(Event::WEBHOOK_CLASS_NAME)
->setQueue(Event::WEBHOOK_QUEUE_NAME)
$queueForWebhooks
->from($queueForEvents)
->trigger();
$queueForFunctions
->from($queueForEvents)
->trigger();
/**
* Trigger realtime.
*/
if ($project->getId() !== 'console') {
$allEvents = Event::generateEvents($queueForEvents->getEvent(), $queueForEvents->getParams());
$payload = new Document($queueForEvents->getPayload());
$db = $queueForEvents->getContext('database');
$collection = $queueForEvents->getContext('collection');
$bucket = $queueForEvents->getContext('bucket');
$target = Realtime::fromPayload(
// Pass first, most verbose event pattern
event: $allEvents[0],
payload: $payload,
project: $project,
database: $db,
collection: $collection,
bucket: $bucket,
);
Realtime::send(
projectId: $target['projectId'] ?? $project->getId(),
payload: $queueForEvents->getRealtimePayload(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles'],
options: [
'permissionsChanged' => $target['permissionsChanged'],
'userId' => $queueForEvents->getParam('userId')
]
);
$queueForRealtime
->from($queueForEvents)
->trigger();
}
}

14
app/init.php
View file

@ -31,7 +31,9 @@ use Appwrite\Event\Func;
use Appwrite\Event\Mail;
use Appwrite\Event\Messaging;
use Appwrite\Event\Migration;
use Appwrite\Event\Realtime;
use Appwrite\Event\Usage;
use Appwrite\Event\Webhook;
use Appwrite\Extend\Exception;
use Appwrite\Functions\Specification;
use Appwrite\GraphQL\Promises\Adapter\Swoole;
@ -40,6 +42,7 @@ use Appwrite\Hooks\Hooks;
use Appwrite\Network\Validator\Email;
use Appwrite\Network\Validator\Origin;
use Appwrite\OpenSSL\OpenSSL;
use Appwrite\PubSub\Adapter\Redis as PubSub;
use Appwrite\URL\URL as AppwriteURL;
use Appwrite\Utopia\Request;
use MaxMind\Db\Reader;
@ -971,7 +974,10 @@ $register->set('pools', function () {
$adapter->setDatabase($dsn->getPath());
break;
case 'pubsub':
$adapter = $resource();
$adapter = match ($dsn->getScheme()) {
'redis' => new PubSub($resource()),
default => null
};
break;
case 'queue':
$adapter = match ($dsn->getScheme()) {
@ -1134,6 +1140,12 @@ App::setResource('queueForDeletes', function (Connection $queue) {
App::setResource('queueForEvents', function (Connection $queue) {
return new Event($queue);
}, ['queue']);
App::setResource('queueForWebhooks', function (Connection $queue) {
return new Webhook($queue);
}, ['queue']);
App::setResource('queueForRealtime', function () {
return new Realtime();
}, []);
App::setResource('queueForAudits', function (Connection $queue) {
return new Audit($queue);
}, ['queue']);

9
app/realtime.php
View file

@ -367,17 +367,16 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
}
$start = time();
$redis = $register->get('pools')->get('pubsub')->pop()->getResource(); /** @var Redis $redis */
$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);
if ($redis->ping(true)) {
/** @var \Appwrite\PubSub\Adapter $pubsub */
$pubsub = $register->get('pools')->get('pubsub')->pop()->getResource();
if ($pubsub->ping(true)) {
$attempts = 0;
Console::success('Pub/sub connection established (worker: ' . $workerId . ')');
} else {
Console::error('Pub/sub failed (worker: ' . $workerId . ')');
}
$redis->subscribe(['realtime'], function (Redis $redis, string $channel, string $payload) use ($server, $workerId, $stats, $register, $realtime) {
$pubsub->subscribe(['realtime'], function (mixed $redis, string $channel, string $payload) use ($server, $workerId, $stats, $register, $realtime) {
$event = json_decode($payload, true);
if ($event['permissionsChanged'] && isset($event['userId'])) {

2
composer.json
View file

@ -51,7 +51,7 @@
"utopia-php/cache": "0.11.*",
"utopia-php/cli": "0.15.*",
"utopia-php/config": "0.2.*",
"utopia-php/database": "dev-feat-migrations as 0.53.16",
"utopia-php/database": "0.53.16",
"utopia-php/domains": "0.5.*",
"utopia-php/dsn": "0.2.1",
"utopia-php/framework": "0.33.*",

2
docker-compose.yml
View file

@ -1053,4 +1053,4 @@ volumes:
appwrite-certificates:
appwrite-functions:
appwrite-builds:
appwrite-config:
appwrite-config:

42
src/Appwrite/Event/Event.php
View file

@ -204,19 +204,6 @@ class Event
return $this->payload;
}
public function getRealtimePayload(): array
{
$payload = [];
foreach ($this->payload as $key => $value) {
if (!isset($this->sensitive[$key])) {
$payload[$key] = $value;
}
}
return $payload;
}
/**
* Set context for this event.
*
@ -315,10 +302,6 @@ class Event
*/
public function trigger(): string|bool
{
if ($this->paused) {
return false;
}
$client = new Client($this->queue, $this->connection);
return $client->enqueue([
@ -530,20 +513,21 @@ class Event
}
/**
* Get the value of paused
* Generate a function event from a base event
*
* @param Event $event
*
* @return self
*
*/
public function isPaused(): bool
public function from(Event $event): self
{
return $this->paused;
}
/**
* Set the value of paused
*/
public function setPaused(bool $paused): self
{
$this->paused = $paused;
$this->project = $event->getProject();
$this->user = $event->getUser();
$this->payload = $event->getPayload();
$this->event = $event->getEvent();
$this->params = $event->getParams();
$this->context = $event->context;
return $this;
}
}

22
src/Appwrite/Event/Func.php
View file

@ -213,10 +213,6 @@ class Func extends Event
*/
public function trigger(): string|bool
{
if ($this->paused) {
return false;
}
$client = new Client($this->queue, $this->connection);
$events = $this->getEvent() ? Event::generateEvents($this->getEvent(), $this->getParams()) : null;
@ -238,22 +234,4 @@ class Func extends Event
'method' => $this->method,
]);
}
/**
* Generate a function event from a base event
*
* @param Event $event
*
* @return self
*
*/
public function from(Event $event): self
{
$this->project = $event->getProject();
$this->user = $event->getUser();
$this->payload = $event->getPayload();
$this->event = $event->getEvent();
$this->params = $event->getParams();
return $this;
}
}

70
src/Appwrite/Event/Realtime.php Normal file
View file

@ -0,0 +1,70 @@
<?php
namespace Appwrite\Event;
use Appwrite\Messaging\Adapter\Realtime as RealtimeAdapter;
use Utopia\Database\Document;
class Realtime extends Event
{
public function __construct()
{
}
public function getRealtimePayload(): array
{
$payload = [];
foreach ($this->payload as $key => $value) {
if (!isset($this->sensitive[$key])) {
$payload[$key] = $value;
}
}
return $payload;
}
/**
* Execute Event.
*
* @return string|bool
* @throws InvalidArgumentException
*/
public function trigger(): string|bool
{
if (empty($this->event)) {
return false;
}
$allEvents = Event::generateEvents($this->getEvent(), $this->getParams());
$payload = new Document($this->getPayload());
$db = $this->getContext('database');
$collection = $this->getContext('collection');
$bucket = $this->getContext('bucket');
$target = RealtimeAdapter::fromPayload(
// Pass first, most verbose event pattern
event: $allEvents[0],
payload: $payload,
project: $this->getProject(),
database: $db,
collection: $collection,
bucket: $bucket,
);
RealtimeAdapter::send(
projectId: $target['projectId'] ?? $this->getProject()->getId(),
payload: $this->getRealtimePayload(),
events: $allEvents,
channels: $target['channels'],
roles: $target['roles'],
options: [
'permissionsChanged' => $target['permissionsChanged'],
'userId' => $this->getParam('userId')
]
);
return true;
}
}

17
src/Appwrite/Event/Webhook.php Normal file
View file

@ -0,0 +1,17 @@
<?php
namespace Appwrite\Event;
use Utopia\Queue\Connection;
class Webhook extends Event
{
public function __construct(protected Connection $connection)
{
parent::__construct($connection);
$this
->setQueue(Event::WEBHOOK_QUEUE_NAME)
->setClass(Event::WEBHOOK_CLASS_NAME);
}
}

35
src/Appwrite/Messaging/Adapter/Realtime.php
View file

@ -7,7 +7,6 @@ use Utopia\Database\DateTime;
use Utopia\Database\Document;
use Utopia\Database\Helpers\ID;
use Utopia\Database\Helpers\Role;
use Utopia\System\System;
class Realtime extends Adapter
{
@ -139,20 +138,26 @@ class Realtime extends Adapter
$permissionsChanged = array_key_exists('permissionsChanged', $options) && $options['permissionsChanged'];
$userId = array_key_exists('userId', $options) ? $options['userId'] : null;
$redis = new \Redis(); //TODO: make this part of the constructor
$redis->connect(System::getEnv('_APP_REDIS_HOST', ''), System::getEnv('_APP_REDIS_PORT', ''));
$redis->publish('realtime', json_encode([
'project' => $projectId,
'roles' => $roles,
'permissionsChanged' => $permissionsChanged,
'userId' => $userId,
'data' => [
'events' => $events,
'channels' => $channels,
'timestamp' => DateTime::formatTz(DateTime::now()),
'payload' => $payload
]
]));
global $register;
$pubsub = $register->get('pools')->get('pubsub')->pop();
try {
/** @var \Appwrite\PubSub\Adapter $redis */
$redis = $pubsub->getResource();
$redis->publish('realtime', json_encode([
'project' => $projectId,
'roles' => $roles,
'permissionsChanged' => $permissionsChanged,
'userId' => $userId,
'data' => [
'events' => $events,
'channels' => $channels,
'timestamp' => DateTime::formatTz(DateTime::now()),
'payload' => $payload
]
]));
} finally {
$pubsub->reclaim();
}
}
/**

2
src/Appwrite/Platform/Tasks/Doctor.php
View file

@ -3,6 +3,7 @@
namespace Appwrite\Platform\Tasks;
use Appwrite\ClamAV\Network;
use Appwrite\PubSub\Adapter;
use Utopia\App;
use Utopia\CLI\Console;
use Utopia\Config\Config;
@ -158,6 +159,7 @@ class Doctor extends Action
foreach ($configs as $key => $config) {
foreach ($config as $pool) {
try {
/** @var Adapter $adapter */
$adapter = $pools->get($pool)->pop()->getResource();
if ($adapter->ping()) {

13
src/Appwrite/PubSub/Adapter.php Normal file
View file

@ -0,0 +1,13 @@
<?php
namespace Appwrite\PubSub;
interface Adapter
{
public function ping($message = null): bool;
public function subscribe($channels, $callback);
public function publish($channel, $message);
}

31
src/Appwrite/PubSub/Adapter/Redis.php Normal file
View file

@ -0,0 +1,31 @@
<?php
namespace Appwrite\PubSub\Adapter;
use Appwrite\PubSub\Adapter;
class Redis implements Adapter
{
private \Redis $client;
public function __construct(\Redis $client)
{
$this->client = $client;
}
public function ping($message = null): bool
{
return $this->client->ping($message);
}
public function subscribe($channels, $callback)
{
return $this->client->subscribe($channels, $callback);
}
public function publish($channel, $message)
{
return $this->client->publish($channel, $message);
}
}

12
tests/e2e/Services/Webhooks/WebhooksBase.php
View file

@ -901,6 +901,17 @@ trait WebhooksBase
$teamId = $data['teamId'] ?? '';
$email = uniqid() . 'friend@localhost.test';
// Create user to ensure team event is triggered after user event
$user = $this->client->call(Client::METHOD_POST, '/account', array_merge([
'content-type' => 'application/json',
'x-appwrite-project' => $this->getProject()['$id'],
], $this->getHeaders()), [
'userId' => ID::unique(),
'email' => $email,
'password' => 'password',
'name' => 'Friend User',
]);
/**
* Test for SUCCESS
*/
@ -909,7 +920,6 @@ trait WebhooksBase
'x-appwrite-project' => $this->getProject()['$id'],
], $this->getHeaders()), [
'email' => $email,
'name' => 'Friend User',
'roles' => ['admin', 'editor'],
'url' => 'http://localhost:5000/join-us#title'
]);

19
tests/unit/Event/EventTest.php
View file

@ -3,13 +3,9 @@
namespace Tests\Unit\Event;
use Appwrite\Event\Event;
use Appwrite\URL\URL;
use InvalidArgumentException;
use PHPUnit\Framework\TestCase;
use Utopia\DSN\DSN;
use Utopia\Queue;
use Utopia\Queue\Client;
use Utopia\System\System;
require_once __DIR__ . '/../../../app/init.php';
@ -20,19 +16,8 @@ class EventTest extends TestCase
public function setUp(): void
{
$fallbackForRedis = 'redis_main=' . URL::unparse([
'scheme' => 'redis',
'host' => System::getEnv('_APP_REDIS_HOST', 'redis'),
'port' => System::getEnv('_APP_REDIS_PORT', '6379'),
'user' => System::getEnv('_APP_REDIS_USER', ''),
'pass' => System::getEnv('_APP_REDIS_PASS', ''),
]);
$dsn = System::getEnv('_APP_CONNECTIONS_QUEUE', $fallbackForRedis);
$dsn = explode('=', $dsn);
$dsn = $dsn[1] ?? '';
$dsn = new DSN($dsn);
$connection = new Queue\Connection\Redis($dsn->getHost(), $dsn->getPort());
global $register;
$connection = $register->get('pools')->get('queue')->pop()->getResource();
$this->queue = 'v1-tests' . uniqid();
$this->object = new Event($connection);
$this->object->setClass('TestsV1');

19
tests/unit/Usage/StatsTest.php
View file

@ -2,13 +2,9 @@
namespace Tests\Unit\Usage;
use Appwrite\URL\URL as AppwriteURL;
use PHPUnit\Framework\TestCase;
use Utopia\DSN\DSN;
use Utopia\Queue;
use Utopia\Queue\Client;
use Utopia\Queue\Connection;
use Utopia\System\System;
class StatsTest extends TestCase
{
@ -19,18 +15,9 @@ class StatsTest extends TestCase
public function setUp(): void
{
$env = System::getEnv('_APP_CONNECTIONS_QUEUE', 'redis_main=' . AppwriteURL::unparse([
'scheme' => 'redis',
'host' => System::getEnv('_APP_REDIS_HOST', 'redis'),
'port' => System::getEnv('_APP_REDIS_PORT', '6379'),
'user' => System::getEnv('_APP_REDIS_USER', ''),
'pass' => System::getEnv('_APP_REDIS_PASS', ''),
]));
$dsn = explode('=', $env);
$dsn = count($dsn) > 1 ? $dsn[1] : $dsn[0];
$dsn = new DSN($dsn);
$this->connection = new Queue\Connection\Redis($dsn->getHost(), $dsn->getPort());
global $register;
$connection = $register->get('pools')->get('queue')->pop()->getResource();
$this->connection = $connection;
$this->client = new Client(self::QUEUE_NAME, $this->connection);
}