Lines 100.00% 78 / 78
Methods 100.00% 13 / 13
Classes 100.00% 1 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 enqueue 100.00% 5 / 5 100.00% 1 / 1 5
 pushStateChange 100.00% 12 / 12 100.00% 1 / 1 5
 takeNext 100.00% 11 / 11 100.00% 1 / 1 4
 clear 100.00% 5 / 5 100.00% 1 / 1 1
 buildStatus 100.00% 7 / 7 100.00% 1 / 1 1
 shiftEntry 100.00% 12 / 12 100.00% 1 / 1 4
 forgetStateChange 100.00% 6 / 6 100.00% 1 / 1 3
 headIsStateChange 100.00% 2 / 2 100.00% 1 / 1 2
 pushDroppable 100.00% 6 / 6 100.00% 1 / 1 2
 replaceOrAppendLatest 100.00% 5 / 5 100.00% 1 / 1 2
 pushSegmentBoundary 100.00% 3 / 3 100.00% 1 / 1 1
 appendToQueue 100.00% 3 / 3 100.00% 1 / 1 1
15final class WorkerOutbox
16{
17    /** @var array<int, BrokerMessage|EncodedStateChange> */
18    private array $queue = [];
19
20    private int $nextSequence = 0;
21
22    // Per entity: the first queued change and one merged tail; a segment boundary resets both.
23    /** @var array<string, int> */
24    private array $firstStateChanges = [];
25
26    /** @var array<string, int> */
27    private array $lastStateChanges = [];
28
29    /** @var array<class-string<BrokerMessage>, int> */
30    private array $latestOnlySequences = [];
31
32    /** @var array<int, true> in queue order */
33    private array $droppable = [];
34
35    public private(set) int $dropped = 0;
36
37    public private(set) int $coalescedStateChanges = 0;
38
39    private int $batchesSent = 0;
40
41    private int $largestBatch = 0;
42
43    public function __construct(private readonly OutboxLimits $limits) {}
44
45    public function enqueue(BrokerMessage $message): void
46    {
47        match ($message->getOutboxDelivery()) {
48            OutboxDelivery::Droppable => $this->pushDroppable($message),
49            OutboxDelivery::StateSegmentBoundary => $this->pushSegmentBoundary($message),
50            OutboxDelivery::LatestOnly => $this->replaceOrAppendLatest($message),
51            OutboxDelivery::Plain => $this->appendToQueue($message),
52        };
53    }
54
55    public function pushStateChange(EncodedStateChange $change): void
56    {
57        $entityId = $change->change->entityId->value;
58
59        if (!isset($this->firstStateChanges[$entityId])) {
60            $this->firstStateChanges[$entityId] = $this->appendToQueue($change);
61
62            return;
63        }
64
65        $lastSequence = $this->lastStateChanges[$entityId] ?? null;
66        $last = $lastSequence === null ? null : $this->queue[$lastSequence];
67
68        if ($lastSequence === null || !$last instanceof EncodedStateChange) {
69            $this->lastStateChanges[$entityId] = $this->appendToQueue($change);
70
71            return;
72        }
73
74        // Re-appended at the end, so the merged change never overtakes a later event.
75        unset($this->queue[$lastSequence]);
76
77        $this->lastStateChanges[$entityId] = $this->appendToQueue(new EncodedStateChange($change->change->withFrom($last->change->from)));
78
79        ++$this->coalescedStateChanges;
80    }
81
82    public function takeNext(): ?BrokerMessage
83    {
84        $entry = $this->shiftEntry();
85
86        if (!$entry instanceof EncodedStateChange) {
87            return $entry;
88        }
89
90        $batch = [$entry];
91
92        while (\count($batch) < $this->limits->stateBatch && $this->headIsStateChange()) {
93            $next = $this->shiftEntry();
94            \assert($next instanceof EncodedStateChange);
95            $batch[] = $next;
96        }
97
98        ++$this->batchesSent;
99        $this->largestBatch = max($this->largestBatch, \count($batch));
100
101        return new StateChangeBatch(StateChangesFragment::fromEncodedChanges(EncodedStateChangeCollection::fromList($batch)));
102    }
103
104    public function clear(): void
105    {
106        $this->queue = [];
107        $this->firstStateChanges = [];
108        $this->lastStateChanges = [];
109        $this->droppable = [];
110        $this->latestOnlySequences = [];
111    }
112
113    public function buildStatus(): OutboxStatus
114    {
115        return new OutboxStatus(
116            queued: \count($this->queue),
117            dropped: $this->dropped,
118            coalescedStateChanges: $this->coalescedStateChanges,
119            stateBatchesSent: $this->batchesSent,
120            largestStateBatch: $this->largestBatch,
121        );
122    }
123
124    private function shiftEntry(): BrokerMessage|EncodedStateChange|null
125    {
126        $sequence = array_key_first($this->queue);
127
128        if ($sequence === null) {
129            return null;
130        }
131
132        $entry = $this->queue[$sequence];
133        unset($this->queue[$sequence]);
134
135        if ($entry instanceof EncodedStateChange) {
136            $this->forgetStateChange($entry->change->entityId->value, $sequence);
137
138            return $entry;
139        }
140
141        unset($this->droppable[$sequence]);
142
143        if (($this->latestOnlySequences[$entry::class] ?? null) === $sequence) {
144            unset($this->latestOnlySequences[$entry::class]);
145        }
146
147        return $entry;
148    }
149
150    private function forgetStateChange(string $entityId, int $sequence): void
151    {
152        if (($this->firstStateChanges[$entityId] ?? null) !== $sequence) {
153            return;
154        }
155
156        unset($this->firstStateChanges[$entityId]);
157
158        if (isset($this->lastStateChanges[$entityId])) {
159            $this->firstStateChanges[$entityId] = $this->lastStateChanges[$entityId];
160            unset($this->lastStateChanges[$entityId]);
161        }
162    }
163
164    private function headIsStateChange(): bool
165    {
166        $sequence = array_key_first($this->queue);
167
168        return $sequence !== null && $this->queue[$sequence] instanceof EncodedStateChange;
169    }
170
171    private function pushDroppable(BrokerMessage $message): void
172    {
173        $this->droppable[$this->appendToQueue($message)] = true;
174
175        if (\count($this->droppable) <= $this->limits->eventBuffer) {
176            return;
177        }
178
179        $oldest = array_key_first($this->droppable);
180        unset($this->droppable[$oldest], $this->queue[$oldest]);
181        ++$this->dropped;
182    }
183
184    private function replaceOrAppendLatest(BrokerMessage $message): void
185    {
186        $queuedSequence = $this->latestOnlySequences[$message::class] ?? null;
187
188        if ($queuedSequence !== null) {
189            $this->queue[$queuedSequence] = $message;
190
191            return;
192        }
193
194        $this->latestOnlySequences[$message::class] = $this->appendToQueue($message);
195    }
196
197    private function pushSegmentBoundary(BrokerMessage $message): void
198    {
199        $this->appendToQueue($message);
200        $this->firstStateChanges = [];
201        $this->lastStateChanges = [];
202    }
203
204    private function appendToQueue(BrokerMessage|EncodedStateChange $entry): int
205    {
206        $sequence = $this->nextSequence++;
207        $this->queue[$sequence] = $entry;
208
209        return $sequence;
210    }
211}