Merge Audits

app/cli.php

@@ -5,6 +5,7 @@ require_once __DIR__ . '/controllers/general.php';
 
 use Appwrite\Event\Delete;
 use Appwrite\Event\Func;
+use Appwrite\Event\Audit;
 use Appwrite\Platform\Appwrite;
 use Utopia\CLI\CLI;
 use Utopia\Database\Validator\Authorization;
@@ -154,6 +155,10 @@ CLI::setResource('queueForFunctions', function (Connection $queue) {
     return new Func($queue);
 }, ['queue']);
 
+CLI::setResource('audits', function (Connection $queue) {
+    return new Audit($queue);
+}, ['queue']);
+
 CLI::setResource('logError', function (Registry $register) {
     return function (Throwable $error, string $namespace, string $action) use ($register) {
         $logger = $register->get('logger');

app/init.php

@@ -848,7 +848,6 @@ App::setResource('register', fn() => $register);
 App::setResource('locale', fn() => new Locale(App::getEnv('_APP_LOCALE', 'en')));
 
 // Queues
-App::setResource('audits', fn() => new Audit());
 App::setResource('mails', fn() => new Mail());
 App::setResource('deletes', function (Connection $queue) {
     return new Delete($queue);
@@ -858,6 +857,12 @@ App::setResource('messaging', fn() => new Phone());
 App::setResource('queue', function (Group $pools) {
     return $pools->get('queue')->pop()->getResource();
 }, ['pools']);
+App::setResource('events', function (Connection $queue) {
+    return new Event('', '', $queue);
+}, ['pools']);
+App::setResource('audits', function (Connection $queue) {
+    return new Audit($queue);
+}, ['pools']);
 App::setResource('events', function (Connection $queue) {
     return new Event('', '', $queue);
 }, ['queue']);

app/worker.php

@@ -3,6 +3,7 @@
 require_once __DIR__ . '/init.php';
 
 use Appwrite\Event\Event;
+use Appwrite\Event\Audit;
 use Appwrite\Event\Func;
 use Swoole\Runtime;
 use Utopia\App;
@@ -24,15 +25,14 @@ Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
 
 global $register;
 
-Server::setResource('register', fn() => $register);
+Server::setResource('register', fn () => $register);
 
 Server::setResource('dbForConsole', function (Cache $cache, Registry $register) {
     $pools = $register->get('pools');
     $database = $pools
         ->get('console')
         ->pop()
-        ->getResource()
+        ->getResource();
-    ;
 
     $adapter = new Database($database, $cache);
     $adapter->setNamespace('console');
@@ -52,8 +52,7 @@ Server::setResource('dbForProject', function (Cache $cache, Registry $register,
     $database = $pools
         ->get($project->getAttribute('database'))
         ->pop()
-        ->getResource()
+        ->getResource();
-    ;
 
     $adapter = new Database($database, $cache);
     $adapter->setNamespace('_' . $project->getInternalId());
@@ -69,8 +68,7 @@ Server::setResource('cache', function (Registry $register) {
         $adapters[] = $pools
             ->get($value)
             ->pop()
-            ->getResource()
+            ->getResource();
-        ;
     }
 
     return new Cache(new Sharding($adapters));
@@ -98,6 +96,16 @@ Server::setResource('events', function (Registry $register) {
     );
 }, ['register']);
 
+Server::setResource('audits', function (Registry $register) {
+    $pools = $register->get('pools');
+    return new Audit(
+        $pools
+            ->get('queue')
+            ->pop()
+            ->getResource()
+    );
+}, ['register']);
+
 Server::setResource('logger', function ($register) {
     return $register->get('logger');
 }, ['register']);

app/workers/audits.php

@@ -1,42 +1,32 @@
 <?php
 
-use Appwrite\Resque\Worker;
+require_once __DIR__ . '/../worker.php';
+
 use Utopia\Audit\Audit;
-use Utopia\CLI\Console;
+use Utopia\Database\Database;
 use Utopia\Database\Document;
+use Utopia\Database\Validator\Authorization;
+use Utopia\Queue\Message;
+use Utopia\Queue\Server;
 
-require_once __DIR__ . '/../init.php';
+Authorization::disable();
-
+Authorization::setDefaultStatus(false);
-Console::title('Audits V1 Worker');
-Console::success(APP_NAME . ' audits worker v1 has started');
-
-class AuditsV1 extends Worker
-{
-    public function getName(): string
-    {
-        return "audits";
-    }
-
-    public function init(): void
-    {
-    }
-
-    public function run(): void
-    {
-        $event = $this->args['event'];
-        $payload = $this->args['payload'];
-        $mode = $this->args['mode'];
-        $resource = $this->args['resource'];
-        $userAgent = $this->args['userAgent'];
-        $ip = $this->args['ip'];
-
-        $user = new Document($this->args['user']);
-        $project = new Document($this->args['project']);
 
+Server::setResource('execute', function () {
+    return function (
+        Database $dbForProject,
+        string $event,
+        array $payload,
+        string $mode,
+        string $resource,
+        string $userAgent,
+        string $ip,
+        Document $user,
+        Document $project
+    ) {
         $userName = $user->getAttribute('name', '');
         $userEmail = $user->getAttribute('email', '');
 
-        $dbForProject = $this->getProjectDB($project);
         $audit = new Audit($dbForProject);
         $audit->log(
             userId: $user->getId(),
@@ -53,9 +43,41 @@ class AuditsV1 extends Worker
                 'data' => $payload,
             ]
         );
-    }
+    };
+});
 
-    public function shutdown(): void
+$server->job()
-    {
+    ->inject('message')
-    }
+    ->inject('dbForProject')
-}
+    ->inject('execute')
+    ->action(function (Message $message, Database $dbForProject, callable $execute) {
+        $payload = $message->getPayload() ?? [];
+
+        if (empty($payload)) {
+            throw new Exception('Missing payload');
+        }
+
+        $event = $payload['event'] ?? '';
+        $auditPayload = $payload['payload'] ?? '';
+        $mode = $payload['mode'] ?? '';
+        $resource = $payload['resource'] ?? '';
+        $userAgent = $payload['userAgent'] ?? '';
+        $ip = $payload['ip'] ?? '';
+        $project = new Document($payload['project'] ?? []);
+        $user = new Document($payload['user'] ?? []);
+
+        $execute(
+            $dbForProject,
+            $event,
+            $auditPayload,
+            $mode,
+            $resource,
+            $userAgent,
+            $ip,
+            $user,
+            $project
+        );
+    });
+
+$server->workerStart();
+$server->start();

src/Appwrite/Event/Audit.php

@@ -3,6 +3,8 @@
 namespace Appwrite\Event;
 
 use Resque;
+use Utopia\Queue\Client;
+use Utopia\Queue\Connection;
 
 class Audit extends Event
 {
@@ -11,7 +13,7 @@ class Audit extends Event
     protected string $userAgent = '';
     protected string $ip = '';
 
-    public function __construct()
+    public function __construct(protected Connection $connection)
     {
         parent::__construct(Event::AUDITS_QUEUE_NAME, Event::AUDITS_CLASS_NAME);
     }
@@ -116,7 +118,9 @@ class Audit extends Event
      */
     public function trigger(): string|bool
     {
-        return Resque::enqueue($this->queue, $this->class, [
+        $client = new Client($this->queue, $this->connection);
+
+        return $client->enqueue([
             'project' => $this->project,
             'user' => $this->user,
             'payload' => $this->payload,