Add an opt-in transactional queue outbox

AGENTS.md

@@ -119,7 +119,7 @@ During implementation, run new or changed test files immediately. After completi
 
 Use checks that match the change. For isolated changes, run `composer lint:fix`, `composer analyse`, and the affected tests. Run a single affected test file with PHPUnit. Use ParaTest when the affected tests span multiple files. Only run `composer fix` when changes could affect code beyond the affected tests; it already runs formatting, analysis, and all test suites, so do not run those checks separately first.
 
-If a check fails, use targeted checks while correcting the issue, then run the failed check and each remaining check. Rerun an earlier check only if the correction could affect it.
+After a failure, fix the cause and rerun only affected tests, formatting or static analysis. Then finish any verification not yet run. Repeat the full test suite only if the fix itself has framework-wide impact.
 
 ## Development Conventions
 

src/broadcasting/src/BroadcastEvent.php

@@ -7,6 +7,7 @@
 use Hypervel\Bus\Queueable;
 use Hypervel\Contracts\Broadcasting\Factory as BroadcastingFactory;
 use Hypervel\Contracts\Queue\ShouldQueue;
+use Hypervel\Contracts\Queue\ShouldQueueViaOutbox;
 use Hypervel\Contracts\Support\Arrayable;
 use Hypervel\Queue\Attributes\Backoff;
 use Hypervel\Queue\Attributes\DeleteWhenMissingModels;
@@ -67,6 +68,17 @@ public function __construct(mixed $event)
         $this->timeout = $this->getAttributeValue($event, Timeout::class, 'timeout');
         $this->backoff = $this->getAttributeValue($event, Backoff::class, 'backoff');
         $this->afterCommit = property_exists($event, 'afterCommit') ? $event->afterCommit : null;
+
+        if (isset($event->viaOutbox)) {
+            $this->viaOutbox = $event->viaOutbox;
+        } elseif ($event instanceof ShouldQueueViaOutbox) {
+            $this->viaOutbox = true;
+        }
+
+        if (isset($event->outboxDatabase)) {
+            $this->outboxDatabase = $event->outboxDatabase;
+        }
+
         $this->maxExceptions = $this->getAttributeValue($event, MaxExceptions::class, 'maxExceptions');
         $this->deleteWhenMissingModels = $this->getAttributeValue($event, DeleteWhenMissingModels::class, 'deleteWhenMissingModels', $this->deleteWhenMissingModels);
     }
@@ -115,6 +127,10 @@ protected function getPayloadFromEvent(mixed $event): array
         $payload = [];
 
         foreach ((new ReflectionClass($event))->getProperties(ReflectionProperty::IS_PUBLIC) as $property) {
+            if (! $property->isInitialized($event)) {
+                continue;
+            }
+
             $payload[$property->getName()] = $this->formatProperty($property->getValue($event));
         }
 

src/bus/src/Batch.php

@@ -12,12 +12,14 @@
 use Hypervel\Contracts\Events\Dispatcher;
 use Hypervel\Contracts\Queue\Factory as QueueFactory;
 use Hypervel\Contracts\Support\Arrayable;
+use Hypervel\Database\ConnectionResolverInterface;
 use Hypervel\Queue\CallQueuedClosure;
 use Hypervel\Support\Arr;
 use Hypervel\Support\CarbonImmutable;
 use Hypervel\Support\Collection;
 use Hypervel\Support\Enumerable;
 use JsonSerializable;
+use LogicException;
 use Throwable;
 
 class Batch implements Arrayable, JsonSerializable
@@ -57,8 +59,10 @@ public function fresh(): ?Batch
     public function add(array|object $jobs): ?Batch
     {
         $count = 0;
+        $owner = null;
+        $resolver = null;
 
-        $jobs = Collection::wrap($jobs)->map(function ($job) use (&$count) {
+        $jobs = Collection::wrap($jobs)->map(function ($job) use (&$count, &$owner, &$resolver) {
             $job = $job instanceof Closure ? CallQueuedClosure::create($job) : $job;
 
             if (is_array($job)) {
@@ -76,11 +80,37 @@ public function add(array|object $jobs): ?Batch
                     $first->allOnConnection($this->options['connection']);
                 }
 
-                return $first->chain($chain->slice(1)->values()->all());
+                $job = $first->chain($chain->slice(1)->values()->all());
+            } else {
+                $job->withBatchId($this->id);
+
+                ++$count;
             }
-            $job->withBatchId($this->id);
 
-            ++$count;
+            if (array_key_exists('viaOutbox', $this->options) || array_key_exists('outboxDatabase', $this->options)) {
+                // Preserve initialized nulls as well as explicit values on each job.
+                $properties = get_object_vars($job);
+
+                foreach (['viaOutbox', 'outboxDatabase'] as $property) {
+                    if (array_key_exists($property, $this->options) && ! array_key_exists($property, $properties)) {
+                        $job->{$property} = $this->options[$property];
+                    }
+                }
+            }
+
+            // The counter transaction supplies the implicit SQL owner; declared owners must not bypass it.
+            if (($database = $job->outboxDatabase ?? null) !== null
+                && ($job->viaOutbox ?? null) !== false
+                && $this->repository instanceof DatabaseBatchRepository) {
+                $owner ??= $this->repository->getConnection()->getNameWithReadWriteType();
+                /** @var ConnectionResolverInterface $resolver */
+                $resolver ??= Container::getInstance()->make('db');
+                $selectedOwner = $resolver->connection($database)->getNameWithReadWriteType();
+
+                if ($selectedOwner !== $owner) {
+                    throw new LogicException("Outbox database [{$selectedOwner}] does not match the batch database [{$owner}]. Use the batch database or omit the explicit outbox owner.");
+                }
+            }
 
             return $job;
         });

src/bus/src/Batchable.php

@@ -11,6 +11,8 @@
 
 trait Batchable
 {
+    use InteractsWithQueueOutbox;
+
     /**
      * The batch ID (if applicable).
      */

src/bus/src/DebounceLock.php

@@ -45,13 +45,14 @@ public function acquire(mixed $job, ?int $debounceFor = null, ?int $maxWait = nu
     public function acquireForDispatch(object $job, ?int $debounceFor = null, ?int $maxWait = null): array
     {
         [$cache, $key, $ttl, $maxWait] = $this->resolveLockValues($job, $debounceFor, $maxWait);
+        $releaseContext = DispatchLockContext::captureReleaseContext();
         $result = $this->acquireResolvedLock($cache, $key, $ttl, $maxWait);
 
         if (isset(class_uses_recursive($job)[Queueable::class])) {
             $job->debounceOwner = $result['owner'];
         }
 
-        DispatchLockContext::registerDebounce($job, $cache, $key, $result['owner']);
+        DispatchLockContext::registerDebounce($job, $cache, $key, $result['owner'], $releaseContext);
 
         return $result;
     }