WIP: State Reconstructor

src/Contracts/StoresEvents.php

@@ -7,6 +7,7 @@
 use Ramsey\Uuid\UuidInterface;
 use Symfony\Component\Uid\AbstractUid;
 use Thunk\Verbs\Event;
+use Thunk\Verbs\Lifecycle\AggregateStateSummary;
 use Thunk\Verbs\State;
 
 interface StoresEvents
@@ -16,6 +17,10 @@ public function read(
         Bits|UuidInterface|AbstractUid|int|string|null $after_id = null,
     ): LazyCollection;
 
+    public function get(iterable $ids): LazyCollection;
+
     /** @param  Event[]  $events */
     public function write(array $events): bool;
+
+    public function summarize(State ...$states): AggregateStateSummary;
 }

src/Contracts/StoresSnapshots.php

@@ -17,4 +17,6 @@ public function loadSingleton(string $type): ?State;
     public function write(array $states): bool;
 
     public function reset(): bool;
+
+    public function delete(Bits|UuidInterface|AbstractUid|int|string ...$ids): bool;
 }

src/Event.php

@@ -13,7 +13,6 @@
 use Thunk\Verbs\Support\EventStateRegistry;
 use Thunk\Verbs\Support\PendingEvent;
 use Thunk\Verbs\Support\StateCollection;
-use WeakMap;
 
 /**
  * @method static static fire(...$args)
@@ -42,11 +41,7 @@ public function metadata(?string $key = null, mixed $default = null): mixed
 
     public function states(): StateCollection
     {
-        // TODO: This is a bit hacky, but is probably OK right now
-
-        static $map = new WeakMap;
-
-        return $map[$this] ??= app(EventStateRegistry::class)->getStates($this);
+        return app(EventStateRegistry::class)->getStates($this);
     }
 
     /**

src/Lifecycle/AggregateStateSummary.php

@@ -0,0 +1,90 @@
+<?php
+
+namespace Thunk\Verbs\Lifecycle;
+
+use Illuminate\Database\Eloquent\Builder;
+use Illuminate\Support\Collection;
+use Thunk\Verbs\Models\VerbStateEvent;
+use Thunk\Verbs\State;
+use Thunk\Verbs\Support\StateIdentity;
+
+class AggregateStateSummary
+{
+    public static function summarize(State ...$states): static
+    {
+        $summary = new static(
+            original_states: Collection::make($states),
+            related_event_ids: new Collection,
+            related_states: Collection::make($states)->map(StateIdentity::from(...)),
+        );
+
+        return $summary->discover();
+    }
+
+    /**
+     * @param  Collection<int, State>  $original_states
+     * @param  Collection<int, int>  $related_event_ids
+     * @param  Collection<int, StateIdentity>  $related_states
+     */
+    public function __construct(
+        public Collection $original_states = new Collection,
+        public Collection $related_event_ids = new Collection,
+        public Collection $related_states = new Collection,
+    ) {}
+
+    protected function discover(): static
+    {
+        $this->discoverNewEventIds();
+
+        do {
+            $continue = $this->discoverNewStates() && $this->discoverNewEventIds();
+        } while ($continue);
+
+        $this->related_event_ids = $this->related_event_ids->sort();
+
+        return $this;
+    }
+
+    protected function discoverNewEventIds(): bool
+    {
+        $new_event_ids = VerbStateEvent::query()
+            ->distinct()
+            ->select('event_id')
+            ->whereNotIn('event_id', $this->related_event_ids)
+            ->where(fn (Builder $query) => $this->related_states->each(
+                fn ($state) => $query->orWhere(fn (Builder $query) => $this->addConstraint($state, $query)))
+            )
+            ->toBase()
+            ->pluck('event_id');
+
+        $this->related_event_ids = $this->related_event_ids->merge($new_event_ids);
+
+        return $new_event_ids->isNotEmpty();
+    }
+
+    protected function discoverNewStates(): bool
+    {
+        $discovered_states = VerbStateEvent::query()
+            ->orderBy('id')
+            ->distinct()
+            ->select(['state_id', 'state_type'])
+            ->whereIn('event_id', $this->related_event_ids)
+            ->where(fn (Builder $query) => $this->related_states->each(
+                fn ($state) => $query->whereNot(fn (Builder $query) => $this->addConstraint($state, $query)))
+            )
+            ->toBase()
+            ->chunkMap(StateIdentity::from(...));
+
+        $this->related_states = $this->related_states->merge($discovered_states);
+
+        return $discovered_states->isNotEmpty();
+    }
+
+    protected function addConstraint(StateIdentity $state, Builder $query): Builder
+    {
+        $query->where('state_type', '=', $state->state_type);
+        $query->where('state_id', '=', $state->state_id);
+
+        return $query;
+    }
+}

