Event Sourcing and CQRS in Laravel: Deep Dive into Core Components

namespace App\EventSourcing;
 
use Illuminate\Support\Carbon;
use Illuminate\Support\Str;
 
abstract class DomainEvent
{
    private string $eventId;
    private array $metadata;
 
    public function __construct(
        private string $aggregateId,
        private array $payload,
        private ?Carbon $occurredAt = null
    ) {
        $this->eventId = Str::uuid()->toString();
        $this->occurredAt = $occurredAt ?? now();
        $this->metadata = [
            'user_id' => auth()->id(),
            'ip_address' => request()->ip(),
            'environment' => app()->environment(),
            'correlation_id' => Str::uuid()->toString(),
        ];
    }
 
    abstract public function getEventName(): string;
    
    public function getAggregateId(): string
    {
        return $this->aggregateId;
    }
    
    public function getPayload(): array
    {
        return $this->payload;
    }
    
    public function getMetadata(): array
    {
        return $this->metadata;
    }
    
    public function getOccurredAt(): Carbon
    {
        return $this->occurredAt;
    }
    
    public function getEventId(): string
    {
        return $this->eventId;
    }
 
    public function toArray(): array
    {
        return [
            'event_id' => $this->eventId,
            'event_type' => $this->getEventName(),
            'aggregate_id' => $this->aggregateId,
            'payload' => $this->payload,
            'metadata' => $this->metadata,
            'occurred_at' => $this->occurredAt,
        ];
    }
}

namespace App\EventSourcing;
 
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use App\Exceptions\ConcurrencyException;
 
class EventStore
{
    private const CACHE_TTL = 3600; // 1 hour
 
    public function __construct(
        private string $tableName = 'event_store',
        private bool $useCache = true
    ) {}
    
    public function append(DomainEvent $event, int $expectedVersion = null): void
    {
        DB::beginTransaction();
        
        try {
            // Concurrency check
            if ($expectedVersion !== null) {
                $currentVersion = $this->getCurrentVersion($event->getAggregateId());
                if ($currentVersion !== $expectedVersion) {
                    throw new ConcurrencyException(
                        "Expected version {$expectedVersion}, but current version is {$currentVersion}"
                    );
                }
            }
            
            // Persist the event
            DB::table($this->tableName)->insert([
                'event_id' => $event->getEventId(),
                'aggregate_id' => $event->getAggregateId(),
                'event_type' => $event->getEventName(),
                'payload' => json_encode($event->getPayload()),
                'metadata' => json_encode($event->getMetadata()),
                'occurred_at' => $event->getOccurredAt(),
                'version' => $this->getNextVersion($event->getAggregateId()),
            ]);
            
            // Cache management
            if ($this->useCache) {
                $this->clearCache($event->getAggregateId());
            }
            
            DB::commit();
        } catch (\Exception $e) {
            DB::rollBack();
            throw $e;
        }
    }
    
    public function getEventsForAggregate(string $aggregateId): Collection
    {
        $cacheKey = "events.{$aggregateId}";
        
        if ($this->useCache && Cache::has($cacheKey)) {
            return Cache::get($cacheKey);
        }
        
        $events = DB::table($this->tableName)
            ->where('aggregate_id', $aggregateId)
            ->orderBy('version')
            ->get()
            ->map(function ($event) {
                return $this->deserializeEvent($event);
            });
            
        if ($this->useCache) {
            Cache::put($cacheKey, $events, self::CACHE_TTL);
        }
        
        return $events;
    }
}