Merge pull request #7273 from appwrite/damodar/pea-15-refactor-maintenance-deletes-worker

PEA-15 Refactor Deletes and maintenance worker

app/worker.php

@@ -72,6 +72,17 @@ Server::setResource('dbForProject', function (Cache $cache, Registry $register,
     return $adapter;
 }, ['cache', 'register', 'message', 'dbForConsole']);
 
+Server::setResource('project', function (Message $message, Database $dbForConsole) {
+    $payload = $message->getPayload() ?? [];
+    $project = new Document($payload['project'] ?? []);
+
+    if ($project->getId() === 'console') {
+        return $project;
+    }
+    return $dbForConsole->getDocument('projects', $project->getId());
+    ;
+}, ['message', 'dbForConsole']);
+
 Server::setResource('getProjectDB', function (Group $pools, Database $dbForConsole, $cache) {
     $databases = []; // TODO: @Meldiron This should probably be responsibility of utopia-php/pools
 
@@ -103,22 +114,16 @@ Server::setResource('getProjectDB', function (Group $pools, Database $dbForConso
     };
 }, ['pools', 'dbForConsole', 'cache']);
 
-Server::setResource('getProjectAbuseRetention', function () {
+Server::setResource('abuseRetention', function () {
-    return function (Document $project) {
+    return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_ABUSE', 86400));
-        return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_ABUSE', 86400));
-    };
 });
 
-Server::setResource('getProjectAuditRetention', function () {
+Server::setResource('auditRetention', function () {
-    return function (Document $project) {
+    return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_AUDIT', 1209600));
-        return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_AUDIT', 1209600));
-    };
 });
 
-Server::setResource('getProjectExecutionRetention', function () {
+Server::setResource('executionRetention', function () {
-    return function (Document $project) {
+    return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_EXECUTION', 1209600));
-        return DateTime::addSeconds(new \DateTime(), -1 * App::getEnv('_APP_MAINTENANCE_RETENTION_EXECUTION', 1209600));
-    };
 });
 
 Server::setResource('cache', function (Registry $register) {

src/Appwrite/Platform/Tasks/Maintenance.php

@@ -44,18 +44,51 @@ class Maintenance extends Action
             $time = DateTime::now();
 
             Console::info("[{$time}] Notifying workers with maintenance tasks every {$interval} seconds");
-            $this->notifyDeleteExecutionLogs($queueForDeletes);
+
-            $this->notifyDeleteAbuseLogs($queueForDeletes);
+            $this->foreachProject($dbForConsole, function (Document $project) use ($queueForDeletes, $usageStatsRetentionHourly) {
-            $this->notifyDeleteAuditLogs($queueForDeletes);
+                $queueForDeletes->setProject($project);
-            $this->notifyDeleteUsageStats($usageStatsRetentionHourly, $queueForDeletes);
+
+                $this->notifyDeleteExecutionLogs($queueForDeletes);
+                $this->notifyDeleteAbuseLogs($queueForDeletes);
+                $this->notifyDeleteAuditLogs($queueForDeletes);
+                $this->notifyDeleteUsageStats($usageStatsRetentionHourly, $queueForDeletes);
+                $this->notifyDeleteExpiredSessions($queueForDeletes);
+            });
+
             $this->notifyDeleteConnections($queueForDeletes);
-            $this->notifyDeleteExpiredSessions($queueForDeletes);
             $this->renewCertificates($dbForConsole, $queueForCertificates);
             $this->notifyDeleteCache($cacheRetention, $queueForDeletes);
             $this->notifyDeleteSchedules($schedulesDeletionRetention, $queueForDeletes);
         }, $interval);
     }
 
+    protected function foreachProject(Database $dbForConsole, callable $callback): void
+    {
+        // TODO: @Meldiron name of this method no longer matches. It does not delete, and it gives whole document
+        $count = 0;
+        $chunk = 0;
+        $limit = 50;
+        $sum = $limit;
+        $executionStart = \microtime(true);
+
+        while ($sum === $limit) {
+            $projects = $dbForConsole->find('projects', [Query::limit($limit), Query::offset($chunk * $limit)]);
+
+            $chunk++;
+
+            /** @var string[] $projectIds */
+            $sum = count($projects);
+
+            foreach ($projects as $project) {
+                $callback($project);
+                $count++;
+            }
+        }
+
+        $executionEnd = \microtime(true);
+        Console::info("Found {$count} projects " . ($executionEnd - $executionStart) . " seconds");
+    }
+
     private function notifyDeleteExecutionLogs(Delete $queueForDeletes): void
     {
         ($queueForDeletes)

src/Appwrite/Platform/Workers/Deletes.php

@@ -2,6 +2,7 @@
 
 namespace Appwrite\Platform\Workers;
 
+use Appwrite\Auth\Auth;
 use Executor\Executor;
 use Throwable;
 use Utopia\Abuse\Abuse;
@@ -45,17 +46,17 @@ class Deletes extends Action
             ->inject('getFunctionsDevice')
             ->inject('getBuildsDevice')
             ->inject('getCacheDevice')
-            ->inject('getProjectAbuseRetention')
+            ->inject('abuseRetention')
-            ->inject('getProjectExecutionRetention')
+            ->inject('executionRetention')
-            ->inject('getProjectAuditRetention')
+            ->inject('auditRetention')
-            ->callback(fn ($message, $dbForConsole, callable $getProjectDB, callable $getFilesDevice, callable $getFunctionsDevice, callable $getBuildsDevice, callable $getCacheDevice, callable $getProjectAbuseRetention, callable $getProjectExecutionRetention, callable $getProjectAuditRetention) => $this->action($message, $dbForConsole, $getProjectDB, $getFilesDevice, $getFunctionsDevice, $getBuildsDevice, $getCacheDevice, $getProjectAbuseRetention, $getProjectExecutionRetention, $getProjectAuditRetention));
+            ->callback(fn ($message, $dbForConsole, callable $getProjectDB, callable $getFilesDevice, callable $getFunctionsDevice, callable $getBuildsDevice, callable $getCacheDevice, string $abuseRetention, string $executionRetention, string $auditRetention) => $this->action($message, $dbForConsole, $getProjectDB, $getFilesDevice, $getFunctionsDevice, $getBuildsDevice, $getCacheDevice, $abuseRetention, $executionRetention, $auditRetention));
     }
 
     /**
      * @throws Exception
      * @throws Throwable
      */
-    public function action(Message $message, Database $dbForConsole, callable $getProjectDB, callable $getFilesDevice, callable $getFunctionsDevice, callable $getBuildsDevice, callable $getCacheDevice, callable $getProjectAbuseRetention, callable $getProjectExecutionRetention, callable $getProjectAuditRetention): void
+    public function action(Message $message, Database $dbForConsole, callable $getProjectDB, callable $getFilesDevice, callable $getFunctionsDevice, callable $getBuildsDevice, callable $getCacheDevice, string $abuseRetention, string $executionRetention, string $auditRetention): void
     {
         $payload = $message->getPayload() ?? [];
 
@@ -117,12 +118,12 @@ class Deletes extends Action
                 break;
 
             case DELETE_TYPE_EXECUTIONS:
-                $this->deleteExecutionLogs($dbForConsole, $getProjectDB, $getProjectExecutionRetention);
+                $this->deleteExecutionLogs($project, $getProjectDB, $executionRetention);
                 break;
 
             case DELETE_TYPE_AUDIT:
-                if (!empty($datetime)) {
+                if (!$project->isEmpty()) {
-                    $this->deleteAuditLogs($dbForConsole, $getProjectDB, $getProjectAuditRetention);
+                    $this->deleteAuditLogs($project, $getProjectDB, $auditRetention);
                 }
 
                 if (!$document->isEmpty()) {
@@ -130,7 +131,7 @@ class Deletes extends Action
                 }
                 break;
             case DELETE_TYPE_ABUSE:
-                $this->deleteAbuseLogs($dbForConsole, $getProjectDB, $getProjectAbuseRetention);
+                $this->deleteAbuseLogs($project, $getProjectDB, $abuseRetention);
                 break;
 
             case DELETE_TYPE_REALTIME:
@@ -138,10 +139,10 @@ class Deletes extends Action
                 break;
 
             case DELETE_TYPE_SESSIONS:
-                $this->deleteExpiredSessions($dbForConsole, $getProjectDB);
+                $this->deleteExpiredSessions($project, $getProjectDB);
                 break;
             case DELETE_TYPE_USAGE:
-                $this->deleteUsageStats($dbForConsole, $getProjectDB, $hourlyUsageRetentionDatetime);
+                $this->deleteUsageStats($project, $getProjectDB, $hourlyUsageRetentionDatetime);
                 break;
             case DELETE_TYPE_CACHE_BY_RESOURCE:
                 $this->deleteCacheByResource($project, $getProjectDB, $resource);
@@ -340,16 +341,14 @@ class Deletes extends Action
      * @return void
      * @throws Exception
      */
-    private function deleteUsageStats(Database $dbForConsole, callable $getProjectDB, string $hourlyUsageRetentionDatetime): void
+    private function deleteUsageStats(Document $project, callable $getProjectDB, string $hourlyUsageRetentionDatetime): void
     {
-        $this->deleteForProjectIds($dbForConsole, function (Document $project) use ($getProjectDB, $hourlyUsageRetentionDatetime) {
+        $dbForProject = $getProjectDB($project);
-            $dbForProject = $getProjectDB($project);
+        // Delete Usage stats
-            // Delete Usage stats
+        $this->deleteByGroup('stats', [
-            $this->deleteByGroup('stats', [
+            Query::lessThan('time', $hourlyUsageRetentionDatetime),
-                Query::lessThan('time', $hourlyUsageRetentionDatetime),
+            Query::equal('period', ['1h']),
-                Query::equal('period', ['1h']),
+        ], $dbForProject);
-            ], $dbForProject);
-        });
     }
 
     /**
@@ -546,16 +545,14 @@ class Deletes extends Action
      * @return void
      * @throws Exception
      */
-    private function deleteExecutionLogs(database $dbForConsole, callable $getProjectDB, callable $getProjectExecutionRetention): void
+    private function deleteExecutionLogs(Document $project, callable $getProjectDB, callable $getProjectExecutionRetention): void
     {
-        $this->deleteForProjectIds($dbForConsole, function (Document $project) use ($getProjectDB, $getProjectExecutionRetention) {
+        $dbForProject = $getProjectDB($project);
-            $dbForProject = $getProjectDB($project);
+        $datetime = $getProjectExecutionRetention($project);
-            $datetime = $getProjectExecutionRetention($project);
+        // Delete Executions
-            // Delete Executions
+        $this->deleteByGroup('executions', [
-            $this->deleteByGroup('executions', [
+            Query::lessThan('$createdAt', $datetime)
-                Query::lessThan('$createdAt', $datetime)
+        ], $dbForProject);
-            ], $dbForProject);
-        });
     }
 
     /**
@@ -564,20 +561,16 @@ class Deletes extends Action
      * @return void
      * @throws Exception|Throwable
      */
-    private function deleteExpiredSessions(Database $dbForConsole, callable $getProjectDB): void
+    private function deleteExpiredSessions(Document $project, callable $getProjectDB): void
     {
+        $dbForProject = $getProjectDB($project);
+        $duration = $project->getAttribute('auths', [])['duration'] ?? Auth::TOKEN_EXPIRATION_LOGIN_LONG;
+        $expired = DateTime::addSeconds(new \DateTime(), -1 * $duration);
 
-        $this->deleteForProjectIds($dbForConsole, function (Document $project) use ($dbForConsole, $getProjectDB) {
+        // Delete Sessions
-            $dbForProject = $getProjectDB($project);
+        $this->deleteByGroup('sessions', [
-            $project = $dbForConsole->getDocument('projects', $project->getId());
+            Query::lessThan('$createdAt', $expired)
-            $duration = $project->getAttribute('auths', [])['duration'] ?? Auth::TOKEN_EXPIRATION_LOGIN_LONG;
+        ], $dbForProject);
-            $expired = DateTime::addSeconds(new \DateTime(), -1 * $duration);
-
-            // Delete Sessions
-            $this->deleteByGroup('sessions', [
-                Query::lessThan('$createdAt', $expired)
-            ], $dbForProject);
-        });
     }
 
     /**
@@ -601,19 +594,16 @@ class Deletes extends Action
      * @return void
      * @throws Exception
      */
-    private function deleteAbuseLogs(Database $dbForConsole, callable $getProjectDB, callable $getProjectAbuseRetention): void
+    private function deleteAbuseLogs(Document $project, callable $getProjectDB, string $abuseRetention): void
     {
-        $this->deleteForProjectIds($dbForConsole, function (Document $project) use ($getProjectDB, $getProjectAbuseRetention) {
+        $projectId = $project->getId();
-            $projectId = $project->getId();
+        $dbForProject = $getProjectDB($project);
-            $dbForProject = $getProjectDB($project);
+        $timeLimit = new TimeLimit("", 0, 1, $dbForProject);
-            $datetime = $getProjectAbuseRetention($project);
+        $abuse = new Abuse($timeLimit);
-            $timeLimit = new TimeLimit("", 0, 1, $dbForProject);
+        $status = $abuse->cleanup($abuseRetention);
-            $abuse = new Abuse($timeLimit);
+        if (!$status) {
-            $status = $abuse->cleanup($datetime);
+            throw new Exception('Failed to delete Abuse logs for project ' . $projectId);
-            if (!$status) {
+        }
-                throw new Exception('Failed to delete Abuse logs for project ' . $projectId);
-            }
-        });
     }
 
     /**
@@ -623,18 +613,15 @@ class Deletes extends Action
      * @return void
      * @throws Exception
      */
-    private function deleteAuditLogs(Database $dbForConsole, callable $getProjectDB, callable $getProjectAuditRetention): void
+    private function deleteAuditLogs(Document $project, callable $getProjectDB, string $auditRetention): void
     {
-        $this->deleteForProjectIds($dbForConsole, function (Document $project) use ($getProjectDB, $getProjectAuditRetention) {
+        $projectId = $project->getId();
-            $projectId = $project->getId();
+        $dbForProject = $getProjectDB($project);
-            $dbForProject = $getProjectDB($project);
+        $audit = new Audit($dbForProject);
-            $datetime = $getProjectAuditRetention($project);
+        $status = $audit->cleanup($auditRetention);
-            $audit = new Audit($dbForProject);
+        if (!$status) {
-            $status = $audit->cleanup($datetime);
+            throw new Exception('Failed to delete Audit logs for project' . $projectId);
-            if (!$status) {
+        }
-                throw new Exception('Failed to delete Audit logs for project' . $projectId);
-            }
-        });
     }
 
     /**
@@ -869,39 +856,6 @@ class Deletes extends Action
         }
     }
 
-    /**
-     * @param Database $dbForConsole
-     * @param callable $callback
-     * @throws Exception
-     */
-    private function deleteForProjectIds(database $dbForConsole, callable $callback): void
-    {
-        // TODO: @Meldiron name of this method no longer matches. It does not delete, and it gives whole document
-        $count = 0;
-        $chunk = 0;
-        $limit = 50;
-        $sum = $limit;
-        $executionStart = \microtime(true);
-
-        while ($sum === $limit) {
-            $projects = $dbForConsole->find('projects', [Query::limit($limit), Query::offset($chunk * $limit)]);
-
-            $chunk++;
-
-            /** @var string[] $projectIds */
-            $sum = count($projects);
-
-            Console::info('Executing delete function for chunk #' . $chunk . '. Found ' . $sum . ' projects');
-            foreach ($projects as $project) {
-                $callback($project);
-                $count++;
-            }
-        }
-
-        $executionEnd = \microtime(true);
-        Console::info("Found {$count} projects " . ($executionEnd - $executionStart) . " seconds");
-    }
-
     /**
      * @param string $collection collectionID
      * @param array $queries