Lines 97.43% 38 / 39
Functions and Methods 90.00% 9 / 10
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
StateCacheSync 97.43% 38 / 39 90.00% 9 / 10 20 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 seedFromSnapshot 100.00% 7 / 7 100.00% 1 / 1 2
 awaitFirstSeed 100.00% 4 / 4 100.00% 1 / 1 2
 isOlderThan 100.00% 1 / 1 100.00% 1 / 1 1
 replaceWithResyncedStates 100.00% 4 / 4 100.00% 1 / 1 1
 applyStateChange 0.00% 0 / 1 0.00% 0 / 1 2
 countEntities 100.00% 1 / 1 100.00% 1 / 1 1
 buildResyncChanges 100.00% 9 / 9 100.00% 1 / 1 6
 hasChanged 100.00% 3 / 3 100.00% 1 / 1 4
 buildResyncChange 100.00% 8 / 8 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Worker;
6
7use Amp\Cancellation;
8use Amp\CancelledException;
9use Amp\DeferredFuture;
10use Psr\Log\LoggerInterface;
11use Stewart\Contracts\Entity\EntityId;
12use Stewart\Contracts\State\Collection\EntityStateCollection;
13use Stewart\Contracts\State\EntityState;
14use Stewart\Contracts\State\StateChange;
15use Stewart\Contracts\State\StateChangeOrigin;
16use Stewart\Contracts\Time\Instant;
17use Stewart\Runtime\State\StateCache;
18use Stewart\Support\State\Collection\StateChangeCollection;
19
20final class StateCacheSync
21{
22    private int $revision = 0;
23
24    private bool $seeded = false;
25
26    /** @var DeferredFuture<null> */
27    private readonly DeferredFuture $firstSeed;
28
29    public function __construct(
30        private readonly StateCache $stateCache,
31        private readonly LoggerInterface $logger,
32    ) {
33        $this->firstSeed = new DeferredFuture();
34    }
35
36    public function seedFromSnapshot(EntityStateCollection $states, int $revision): void
37    {
38        $this->stateCache->replaceAll($states);
39        $this->revision = $revision;
40
41        if ($this->seeded) {
42            return;
43        }
44
45        $this->seeded = true;
46
47        $this->logger->debug('State cache seeded', ['entities' => $this->stateCache->count(), 'revision' => $revision]);
48
49        $this->firstSeed->complete();
50    }
51
52    public function awaitFirstSeed(Cancellation $cancellation): bool
53    {
54        try {
55            $this->firstSeed->getFuture()->await($cancellation);
56        } catch (CancelledException) {
57            return false;
58        }
59
60        return true;
61    }
62
63    public function isOlderThan(int $revision): bool
64    {
65        return $revision > $this->revision;
66    }
67
68    public function replaceWithResyncedStates(EntityStateCollection $states, int $revision, Instant $resyncedAt): StateChangeCollection
69    {
70        $reconstructed = $this->buildResyncChanges($this->stateCache->getAllStates(), $states, $resyncedAt);
71
72        $this->stateCache->replaceAll($states);
73        $this->revision = $revision;
74
75        return $reconstructed;
76    }
77
78    public function applyStateChange(StateChange $change): void
79    {
80        $this->stateCache->applyChange($change);
81    }
82
83    public function countEntities(): int
84    {
85        return $this->stateCache->count();
86    }
87
88    private function buildResyncChanges(EntityStateCollection $before, EntityStateCollection $after, Instant $resyncedAt): StateChangeCollection
89    {
90        $changes = [];
91
92        foreach ($after as $state) {
93            $previous = $before->find($state->entityId);
94
95            if ($previous === null || $this->hasChanged($previous, $state)) {
96                $changes[] = $this->buildResyncChange($state->entityId, $previous, $state, $resyncedAt);
97            }
98        }
99
100        foreach ($before as $state) {
101            if ($after->find($state->entityId) === null) {
102                $changes[] = $this->buildResyncChange($state->entityId, $state, null, $resyncedAt);
103            }
104        }
105
106        return StateChangeCollection::fromList($changes);
107    }
108
109    private function hasChanged(EntityState $before, EntityState $after): bool
110    {
111        // HA bumps last_updated on any write; the fallback covers states without it.
112        if ($before->lastUpdatedAt !== null && $after->lastUpdatedAt !== null) {
113            return !$before->lastUpdatedAt->equals($after->lastUpdatedAt);
114        }
115
116        return $before->state !== $after->state || $before->attributes !== $after->attributes;
117    }
118
119    private function buildResyncChange(EntityId $entityId, ?EntityState $from, ?EntityState $to, Instant $resyncedAt): StateChange
120    {
121        return new StateChange(
122            entityId: $entityId,
123            from: $from,
124            to: $to,
125            firedAt: $to->lastUpdatedAt ?? $resyncedAt,
126            context: null,
127            origin: StateChangeOrigin::Resync,
128        );
129    }
130}