src/Lifecycle/EventStore.php

@@ -5,6 +5,7 @@
 use Glhd\Bits\Bits;
 use Illuminate\Database\Eloquent\Builder;
 use Illuminate\Database\Query\Builder as BaseBuilder;
+use Illuminate\Database\Query\Expression;
 use Illuminate\Support\Collection;
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\LazyCollection;
@@ -35,6 +36,15 @@ public function read(
             ->map(fn (VerbEvent $model) => $model->event());
     }
 
+    public function get(iterable $ids): LazyCollection
+    {
+        return VerbEvent::query()
+            ->whereIn('id', collect($ids))
+            ->lazyById()
+            ->each(fn (VerbEvent $model) => $this->metadata->set($model->event(), $model->metadata()))
+            ->map(fn (VerbEvent $model) => $model->event());
+    }
+
     public function write(array $events): bool
     {
         if (empty($events)) {
@@ -47,6 +57,11 @@ public function write(array $events): bool
             && VerbStateEvent::insert($this->formatRelationshipsForWrite($events));
     }
 
+    public function summarize(State ...$states): AggregateStateSummary
+    {
+        return AggregateStateSummary::summarize(...$states);
+    }
+
     protected function readEvents(
         ?State $state,
         Bits|UuidInterface|AbstractUid|int|string|null $after_id,
@@ -78,11 +93,7 @@ protected function guardAgainstConcurrentWrites(array $events): void
         $query->select([
             'state_type',
             'state_id',
-            DB::raw(sprintf(
-                'max(%s) as %s',
-                $query->getGrammar()->wrap('event_id'),
-                $query->getGrammar()->wrapTable('max_event_id')
-            )),
+            $this->aggregateExpression($query, 'event_id', 'max'),
         ]);
 
         $query->groupBy('state_type', 'state_id');
@@ -148,4 +159,14 @@ protected function formatRelationshipsForWrite(array $event_objects): array
             ->values()
             ->all();
     }
+
+    protected function aggregateExpression(BaseBuilder $query, string $column, string $function): Expression
+    {
+        return DB::raw(sprintf(
+            '%s(%s) as %s',
+            $function,
+            $query->getGrammar()->wrap($column),
+            $query->getGrammar()->wrapTable("{$function}_{$column}")
+        ));
+    }
 }

src/Lifecycle/NullSnapshotStore.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace Thunk\Verbs\Lifecycle;
+
+use Glhd\Bits\Bits;
+use Ramsey\Uuid\UuidInterface;
+use Symfony\Component\Uid\AbstractUid;
+use Thunk\Verbs\Contracts\StoresSnapshots;
+use Thunk\Verbs\State;
+use Thunk\Verbs\Support\StateCollection;
+
+class NullSnapshotStore implements StoresSnapshots
+{
+    public function load(Bits|UuidInterface|AbstractUid|iterable|int|string $id, string $type): State|StateCollection|null
+    {
+        return null;
+    }
+
+    public function loadSingleton(string $type): ?State
+    {
+        return null;
+    }
+
+    public function write(array $states): bool
+    {
+        return true;
+    }
+
+    public function delete(Bits|UuidInterface|AbstractUid|int|string ...$ids): bool
+    {
+        return true;
+    }
+
+    public function reset(): bool
+    {
+        return true;
+    }
+}

src/Lifecycle/SnapshotStore.php

@@ -64,6 +64,13 @@ public function write(array $states): bool
         return true;
     }
 
