Flux event-driven

ASYDATA decuplează modulele și procesele prin evenimente. Un domain event ridicat în timpul unei unități de lucru ajunge garantat pe topicul Kafka firm.events (prin outbox pattern), de unde mai mulți consumeri independenți îl preiau — fără ca producătorul să știe cine ascultă.

Principiul de bază: niciun consumator nu depinde de altul, iar producătorul nu depinde de niciun consumator. WebApi scrie evenimentul și își vede de treabă; worker-ul Telegram și cel de Web Push reacționează separat, fiecare cu offset-ul lui.

Fără MediatR / Wolverine

Nu se folosește niciun framework de messaging (MediatR, Wolverine, MassTransit — toate respinse în CLAUDE.md / docs/DECISIONS.md). Outbox-ul e propriu, iar publicarea se face cu Confluent.Kafka direct. Diferă de docs/ARCHITECTURE.md, care propunea Wolverine pentru mediator + transport.

Fluxul complet

[1] Domeniu          aggregate.Raise(new InvoiceReceivedEvent(...))
                            │  (AggregateRoot.DomainEvents)
                            ▼
[2] SaveChanges      OutboxSaveChangesInterceptor
                       → serializează fiecare event → rând în `outbox_events`
                       → ACEEAȘI tranzacție EF Core (atomic cu schimbarea de stare)
                            │
                            ▼ commit
[3] Relay            OutboxRelayService (BackgroundService, poll 500ms)
                       → citește rândurile cu processed_at IS NULL
                       → KafkaEventPublisher.PublishAsync(...)
                       → setează processed_at = now
                            │
                            ▼
[4] Kafka            topic `firm.events`
                       Key = firm_id   header event-type = CLR FullName
                            │
              ┌─────────────┴──────────────┐
              ▼                             ▼
[5] FirmEventsConsumer            WebPushConsumer
    group: telegram-worker        group: webpush-worker
    → Telegram Bot API            → browser Web Push

[1] Domain event — AggregateRoot.Raise

Entitățile de domeniu moștenesc AggregateRoot<TId> (asydata-app-backend-dotnet/src/Asy.Domain/SharedKernel/AggregateRoot.cs) și ridică evenimente cu Raise(...):

public abstract class AggregateRoot<TId> where TId : struct
{
    private readonly List<IDomainEvent> _domainEvents = [];
    public IReadOnlyList<IDomainEvent> DomainEvents => _domainEvents;
    protected void Raise(IDomainEvent @event) => _domainEvents.Add(@event);
    public void ClearDomainEvents() => _domainEvents.Clear();
}

Evenimentele se acumulează în memorie pe agregat; nu se publică nimic încă.

[2] Interceptor → tabela outbox_events

OutboxSaveChangesInterceptor (asydata-app-backend-dotnet/src/Asy.Infrastructure/Persistence/Outbox/OutboxSaveChangesInterceptor.cs) prinde SaveChangesAsync, parcurge change tracker-ul, ia DomainEvents de pe fiecare agregat, serializează fiecare eveniment într-un rând OutboxEvent adăugat în aceeași tranzacție, apoi golește colecția (ca să nu se republice).

Rândul (asydata-app-backend-dotnet/src/Asy.Domain/SharedKernel/OutboxEvent.cs) conține:

Câmp	Conținut
EventType	numele CLR complet al evenimentului (ex. Asy.Domain.Efactura.Events.InvoiceReceived)
Topic	topicul Kafka (default firm.events)
PartitionKey	cheia de partiție — de regulă firm_id
PayloadJson	evenimentul serializat
ProcessedAt	NULL până la publicare
AttemptCount / LastError	pentru retry

Topicul și cheia se pot personaliza per eveniment prin interfețele IRoutedEvent (override topic) și IPartitionedEvent (override partition key); altfel se folosesc default-urile (firm.events, respectiv Id-ul agregatului).

Garanția outbox

tranzacție: { schimbare agregat  +  rând outbox }  → commit împreună / rollback împreună

Dacă tranzacția se commit-ă, rândul outbox este salvat → evenimentul va fi publicat (eventually). Dacă eșuează, totul se anulează. Elimină cele două situații clasice: „am scris în DB dar n-am publicat” și „am publicat dar n-am scris”. Publicarea se face doar după commit.

