diff --git a/Dockerfile b/Dockerfile index 88d5ed030b..4b5ac3fc62 100755 --- a/Dockerfile +++ b/Dockerfile @@ -44,12 +44,14 @@ COPY ./dev /usr/src/code/dev # Set Volumes RUN mkdir -p /storage/uploads && \ + mkdir -p /storage/csv-imports && \ mkdir -p /storage/cache && \ mkdir -p /storage/config && \ mkdir -p /storage/certificates && \ mkdir -p /storage/functions && \ mkdir -p /storage/debug && \ chown -Rf www-data.www-data /storage/uploads && chmod -Rf 0755 /storage/uploads && \ + chown -Rf www-data.www-data /storage/csv-imports && chmod -Rf 0755 /storage/csv-imports && \ chown -Rf www-data.www-data /storage/cache && chmod -Rf 0755 /storage/cache && \ chown -Rf www-data.www-data /storage/config && chmod -Rf 0755 /storage/config && \ chown -Rf www-data.www-data /storage/certificates && chmod -Rf 0755 /storage/certificates && \ diff --git a/app/config/collections/projects.php b/app/config/collections/projects.php index 851b467159..d28b2a7e38 100644 --- a/app/config/collections/projects.php +++ b/app/config/collections/projects.php @@ -1894,6 +1894,13 @@ return [ 'lengths' => [Database::LENGTH_KEY], 'orders' => [Database::ORDER_ASC], ], + [ + '$id' => '_key_resource_id', + 'type' => Database::INDEX_KEY, + 'attributes' => ['resourceId'], + 'lengths' => [Database::LENGTH_KEY], + 'orders' => [Database::ORDER_DESC], + ], [ '$id' => ID::custom('_fulltext_search'), 'type' => Database::INDEX_FULLTEXT, @@ -1935,7 +1942,7 @@ return [ '$id' => ID::custom('status'), 'type' => Database::VAR_STRING, 'format' => '', - 'size' => 16, + 'size' => Database::LENGTH_KEY, 'signed' => true, 'required' => false, 'default' => null, @@ -1987,14 +1994,14 @@ return [ 'filters' => [], ], [ - '$id' => ID::custom('error'), + '$id' => ID::custom('errors'), 'type' => Database::VAR_STRING, 'format' => '', - 'size' => 2048, + 'size' => 65535, 'signed' => true, - 'required' => false, + 'required' => true, 'default' => null, - 'array' => false, + 'array' => true, 'filters' => [], ], ], diff --git a/app/controllers/api/migrations.php b/app/controllers/api/migrations.php index 10fc48bd33..90f0930ff2 100644 --- a/app/controllers/api/migrations.php +++ b/app/controllers/api/migrations.php @@ -4,12 +4,12 @@ use Appwrite\Auth\Auth; use Appwrite\Event\Event; use Appwrite\Event\Migration; use Appwrite\Extend\Exception; +use Appwrite\OpenSSL\OpenSSL; use Appwrite\SDK\AuthType; use Appwrite\SDK\ContentType; use Appwrite\SDK\Method; use Appwrite\SDK\Response as SDKResponse; use Appwrite\Utopia\Database\Validator\Queries\Migrations; -use Appwrite\Utopia\Request; use Appwrite\Utopia\Response; use Utopia\App; use Utopia\Database\Database; @@ -22,10 +22,16 @@ use Utopia\Database\Validator\Query\Cursor; use Utopia\Database\Validator\UID; use Utopia\Migration\Resource; use Utopia\Migration\Sources\Appwrite; +use Utopia\Migration\Sources\Csv; use Utopia\Migration\Sources\Firebase; use Utopia\Migration\Sources\NHost; use Utopia\Migration\Sources\Supabase; +use Utopia\Migration\Transfer; +use Utopia\Storage\Compression\Algorithms\GZIP; +use Utopia\Storage\Compression\Algorithms\Zstd; +use Utopia\Storage\Compression\Compression; use Utopia\Storage\Device; +use Utopia\System\System; use Utopia\Validator\ArrayList; use Utopia\Validator\Integer; use Utopia\Validator\Text; @@ -314,28 +320,38 @@ App::post('/v1/migrations/csv') )) ->param('bucketId', '', new UID(), 'Storage bucket unique ID. You can create a new storage bucket using the Storage service [server integration](https://appwrite.io/docs/server/storage#createBucket).') ->param('fileId', '', new UID(), 'File ID.') - ->param('resourceId', null, new UID(), 'Composite ID in the format {databaseId:collectionId}, identifying a collection within a database.') - ->inject('request') + ->param('resourceId', null, new Text(75), 'Composite ID in the format {databaseId:collectionId}, identifying a collection within a database.') ->inject('response') ->inject('dbForProject') ->inject('project') ->inject('deviceForFiles') - ->inject('deviceForLocal') - ->inject('$queueForEvents') + ->inject('deviceForCsvImports') + ->inject('queueForEvents') ->inject('queueForMigrations') - ->action(function (string $bucketId, string $fileId, string $resourceId, Request $request, Response $response, Database $dbForProject, Document $project, Device $deviceForFiles, Migration $queueForEvents, Migration $queueForMigrations) { - - // TODO: Check if there's already a migrations worker process running for CSV Import for the same collection. - // If so, short-circuit and cancel the task early on because console may not allow it but API will. - // if (inProgress(resourceId)) { - // throw some exception. - //} - - $bucket = Authorization::skip(fn () => $dbForProject->getDocument('buckets', $bucketId)); - + ->action(function (string $bucketId, string $fileId, string $resourceId, Response $response, Database $dbForProject, Document $project, Device $deviceForFiles, Device $deviceForCsvImports, Event $queueForEvents, Migration $queueForMigrations) { $isAPIKey = Auth::isAppUser(Authorization::getRoles()); $isPrivilegedUser = Auth::isPrivilegedUser(Authorization::getRoles()); + // Check if migration/import is already in progress! + // if for some reason the worker crashes, the stage will always be `init`, what do we do? + $isInProgress = Authorization::skip(function () use ($dbForProject, $resourceId) { + $exists = $dbForProject->findOne( + 'migrations', + [ + Query::notEqual('stage', 'finished'), + Query::equal('resourceId', [$resourceId]), + ] + ); + + return !$exists->isEmpty(); + }); + + if ($isInProgress || (!$isAPIKey && !$isPrivilegedUser)) { + throw new Exception(Exception::MIGRATION_IN_PROGRESS, 'An import is already in progress for this collection.'); + } + + $bucket = Authorization::skip(fn () => $dbForProject->getDocument('buckets', $bucketId)); + if ($bucket->isEmpty() || (!$isAPIKey && !$isPrivilegedUser)) { throw new Exception(Exception::STORAGE_BUCKET_NOT_FOUND); } @@ -346,36 +362,72 @@ App::post('/v1/migrations/csv') } $path = $file->getAttribute('path', ''); - if (!$deviceForFiles->exists($path)) { throw new Exception(Exception::STORAGE_FILE_NOT_FOUND, 'File not found in ' . $path); } - // TODO: send path migrations/csv worker + // read file content. + $source = $deviceForFiles->read($path); + + // decrypt + if (!empty($file->getAttribute('openSSLCipher'))) { + $source = OpenSSL::decrypt( + $source, + $file->getAttribute('openSSLCipher'), + System::getEnv('_APP_OPENSSL_KEY_V' . $file->getAttribute('openSSLVersion')), + 0, + \hex2bin($file->getAttribute('openSSLIV')), + \hex2bin($file->getAttribute('openSSLTag')) + ); + } + + // decompress + switch ($file->getAttribute('algorithm', Compression::NONE)) { + case Compression::ZSTD: + $compressor = new Zstd(); + $source = $compressor->decompress($source); + break; + case Compression::GZIP: + $compressor = new GZIP(); + $source = $compressor->decompress($source); + break; + } + + // copy to temporary folder + $migrationId = ID::unique(); + $path = $deviceForCsvImports->getRoot() . '/' . $migrationId . '_' . $fileId . '.csv'; + $deviceForCsvImports->write($path, $source, 'text/csv'); + $fileSize = $deviceForCsvImports->getFileSize($path); + $resources = Transfer::extractServices([Transfer::GROUP_DATABASES]); + $migration = $dbForProject->createDocument('migrations', new Document([ - '$id' => ID::unique(), + '$id' => $migrationId, 'status' => 'pending', 'stage' => 'init', - // TODO: add stuff to migration library - 'source' => SourcesCSV::getName(), - 'destination' => DestinationsCSV::getName(), - 'resources' => [Resource::TYPE_DOCUMENT], + 'source' => Csv::getName(), + 'destination' => Appwrite::class::getName(), + 'resources' => $resources, 'resourceId' => $resourceId, 'resourceType' => Resource::TYPE_DATABASE, 'statusCounters' => [], 'resourceData' => [], 'errors' => [], - 'credentials' => [], + 'credentials' => [ + 'path' => $path, + 'size' => $fileSize, + ], ])); - // TODO: use migrationId or importId? $queueForEvents->setParam('migrationId', $migration->getId()); - // Trigger Import $queueForMigrations ->setMigration($migration) ->setProject($project) ->trigger(); + + $response + ->setStatusCode(Response::STATUS_CODE_ACCEPTED) + ->dynamic($migration, Response::MODEL_MIGRATION); }); App::get('/v1/migrations') diff --git a/app/init/constants.php b/app/init/constants.php index 5e4edfd97d..2deeff1c95 100644 --- a/app/init/constants.php +++ b/app/init/constants.php @@ -49,6 +49,7 @@ const APP_STORAGE_UPLOADS = '/storage/uploads'; const APP_STORAGE_FUNCTIONS = '/storage/functions'; const APP_STORAGE_BUILDS = '/storage/builds'; const APP_STORAGE_CACHE = '/storage/cache'; +const APP_STORAGE_CSV_IMPORTS = '/storage/csv-imports'; // Temporary storage for csv imports const APP_STORAGE_CERTIFICATES = '/storage/certificates'; const APP_STORAGE_CONFIG = '/storage/config'; const APP_STORAGE_READ_BUFFER = 20 * (1000 * 1000); //20MB other names `APP_STORAGE_MEMORY_LIMIT`, `APP_STORAGE_MEMORY_BUFFER`, `APP_STORAGE_READ_LIMIT`, `APP_STORAGE_BUFFER_LIMIT` diff --git a/app/init/resources.php b/app/init/resources.php index 2360179913..6eae238fee 100644 --- a/app/init/resources.php +++ b/app/init/resources.php @@ -508,6 +508,10 @@ App::setResource('deviceForFiles', function ($project) { return getDevice(APP_STORAGE_UPLOADS . '/app-' . $project->getId()); }, ['project']); +App::setResource('deviceForCsvImports', function (Document $project) { + return getDevice(APP_STORAGE_CSV_IMPORTS . '/app-' . $project->getId()); +}, ['project']); + App::setResource('deviceForFunctions', function ($project) { return getDevice(APP_STORAGE_FUNCTIONS . '/app-' . $project->getId()); }, ['project']); diff --git a/app/worker.php b/app/worker.php index 6a51ee55be..4e865858a0 100644 --- a/app/worker.php +++ b/app/worker.php @@ -339,6 +339,10 @@ Server::setResource('pools', function (Registry $register) { return $register->get('pools'); }, ['register']); +Server::setResource('deviceForCsvImports', function (Document $project) { + return getDevice(APP_STORAGE_CSV_IMPORTS . '/app-' . $project->getId()); +}, ['project']); + Server::setResource('deviceForFunctions', function (Document $project) { return getDevice(APP_STORAGE_FUNCTIONS . '/app-' . $project->getId()); }, ['project']); diff --git a/composer.json b/composer.json index 3920351c06..4c26b19d1e 100644 --- a/composer.json +++ b/composer.json @@ -60,7 +60,7 @@ "utopia-php/locale": "0.4.*", "utopia-php/logger": "0.6.*", "utopia-php/messaging": "0.16.*", - "utopia-php/migration": "0.8.*", + "utopia-php/migration": "dev-feat-csv", "utopia-php/orchestration": "0.9.*", "utopia-php/platform": "0.7.*", "utopia-php/pools": "0.8.*", @@ -91,6 +91,12 @@ "laravel/pint": "1.*", "phpbench/phpbench": "1.*" }, + "repositories": [ + { + "type": "git", + "url": "https://github.com/utopia-php/migration" + } + ], "provide": { "ext-phpiredis": "*" }, diff --git a/composer.lock b/composer.lock index 8db4706bb5..63323ad236 100644 --- a/composer.lock +++ b/composer.lock @@ -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": "6a54c8bc4f9f14cd3883f55880864630", + "content-hash": "cfb5c437126bf194a6fe7225961c1582", "packages": [ { "name": "adhocore/jwt", @@ -1365,16 +1365,16 @@ }, { "name": "open-telemetry/sdk", - "version": "1.2.2", + "version": "1.2.3", "source": { "type": "git", "url": "https://github.com/opentelemetry-php/sdk.git", - "reference": "37eec0fe47ddd627911f318f29b6cd48196be0c0" + "reference": "0e7804c176c4b09d95b7985400aa38ce544cb7fc" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/opentelemetry-php/sdk/zipball/37eec0fe47ddd627911f318f29b6cd48196be0c0", - "reference": "37eec0fe47ddd627911f318f29b6cd48196be0c0", + "url": "https://api.github.com/repos/opentelemetry-php/sdk/zipball/0e7804c176c4b09d95b7985400aa38ce544cb7fc", + "reference": "0e7804c176c4b09d95b7985400aa38ce544cb7fc", "shasum": "" }, "require": { @@ -1451,7 +1451,7 @@ "issues": "https://github.com/open-telemetry/opentelemetry-php/issues", "source": "https://github.com/open-telemetry/opentelemetry-php" }, - "time": "2025-01-29T21:40:28+00:00" + "time": "2025-04-08T09:55:41+00:00" }, { "name": "open-telemetry/sem-conv", @@ -3951,17 +3951,11 @@ }, { "name": "utopia-php/migration", - "version": "0.8.4", + "version": "dev-feat-csv", "source": { "type": "git", - "url": "https://github.com/utopia-php/migration.git", - "reference": "845fd04ccf5e0edb03c184b864e0596080a432b8" - }, - "dist": { - "type": "zip", - "url": "https://api.github.com/repos/utopia-php/migration/zipball/845fd04ccf5e0edb03c184b864e0596080a432b8", - "reference": "845fd04ccf5e0edb03c184b864e0596080a432b8", - "shasum": "" + "url": "https://github.com/utopia-php/migration", + "reference": "5c4e6c61c393d176348e88b65730a14b52f4ea2e" }, "require": { "appwrite/appwrite": "11.*", @@ -3987,7 +3981,25 @@ "Utopia\\Migration\\": "src/Migration" } }, - "notification-url": "https://packagist.org/downloads/", + "autoload-dev": { + "psr-4": { + "Utopia\\Tests\\": "tests/Migration" + } + }, + "scripts": { + "test": [ + "./vendor/bin/phpunit" + ], + "lint": [ + "./vendor/bin/pint --test" + ], + "format": [ + "./vendor/bin/pint" + ], + "check": [ + "./vendor/bin/phpstan analyse --level 3 src tests --memory-limit 2G" + ] + }, "license": [ "MIT" ], @@ -3999,11 +4011,7 @@ "upf", "utopia" ], - "support": { - "issues": "https://github.com/utopia-php/migration/issues", - "source": "https://github.com/utopia-php/migration/tree/0.8.4" - }, - "time": "2025-03-28T02:08:22+00:00" + "time": "2025-04-08T08:38:41+00:00" }, { "name": "utopia-php/orchestration", @@ -8126,7 +8134,9 @@ ], "aliases": [], "minimum-stability": "stable", - "stability-flags": {}, + "stability-flags": { + "utopia-php/migration": 20 + }, "prefer-stable": false, "prefer-lowest": false, "platform": { diff --git a/docker-compose.yml b/docker-compose.yml index 8c8a364f30..4181cc6564 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -72,6 +72,7 @@ services: - traefik.http.routers.appwrite_api_https.tls=true volumes: - appwrite-uploads:/storage/uploads:rw + - appwrite-csv-imports:/storage/csv-imports:rw - appwrite-cache:/storage/cache:rw - appwrite-config:/storage/config:rw - appwrite-certificates:/storage/certificates:rw @@ -204,7 +205,7 @@ services: appwrite-console: <<: *x-logging container_name: appwrite-console - image: appwrite/console:5.2.53 + image: appwrite/console:5.2.56 restart: unless-stopped networks: - appwrite @@ -672,6 +673,8 @@ services: - ./app:/usr/src/code/app - ./src:/usr/src/code/src - ./tests:/usr/src/code/tests + # for csv import access + - appwrite-csv-imports:/storage/csv-imports:rw depends_on: - mariadb environment: @@ -1132,6 +1135,7 @@ volumes: appwrite-redis: appwrite-cache: appwrite-uploads: + appwrite-csv-imports: appwrite-certificates: appwrite-functions: appwrite-builds: diff --git a/src/Appwrite/Platform/Workers/Migrations.php b/src/Appwrite/Platform/Workers/Migrations.php index 4939dc8143..b670c79822 100644 --- a/src/Appwrite/Platform/Workers/Migrations.php +++ b/src/Appwrite/Platform/Workers/Migrations.php @@ -4,10 +4,12 @@ namespace Appwrite\Platform\Workers; use Ahc\Jwt\JWT; use Appwrite\Event\Realtime; +use Appwrite\ID; use Exception; use Utopia\CLI\Console; use Utopia\Config\Config; use Utopia\Database\Database; +use Utopia\Database\DateTime; use Utopia\Database\Document; use Utopia\Database\Exception\Authorization; use Utopia\Database\Exception\Conflict; @@ -18,12 +20,14 @@ use Utopia\Migration\Destinations\Appwrite as DestinationAppwrite; use Utopia\Migration\Exception as MigrationException; use Utopia\Migration\Source; use Utopia\Migration\Sources\Appwrite as SourceAppwrite; +use Utopia\Migration\Sources\Csv; use Utopia\Migration\Sources\Firebase; use Utopia\Migration\Sources\NHost; use Utopia\Migration\Sources\Supabase; use Utopia\Migration\Transfer; use Utopia\Platform\Action; use Utopia\Queue\Message; +use Utopia\Storage\Device; use Utopia\System\System; class Migrations extends Action @@ -32,6 +36,8 @@ class Migrations extends Action protected Database $dbForPlatform; + protected Device $deviceForCsvImports; + protected Document $project; /** @@ -57,15 +63,17 @@ class Migrations extends Action ->inject('dbForPlatform') ->inject('logError') ->inject('queueForRealtime') - ->callback(fn (Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime) => $this->action($message, $project, $dbForProject, $dbForPlatform, $logError, $queueForRealtime)); + ->inject('deviceForCsvImports') + ->callback(fn (Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime, Device $deviceForCsvImports) => $this->action($message, $project, $dbForProject, $dbForPlatform, $logError, $queueForRealtime, $deviceForCsvImports)); } /** * @throws Exception */ - public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime): void + public function action(Message $message, Document $project, Database $dbForProject, Database $dbForPlatform, callable $logError, Realtime $queueForRealtime, Device $deviceForCsvImports): void { $payload = $message->getPayload() ?? []; + $this->deviceForCsvImports = $deviceForCsvImports; if (empty($payload)) { throw new Exception('Missing payload'); @@ -99,6 +107,7 @@ class Migrations extends Action protected function processSource(Document $migration): Source { $source = $migration->getAttribute('source'); + $resourceId = $migration->getAttribute('resourceId'); $credentials = $migration->getAttribute('credentials'); return match ($source) { @@ -128,6 +137,11 @@ class Migrations extends Action $credentials['endpoint'] === 'http://localhost/v1' ? 'http://appwrite/v1' : $credentials['endpoint'], $credentials['apiKey'], ), + Csv::getName() => new Csv( + $resourceId, + $credentials['path'], + $this->deviceForCsvImports + ), default => throw new \Exception('Invalid source type'), }; } @@ -222,8 +236,23 @@ class Migrations extends Action $projectDocument = $this->dbForPlatform->getDocument('projects', $project->getId()); $tempAPIKey = $this->generateAPIKey($projectDocument); + $importDocument = null; $transfer = $source = $destination = null; + if ($migration->getAttribute('source') === Csv::getName()) { + $fileSize = $migration->getAttribute('credentials', [])['size'] ?? 0; + $importDocument = new Document([ + '$id' => ID::unique(), + 'size' => $fileSize, // uncompressed and decrypted file size + 'startedAt' => DateTime::now(), + 'migrationId' => $migration->getId(), + 'migrationInternalId' => $migration->getInternalId(), + 'resourceId' => $migration->getAttribute('resourceId', ''), + 'resourceType' => $migration->getAttribute('resourceType', ''), + 'errors' => [], + ]); + } + try { if ( $migration->getAttribute('source') === SourceAppwrite::getName() && @@ -337,6 +366,7 @@ class Migrations extends Action } $migration->setAttribute('errors', $errorMessages); + $importDocument?->setAttribute('errors', $errorMessages); } } finally { $this->updateMigrationDocument($migration, $projectDocument, $queueForRealtime); @@ -379,6 +409,12 @@ class Migrations extends Action $destination?->success(); $source?->success(); } + + if ($migration->getAttribute('source') === Csv::getName()) { + // make and save the import document to database + $importDocument->setAttribute('status', $migration->getAttribute('status', '')); + $this->dbForProject->createDocument('imports', $importDocument); + } } } }