+    public function delete(Bits|UuidInterface|AbstractUid|int|string ...$ids): bool
+    {
+        $ids = array_map(Id::from(...), $ids);
+
+        return VerbSnapshot::whereIn('state_id', $ids)->delete() === true;
+    }
+
     public function reset(): bool
     {
         VerbSnapshot::truncate();

src/Lifecycle/StateManager.php

@@ -11,13 +11,17 @@
 use Thunk\Verbs\Contracts\StoresSnapshots;
 use Thunk\Verbs\Event;
 use Thunk\Verbs\Facades\Id;
+use Thunk\Verbs\Models\VerbStateEvent;
 use Thunk\Verbs\State;
+use Thunk\Verbs\Support\EventStateRegistry;
 use Thunk\Verbs\Support\StateCollection;
 use Thunk\Verbs\Support\StateInstanceCache;
 use UnexpectedValueException;
 
 class StateManager
 {
+    protected bool $is_reconstituting = false;
+
     protected bool $is_replaying = false;
 
     public function __construct(
@@ -42,6 +46,12 @@ public function register(State $state): State
      */
     public function load(Bits|UuidInterface|AbstractUid|iterable|int|string $id, string $type): StateCollection|State
     {
+        // FIXME: This was not written to support loading multiple states
+        // $summary = $this->events->summarize($state);
+        // if ($summary->out_of_sync) {
+        //     $this->snapshots->delete(...$summary->related_state_ids);
+        // }
+
         return is_iterable($id)
             ? $this->loadMany($id, $type)
             : $this->loadOne($id, $type);
@@ -110,6 +120,8 @@ public function setReplaying(bool $replaying): static
     public function reset(bool $include_storage = false): static
     {
         $this->states->reset();
+        app(EventStateRegistry::class)->reset();
+
         $this->is_replaying = false;
 
         if ($include_storage) {
@@ -154,7 +166,7 @@ protected function loadOne(Bits|UuidInterface|AbstractUid|int|string $id, string
         $this->remember($state);
         $this->reconstitute($state);
 
-        return $state;
+        return $this->states->get($key); // FIXME
     }
 
     /** @param  class-string<State>  $type */
@@ -181,30 +193,75 @@ protected function loadMany(iterable $ids, string $type): StateCollection
 
         // At this point, all the states should be in our cache, so we can just load everything
         return StateCollection::make(
-            $ids->map(fn ($id) => $this->states->get($this->key($id, $type)))
+            $ids->map(fn ($id) => $this->states->get($this->key($id, $type))),
         );
     }
 
     protected function reconstitute(State $state): static
     {
-        // When we're replaying, the Broker is in charge of applying the correct events
-        // to the State, so we only need to do it *outside* of replays.
-        if (! $this->is_replaying) {
-            $this->events
-                ->read(state: $state, after_id: $state->last_event_id)
-                ->each(fn (Event $event) => $this->dispatcher->apply($event));
-
-            // It's possible for an event to mutate state out of order when reconstituting,
-            // so as a precaution, we'll clear all other states from the store and reload
-            // them from snapshots as needed in the rest of the request.
-            // FIXME: We still need to figure this out
-            // $this->states->reset();
-            // $this->remember($state);
+        // FIXME: Only run this if the state is out of date
+        if (! $this->needsReconstituting($state)) {
+            // dump('skipping: everything in sync');
+            return $this;
+        }
+
+        if (! $this->is_replaying && ! $this->is_reconstituting) {
+            $real_registry = app(EventStateRegistry::class);
+
+            try {
+                $this->is_reconstituting = true;
+
+                $summary = $this->events->summarize($state);
+
+                [$temp_manager] = $this->bindNewEmptyStateManager();
+
+                $this->events
+                    ->get($summary->related_event_ids)
+                    ->each($this->dispatcher->apply(...));
+
+                foreach ($temp_manager->states->all() as $key => $state) {
+                    $this->states->put($key, $state);
+                }
+
+            } finally {
+                $this->is_reconstituting = false;
+
+                app()->instance(StateManager::class, $this);
+                app()->instance(EventStateRegistry::class, $real_registry);
+            }
         }
 
         return $this;
     }
 
+    protected function needsReconstituting(State $state): bool
+    {
+        $max_id = VerbStateEvent::query()
+            ->where('state_id', $state->id)
+            ->where('state_type', $state::class)
+            ->max('event_id');
+
+        return $max_id !== $state->last_event_id;
+    }
+
+    protected function bindNewEmptyStateManager()
+    {
+        $temp_manager = new StateManager(
+            dispatcher: $this->dispatcher,
+            snapshots: new NullSnapshotStore,
+            events: $this->events,
+            states: new StateInstanceCache,
+        );
+        $temp_manager->is_reconstituting = true; // FIXME
+
+        $temp_registry = new EventStateRegistry($temp_manager);
+
+        app()->instance(StateManager::class, $temp_manager);
+        app()->instance(EventStateRegistry::class, $temp_registry);
+
+        return [$temp_manager, $temp_registry];
+    }
+
     protected function remember(State $state): State
     {
         $key = $this->key($state->id, $state::class);