feat(events): domain event outbox and the two reports that close the loop
Tasks 07 through 13 each changed something the rest of the system might want to know about, with no contract for saying so. And task 05 shipped a powerful segment editor with no feedback on whether a clinic defined its segments right. Events - A closed list of names, because a consumer branches on the string and a one-letter typo would produce an event nobody hears and no error either - Payloads carry uuids and scalars only; non-scalars are dropped, not serialised, so a consumer always fetches fresh rather than reading a stale detached entity - record() deliberately does not flush: the event row commits with the change it describes, so a rolled-back transaction leaves no event behind. A test pins exactly that - app:events:publish drains the outbox; five failed attempts park a row with its error rather than deleting it, because a silently dropped event is a loss with no trace. app:events:prune only ever removes published rows Reports - Resource utilisation separates available, occupied and active minutes. The gap between occupied and active is what exposes a bad segment definition, and available is multiplied by capacity so a three-chair room does not read as permanently over 100% - A resource with no calendar reports utilization: null, not zero — dividing by zero means something different from being idle - Plan accuracy compares planned against actual duration per service and flags both directions: running short wastes capacity that could have been sold. Its row links straight to editing that service's segments, because a report with no route to a fix does not get read Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
hamed
@@ -0,0 +1,65 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event\Command;
|
||||
|
||||
use Doctrine\DBAL\Connection;
|
||||
use Symfony\Component\Console\Attribute\AsCommand;
|
||||
use Symfony\Component\Console\Command\Command;
|
||||
use Symfony\Component\Console\Input\InputInterface;
|
||||
use Symfony\Component\Console\Input\InputOption;
|
||||
use Symfony\Component\Console\Output\OutputInterface;
|
||||
use Symfony\Component\Console\Style\SymfonyStyle;
|
||||
|
||||
/**
|
||||
* پاکسازی رویدادهای **منتشرشدهٔ** قدیمی.
|
||||
*
|
||||
* ردیف منتشرنشده هرگز حذف نمیشود، حتی اگر سالخورده باشد: آن یک رویداد گمشده است و
|
||||
* حذفش یعنی پاک کردن مدرکِ همان گمشدن.
|
||||
*/
|
||||
#[AsCommand(name: 'app:events:prune', description: 'Delete published domain events older than a retention window.')]
|
||||
class PruneDomainEventsCommand extends Command
|
||||
{
|
||||
public function __construct(private readonly Connection $connection)
|
||||
{
|
||||
parent::__construct();
|
||||
}
|
||||
|
||||
protected function configure(): void
|
||||
{
|
||||
$this
|
||||
->addOption('days', null, InputOption::VALUE_REQUIRED, 'Retention window in days', '180')
|
||||
->addOption('dry-run', null, InputOption::VALUE_NONE, 'Report without deleting');
|
||||
}
|
||||
|
||||
protected function execute(InputInterface $input, OutputInterface $output): int
|
||||
{
|
||||
$io = new SymfonyStyle($input, $output);
|
||||
$before = time() - max(1, (int) $input->getOption('days')) * 86400;
|
||||
|
||||
$count = (int) $this->connection->fetchOne(
|
||||
'SELECT COUNT(*) FROM domain_events WHERE published_at IS NOT NULL AND occurred_at < ?',
|
||||
[$before],
|
||||
);
|
||||
|
||||
if ($count === 0) {
|
||||
$io->success('رویداد قابل حذفی نیست.');
|
||||
|
||||
return Command::SUCCESS;
|
||||
}
|
||||
|
||||
if ($input->getOption('dry-run')) {
|
||||
$io->note(sprintf('%d رویداد حذف میشد.', $count));
|
||||
|
||||
return Command::SUCCESS;
|
||||
}
|
||||
|
||||
$this->connection->executeStatement(
|
||||
'DELETE FROM domain_events WHERE published_at IS NOT NULL AND occurred_at < ?',
|
||||
[$before],
|
||||
);
|
||||
|
||||
$io->success(sprintf('%d رویداد حذف شد.', $count));
|
||||
|
||||
return Command::SUCCESS;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event\Command;
|
||||
|
||||
use App\Shared\Event\Entity\DomainEventLog;
|
||||
use App\Shared\Event\Repository\DomainEventLogRepository;
|
||||
use Doctrine\ORM\EntityManagerInterface;
|
||||
use Symfony\Component\Console\Attribute\AsCommand;
|
||||
use Symfony\Component\Console\Command\Command;
|
||||
use Symfony\Component\Console\Input\InputInterface;
|
||||
use Symfony\Component\Console\Input\InputOption;
|
||||
use Symfony\Component\Console\Output\OutputInterface;
|
||||
use Symfony\Component\Console\Style\SymfonyStyle;
|
||||
use Symfony\Component\Messenger\MessageBusInterface;
|
||||
|
||||
/**
|
||||
* انتشار صندوق خروجی: ردیفهای `published_at IS NULL` به messenger میروند.
|
||||
*
|
||||
* شکست انتشار ردیف را نمیکشد؛ `attempts` بالا میرود و خطا ثبت میشود. بعد از سقف
|
||||
* تلاش، ردیف با خطایش باقی میماند تا ادمین ببیند — حذف خاموش یعنی رویداد گمشدهٔ بیرد.
|
||||
*/
|
||||
#[AsCommand(name: 'app:events:publish', description: 'Publish pending domain events from the outbox.')]
|
||||
class PublishDomainEventsCommand extends Command
|
||||
{
|
||||
public function __construct(
|
||||
private readonly DomainEventLogRepository $events,
|
||||
private readonly MessageBusInterface $bus,
|
||||
private readonly EntityManagerInterface $em,
|
||||
) {
|
||||
parent::__construct();
|
||||
}
|
||||
|
||||
protected function configure(): void
|
||||
{
|
||||
$this->addOption('limit', null, InputOption::VALUE_REQUIRED, 'How many events to publish per run', '100');
|
||||
}
|
||||
|
||||
protected function execute(InputInterface $input, OutputInterface $output): int
|
||||
{
|
||||
$io = new SymfonyStyle($input, $output);
|
||||
$pending = $this->events->findPending(max(1, (int) $input->getOption('limit')));
|
||||
|
||||
$published = 0;
|
||||
$failed = 0;
|
||||
|
||||
foreach ($pending as $event) {
|
||||
try {
|
||||
$this->bus->dispatch(new \App\Shared\Event\Message\DomainEventMessage(
|
||||
$event->getUuid(),
|
||||
$event->getName(),
|
||||
$event->getEntityType(),
|
||||
$event->getEntityId(),
|
||||
$event->getPayload(),
|
||||
$event->getOccurredAt(),
|
||||
));
|
||||
|
||||
$event->markPublished();
|
||||
$published++;
|
||||
} catch (\Throwable $e) {
|
||||
$event->markFailed($e->getMessage());
|
||||
$failed++;
|
||||
}
|
||||
}
|
||||
|
||||
if ($pending !== []) {
|
||||
$this->em->flush();
|
||||
}
|
||||
|
||||
$io->success(sprintf('%d رویداد منتشر شد، %d ناموفق.', $published, $failed));
|
||||
|
||||
return Command::SUCCESS;
|
||||
}
|
||||
|
||||
/** @return DomainEventLog[] */
|
||||
public function pending(int $limit = 100): array
|
||||
{
|
||||
return $this->events->findPending($limit);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event;
|
||||
|
||||
use App\Shared\Event\Entity\DomainEventLog;
|
||||
use Doctrine\ORM\EntityManagerInterface;
|
||||
|
||||
/**
|
||||
* تنها نقطهٔ ثبت رویداد دامنه.
|
||||
*
|
||||
* `record()` عمداً **flush نمیکند**: ردیف رویداد باید در همان تراکنشی commit شود که
|
||||
* خودِ تغییر را انجام میدهد. اگر اینجا flush میکردیم، rollbackِ تراکنش اصلی رویدادی
|
||||
* را جا میگذاشت که هرگز اتفاق نیفتاده.
|
||||
*/
|
||||
final class DomainEventPublisher
|
||||
{
|
||||
public function __construct(
|
||||
private readonly EntityManagerInterface $em,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* @param array<string, mixed> $payload فقط uuid و اسکالر
|
||||
*/
|
||||
public function record(string $entityType, int $entityId, string $name, array $payload, ?int $occurredAt = null): DomainEventLog
|
||||
{
|
||||
if (!in_array($name, DomainEvents::ALL, true)) {
|
||||
throw new \InvalidArgumentException(sprintf('Unknown domain event "%s".', $name));
|
||||
}
|
||||
|
||||
$event = new DomainEventLog($entityType, $entityId, $name, $payload, $occurredAt);
|
||||
|
||||
$this->em->persist($event);
|
||||
|
||||
return $event;
|
||||
}
|
||||
|
||||
/**
|
||||
* ثبت + flush — برای جاهایی که فراخوان تراکنش باز ندارد.
|
||||
*
|
||||
* @param array<string, mixed> $payload
|
||||
*/
|
||||
public function recordAndFlush(string $entityType, int $entityId, string $name, array $payload, ?int $occurredAt = null): DomainEventLog
|
||||
{
|
||||
$event = $this->record($entityType, $entityId, $name, $payload, $occurredAt);
|
||||
$this->em->flush();
|
||||
|
||||
return $event;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event;
|
||||
|
||||
/**
|
||||
* فهرست بستهٔ نام رویدادها — بند ۱۶ مستند.
|
||||
*
|
||||
* نام رویداد قرارداد عمومی است: مصرفکننده روی رشته شرط میگذارد. تایپوی یک حرفی
|
||||
* یعنی رویدادی که هیچکس نمیشنود و هیچ خطایی هم نمیدهد، پس فهرست بسته است.
|
||||
*/
|
||||
final class DomainEvents
|
||||
{
|
||||
public const HOLD_CREATED = 'HoldCreated';
|
||||
public const APPOINTMENT_BOOKED = 'AppointmentBooked';
|
||||
public const APPOINTMENT_CANCELLED = 'AppointmentCancelled';
|
||||
public const APPOINTMENT_RESCHEDULED = 'AppointmentRescheduled';
|
||||
public const PATIENT_NO_SHOW = 'PatientNoShow';
|
||||
public const APPOINTMENT_COMPLETED = 'AppointmentCompleted';
|
||||
public const RESOURCE_BLOCKED = 'ResourceBlocked';
|
||||
public const RESOURCE_RELEASED = 'ResourceReleased';
|
||||
public const COURSE_STARTED = 'CourseStarted';
|
||||
public const COURSE_SESSION_COMPLETED = 'CourseSessionCompleted';
|
||||
public const COURSE_COMPLETED = 'CourseCompleted';
|
||||
public const PACKAGE_PURCHASED = 'PackagePurchased';
|
||||
public const CREDIT_CONSUMED = 'CreditConsumed';
|
||||
public const CREDIT_REFUNDED = 'CreditRefunded';
|
||||
|
||||
public const ALL = [
|
||||
self::HOLD_CREATED,
|
||||
self::APPOINTMENT_BOOKED,
|
||||
self::APPOINTMENT_CANCELLED,
|
||||
self::APPOINTMENT_RESCHEDULED,
|
||||
self::PATIENT_NO_SHOW,
|
||||
self::APPOINTMENT_COMPLETED,
|
||||
self::RESOURCE_BLOCKED,
|
||||
self::RESOURCE_RELEASED,
|
||||
self::COURSE_STARTED,
|
||||
self::COURSE_SESSION_COMPLETED,
|
||||
self::COURSE_COMPLETED,
|
||||
self::PACKAGE_PURCHASED,
|
||||
self::CREDIT_CONSUMED,
|
||||
self::CREDIT_REFUNDED,
|
||||
];
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event\Entity;
|
||||
|
||||
use App\Shared\Event\Repository\DomainEventLogRepository;
|
||||
use App\Shared\Tenant\TenantOwnedTrait;
|
||||
use Doctrine\ORM\Mapping as ORM;
|
||||
use Symfony\Component\Uid\Uuid;
|
||||
|
||||
/**
|
||||
* صندوق خروجی رویدادهای دامنه (outbox).
|
||||
*
|
||||
* ردیف رویداد **در همان تراکنشی** نوشته میشود که خودِ تغییر را انجام میدهد، و یک
|
||||
* worker بعداً منتشرش میکند. بدون این الگو دو حالت شکست ممکن است:
|
||||
*
|
||||
* | حالت | نتیجه |
|
||||
* |---|---|
|
||||
* | انتشار پیش از commit، بعد rollback | پیامک رفته، نوبتی وجود ندارد |
|
||||
* | commit موفق، انتشار شکست خورد | نوبت هست، هیچکس مطلع نشد |
|
||||
*
|
||||
* با outbox حداکثر **تأخیر** داریم، هرگز گمشدن.
|
||||
*
|
||||
* این جدول با `AppointmentEvent` موجود اشتباه نشود: آن تاریخچهٔ وضعیت یک نوبت است،
|
||||
* این اعلان تغییر به بیرونِ دامنه.
|
||||
*/
|
||||
#[ORM\Entity(repositoryClass: DomainEventLogRepository::class)]
|
||||
#[ORM\Table(name: 'domain_events')]
|
||||
#[ORM\Index(columns: ['published_at', 'occurred_at'], name: 'idx_de_pending')]
|
||||
#[ORM\Index(columns: ['entity_type', 'entity_id', 'occurred_at'], name: 'idx_de_tenant')]
|
||||
#[ORM\Index(columns: ['name', 'occurred_at'], name: 'idx_de_name')]
|
||||
class DomainEventLog
|
||||
{
|
||||
use TenantOwnedTrait;
|
||||
|
||||
/** سقف تلاش — ردیف مرده با خطایش میماند تا دیده شود، حذف نمیشود. */
|
||||
public const MAX_ATTEMPTS = 5;
|
||||
|
||||
#[ORM\Id]
|
||||
#[ORM\GeneratedValue]
|
||||
#[ORM\Column(type: 'bigint')]
|
||||
private ?string $id = null;
|
||||
|
||||
/** شناسهٔ idempotency برای مصرفکننده. */
|
||||
#[ORM\Column(type: 'string', length: 36, unique: true)]
|
||||
private string $uuid;
|
||||
|
||||
#[ORM\Column(type: 'string', length: 60)]
|
||||
private string $name;
|
||||
|
||||
/** @var array<string, scalar|null> */
|
||||
#[ORM\Column(type: 'json')]
|
||||
private array $payload;
|
||||
|
||||
/** زمان **وقوع**، نه انتشار. */
|
||||
#[ORM\Column(name: 'occurred_at', type: 'integer')]
|
||||
private int $occurredAt;
|
||||
|
||||
#[ORM\Column(name: 'published_at', type: 'integer', nullable: true)]
|
||||
private ?int $publishedAt = null;
|
||||
|
||||
#[ORM\Column(type: 'smallint', options: ['default' => 0])]
|
||||
private int $attempts = 0;
|
||||
|
||||
#[ORM\Column(name: 'last_error', type: 'string', length: 255, nullable: true)]
|
||||
private ?string $lastError = null;
|
||||
|
||||
/** @param array<string, mixed> $payload */
|
||||
public function __construct(string $entityType, int $entityId, string $name, array $payload, ?int $occurredAt = null)
|
||||
{
|
||||
$this->uuid = Uuid::v4()->toRfc4122();
|
||||
$this->name = $name;
|
||||
$this->payload = self::scalarsOnly($payload);
|
||||
$this->occurredAt = $occurredAt ?? time();
|
||||
|
||||
$this->assignTenantPair($entityType, $entityId);
|
||||
}
|
||||
|
||||
/**
|
||||
* هیچ entity ای در رویداد نیست — فقط uuid و اسکالر.
|
||||
*
|
||||
* entity در پیام async یعنی سریالسازی، detach شدن، و دادهٔ کهنه؛ مصرفکننده باید
|
||||
* خودش با uuid واکشی کند تا همیشه تازهترین حالت را ببیند.
|
||||
*
|
||||
* @param array<string, mixed> $payload
|
||||
* @return array<string, scalar|null>
|
||||
*/
|
||||
private static function scalarsOnly(array $payload): array
|
||||
{
|
||||
return array_filter($payload, static fn (mixed $v): bool => is_scalar($v) || $v === null);
|
||||
}
|
||||
|
||||
public function getId(): ?string { return $this->id; }
|
||||
public function getUuid(): string { return $this->uuid; }
|
||||
public function getName(): string { return $this->name; }
|
||||
public function getPayload(): array { return $this->payload; }
|
||||
public function getOccurredAt(): int { return $this->occurredAt; }
|
||||
public function getPublishedAt(): ?int { return $this->publishedAt; }
|
||||
public function getAttempts(): int { return $this->attempts; }
|
||||
public function getLastError(): ?string { return $this->lastError; }
|
||||
public function isPublished(): bool { return $this->publishedAt !== null; }
|
||||
|
||||
public function markPublished(?int $at = null): self
|
||||
{
|
||||
$this->publishedAt = $at ?? time();
|
||||
$this->lastError = null;
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
public function markFailed(string $error): self
|
||||
{
|
||||
$this->attempts++;
|
||||
$this->lastError = mb_substr($error, 0, 255);
|
||||
|
||||
return $this;
|
||||
}
|
||||
|
||||
/** @return array<string, mixed> */
|
||||
public function toArray(): array
|
||||
{
|
||||
return [
|
||||
'uuid' => $this->uuid,
|
||||
'name' => $this->name,
|
||||
'payload' => (object) $this->payload,
|
||||
'occurred_at' => $this->occurredAt,
|
||||
'published_at' => $this->publishedAt,
|
||||
'attempts' => $this->attempts,
|
||||
'last_error' => $this->lastError,
|
||||
];
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event\Message;
|
||||
|
||||
/**
|
||||
* پیام async یک رویداد دامنه.
|
||||
*
|
||||
* `uuid` شناسهٔ idempotency است: messenger ممکن است پیام را دوباره تحویل بدهد، و
|
||||
* **مصرفکننده** باید تکراری را تشخیص بدهد — نه اینکه رویداد تضمین یکتایی بدهد.
|
||||
*/
|
||||
final readonly class DomainEventMessage
|
||||
{
|
||||
/** @param array<string, scalar|null> $payload */
|
||||
public function __construct(
|
||||
public string $uuid,
|
||||
public string $name,
|
||||
public string $entityType,
|
||||
public int $entityId,
|
||||
public array $payload,
|
||||
public int $occurredAt,
|
||||
) {}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event\MessageHandler;
|
||||
|
||||
use App\Shared\Event\Message\DomainEventMessage;
|
||||
use Psr\Log\LoggerInterface;
|
||||
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
|
||||
|
||||
/**
|
||||
* درزِ اتصال مصرفکنندهها.
|
||||
*
|
||||
* خودش کاری جز ثبت لاگ نمیکند و **نباید بکند**: پیامک، حسابداری و گزارش هر کدام
|
||||
* مصرفکنندهٔ خودشان را کنار این ثبت میکنند. وجودش لازم است چون messenger پیامِ
|
||||
* بدون handler را خطا میدهد، و آن خطا در صندوق خروجی بهعنوان «شکست انتشار» ثبت
|
||||
* میشد — یعنی یک ایراد پیکربندی، شبیه یک رویداد گمشده به نظر میرسید.
|
||||
*
|
||||
* مصرفکنندهٔ تازه باید **idempotent** باشد: messenger ممکن است پیام را دوباره تحویل
|
||||
* بدهد و `DomainEventMessage::$uuid` همان شناسهای است که با آن تکراری را میشناسد.
|
||||
*/
|
||||
#[AsMessageHandler]
|
||||
final class DomainEventHandler
|
||||
{
|
||||
public function __construct(
|
||||
private readonly LoggerInterface $logger,
|
||||
) {}
|
||||
|
||||
public function __invoke(DomainEventMessage $message): void
|
||||
{
|
||||
$this->logger->info('domain event published', [
|
||||
'uuid' => $message->uuid,
|
||||
'name' => $message->name,
|
||||
'entity_type' => $message->entityType,
|
||||
'entity_id' => $message->entityId,
|
||||
]);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
<?php
|
||||
|
||||
namespace App\Shared\Event\Repository;
|
||||
|
||||
use App\Shared\Event\Entity\DomainEventLog;
|
||||
use Doctrine\Bundle\DoctrineBundle\Repository\ServiceEntityRepository;
|
||||
use Doctrine\Persistence\ManagerRegistry;
|
||||
|
||||
/** @extends ServiceEntityRepository<DomainEventLog> */
|
||||
class DomainEventLogRepository extends ServiceEntityRepository
|
||||
{
|
||||
public function __construct(ManagerRegistry $registry)
|
||||
{
|
||||
parent::__construct($registry, DomainEventLog::class);
|
||||
}
|
||||
|
||||
/**
|
||||
* ردیفهای منتشرنشدهای که هنوز سقف تلاش را رد نکردهاند.
|
||||
*
|
||||
* @return DomainEventLog[]
|
||||
*/
|
||||
public function findPending(int $limit = 100): array
|
||||
{
|
||||
return $this->createQueryBuilder('e')
|
||||
->where('e.publishedAt IS NULL')
|
||||
->andWhere('e.attempts < :max')
|
||||
->setParameter('max', DomainEventLog::MAX_ATTEMPTS)
|
||||
->orderBy('e.occurredAt', 'ASC')
|
||||
->addOrderBy('e.id', 'ASC')
|
||||
->setMaxResults($limit)
|
||||
->getQuery()
|
||||
->getResult();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return DomainEventLog[]
|
||||
*/
|
||||
public function search(?string $name, ?string $entityType, ?int $entityId, int $limit = 100): array
|
||||
{
|
||||
$qb = $this->createQueryBuilder('e')
|
||||
->orderBy('e.occurredAt', 'DESC')
|
||||
->addOrderBy('e.id', 'DESC')
|
||||
->setMaxResults(min($limit, 500));
|
||||
|
||||
if ($name !== null && $name !== '') {
|
||||
$qb->andWhere('e.name = :name')->setParameter('name', $name);
|
||||
}
|
||||
|
||||
if ($entityType !== null && $entityId !== null) {
|
||||
$qb->andWhere('e.entityType = :type')
|
||||
->andWhere('e.entityId = :id')
|
||||
->setParameter('type', $entityType)
|
||||
->setParameter('id', $entityId);
|
||||
}
|
||||
|
||||
return $qb->getQuery()->getResult();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user