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

Merge branch 'refactor-usage-sn' of https://github.com/appwrite/appwrite into refactor-usage-sn-collection-not-found

Browse Source 
 Conflicts:
	app/controllers/general.php
This commit is contained in:
fogelito
2024-04-11 11:26:50 +03:00
parent 7b25719f4f 3f92987002
commit 9b3ce793dd
14 changed files with 597 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/config/errors.php
+1 -1
View File
@@ -639,7 +639,7 @@ return [
Exception::PROJECT_PROVIDER_UNSUPPORTED => [
'name' => Exception::PROJECT_PROVIDER_UNSUPPORTED,
'description' => 'The chosen OAuth provider is unsupported. Please check the <a href="/docs/client/account?sdk=web-default#accountCreateOAuth2Session">Create OAuth2 Session docs</a> for the complete list of supported OAuth providers.',
'code' => 501,
'code' => 400,
],
Exception::PROJECT_INVALID_SUCCESS_URL => [
'name' => Exception::PROJECT_INVALID_SUCCESS_URL,
app/controllers/api/projects.php
+5 -1
View File
@@ -4,6 +4,7 @@ use Appwrite\Auth\Auth;
use Appwrite\Event\Delete;
use Appwrite\Event\Validator\Event;
use Appwrite\Extend\Exception;
use Appwrite\Hooks\Hooks;
use Appwrite\Network\Validator\Email;
use Appwrite\Network\Validator\Origin;
use Appwrite\Template\Template;
@@ -75,7 +76,8 @@ App::post('/v1/projects')
->inject('dbForConsole')
->inject('cache')
->inject('pools')
->action(function (string $projectId, string $name, string $teamId, string $region, string $description, string $logo, string $url, string $legalName, string $legalCountry, string $legalState, string $legalCity, string $legalAddress, string $legalTaxId, Response $response, Database $dbForConsole, Cache $cache, Group $pools) {
->inject('hooks')
->action(function (string $projectId, string $name, string $teamId, string $region, string $description, string $logo, string $url, string $legalName, string $legalCountry, string $legalState, string $legalCity, string $legalAddress, string $legalTaxId, Response $response, Database $dbForConsole, Cache $cache, Group $pools, Hooks $hooks) {
$team = $dbForConsole->getDocument('teams', $teamId);
@@ -217,6 +219,8 @@ App::post('/v1/projects')
$dbForProject->createCollection($key, $attributes, $indexes);
}
$hooks->trigger('afterProjectCreation', [ $project, $pools, $cache ]);
$response
->setStatusCode(Response::STATUS_CODE_CREATED)
->dynamic($project, Response::MODEL_PROJECT);
app/controllers/general.php
+11 -8
View File
@@ -376,6 +376,16 @@ App::init()
}
});
App::init()
->groups(['api'])
->inject('project')
->inject('mode')
->action(function (Document $project, string $mode) {
if ($mode === APP_MODE_ADMIN && $project->getId() === 'console') {
throw new AppwriteException(AppwriteException::GENERAL_BAD_REQUEST, 'Admin mode is not allowed for console project');
}
});
App::init()
->groups(['api', 'web'])
->inject('utopia')
@@ -639,7 +649,7 @@ App::init()
: Role::users()->toString();
// Add user roles
$memberships = $user->find('teamId', $project->getAttribute('teamId'), 'memberships');
$memberships = $user->find('teamInternalId', $project->getAttribute('teamInternalId'), 'memberships');
if ($memberships) {
foreach ($memberships->getAttribute('roles', []) as $memberRole) {
@@ -829,31 +839,24 @@ App::error()
}
break;
case 'Utopia\Database\Exception\Conflict':
\var_dump('Wrapping conflict exception');
$error = new AppwriteException(AppwriteException::DOCUMENT_UPDATE_CONFLICT, previous: $error);
break;
case 'Utopia\Database\Exception\Timeout':
\var_dump('Wrapping timeout exception');
$error = new AppwriteException(AppwriteException::DATABASE_TIMEOUT, previous: $error);
break;
case 'Utopia\Database\Exception\Query':
\var_dump('Wrapping query exception');
$error = new AppwriteException(AppwriteException::GENERAL_QUERY_INVALID, $error->getMessage(), previous: $error);
break;
case 'Utopia\Database\Exception\Structure':
\var_dump('Wrapping structure exception');
$error = new AppwriteException(AppwriteException::DOCUMENT_INVALID_STRUCTURE, $error->getMessage(), previous: $error);
break;
case 'Utopia\Database\Exception\Duplicate':
\var_dump('Wrapping duplicate exception');
$error = new AppwriteException(AppwriteException::DOCUMENT_ALREADY_EXISTS);
break;
case 'Utopia\Database\Exception\Restricted':
\var_dump('Wrapping restricted exception');
$error = new AppwriteException(AppwriteException::DOCUMENT_DELETE_RESTRICTED);
break;
case 'Utopia\Database\Exception\Authorization':
\var_dump('Wrapping authorization exception');
$error = new AppwriteException(AppwriteException::USER_UNAUTHORIZED);
break;
}
app/http.php
+9 -1
View File
@@ -2,6 +2,7 @@
require_once __DIR__ . '/../vendor/autoload.php';
use Appwrite\Utopia\Pools\Connections;
use Appwrite\Utopia\Response;
use Swoole\Process;
use Swoole\Http\Server;
@@ -323,7 +324,14 @@ $http->on('request', function (SwooleRequest $swooleRequest, SwooleResponse $swo
$swooleResponse->end(\json_encode($output));
} finally {
$pools->reclaim();
/**
* @var Connections $connections
*/
$connections = $app->getResource('connections');
if (!empty($connections)) {
$connections->reclaim();
}
}
});
app/init.php
+51 -35
View File
@@ -33,6 +33,7 @@ use Appwrite\Network\Validator\Email;
use Appwrite\Network\Validator\Origin;
use Appwrite\OpenSSL\OpenSSL;
use Appwrite\URL\URL as AppwriteURL;
use Appwrite\Utopia\Pools\Connections;
use Utopia\App;
use Utopia\Logger\Logger;
use Utopia\Cache\Adapter\Redis as RedisCache;
@@ -881,11 +882,14 @@ App::setResource('locale', fn() => new Locale(App::getEnv('_APP_LOCALE', 'en')))
App::setResource('localeCodes', function () {
return array_map(fn($locale) => $locale['code'], Config::getParam('locale-codes', []));
});
// Queues
App::setResource('queue', function (Group $pools) {
return $pools->get('queue')->pop()->getResource();
}, ['pools']);
App::setResource('queue', function (Group $pools, Connections $connections) {
$connection = $pools->get('queue')->pop();
$connections->add($connection);
return $connection->getResource();
}, ['pools', 'connections']);
App::setResource('queueForMessaging', function (Connection $queue) {
return new Phone($queue);
}, ['queue']);
@@ -1126,15 +1130,22 @@ App::setResource('console', function () {
]);
}, []);
App::setResource('dbForProject', function (Group $pools, Database $dbForConsole, Cache $cache, Document $project) {
App::setResource('connections', function () {
return new Connections();
});
App::setResource('dbForProject', function (Group $pools, Database $dbForConsole, Cache $cache, Document $project, Connections $connections) {
if ($project->isEmpty() || $project->getId() === 'console') {
return $dbForConsole;
}
$dbAdapter = $pools
$connection = $pools
->get($project->getAttribute('database'))
->pop()
->getResource();
->pop();
$connections->add($connection);
$dbAdapter = $connection->getResource();
$database = new Database($dbAdapter, $cache);
@@ -1145,14 +1156,16 @@ App::setResource('dbForProject', function (Group $pools, Database $dbForConsole,
->setTimeout(APP_DATABASE_TIMEOUT_MILLISECONDS);
return $database;
}, ['pools', 'dbForConsole', 'cache', 'project']);
}, ['pools', 'dbForConsole', 'cache', 'project', 'connections']);
App::setResource('dbForConsole', function (Group $pools, Cache $cache) {
$dbAdapter = $pools
App::setResource('dbForConsole', function (Group $pools, Cache $cache, Connections $connections) {
$connection = $pools
->get('console')
->pop()
->getResource()
;
->pop();
$connections->add($connection);
$dbAdapter = $connection->getResource();
$database = new Database($dbAdapter, $cache);
@@ -1163,12 +1176,12 @@ App::setResource('dbForConsole', function (Group $pools, Cache $cache) {
->setTimeout(APP_DATABASE_TIMEOUT_MILLISECONDS);
return $database;
}, ['pools', 'cache']);
}, ['pools', 'cache', 'connections']);
App::setResource('getProjectDB', function (Group $pools, Database $dbForConsole, $cache) {
App::setResource('getProjectDB', function (Group $pools, Database $dbForConsole, Cache $cache, Connections $connections) {
$databases = []; // TODO: @Meldiron This should probably be responsibility of utopia-php/pools
$getProjectDB = function (Document $project) use ($pools, $dbForConsole, $cache, &$databases) {
$getProjectDB = function (Document $project) use ($pools, $dbForConsole, $cache, $connections, &$databases) {
if ($project->isEmpty() || $project->getId() === 'console') {
return $dbForConsole;
}
@@ -1180,48 +1193,47 @@ App::setResource('getProjectDB', function (Group $pools, Database $dbForConsole,
$database
->setNamespace('_' . $project->getInternalId())
->setMetadata('host', \gethostname())
->setMetadata('project', $project->getId())
->setTimeout(APP_DATABASE_TIMEOUT_MILLISECONDS);
return $database;
}
$dbAdapter = $pools
$connection = $pools
->get($databaseName)
->pop()
->getResource();
->pop();
$connections->add($connection);
$dbAdapter = $connection->getResource();
$database = new Database($dbAdapter, $cache);
$databases[$databaseName] = $database;
$database
->setNamespace('_' . $project->getInternalId())
->setMetadata('host', \gethostname())
->setMetadata('project', $project->getId())
->setTimeout(APP_DATABASE_TIMEOUT_MILLISECONDS);
return $database;
};
return $getProjectDB;
}, ['pools', 'dbForConsole', 'cache']);
}, ['pools', 'dbForConsole', 'cache', 'connections']);
App::setResource('cache', function (Group $pools) {
App::setResource('cache', function (Group $pools, Connections $connections) {
$list = Config::getParam('pools-cache', []);
$adapters = [];
foreach ($list as $value) {
$adapters[] = $pools
$connection = $pools
->get($value)
->pop()
->getResource()
;
->pop();
$connections->add($connection);
$adapters[] = $connection->getResource();
}
return new Cache(new Sharding($adapters));
}, ['pools']);
}, ['pools', 'connections']);
App::setResource('deviceLocal', function () {
return new Local();
@@ -1266,7 +1278,9 @@ function getDevice($root): Device
case Storage::DEVICE_S3:
return new S3($root, $accessKey, $accessSecret, $bucket, $region, $acl);
case STORAGE::DEVICE_DO_SPACES:
return new DOSpaces($root, $accessKey, $accessSecret, $bucket, $region, $acl);
$device = new DOSpaces($root, $accessKey, $accessSecret, $bucket, $region, $acl);
$device->setHttpVersion(S3::HTTP_VERSION_1_1);
return $device;
case Storage::DEVICE_BACKBLAZE:
return new Backblaze($root, $accessKey, $accessSecret, $bucket, $region, $acl);
case Storage::DEVICE_LINODE:
@@ -1295,7 +1309,9 @@ function getDevice($root): Device
$doSpacesRegion = App::getEnv('_APP_STORAGE_DO_SPACES_REGION', '');
$doSpacesBucket = App::getEnv('_APP_STORAGE_DO_SPACES_BUCKET', '');
$doSpacesAcl = 'private';
return new DOSpaces($root, $doSpacesAccessKey, $doSpacesSecretKey, $doSpacesBucket, $doSpacesRegion, $doSpacesAcl);
$device = new DOSpaces($root, $doSpacesAccessKey, $doSpacesSecretKey, $doSpacesBucket, $doSpacesRegion, $doSpacesAcl);
$device->setHttpVersion(S3::HTTP_VERSION_1_1);
return $device;
case Storage::DEVICE_BACKBLAZE:
$backblazeAccessKey = App::getEnv('_APP_STORAGE_BACKBLAZE_ACCESS_KEY', '');
$backblazeSecretKey = App::getEnv('_APP_STORAGE_BACKBLAZE_SECRET', '');
app/realtime.php
+294 -125
View File
@@ -26,84 +26,146 @@ use Utopia\Cache\Adapter\Sharding;
use Utopia\Cache\Cache;
use Utopia\Config\Config;
use Utopia\Database\Database;
use Utopia\Pools\Group;
use Utopia\Registry\Registry;
use Utopia\WebSocket\Server;
use Utopia\WebSocket\Adapter;
/**
* @var \Utopia\Registry\Registry $register
* @var Registry $register
*/
require_once __DIR__ . '/init.php';
Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
Runtime::enableCoroutine();
function getConsoleDB(): Database
{
global $register;
$redisConnections = [];
/** @var \Utopia\Pools\Group $pools */
$pools = $register->get('pools');
// Allows overriding
if (!function_exists('getConsoleDB')) {
/**
* @return array{Database, callable}
* @throws Exception|\Exception
*/
function getConsoleDB(): array
{
global $register;
$dbAdapter = $pools
->get('console')
->pop()
->getResource()
;
/** @var Group $pools */
$pools = $register->get('pools');
$database = new Database($dbAdapter, getCache());
$dbConnection = $pools
->get('console')
->pop();
$database
->setNamespace('_console')
->setMetadata('host', \gethostname())
->setMetadata('project', '_console');
$dbAdapter = $dbConnection->getResource();
return $database;
[$cache, $reclaimCache] = getCache();
$database = new Database($dbAdapter, $cache);
$database->setNamespace('_console');
return [$database, function () use ($dbConnection, $reclaimCache) {
$dbConnection->reclaim();
$reclaimCache();
}];
}
}
function getProjectDB(Document $project): Database
{
global $register;
// Allows overriding
if (!function_exists('getProjectDB')) {
/**
* @param Document $project
* @return array{Database, callable}
* @throws Exception
*/
function getProjectDB(Document $project): array
{
global $register;
/** @var \Utopia\Pools\Group $pools */
$pools = $register->get('pools');
/** @var Group $pools */
$pools = $register->get('pools');
if ($project->isEmpty() || $project->getId() === 'console') {
return getConsoleDB();
if ($project->isEmpty() || $project->getId() === 'console') {
return getConsoleDB();
}
$dbConnection = $pools
->get($project->getAttribute('database'))
->pop();
$dbAdapter = $dbConnection->getResource();
[$cache, $reclaimCache] = getCache();
$database = new Database($dbAdapter, $cache);
$database->setNamespace('_' . $project->getInternalId());
return [$database, function () use ($dbConnection, $reclaimCache) {
$dbConnection->reclaim();
$reclaimCache();
}];
}
$dbAdapter = $pools
->get($project->getAttribute('database'))
->pop()
->getResource()
;
$database = new Database($dbAdapter, getCache());
$database
->setNamespace('_' . $project->getInternalId())
->setMetadata('host', \gethostname())
->setMetadata('project', $project->getId());
return $database;
}
function getCache(): Cache
{
global $register;
// Allows overriding
if (!function_exists('getCache')) {
/**
* @return array{Cache, callable}
* @throws Exception|\Exception
*/
function getCache(): array
{
global $register;
$pools = $register->get('pools'); /** @var \Utopia\Pools\Group $pools */
/** @var Group $pools */
$pools = $register->get('pools');
$list = Config::getParam('pools-cache', []);
$adapters = [];
$list = Config::getParam('pools-cache', []);
foreach ($list as $value) {
$adapters[] = $pools
->get($value)
->pop()
->getResource()
;
$connections = [];
$adapters = [];
foreach ($list as $value) {
$connection = $pools
->get($value)
->pop();
$connections[] = $connection;
$adapters[] = $connection->getResource();
}
$cache = new Cache(new Sharding($adapters));
return [$cache, function () use ($connections) {
foreach ($connections as $connection) {
$connection->reclaim();
}
}];
}
}
return new Cache(new Sharding($adapters));
if (!function_exists('getPubSub')) {
/**
* @return array{Redis, callable}
* @throws Exception|\Exception
*/
function getPubSub(): array
{
global $register;
/** @var Group $pools */
$pools = $register->get('pools');
$connection = $pools
->get('pubsub')
->pop();
$redis = $connection->getResource();
return [$redis, function () use ($connection) {
$connection->reclaim();
}];
}
}
$realtime = new Realtime();
@@ -124,13 +186,17 @@ $statsDocument = null;
$workerNumber = swoole_cpu_num() * intval(App::getEnv('_APP_WORKER_PER_CORE', 6));
$adapter = new Adapter\Swoole(port: App::getEnv('PORT', 80));
$adapter
->setPackageMaxLength(64000) // Default maximum Package Size (64kb)
->setWorkerNumber($workerNumber);
$server = new Server($adapter);
$logError = function (Throwable $error, string $action) use ($register) {
function logError(Throwable $error, string $action): void
{
global $register;
$logger = $register->get('logger');
if ($logger && !$error instanceof Exception) {
@@ -164,11 +230,13 @@ $logError = function (Throwable $error, string $action) use ($register) {
Console::error('[Error] Message: ' . $error->getMessage());
Console::error('[Error] File: ' . $error->getFile());
Console::error('[Error] Line: ' . $error->getLine());
};
}
$server->error($logError);
$server->error(function (Throwable $th, string $method) {
logError($th, $method);
});
$server->onStart(function () use ($stats, $register, $containerId, &$statsDocument, $logError) {
$server->onStart(function () use ($stats, $register, $containerId, &$statsDocument) {
sleep(5); // wait for the initial database schema to be ready
Console::success('Server started successfully');
@@ -177,11 +245,17 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume
*/
go(function () use ($register, $containerId, &$statsDocument) {
$attempts = 0;
$database = getConsoleDB();
do {
try {
/**
* @var Database $database
* @var callable $reclaim
*/
[$database, $reclaim] = getConsoleDB();
$attempts++;
$document = new Document([
'$id' => ID::unique(),
'$collection' => ID::custom('realtime'),
@@ -191,102 +265,131 @@ $server->onStart(function () use ($stats, $register, $containerId, &$statsDocume
'value' => '{}'
]);
$statsDocument = Authorization::skip(fn () => $database->createDocument('realtime', $document));
$statsDocument = Authorization::skip(function () use ($database, $document) {
return $database->createDocument('realtime', $document);
});
break;
} catch (Throwable) {
Console::warning("Collection not ready. Retrying connection ({$attempts})...");
sleep(DATABASE_RECONNECT_SLEEP);
}
} while (true);
$register->get('pools')->reclaim();
if (isset($reclaim)) {
$reclaim();
}
});
/**
* Save current connections to the Database every 5 seconds.
*/
Timer::tick(5000, function () use ($register, $stats, &$statsDocument, $logError) {
Timer::tick(5000, function () use ($register, $stats, &$statsDocument) {
$payload = [];
foreach ($stats as $projectId => $value) {
$payload[$projectId] = $stats->get($projectId, 'connectionsTotal');
}
if (empty($payload) || empty($statsDocument)) {
return;
}
try {
$database = getConsoleDB();
/**
* @var Database $database
* @var callable $reclaim
*/
[$database, $reclaim] = getConsoleDB();
$statsDocument
->setAttribute('timestamp', DateTime::now())
->setAttribute('value', json_encode($payload));
Authorization::skip(fn () => $database->updateDocument('realtime', $statsDocument->getId(), $statsDocument));
Authorization::skip(function () use ($database, $statsDocument) {
$database->updateDocument('realtime', $statsDocument->getId(), $statsDocument);
});
} catch (Throwable $th) {
call_user_func($logError, $th, "updateWorkerDocument");
logError($th, 'updateWorkerDocument');
} finally {
$register->get('pools')->reclaim();
if (isset($reclaim)) {
$reclaim();
}
}
});
});
$server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $realtime, $logError) {
$server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $realtime, &$redisConnections) {
Console::success('Worker ' . $workerId . ' started successfully');
$attempts = 0;
$start = time();
Timer::tick(5000, function () use ($server, $register, $realtime, $stats, $logError) {
Timer::tick(5000, function () use ($server, $register, $realtime, $stats) {
/**
* Sending current connections to project channels on the console project every 5 seconds.
*/
if ($realtime->hasSubscriber('console', Role::users()->toString(), 'project')) {
$database = getConsoleDB();
try {
/**
* @var Database $database
* @var callable $reclaim
*/
[$database, $reclaim] = getConsoleDB();
$payload = [];
$payload = [];
$list = Authorization::skip(fn () => $database->find('realtime', [
Query::greaterThan('timestamp', DateTime::addSeconds(new \DateTime(), -15)),
]));
$list = Authorization::skip(function () use ($database) {
return $database->find('realtime', [
Query::greaterThan('timestamp', DateTime::addSeconds(new \DateTime(), -15)),
]);
});
/**
* Aggregate stats across containers.
*/
foreach ($list as $document) {
foreach (json_decode($document->getAttribute('value')) as $projectId => $value) {
if (array_key_exists($projectId, $payload)) {
$payload[$projectId] += $value;
} else {
$payload[$projectId] = $value;
/**
* Aggregate stats across containers.
*/
foreach ($list as $document) {
foreach (json_decode($document->getAttribute('value')) as $projectId => $value) {
if (array_key_exists($projectId, $payload)) {
$payload[$projectId] += $value;
} else {
$payload[$projectId] = $value;
}
}
}
}
foreach ($stats as $projectId => $value) {
if (!array_key_exists($projectId, $payload)) {
continue;
}
foreach ($stats as $projectId => $value) {
if (!array_key_exists($projectId, $payload)) {
continue;
}
$event = [
'project' => 'console',
'roles' => ['team:' . $stats->get($projectId, 'teamId')],
'data' => [
'events' => ['stats.connections'],
'channels' => ['project'],
'timestamp' => DateTime::formatTz(DateTime::now()),
'payload' => [
$projectId => $payload[$projectId]
$event = [
'project' => 'console',
'roles' => ['team:' . $stats->get($projectId, 'teamId')],
'data' => [
'events' => ['stats.connections'],
'channels' => ['project'],
'timestamp' => DateTime::formatTz(DateTime::now()),
'payload' => [
$projectId => $payload[$projectId]
]
]
]
];
];
$server->send($realtime->getSubscribers($event), json_encode([
'type' => 'event',
'data' => $event['data']
]));
$server->send($realtime->getSubscribers($event), json_encode([
'type' => 'event',
'data' => $event['data']
]));
}
} catch (Throwable $th) {
logError($th, 'sendStats');
} finally {
if (isset($reclaim)) {
$reclaim();
}
}
$register->get('pools')->reclaim();
}
/**
* Sending test message for SDK E2E tests every 5 seconds.
*/
@@ -318,9 +421,17 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
Attempting restart in 5 seconds (attempt #' . $attempts . ')');
sleep(5); // 5 sec delay between connection attempts
}
$start = time();
$redis = $register->get('pools')->get('pubsub')->pop()->getResource(); /** @var Redis $redis */
/**
* @var Redis $redis
* @var callable $reclaimForRedis
*/
[$redis, $reclaimForRedis] = getPubSub();
$redisConnections[$workerId] = [$redis, $reclaimForRedis];
$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);
if ($redis->ping(true)) {
@@ -339,17 +450,36 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
if ($realtime->hasSubscriber($projectId, 'user:' . $userId)) {
$connection = array_key_first(reset($realtime->subscriptions[$projectId]['user:' . $userId]));
$consoleDatabase = getConsoleDB();
$project = Authorization::skip(fn() => $consoleDatabase->getDocument('projects', $projectId));
$database = getProjectDB($project);
$user = $database->getDocument('users', $userId);
/**
* @var Database $dbForConsole
* @var Database $dbForProject
* @var callable $reclaimForConsole
* @var callable $reclaimForProject
*/
[$dbForConsole, $reclaimForConsole] = getConsoleDB();
$project = Authorization::skip(function () use ($dbForConsole, $projectId) {
return $dbForConsole->getDocument('projects', $projectId);
});
[$dbForProject, $reclaimForProject] = getProjectDB($project);
$user = $dbForProject->getDocument('users', $userId);
$roles = Auth::getRoles($user);
$realtime->subscribe($projectId, $connection, $roles, $realtime->connections[$connection]['channels']);
$register->get('pools')->reclaim();
/**
* If we successfully reclaim, clear the callbacks
* so the finally block doesn't try to reclaim again.
*/
$reclaimForConsole();
$reclaimForConsole = null;
$reclaimForProject();
$reclaimForProject = null;
}
}
@@ -374,30 +504,54 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
}
});
} catch (Throwable $th) {
call_user_func($logError, $th, "pubSubConnection");
logError($th, 'pubSubConnection');
Console::error('Pub/sub error: ' . $th->getMessage());
$attempts++;
sleep(DATABASE_RECONNECT_SLEEP);
continue;
} finally {
$register->get('pools')->reclaim();
if (isset($reclaimForConsole)) {
$reclaimForConsole();
}
if (isset($reclaimForProject)) {
$reclaimForProject();
}
}
}
Console::error('Failed to restart pub/sub...');
});
$server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $register, $stats, &$realtime, $logError) {
$server->onWorkerStop(function (int $workerId) use ($redisConnections) {
/**
* @var Redis $redis
* @var callable $reclaim
*/
[$redis, $reclaim] = $redisConnections[$workerId] ?? null;
$redis?->unsubscribe(['realtime']);
if ($reclaim) {
$reclaim();
}
});
$server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $register, $stats, &$realtime) {
$app = new App('UTC');
$request = new Request($request);
$response = new Response(new SwooleResponse());
Console::info("Connection open (user: {$connection})");
App::setResource('pools', fn() => $register->get('pools'));
App::setResource('request', fn() => $request);
App::setResource('response', fn() => $response);
App::setResource('pools', function () use ($register) {
return $register->get('pools');
});
App::setResource('request', function () use ($request) {
return $request;
});
App::setResource('response', function () use ($response) {
return $response;
});
try {
/** @var Document $project */
@@ -410,9 +564,13 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
throw new Exception(Exception::REALTIME_POLICY_VIOLATION, 'Missing or unknown project ID');
}
$dbForProject = getProjectDB($project);
$console = $app->getResource('console'); /** @var Document $console */
$user = $app->getResource('user'); /** @var Document $user */
[$dbForProject, $reclaimForProject] = getProjectDB($project);
/** @var Document $console */
$console = $app->getResource('console');
/** @var Document $user */
$user = $app->getResource('user');
/*
* Abuse Check
@@ -472,7 +630,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
$stats->incr($project->getId(), 'connections');
$stats->incr($project->getId(), 'connectionsTotal');
} catch (Throwable $th) {
call_user_func($logError, $th, "initServer");
logError($th, 'initServer');
$response = [
'type' => 'error',
@@ -491,7 +649,9 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
Console::error('[Error] Message: ' . $response['data']['message']);
}
} finally {
$register->get('pools')->reclaim();
if (isset($reclaimForProject)) {
$reclaimForProject();
}
}
});
@@ -499,11 +659,14 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
try {
$response = new Response(new SwooleResponse());
$projectId = $realtime->connections[$connection]['projectId'];
$database = getConsoleDB();
[$database, $reclaimForConsole] = getConsoleDB();
if ($projectId !== 'console') {
$project = Authorization::skip(fn() => $database->getDocument('projects', $projectId));
$database = getProjectDB($project);
$project = Authorization::skip(function () use ($database, $projectId) {
return $database->getDocument('projects', $projectId);
});
[$database, $reclaimForProject] = getProjectDB($project);
} else {
$project = null;
}
@@ -589,7 +752,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
$server->close($connection, $th->getCode());
}
} finally {
$register->get('pools')->reclaim();
if (isset($reclaimForConsole)) {
$reclaimForConsole();
}
if (isset($reclaimForProject)) {
$reclaimForProject();
}
}
});
@@ -597,6 +765,7 @@ $server->onClose(function (int $connection) use ($realtime, $stats) {
if (array_key_exists($connection, $realtime->connections)) {
$stats->decr($realtime->connections[$connection]['projectId'], 'connectionsTotal');
}
$realtime->unsubscribe($connection);
Console::info('Connection close: ' . $connection);
composer.json
+2 -2
View File
@@ -50,7 +50,7 @@
"utopia-php/cache": "0.9.*",
"utopia-php/cli": "0.15.*",
"utopia-php/config": "0.2.*",
"utopia-php/database": "0.45.7",
"utopia-php/database": "0.45.12",
"utopia-php/domains": "0.5.*",
"utopia-php/dsn": "0.1.*",
"utopia-php/framework": "0.33.*",
@@ -69,7 +69,7 @@
"utopia-php/storage": "0.18.*",
"utopia-php/swoole": "0.5.*",
"utopia-php/vcs": "0.6.*",
"utopia-php/websocket": "0.1.*",
"utopia-php/websocket": "0.2.0",
"matomo/device-detector": "6.1.*",
"dragonmantank/cron-expression": "3.3.2",
"phpmailer/phpmailer": "6.8.0",
composer.lock
Generated
+17 -26
View File
@@ -4,7 +4,7 @@
"Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
"This file is @generated automatically"
],
"content-hash": "95d4c9ccd4b2f958ca247d83b04d1391",
"content-hash": "7ec104df6534b79c176a50541bf5ca24",
"packages": [
{
"name": "adhocore/jwt",
@@ -1191,16 +1191,16 @@
},
{
"name": "utopia-php/database",
"version": "0.45.7",
"version": "0.45.12",
"source": {
"type": "git",
"url": "https://github.com/utopia-php/database.git",
"reference": "3ffcc226cbc5b8747cdc5d6b06ce887cdffee8e1"
"reference": "805efb95b19d555b6cfd2244e36bfe08289f24f5"
},
"dist": {
"type": "zip",
"url": "https://api.github.com/repos/utopia-php/database/zipball/3ffcc226cbc5b8747cdc5d6b06ce887cdffee8e1",
"reference": "3ffcc226cbc5b8747cdc5d6b06ce887cdffee8e1",
"url": "https://api.github.com/repos/utopia-php/database/zipball/805efb95b19d555b6cfd2244e36bfe08289f24f5",
"reference": "805efb95b19d555b6cfd2244e36bfe08289f24f5",
"shasum": ""
},
"require": {
@@ -1208,7 +1208,7 @@
"ext-pdo": "*",
"php": ">=8.0",
"utopia-php/cache": "0.9.*",
"utopia-php/framework": "0.*.*",
"utopia-php/framework": "0.33.*",
"utopia-php/mongo": "0.3.*"
},
"require-dev": {
@@ -1241,9 +1241,9 @@
],
"support": {
"issues": "https://github.com/utopia-php/database/issues",
"source": "https://github.com/utopia-php/database/tree/0.45.7"
"source": "https://github.com/utopia-php/database/tree/0.45.12"
},
"time": "2024-03-13T18:03:15+00:00"
"time": "2024-04-08T07:50:55+00:00"
},
{
"name": "utopia-php/domains",
@@ -2273,26 +2273,27 @@
},
{
"name": "utopia-php/websocket",
"version": "0.1.0",
"version": "0.2.0",
"source": {
"type": "git",
"url": "https://github.com/utopia-php/websocket.git",
"reference": "51fcb86171400d8aa40d76c54593481fd273dab5"
"reference": "e9d0919b321744a61f12563f5791c47ba9f57810"
},
"dist": {
"type": "zip",
"url": "https://api.github.com/repos/utopia-php/websocket/zipball/51fcb86171400d8aa40d76c54593481fd273dab5",
"reference": "51fcb86171400d8aa40d76c54593481fd273dab5",
"url": "https://api.github.com/repos/utopia-php/websocket/zipball/e9d0919b321744a61f12563f5791c47ba9f57810",
"reference": "e9d0919b321744a61f12563f5791c47ba9f57810",
"shasum": ""
},
"require": {
"php": ">=8.0"
},
"require-dev": {
"laravel/pint": "^1.15",
"phpstan/phpstan": "^1.8",
"phpunit/phpunit": "^9.5.5",
"swoole/ide-helper": "4.6.6",
"swoole/ide-helper": "5.1.2",
"textalk/websocket": "1.5.2",
"vimeo/psalm": "^4.8.1",
"workerman/workerman": "^4.0"
},
"type": "library",
@@ -2305,16 +2306,6 @@
"license": [
"MIT"
],
"authors": [
{
"name": "Eldad Fux",
"email": "eldad@appwrite.io"
},
{
"name": "Torsten Dittmann",
"email": "torsten@appwrite.io"
}
],
"description": "A simple abstraction for WebSocket servers.",
"keywords": [
"framework",
@@ -2325,9 +2316,9 @@
],
"support": {
"issues": "https://github.com/utopia-php/websocket/issues",
"source": "https://github.com/utopia-php/websocket/tree/0.1.0"
"source": "https://github.com/utopia-php/websocket/tree/0.2.0"
},
"time": "2021-12-20T10:50:09+00:00"
"time": "2024-04-09T08:28:11+00:00"
},
{
"name": "webmozart/assert",
docker-compose.yml
+1 -1
View File
@@ -751,7 +751,7 @@ services:
hostname: appwrite-executor
<<: *x-logging
stop_signal: SIGINT
image: openruntimes/executor:0.4.5
image: openruntimes/executor:0.5.1
restart: unless-stopped
networks:
- appwrite
src/Appwrite/Hooks/Hooks.php
+5 -2
View File
@@ -16,11 +16,14 @@ class Hooks
/**
* @param mixed[] $params
* @return mixed
*/
public function trigger(string $name, array $params = [])
public function trigger(string $name, array $params = []): mixed
{
if (isset(self::$hooks[$name])) {
call_user_func_array(self::$hooks[$name], $params);
return call_user_func_array(self::$hooks[$name], $params);
}
return null;
}
}
src/Appwrite/Platform/Workers/Functions.php
+79 -8
View File
@@ -194,6 +194,70 @@ class Functions extends Action
}
}
/**
* @param string $message
* @param Document $function
* @param string $trigger
* @param string $path
* @param string $method
* @param Document $user
* @param string|null $jwt
* @param string|null $event
* @throws Exception
*/
private function fail(
string $message,
Database $dbForProject,
Document $function,
string $trigger,
string $path,
string $method,
Document $user,
string $jwt = null,
string $event = null,
): void {
$headers['x-appwrite-trigger'] = $trigger;
$headers['x-appwrite-event'] = $event ?? '';
$headers['x-appwrite-user-id'] = $user->getId() ?? '';
$headers['x-appwrite-user-jwt'] = $jwt ?? '';
$headersFiltered = [];
foreach ($headers as $key => $value) {
if (\in_array(\strtolower($key), FUNCTION_ALLOWLIST_HEADERS_REQUEST)) {
$headersFiltered[] = ['name' => $key, 'value' => $value];
}
}
$executionId = ID::unique();
$execution = new Document([
'$id' => $executionId,
'$permissions' => $user->isEmpty() ? [] : [Permission::read(Role::user($user->getId()))],
'functionInternalId' => $function->getInternalId(),
'functionId' => $function->getId(),
'deploymentInternalId' => '',
'deploymentId' => '',
'trigger' => $trigger,
'status' => 'failed',
'responseStatusCode' => 0,
'responseHeaders' => [],
'requestPath' => $path,
'requestMethod' => $method,
'requestHeaders' => $headersFiltered,
'errors' => $message,
'logs' => '',
'duration' => 0.0,
'search' => implode(' ', [$function->getId(), $executionId]),
]);
if ($function->getAttribute('logging')) {
$execution = $dbForProject->createDocument('executions', $execution);
}
if ($execution->isEmpty()) {
throw new Exception('Failed to create execution');
}
}
/**
* @param Log $log
* @param Database $dbForProject
@@ -248,24 +312,32 @@ class Functions extends Action
$deployment = $dbForProject->getDocument('deployments', $deploymentId);
if ($deployment->getAttribute('resourceId') !== $functionId) {
throw new Exception('Deployment not found. Create deployment before trying to execute a function');
$errorMessage = 'The execution could not be completed because a corresponding deployment was not found. A function deployment needs to be created before it can be executed. Please create a deployment for your function and try again.';
$this->fail($errorMessage, $dbForProject, $function, $trigger, $path, $method, $user, $jwt, $event);
return;
}
if ($deployment->isEmpty()) {
throw new Exception('Deployment not found. Create deployment before trying to execute a function');
$errorMessage = 'The execution could not be completed because a corresponding deployment was not found. A function deployment needs to be created before it can be executed. Please create a deployment for your function and try again.';
$this->fail($errorMessage, $dbForProject, $function, $trigger, $path, $method, $user, $jwt, $event);
return;
}
/** Check if build has exists */
$build = $dbForProject->getDocument('builds', $deployment->getAttribute('buildId', ''));
/** Check if the build exists */
$build = $dbForProject->getDocument('builds', $deployment->getAttribute('buildId', ''));
if ($build->isEmpty()) {
throw new Exception('Build not found');
$errorMessage = 'The execution could not be completed because a corresponding deployment was not found. A function deployment needs to be created before it can be executed. Please create a deployment for your function and try again.';
$this->fail($errorMessage, $dbForProject, $function, $trigger, $path, $method, $user, $jwt, $event);
return;
}
if ($build->getAttribute('status') !== 'ready') {
throw new Exception('Build not ready');
$errorMessage = 'The execution could not be completed because the build is not ready. Please wait for the build to complete and try again.';
$this->fail($errorMessage, $dbForProject, $function, $trigger, $path, $method, $user, $jwt, $event);
return;
}
/** Check if runtime is supported */
/** Check if runtime is supported */
$version = $function->getAttribute('version', 'v2');
$runtimes = Config::getParam($version === 'v2' ? 'runtimes-v2' : 'runtimes', []);
@@ -280,7 +352,6 @@ class Functions extends Action
$headers['x-appwrite-user-id'] = $user->getId() ?? '';
$headers['x-appwrite-user-jwt'] = $jwt ?? '';
/** Create execution or update execution status */
/** Create execution or update execution status */
$execution = $dbForProject->getDocument('executions', $executionId ?? '');
if ($execution->isEmpty()) {
src/Appwrite/Utopia/Pools/Connections.php
+52
View File
@@ -0,0 +1,52 @@
<?php
namespace Appwrite\Utopia\Pools;
use Utopia\Pools\Connection;
class Connections
{
/**
* @var array<Connection>
*/
protected array $connections = [];
/**
* @param Connection $connection
* @return self
*/
public function add(Connection $connection): self
{
$this->connections[$connection->getID()] = $connection;
return $this;
}
/**
* @param string $id
* @return self
*/
public function remove(string $id): self
{
unset($this->connections[$id]);
return $this;
}
public function count(): int
{
return \count($this->connections);
}
/**
* @return self
* @throws \Exception
*/
public function reclaim(): self
{
foreach ($this->connections as $id => $connection) {
$connection->reclaim();
unset($this->connections[$id]);
}
return $this;
}
}
tests/e2e/Services/Console/ModeTest.php
+30
View File
@@ -0,0 +1,30 @@
<?php
namespace Tests\E2E\Services\Console;
use Appwrite\Extend\Exception;
use Tests\E2E\Client;
use Tests\E2E\Scopes\ProjectConsole;
use Tests\E2E\Scopes\Scope;
use Tests\E2E\Scopes\SideClient;
class ModeTest extends Scope
{
use ProjectConsole;
use SideClient;
public function testConsoleWithAdminMode(): void
{
/**
* Test for SUCCESS
*/
$response = $this->client->call(Client::METHOD_GET, '/account', array_merge([
'content-type' => 'application/json',
'x-appwrite-project' => $this->getProject()['$id'],
'x-appwrite-mode' => 'admin',
], $this->getHeaders()));
$this->assertEquals(400, $response['headers']['status-code']);
$this->assertEquals(Exception::GENERAL_BAD_REQUEST, $response['body']['type']);
}
}
tests/unit/Utopia/Pools/ConnectionsTest.php
+40
View File
@@ -0,0 +1,40 @@
<?php
namespace Tests\Unit\Utopia\Pools;
use Appwrite\Utopia\Pools\Connections;
use PHPUnit\Framework\TestCase;
use Utopia\Pools\Connection;
use Utopia\Pools\Pool;
class ConnectionsTest extends TestCase
{
public function testAdd()
{
$connections = new Connections();
$connection = new Connection('resource');
$connections->add($connection);
$this->assertEquals(1, $connections->count());
}
public function testRemove()
{
$connections = new Connections();
$connection = new Connection('resource');
$connections->add($connection);
$connections->remove($connection->getID());
$this->assertEquals(0, $connections->count());
}
public function testReclaim()
{
$connections = new Connections();
$pool = new Pool('test', 1, function () {
return 'resource';
});
$connection = $pool->pop();
$connections->add($connection);
$connections->reclaim();
$this->assertEquals(1, $pool->count());
}
}