[3] Relay → Kafka

OutboxRelayService (asydata-app-backend-dotnet/src/Asy.Infrastructure/Persistence/Outbox/OutboxRelayService.cs) este un BackgroundService care:

• face poll la 500 ms (dacă găsește de lucru, reia imediat; altfel doarme intervalul);
• citește un batch (50 de rânduri) cu processed_at IS NULL, ordonat după created_at (ordine cauzală);
• publică fiecare rând prin IEventPublisher, apoi setează processed_at = now;
• la eșec: incrementează AttemptCount, salvează LastError, lasă rândul neprocesat → retry la următorul tick.

De ce polling și nu CDC/Debezium: pentru volumul ASYDATA, 500 ms e „near real-time” suficient și ieftin (index parțial pe WHERE processed_at IS NULL), fără infrastructură suplimentară.

KafkaEventPublisher (asydata-app-backend-dotnet/src/Asy.Infrastructure/Messaging/KafkaEventPublisher.cs) este un producer Confluent.Kafka singleton. Setări relevante:

Setare	Valoare	Motiv
Acks	All	așteaptă toate replicile in-sync — durabilitate
EnableIdempotence	true	fără duplicate, ordine garantată per partition
CompressionType	Snappy	throughput
MessageSendMaxRetries	10 (RetryBackoffMs = 200)	rezistență la erori tranzitorii

Mesajul are Key = firm_id (partiționare → ordinea evenimentelor unei firme e păstrată pe aceeași partiție) și headere event-type (numele CLR complet) + content-type: application/json.

Fallback fără Kafka

Când Kafka:BootstrapServers nu e configurat (dev fără Docker), DI înregistrează NullEventPublisher în loc de cel Kafka — evenimentele sunt aruncate silențios (doar log Debug). Vezi asydata-app-backend-dotnet/src/Asy.Infrastructure/DependencyInjection.cs.

[4]–[5] Consumeri independenți

Worker-ul Asy.Workers.Telegram (asydata-app-backend-dotnet/src/Workers/Asy.Workers.Telegram/Program.cs) găzduiește doi consumeri separați, fiecare cu propriul consumer group → fiecare primește toate evenimentele, niciunul nu blochează pe celălalt:

Consumator	Grup	Acțiune
FirmEventsConsumer	telegram-worker	rutează evenimentul → mesaje/carduri Telegram
WebPushConsumer	webpush-worker	fan-out → notificări browser Web Push

Setări consumer (ambele): EnableAutoCommit = false → commit manual abia după procesare reușită; AutoOffsetReset = Earliest; SessionTimeoutMs = 30000.

FirmEventsConsumer (asydata-app-backend-dotnet/src/Workers/Asy.Workers.Telegram/FirmEventsConsumer.cs) citește header-ul event-type, extrage firm_id din cheia mesajului, apoi dispecerizează după tipul scurt al evenimentului (ex. InvoiceReceivedEvent, PermissionRequestedEvent, InvoiceEFacturaStatusChangedEvent) — unele evenimente merg către chat-ul cererii, altele se difuzează către toate chat-urile firmei cu toggle-ul activ.

Idempotență

Consumul nu e încă idempotent: la restart de consumator pot apărea mesaje Telegram duplicate (semantică at-most-once + risc mic acceptat). Planul notat în cod: o tabelă telegram_dispatched cu perechi (event_id, chat_id).

Topic și principii

• Un topic principal: firm.events (nu un topic per bounded context). docs/ARCHITECTURE.md propunea topicuri separate (company.events, clients.events, …) + DLQ gestionat de Wolverine; implementarea actuală consolidează pe firm.events, cu rutare prin header-ul event-type.
• Cheia = firm_id → ordine per firmă.
• Decuplare totală: producătorul nu cunoaște consumatorii; consumatorii nu se cunosc între ei.
• Livrare garantată prin outbox (publici doar după commit) + retry în relay.

Legături

• Unde se așază Kafka în peisaj: Context de sistem.
• Ce devine cu evenimentele în Telegram / push: Telegram bot și Notificări și Web Push.
• Extinderea cross-modul prin evenimente: Clean Architecture + module.