Watch "processing" collection and process images

Dockerfile.processing

@@ -0,0 +1,31 @@
+FROM php:8.1-cli-alpine
+
+RUN apk add \
+  ffmpeg \
+  zlib zlib-dev \
+  libpng libpng-dev \
+  libjpeg-turbo libjpeg-turbo-dev \
+  libwebp libwebp-dev \
+  && rm -f /var/cache/apk/*
+
+RUN docker-php-ext-configure gd --with-jpeg --with-webp
+
+RUN docker-php-ext-install -j$(nproc) bcmath gd
+
+RUN mv "$PHP_INI_DIR/php.ini-production" "$PHP_INI_DIR/php.ini"
+
+RUN sed -ri -e 's!memory_limit = .+!memory_limit = -1!g' "$PHP_INI_DIR/php.ini"
+
+RUN curl -sSL https://baltocdn.com/xp-framework/xp-runners/distribution/downloads/e/entrypoint/xp-run-8.6.2.sh > /usr/bin/xp-run
+
+RUN mkdir /app
+
+COPY class.pth /app/
+
+COPY src/ /app/src/
+
+COPY vendor/ /app/vendor/
+
+WORKDIR /app
+
+CMD ["/bin/sh", "/usr/bin/xp-run", "xp.command.CmdRunner", "de.thekid.dialog.processing.Watch"]

src/main/php/de/thekid/dialog/api/Entries.php

@@ -42,6 +42,16 @@ public function create(#[Value] $user, string $id, #[Entity] array<string, mixed
     return $result->entry();
   }
 
+  #[Get('/{id:.+(/.+)?}/images/{name}')]
+  public function media(#[Value] $user, string $id, string $name) {
+    $f= new File($this->storage->folder($id), $name);
+    if ($f->exists()) {
+      return Response::ok()->stream($f->in(), $f->size());
+    } else {
+      return Response::error(404, 'No media named "'.$name.'" in '.$id);
+    }
+  }
+
   #[Put('/{id:.+(/.+)?}/images/{name}')]
   public function upload(#[Value] $user, string $id, string $name, #[Request] $req) {
 

src/main/php/de/thekid/dialog/import/LocalDirectory.php

@@ -1,6 +1,6 @@
 <?php namespace de\thekid\dialog\import;
 
-use de\thekid\dialog\processing\{Files, Images, Videos, ResizeTo};
+use de\thekid\dialog\processing\ProcessingDefaults;
 use io\{Folder, File};
 use lang\{IllegalArgumentException, IllegalStateException, FormatException, Process};
 use peer\http\HttpConnection;
@@ -19,6 +19,8 @@
  * - cover.md: The image to use for the cover page
  */
 class LocalDirectory extends Command {
+  use ProcessingDefaults;
+
   private $origin, $api;
   private $force= false;
 
@@ -56,19 +58,9 @@ private function execute(string $command, array<string> $args, $redirect= null):
 
   /** Runs this command */
   public function run(): int {
-    $files= new Files()
-      ->matching(['.jpg', '.jpeg', '.png', '.webp'], new Images()
-        ->targeting('preview', new ResizeTo(720, 'jpg'))
-        ->targeting('thumb', new ResizeTo(1024, 'webp'))
-        ->targeting('full', new ResizeTo(3840, 'webp'))
-      )
-      ->matching(['.mp4', '.mpeg', '.mov'], new Videos()
-        ->targeting('preview', new ResizeTo(720, 'jpg'))
-        ->targeting('thumb', new ResizeTo(1024, 'webp'))
-      )
-    ;
-
+    $files= $this->files();
     $publish= time();
+
     foreach (Sources::in($this->origin) as $folder => $item) {
       $this->out->writeLine('[+] ', $item);
 

src/main/php/de/thekid/dialog/processing/Collections.php

@@ -0,0 +1,50 @@
+<?php namespace de\thekid\dialog\processing;
+
+use com\mongodb\{Collection, Document};
+use lang\Throwable;
+use util\Date;
+
+class Collections {
+
+  private function __construct(private Collection $collection, private iterable $items) { }
+
+  /**
+   * Returns all documents in the given collection. Includes currently
+   * existing documents as well as anything inserted or updated with
+   * its `state` field set to "new".
+   */
+  public static function watching(Collection $collection): self {
+    $generator= fn() => {
+      static $options= ['fullDocument' => 'updateLookup'];
+
+      // Process all currently existing items
+      yield from $collection->find(['state' => 'new']);
+
+      // Watch the collection for changes
+      foreach ($collection->watch([['$match' => ['fullDocument.state' => 'new']]], $options) as $change) {
+        yield new Document($change['fullDocument']);
+      }
+    };
+    return new self($collection, $generator());
+  }
+
+  public function each(function(Document): iterable $apply): int {
+    $i= 0;
+    foreach ($this->items as $item) {
+      try {
+        foreach ($apply($item) as $state => $value) {
+          $this->collection->update($item->id(), ['$set' => [
+            'state' => $state,
+            'value' => $value,
+            'at'    => Date::now(),
+          ]]);
+        }
+      } catch ($e) {
+        Throwable::wrap($e)->printStackTrace();
+        // TODO: Error handling
+      }
+      $i++;
+    }
+    return $i;
+  } 
+}

src/main/php/de/thekid/dialog/processing/ProcessingDefaults.php

@@ -0,0 +1,18 @@
+<?php namespace de\thekid\dialog\processing;
+
+trait ProcessingDefaults {
+
+  private function files(): Files {
+    return new Files()
+      ->matching(['.jpg', '.jpeg', '.png', '.webp'], new Images()
+        ->targeting('preview', new ResizeTo(720, 'jpg'))
+        ->targeting('thumb', new ResizeTo(1024, 'webp'))
+        ->targeting('full', new ResizeTo(3840, 'webp'))
+      )
+      ->matching(['.mp4', '.mpeg', '.mov'], new Videos()
+        ->targeting('preview', new ResizeTo(720, 'jpg'))
+        ->targeting('thumb', new ResizeTo(1024, 'webp'))
+      )
+    ;
+  }
+}

src/main/php/de/thekid/dialog/processing/Watch.php

@@ -0,0 +1,103 @@
+<?php namespace de\thekid\dialog\processing;
+
+use com\mongodb\{MongoConnection, Error};
+use de\thekid\dialog\Preferences;
+use io\TempFile;
+use io\streams\StreamTransfer;
+use lang\IllegalStateException;
+use util\Objects;
+use util\cmd\{Arg, Command};
+use web\Environment;
+use webservices\rest\{Endpoint, RestUpload};
+
+/**
+ * Watch for changes on the `processing` collection and process media
+ * accordingly.
+ * 
+ * @see   https://github.com/thekid/dialog/issues/44
+ */
+class Watch extends Command {
+  use ProcessingDefaults;
+
+  private const WAIT_BEFORE_RETRYING= 10;
+  private $api;
+
+  /** Sets API endpoint to use */
+  #[Arg(position: 0)]
+  public function usingApi(?string $endpoint= null): void {
+    $this->api= new Endpoint($endpoint ?? getenv('DIALOG_API'));
+  }
+
+  /** Runs forever, consuming the change stream */
+  public function run(): int {
+    $preferences= new Preferences(new Environment('console'), 'config');
+    $collection= new MongoConnection($preferences->get('mongo', 'uri'))
+      ->database($preferences->optional('mongo', 'db', 'dialog'))
+      ->collection('processing')
+    ;
+
+    $this->out->writeLine('> Processing ', $collection);
+    $this->out->writeLine('  ', date('r'), ' - PID ', getmypid(), '; press Ctrl+C to exit');
+    $this->out->writeLine();
+
+    $files= $this->files();
+    process: try {
+      Collections::watching($collection)->each(function($item) use($files) {
+        $this->out->writeLinef(
+          '  [%s %d %.3fkB] %s',
+          date('Y-m-d H:i:s'),
+          getmypid(),
+          memory_get_usage() / 1024,
+          Objects::stringOf($item, '  '),
+        );
+
+        if (null === ($processing= $files->processing($item['file']))) return;
+        $this->out->write('  => Processing ', $processing->kind(), ' ', $item['file']);
+
+        yield 'fetching' => $processing->kind();
+        $resource= $this->api->resource('entries/{slug}/images/{file}', $item);
+        $r= $resource->get();
+        if (200 !== $r->status()) {
+          throw new IllegalStateException($r->content());
+        }
+
+        // Fetch media into temporary file
+        $source= new TempFile($item['file']);
+        using ($s= new StreamTransfer($r->stream(), $source->out())) {
+          $s->transferAll();
+        }
+
+        // Extract meta data from source, then convert source file to targets
+        yield 'extracting' => $source->size();
+        $meta= $processing->meta($source);
+
+        $transfer= [];
+        foreach ($processing->targets($source, filename: $item['file']) as $kind => $target) {
+          yield 'targeting' => $kind;
+          $transfer[$kind]= $target;
+        }
+
+        // Upload processed results and meta data
+        $size= sizeof($transfer);
+        yield 'uploading' => $size;
+        $upload= new RestUpload($this->api, $resource->request('PUT')->waiting(read: 3600));
+        foreach ($meta as $key => $value) {
+          $upload->pass('meta['.$key.']', $value);
+        }
+        foreach ($transfer as $kind => $file) {
+          $upload->transfer($kind, $file->in(), $file->filename);
+        }
+        $r= $upload->finish();
+        $this->out->writeLine(': ', $r->status());
+
+        yield 'finished' => $size;
+        $source->unlink();
+      });
+    } catch (Error $e) {
+      $this->err->writeLine('# Gracefully handling ', $e);
+      sleep(self::WAIT_BEFORE_RETRYING);
+      goto process;
+    }
+    return 0;
+  }
+}

src/test/php/de/thekid/dialog/unittest/FilesTest.php

@@ -0,0 +1,28 @@
+<?php namespace de\thekid\dialog\unittest;
+
+use de\thekid\dialog\processing\{Files, Images};
+use unittest\{Assert, Expect, Test, Values};
+
+class FilesTest {
+
+  #[Test]
+  public function can_create() {
+    new Files();
+  }
+
+  #[Test, Values(['test.jpg', 'IMG_1234.JPG', '20221119-iOS.jpeg'])]
+  public function matching_jpeg_files($filename) {
+    $processing= new Images();
+    $fixture= new Files()->matching(['.jpg', '.jpeg'], $processing);
+
+    Assert::equals($processing, $fixture->processing($filename));
+  }
+
+  #[Test, Values(['test-jpg', 'IMG_1234JPG', 'jpeg', '.jpeg-file'])]
+  public function unmatched_jpeg_files($filename) {
+    $processing= new Images();
+    $fixture= new Files()->matching(['.jpg', '.jpeg'], $processing);
+
+    Assert::null($fixture->processing($filename));
+  }
+}