feat: typed callbacks and errors

README.md

@@ -180,14 +180,43 @@ try {
 }
 ```
 
-## Callback Signatures
+## Callbacks
+
+Verify the signature, then decode the CloudEvent. `CloudEvent` is a plain
+envelope; `CallbackEvent::decode()` turns its data into the callback for its
+type — `JobStart`, `JobLog`, `JobArtifact`, `JobExit`, `JobComplete`, or
+`DeploymentResponse`. A callback that can fail carries an `Error` with a stable
+`code` to branch on and a `message` to show:
 
 ```php
+use OpenRuntimes\Orchestrator\Callback\CloudEvent;
+use OpenRuntimes\Orchestrator\Callback\JobArtifact;
+use OpenRuntimes\Orchestrator\Callback\JobExit;
 use OpenRuntimes\Orchestrator\Callback\Signature;
+use OpenRuntimes\Orchestrator\Enum\CallbackEvent;
+use OpenRuntimes\Orchestrator\Enum\ErrorCode;
+
+if (! Signature::verifyEvent($rawBody, $headers['x-signature-256'] ?? '', $secret)) {
+    return;
+}
 
-$valid = Signature::verifyEvent($rawBody, $headers['x-signature-256'] ?? '', $secret);
+$event = CloudEvent::decode(
+    \json_decode($rawBody, true),
+    fn (string $type, array $data) => CallbackEvent::from($type)->decode($data),
+);
+
+match (true) {
+    $event->data instanceof JobArtifact && $event->data->error !== null
+        => $log->error("{$event->data->artifactId}: {$event->data->error->message}"),
+    $event->data instanceof JobExit && $event->data->error?->code === ErrorCode::JobOom
+        => $log->error('Out of memory'),
+    default => null,
+};
 ```
 
+`CloudEvent::fromArray()` keeps the data as the raw array when you would rather
+read it yourself.
+
 ## Development
 
 ```sh

rector.php

@@ -4,6 +4,7 @@
 
 use Rector\CodeQuality\Rector\Catch_\ThrowWithPreviousExceptionRector;
 use Rector\Config\RectorConfig;
+use Rector\PHPUnit\CodeQuality\Rector\MethodCall\AssertEmptyNullableObjectToAssertInstanceofRector;
 use Rector\Strict\Rector\Empty_\DisallowedEmptyRuleFixerRector;
 
 return RectorConfig::configure()
@@ -22,6 +23,8 @@
         phpunitCodeQuality: true
     )
     ->withSkip([
+        // assertNull on a ?Object return states the contract; assertNotInstanceOf obscures it.
+        AssertEmptyNullableObjectToAssertInstanceofRector::class,
         ThrowWithPreviousExceptionRector::class,
         DisallowedEmptyRuleFixerRector::class,
     ]);

src/Callback/Callback.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenRuntimes\Orchestrator\Callback;
+
+/**
+ * The data of one callback, typed by its event. Decode a raw envelope into
+ * one with CallbackEvent::decode() — see CloudEvent::decode().
+ */
+interface Callback
+{
+    /**
+     * @param  array<string, mixed>  $data
+     */
+    public static function fromArray(array $data): static;
+}

src/Callback/CloudEvent.php

@@ -9,10 +9,16 @@
 use Exception;
 use OpenRuntimes\Orchestrator\Exception\ClientException;
 
+/**
+ * A CloudEvents 1.0 envelope. It carries data without knowing what the data
+ * means: decode() hands the type and raw data to whatever does.
+ *
+ * @template T
+ */
 final readonly class CloudEvent
 {
     /**
-     * @param  array<string, mixed>  $data
+     * @param  T  $data
      */
     public function __construct(
         public string $specVersion,
@@ -22,13 +28,26 @@ public function __construct(
         public string $id,
         public DateTimeInterface $time,
         public string $dataContentType,
-        public array $data,
+        public mixed $data,
     ) {}
 
     /**
      * @param  array<string, mixed>  $payload
+     * @return self<array<string, mixed>>
      */
     public static function fromArray(array $payload): self
+    {
+        return self::decode($payload, static fn (string $type, array $data): array => $data);
+    }
+
+    /**
+     * @template U
+     *
+     * @param  array<string, mixed>  $payload
+     * @param  callable(string, array<string, mixed>): U  $decode
+     * @return self<U>
+     */
+    public static function decode(array $payload, callable $decode): self
     {
         $data = $payload['data'] ?? [];
         if (! isset($payload['time']) || ! \is_string($payload['time']) || $payload['time'] === '') {
@@ -41,15 +60,17 @@ public static function fromArray(array $payload): self
             throw new ClientException('Invalid CloudEvent: malformed time.', previous: $e);
         }
 
+        $type = (string) ($payload['type'] ?? '');
+
         return new self(
             specVersion: (string) ($payload['specversion'] ?? ''),
-            type: (string) ($payload['type'] ?? ''),
+            type: $type,
             source: (string) ($payload['source'] ?? ''),
             subject: (string) ($payload['subject'] ?? ''),
             id: (string) ($payload['id'] ?? ''),
             time: $time,
             dataContentType: (string) ($payload['datacontenttype'] ?? ''),
-            data: \is_array($data) ? $data : [],
+            data: $decode($type, \is_array($data) ? $data : []),
         );
     }
 }

src/Callback/DeploymentResponse.php

@@ -0,0 +1,59 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenRuntimes\Orchestrator\Callback;
+
+use OpenRuntimes\Orchestrator\Exception\ClientException;
+use OpenRuntimes\Orchestrator\Model\Data;
+
+/**
+ * orchestrator.deployment.response — an async request completed. statusCode and
+ * body are null when the request never reached a replica; $error then says
+ * why. A body that is not valid UTF-8 arrives base64-encoded with bodyEncoding
+ * "base64". requestHeaders is null when it was dropped for size.
+ */
+final readonly class DeploymentResponse implements Callback
+{
+    public function __construct(
+        public string $deploymentId,
+        public string $invocationId,
+        public string $requestMethod,
+        public string $requestPath,
+        public bool $requestPathTruncated,
+        /** @var array<string, list<string>>|null */
+        public ?array $requestHeaders,
+        public bool $requestHeadersTruncated,
+        public ?float $durationSeconds,
+        public ?int $statusCode,
+        public ?string $body,
+        public ?string $bodyEncoding,
+        public bool $bodyTruncated,
+        public ?Error $error,
+    ) {}
+
+    public static function fromArray(array $data): static
+    {
+        $headers = $data['requestHeaders'] ?? null;
+        if ($headers !== null && ! \is_array($headers)) {
+            throw new ClientException('Invalid deployment response: requestHeaders must be an object.');
+        }
+
+        /** @var array<string, list<string>>|null $headers */
+        return new self(
+            deploymentId: Data::string($data, 'deploymentId', 'deployment response'),
+            invocationId: Data::string($data, 'invocationId', 'deployment response'),
+            requestMethod: Data::string($data, 'requestMethod', 'deployment response'),
+            requestPath: Data::string($data, 'requestPath', 'deployment response'),
+            requestPathTruncated: Data::bool($data, 'requestPathTruncated', 'deployment response'),
+            requestHeaders: $headers,
+            requestHeadersTruncated: Data::bool($data, 'requestHeadersTruncated', 'deployment response'),
+            durationSeconds: Data::optionalFloat($data, 'durationSeconds', 'deployment response'),
+            statusCode: Data::optionalInt($data, 'statusCode', 'deployment response'),
+            body: Data::optionalString($data, 'body', 'deployment response'),
+            bodyEncoding: Data::optionalString($data, 'bodyEncoding', 'deployment response'),
+            bodyTruncated: Data::bool($data, 'bodyTruncated', 'deployment response'),
+            error: Error::fromData($data),
+        );
+    }
+}

src/Callback/Error.php

@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenRuntimes\Orchestrator\Callback;
+
+use OpenRuntimes\Orchestrator\Enum\ErrorCode;
+use OpenRuntimes\Orchestrator\Exception\ClientException;
+use OpenRuntimes\Orchestrator\Model\Data;
+
+/**
+ * The error carried by a failed callback: a stable code to branch on, and a
+ * sentence about this occurrence to show — never to parse.
+ */
+final readonly class Error
+{
+    public function __construct(
+        public ErrorCode $code,
+        public string $message,
+    ) {}
+
+    /**
+     * The error a callback's data reports, or null when it reports success.
+     *
+     * @param  array<string, mixed>  $data
+     */
+    public static function fromData(array $data): ?self
+    {
+        if (! \array_key_exists('error', $data)) {
+            return null;
+        }
+        if (! \is_array($data['error'])) {
+            throw new ClientException('Invalid callback error: must be an object.');
+        }
+
+        /** @var ErrorCode $code */
+        $code = Data::enum($data['error'], 'code', ErrorCode::class, 'callback error');
+
+        return new self($code, Data::string($data['error'], 'message', 'callback error'));
+    }
+}

src/Callback/JobArtifact.php

@@ -0,0 +1,45 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenRuntimes\Orchestrator\Callback;
+
+use OpenRuntimes\Orchestrator\Model\Data;
+
+/**
+ * orchestrator.job.artifact — one artifact finished. On failure, $error says
+ * why; format and compression are what the artifact was sniffed to be, null
+ * when it was never read far enough to tell.
+ */
+final readonly class JobArtifact implements Callback
+{
+    public function __construct(
+        public string $jobId,
+        public string $artifactId,
+        public string $artifactType,
+        public string $status,
+        public mixed $content,
+        public ?float $durationSeconds,
+        public ?string $format,
+        public ?string $compression,
+        public ?Error $error,
+        /** @var array<string, string> */
+        public array $meta,
+    ) {}
+
+    public static function fromArray(array $data): static
+    {
+        return new self(
+            jobId: Data::string($data, 'jobId', 'job artifact'),
+            artifactId: Data::string($data, 'artifactId', 'job artifact'),
+            artifactType: Data::string($data, 'artifactType', 'job artifact'),
+            status: Data::string($data, 'status', 'job artifact'),
+            content: $data['content'] ?? null,
+            durationSeconds: Data::optionalFloat($data, 'durationSeconds', 'job artifact'),
+            format: Data::optionalString($data, 'format', 'job artifact'),
+            compression: Data::optionalString($data, 'compression', 'job artifact'),
+            error: Error::fromData($data),
+            meta: Data::stringMap($data, 'meta', 'job artifact'),
+        );
+    }
+}

src/Callback/JobComplete.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenRuntimes\Orchestrator\Callback;
+
+use OpenRuntimes\Orchestrator\Model\Data;
+
+/**
+ * orchestrator.job.complete — every post-job artifact has been processed,
+ * successfully or not. No more events follow for this job.
+ */
+final readonly class JobComplete implements Callback
+{
+    public function __construct(
+        public string $jobId,
+        /** @var array<string, string> */
+        public array $meta,
+    ) {}
+
+    public static function fromArray(array $data): static
+    {
+        return new self(
+            jobId: Data::string($data, 'jobId', 'job complete'),
+            meta: Data::stringMap($data, 'meta', 'job complete'),
+        );
+    }
+}

src/Callback/JobExit.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenRuntimes\Orchestrator\Callback;
+
+use OpenRuntimes\Orchestrator\Model\Data;
+
+/**
+ * orchestrator.job.exit — the worker exited. exitCode is -1 when the job failed
+ * before the worker could run; reason is the backend's own detail ("oom") and
+ * null when it has nothing to add beyond the code.
+ */
+final readonly class JobExit implements Callback
+{
+    public function __construct(
+        public string $jobId,
+        public int $exitCode,
+        public ?string $reason,
+        public string $image,
+        public ?float $durationSeconds,
+        public ?Error $error,
+        /** @var array<string, string> */
+        public array $meta,
+    ) {}
+
+    public static function fromArray(array $data): static
+    {
+        return new self(
+            jobId: Data::string($data, 'jobId', 'job exit'),
+            exitCode: Data::int($data, 'exitCode', 'job exit'),
+            reason: Data::optionalString($data, 'reason', 'job exit'),
+            image: Data::string($data, 'image', 'job exit'),
+            durationSeconds: Data::optionalFloat($data, 'durationSeconds', 'job exit'),
+            error: Error::fromData($data),
+            meta: Data::stringMap($data, 'meta', 'job exit'),
+        );
+    }
+}

src/Callback/JobLog.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenRuntimes\Orchestrator\Callback;
+
+use OpenRuntimes\Orchestrator\Model\Data;
+
+/**
+ * orchestrator.job.log — a batch of stdout or stderr lines.
+ */
+final readonly class JobLog implements Callback
+{
+    public function __construct(
+        public string $jobId,
+        /** @var list<string> */
+        public array $lines,
+        public string $stream,
+        /** @var array<string, string> */
+        public array $meta,
+    ) {}
+
+    public static function fromArray(array $data): static
+    {
+        return new self(
+            jobId: Data::string($data, 'jobId', 'job log'),
+            lines: Data::strings($data, 'lines', 'job log'),
+            stream: Data::string($data, 'stream', 'job log'),
+            meta: Data::stringMap($data, 'meta', 'job log'),
+        );
+    }
+}