Event-Driven আর্কিটেকচার ও CQRS

package main

import (
	"encoding/json"
	"fmt"
	"sync"
	"time"
)

// --- Event envelope ---

type EventEnvelope struct {
	GlobalPosition int             `json:"globalPosition"`
	StreamID       string          `json:"streamId"`
	StreamVersion  int             `json:"streamVersion"`
	Type           string          `json:"type"`
	SchemaVersion  int             `json:"schemaVersion"`
	Payload        json.RawMessage `json:"payload"`
	RecordedAt     time.Time       `json:"recordedAt"`
	CorrelationID  string          `json:"correlationId"`
}

type AppendCandidate struct {
	Type          string
	SchemaVersion int
	Payload       any
	CorrelationID string
}

type ConcurrencyError struct {
	StreamID string
	Expected int
	Actual   int
}

func (e *ConcurrencyError) Error() string {
	return fmt.Sprintf("stream %s: expected version %d, found %d", e.StreamID, e.Expected, e.Actual)
}

type Subscriber func(EventEnvelope)

// --- Event store ---

type EventStore struct {
	mu          sync.RWMutex
	log         []EventEnvelope
	versions    map[string]int
	subscribers []Subscriber
}

func NewEventStore() *EventStore {
	return &EventStore{versions: make(map[string]int)}
}

// Append uses optimistic concurrency: expectedVersion -1 means the stream
// must not exist yet. This replaces a distributed lock entirely.
func (s *EventStore) Append(streamID string, expectedVersion int, candidates []AppendCandidate) ([]EventEnvelope, error) {
	s.mu.Lock()

	current, ok := s.versions[streamID]
	if !ok {
		current = -1
	}
	if current != expectedVersion {
		s.mu.Unlock()
		return nil, &ConcurrencyError{StreamID: streamID, Expected: expectedVersion, Actual: current}
	}

	written := make([]EventEnvelope, 0, len(candidates))
	version := current

	for _, c := range candidates {
		raw, err := json.Marshal(c.Payload)
		if err != nil {
			s.mu.Unlock()
			return nil, err
		}
		version++
		env := EventEnvelope{
			GlobalPosition: len(s.log),
			StreamID:       streamID,
			StreamVersion:  version,
			Type:           c.Type,
			SchemaVersion:  c.SchemaVersion,
			Payload:        raw,
			RecordedAt:     time.Now().UTC(),
			CorrelationID:  c.CorrelationID,
		}
		s.log = append(s.log, env)
		written = append(written, env)
	}
	s.versions[streamID] = version

	subs := make([]Subscriber, len(s.subscribers))
	copy(subs, s.subscribers)
	s.mu.Unlock()

	for _, env := range written {
		for _, sub := range subs {
			sub(env)
		}
	}
	return written, nil
}

func (s *EventStore) ReadStream(streamID string) []EventEnvelope {
	s.mu.RLock()
	defer s.mu.RUnlock()
	out := []EventEnvelope{}
	for _, e := range s.log {
		if e.StreamID == streamID {
			out = append(out, e)
		}
	}
	return out
}

func (s *EventStore) ReadAll(from, limit int) []EventEnvelope {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if from >= len(s.log) {
		return nil
	}
	end := from + limit
	if end > len(s.log) {
		end = len(s.log)
	}
	out := make([]EventEnvelope, end-from)
	copy(out, s.log[from:end])
	return out
}

func (s *EventStore) Subscribe(fn Subscriber) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.subscribers = append(s.subscribers, fn)
}

// --- Domain events ---

type ManuscriptRegistered struct {
	Title string `json:"title"`
	Shelf string `json:"shelf"`
}

type ManuscriptBorrowed struct {
	Borrower string `json:"borrower"`
	DueDays  int    `json:"dueDays"`
}

type ManuscriptReturned struct {
	Borrower  string `json:"borrower"`
	Condition string `json:"condition"` // good | damaged
}

// --- Aggregate ---

type ManuscriptState struct {
	Exists     bool
	Title      string
	Shelf      string
	BorrowedBy string
	Version    int
}

func applyEvent(state ManuscriptState, e EventEnvelope) ManuscriptState {
	switch e.Type {
	case "ManuscriptRegistered":
		var p ManuscriptRegistered
		_ = json.Unmarshal(e.Payload, &p)
		state.Exists = true
		state.Title = p.Title
		state.Shelf = p.Shelf
	case "ManuscriptBorrowed":
		var p ManuscriptBorrowed
		_ = json.Unmarshal(e.Payload, &p)
		state.BorrowedBy = p.Borrower
	case "ManuscriptReturned":
		state.BorrowedBy = ""
	}
	state.Version = e.StreamVersion
	return state
}

func rehydrate(events []EventEnvelope) ManuscriptState {
	state := ManuscriptState{Version: -1}
	for _, e := range events {
		state = applyEvent(state, e)
	}
	return state
}

// --- Command handlers ---

type LibraryService struct {
	store *EventStore
}

func (svc *LibraryService) Register(id, title, shelf, correlationID string) error {
	state := rehydrate(svc.store.ReadStream(id))
	if state.Exists {
		return fmt.Errorf("manuscript %s already registered", id)
	}
	_, err := svc.store.Append(id, state.Version, []AppendCandidate{{
		Type:          "ManuscriptRegistered",
		SchemaVersion: 2,
		Payload:       ManuscriptRegistered{Title: title, Shelf: shelf},
		CorrelationID: correlationID,
	}})
	return err
}

func (svc *LibraryService) Borrow(id, borrower, correlationID string) error {
	state := rehydrate(svc.store.ReadStream(id))
	if !state.Exists {
		return fmt.Errorf("manuscript %s not found", id)
	}
	if state.BorrowedBy != "" {
		return fmt.Errorf("already borrowed by %s", state.BorrowedBy)
	}
	_, err := svc.store.Append(id, state.Version, []AppendCandidate{{
		Type:          "ManuscriptBorrowed",
		SchemaVersion: 1,
		Payload:       ManuscriptBorrowed{Borrower: borrower, DueDays: 14},
		CorrelationID: correlationID,
	}})
	return err
}

func (svc *LibraryService) GiveBack(id, borrower, condition, correlationID string) error {
	state := rehydrate(svc.store.ReadStream(id))
	if state.BorrowedBy != borrower {
		return fmt.Errorf("%s does not hold %s", borrower, id)
	}
	_, err := svc.store.Append(id, state.Version, []AppendCandidate{{
		Type:          "ManuscriptReturned",
		SchemaVersion: 1,
		Payload:       ManuscriptReturned{Borrower: borrower, Condition: condition},
		CorrelationID: correlationID,
	}})
	return err
}

// --- Projections ---

type Projection interface {
	Name() string
	Checkpoint() int
	Handle(EventEnvelope)
	Reset()
}

type shelfRow struct {
	Title  string
	Shelf  string
	Holder string
}

type ShelfProjection struct {
	checkpoint int
	rows       map[string]*shelfRow
}

func NewShelfProjection() *ShelfProjection {
	return &ShelfProjection{rows: make(map[string]*shelfRow)}
}

func (p *ShelfProjection) Name() string    { return "shelf-view" }
func (p *ShelfProjection) Checkpoint() int { return p.checkpoint }

func (p *ShelfProjection) Handle(e EventEnvelope) {
	switch e.Type {
	case "ManuscriptRegistered":
		var d ManuscriptRegistered
		_ = json.Unmarshal(e.Payload, &d)
		p.rows[e.StreamID] = &shelfRow{Title: d.Title, Shelf: d.Shelf}
	case "ManuscriptBorrowed":
		var d ManuscriptBorrowed
		_ = json.Unmarshal(e.Payload, &d)
		if row, ok := p.rows[e.StreamID]; ok {
			row.Holder = d.Borrower
		}
	case "ManuscriptReturned":
		if row, ok := p.rows[e.StreamID]; ok {
			row.Holder = ""
		}
	}
	p.checkpoint = e.GlobalPosition + 1
}

func (p *ShelfProjection) Reset() {
	p.rows = make(map[string]*shelfRow)
	p.checkpoint = 0
}

func (p *ShelfProjection) OnShelf(shelf string) []string {
	out := []string{}
	for id, row := range p.rows {
		if row.Shelf == shelf && row.Holder == "" {
			out = append(out, id)
		}
	}
	return out
}

type borrowerRow struct {
	Borrowed int
	Damaged  int
}

type BorrowerProjection struct {
	checkpoint int
	counts     map[string]*borrowerRow
}

func NewBorrowerProjection() *BorrowerProjection {
	return &BorrowerProjection{counts: make(map[string]*borrowerRow)}
}

func (p *BorrowerProjection) Name() string    { return "borrower-view" }
func (p *BorrowerProjection) Checkpoint() int { return p.checkpoint }

func (p *BorrowerProjection) Handle(e EventEnvelope) {
	switch e.Type {
	case "ManuscriptBorrowed":
		var d ManuscriptBorrowed
		_ = json.Unmarshal(e.Payload, &d)
		p.row(d.Borrower).Borrowed++
	case "ManuscriptReturned":
		var d ManuscriptReturned
		_ = json.Unmarshal(e.Payload, &d)
		if d.Condition == "damaged" {
			p.row(d.Borrower).Damaged++
		}
	}
	p.checkpoint = e.GlobalPosition + 1
}

func (p *BorrowerProjection) row(name string) *borrowerRow {
	if r, ok := p.counts[name]; ok {
		return r
	}
	r := &borrowerRow{}
	p.counts[name] = r
	return r
}

func (p *BorrowerProjection) Reset() {
	p.counts = make(map[string]*borrowerRow)
	p.checkpoint = 0
}

func (p *BorrowerProjection) Report() map[string]borrowerRow {
	out := make(map[string]borrowerRow, len(p.counts))
	for k, v := range p.counts {
		out[k] = *v
	}
	return out
}

// --- Projection runner ---

type ProjectionRunner struct {
	store      *EventStore
	projection Projection
}

func (r *ProjectionRunner) CatchUp() int {
	processed := 0
	for {
		batch := r.store.ReadAll(r.projection.Checkpoint(), 200)
		if len(batch) == 0 {
			return processed
		}
		for _, e := range batch {
			r.projection.Handle(e)
			processed++
		}
	}
}

func (r *ProjectionRunner) Rebuild() int {
	fmt.Printf("[projection] rebuilding %s from position 0\n", r.projection.Name())
	r.projection.Reset()
	return r.CatchUp()
}

func main() {
	store := NewEventStore()
	library := &LibraryService{store: store}

	must(library.Register("manuscript-geometry-01", "Kitab al-Handasa", "cordoba-A3", "req-1001"))
	must(library.Borrow("manuscript-geometry-01", "al-biruni", "req-1002"))
	must(library.GiveBack("manuscript-geometry-01", "al-biruni", "damaged", "req-1003"))
	must(library.Borrow("manuscript-geometry-01", "ibn-al-haytham", "req-1004"))

	shelf := NewShelfProjection()
	borrower := NewBorrowerProjection()

	(&ProjectionRunner{store: store, projection: shelf}).CatchUp()
	(&ProjectionRunner{store: store, projection: borrower}).CatchUp()

	fmt.Println("available on cordoba-A3:", shelf.OnShelf("cordoba-A3"))
	fmt.Printf("borrower report: %+v\n", borrower.Report())

	rebuilt := (&ProjectionRunner{store: store, projection: borrower}).Rebuild()
	fmt.Printf("rebuilt from %d events\n", rebuilt)
}

func must(err error) {
	if err != nil {
		panic(err)
	}
}

package main

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/google/uuid"
)

// --- Rows ---

type OrderRow struct {
	ID           string
	Customer     string
	City         string
	AmountDirham int
	Status       string
}

type OutboxRow struct {
	ID          string
	AggregateID string
	Type        string
	Payload     map[string]any
	CreatedAt   time.Time
	PublishedAt *time.Time
	Attempts    int
}

// --- A tiny transactional store ---

type Database struct {
	mu     sync.Mutex
	Orders map[string]OrderRow
	Outbox []*OutboxRow
}

func NewDatabase() *Database {
	return &Database{Orders: make(map[string]OrderRow)}
}

type Transaction struct {
	staged []func()
}

func (t *Transaction) InsertOrder(db *Database, row OrderRow) {
	t.staged = append(t.staged, func() { db.Orders[row.ID] = row })
}

func (t *Transaction) InsertOutbox(db *Database, row *OutboxRow) {
	t.staged = append(t.staged, func() { db.Outbox = append(db.Outbox, row) })
}

// Transaction commits business rows and outbox rows together or not at all.
func (db *Database) Transaction(fn func(tx *Transaction)) {
	db.mu.Lock()
	defer db.mu.Unlock()
	tx := &Transaction{}
	fn(tx)
	for _, apply := range tx.staged {
		apply()
	}
}

// --- Broker ---

type Broker interface {
	Publish(topic, eventID string, payload map[string]any) error
}

type FlakyBroker struct {
	mu        sync.Mutex
	Delivered []struct {
		EventID string
		Payload map[string]any
	}
	FailNext int
}

func (b *FlakyBroker) Publish(topic, eventID string, payload map[string]any) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.FailNext > 0 {
		b.FailNext--
		return errors.New("broker unavailable for " + topic)
	}
	b.Delivered = append(b.Delivered, struct {
		EventID string
		Payload map[string]any
	}{eventID, payload})
	return nil
}

// --- Service ---

type OrderService struct {
	db *Database
}

func (s *OrderService) PlaceOrder(customer, city string, amount int) string {
	orderID := uuid.NewString()

	s.db.Transaction(func(tx *Transaction) {
		tx.InsertOrder(s.db, OrderRow{
			ID: orderID, Customer: customer, City: city,
			AmountDirham: amount, Status: "placed",
		})
		tx.InsertOutbox(s.db, &OutboxRow{
			ID:          uuid.NewString(),
			AggregateID: orderID,
			Type:        "OrderPlaced",
			Payload: map[string]any{
				"orderId": orderID, "customer": customer,
				"city": city, "amountDirham": amount,
			},
			CreatedAt: time.Now(),
		})
	})

	return orderID
}

// --- Relay ---

type OutboxRelay struct {
	db          *Database
	broker      Broker
	batchSize   int
	maxAttempts int
	stop        chan struct{}
}

func NewOutboxRelay(db *Database, b Broker) *OutboxRelay {
	return &OutboxRelay{db: db, broker: b, batchSize: 50, maxAttempts: 8, stop: make(chan struct{})}
}

// DrainOnce publishes first and marks second. A crash in between causes a
// duplicate delivery, which consumers must tolerate. The reverse order
// would silently lose events.
func (r *OutboxRelay) DrainOnce() int {
	r.db.mu.Lock()
	pending := []*OutboxRow{}
	for _, row := range r.db.Outbox {
		if row.PublishedAt == nil && row.Attempts < r.maxAttempts {
			pending = append(pending, row)
			if len(pending) >= r.batchSize {
				break
			}
		}
	}
	r.db.mu.Unlock()

	published := 0
	for _, row := range pending {
		if err := r.broker.Publish(row.Type, row.ID, row.Payload); err != nil {
			row.Attempts++
			backoff := time.Duration(1<<uint(row.Attempts)) * 100 * time.Millisecond
			if backoff > 30*time.Second {
				backoff = 30 * time.Second
			}
			fmt.Printf("[outbox] %s %s failed (attempt %d), retry in %s: %v\n",
				row.Type, row.ID, row.Attempts, backoff, err)
			break // preserve ordering
		}
		now := time.Now()
		row.PublishedAt = &now
		published++
	}
	return published
}

func (r *OutboxRelay) Run(interval time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			r.DrainOnce()
		case <-r.stop:
			return
		}
	}
}

func (r *OutboxRelay) Stop() { close(r.stop) }

func (r *OutboxRelay) DeadLetters() []*OutboxRow {
	out := []*OutboxRow{}
	for _, row := range r.db.Outbox {
		if row.PublishedAt == nil && row.Attempts >= r.maxAttempts {
			out = append(out, row)
		}
	}
	return out
}

// --- Idempotent consumer ---

type ShippingConsumer struct {
	seen      map[string]bool
	Shipments []string
}

func NewShippingConsumer() *ShippingConsumer {
	return &ShippingConsumer{seen: make(map[string]bool)}
}

func (c *ShippingConsumer) Handle(eventID string, payload map[string]any) {
	if c.seen[eventID] {
		fmt.Printf("[shipping] duplicate %s ignored\n", eventID)
		return
	}
	c.seen[eventID] = true
	c.Shipments = append(c.Shipments, fmt.Sprintf("%v -> %v", payload["orderId"], payload["city"]))
}

func main() {
	db := NewDatabase()
	broker := &FlakyBroker{}
	service := &OrderService{db: db}
	relay := NewOutboxRelay(db, broker)
	shipping := NewShippingConsumer()

	service.PlaceOrder("fatima-al-fihri", "fez", 1200)
	service.PlaceOrder("al-khwarizmi", "baghdad", 340)

	broker.FailNext = 1
	fmt.Println("published:", relay.DrainOnce())
	fmt.Println("published:", relay.DrainOnce())

	// Simulate crash-after-publish: the first event arrives twice
	msgs := append([]struct {
		EventID string
		Payload map[string]any
	}{}, broker.Delivered...)
	msgs = append(msgs, broker.Delivered[0])

	for _, m := range msgs {
		shipping.Handle(m.EventID, m.Payload)
	}

	fmt.Println("shipments:", shipping.Shipments)
	fmt.Println("dead letters:", len(relay.DeadLetters()))
}

package main

import (
	"errors"
	"fmt"
	"time"
)

type SagaContext struct {
	SagaID string
	Data   map[string]any
}

type SagaStep interface {
	Name() string
	Execute(ctx *SagaContext) error
	Compensate(ctx *SagaContext) error
	Retries() int
}

type SagaStatus string

const (
	StatusRunning      SagaStatus = "running"
	StatusCompleted    SagaStatus = "completed"
	StatusCompensating SagaStatus = "compensating"
	StatusCompensated  SagaStatus = "compensated"
	StatusStuck        SagaStatus = "stuck"
)

type SagaRecord struct {
	SagaID         string
	Status         SagaStatus
	CompletedSteps []string
	FailedStep     string
	History        []string
}

type SagaOrchestrator struct {
	steps   []SagaStep
	records map[string]*SagaRecord
}

func NewSagaOrchestrator(steps ...SagaStep) *SagaOrchestrator {
	return &SagaOrchestrator{steps: steps, records: make(map[string]*SagaRecord)}
}

func (o *SagaOrchestrator) Run(ctx *SagaContext) *SagaRecord {
	record := &SagaRecord{SagaID: ctx.SagaID, Status: StatusRunning}
	o.records[ctx.SagaID] = record

	for _, step := range o.steps {
		if !o.executeWithRetry(step, ctx, record) {
			record.Status = StatusCompensating
			record.FailedStep = step.Name()
			o.compensate(ctx, record)
			return record
		}
		record.CompletedSteps = append(record.CompletedSteps, step.Name())
		record.History = append(record.History, "ok: "+step.Name())
	}

	record.Status = StatusCompleted
	return record
}

func (o *SagaOrchestrator) executeWithRetry(step SagaStep, ctx *SagaContext, rec *SagaRecord) bool {
	maxAttempts := step.Retries() + 1
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		err := step.Execute(ctx)
		if err == nil {
			return true
		}
		rec.History = append(rec.History,
			fmt.Sprintf("fail(%d/%d): %s - %v", attempt, maxAttempts, step.Name(), err))
		if attempt == maxAttempts {
			return false
		}
		time.Sleep(time.Duration(1<<uint(attempt)) * 50 * time.Millisecond)
	}
	return false
}

// compensate runs in reverse over the steps that actually succeeded.
func (o *SagaOrchestrator) compensate(ctx *SagaContext, rec *SagaRecord) {
	allClean := true

	for i := len(rec.CompletedSteps) - 1; i >= 0; i-- {
		name := rec.CompletedSteps[i]
		var step SagaStep
		for _, s := range o.steps {
			if s.Name() == name {
				step = s
				break
			}
		}
		if step == nil {
			continue
		}
		if err := step.Compensate(ctx); err != nil {
			allClean = false
			rec.History = append(rec.History, "compensation failed: "+name+" - "+err.Error())
			continue
		}
		rec.History = append(rec.History, "compensated: "+name)
	}

	if allClean {
		rec.Status = StatusCompensated
	} else {
		rec.Status = StatusStuck
	}
}

func (o *SagaOrchestrator) Inspect(sagaID string) (*SagaRecord, bool) {
	r, ok := o.records[sagaID]
	return r, ok
}

// --- Concrete steps ---

type PaymentStep struct{}

func (PaymentStep) Name() string { return "authorise-payment" }
func (PaymentStep) Retries() int { return 2 }
func (PaymentStep) Execute(ctx *SagaContext) error {
	ctx.Data["paymentRef"] = "auth-" + ctx.SagaID
	fmt.Printf("[payment] authorised %v dirham for %v\n", ctx.Data["amountDirham"], ctx.Data["customer"])
	return nil
}
func (PaymentStep) Compensate(ctx *SagaContext) error {
	fmt.Printf("[payment] voiding %v\n", ctx.Data["paymentRef"])
	return nil
}

type InventoryStep struct{}

func (InventoryStep) Name() string { return "reserve-stock" }
func (InventoryStep) Retries() int { return 3 }
func (InventoryStep) Execute(ctx *SagaContext) error {
	ctx.Data["reservationRef"] = "res-" + ctx.SagaID
	fmt.Printf("[inventory] reserved %v\n", ctx.Data["sku"])
	return nil
}
func (InventoryStep) Compensate(ctx *SagaContext) error {
	fmt.Printf("[inventory] releasing %v\n", ctx.Data["reservationRef"])
	return nil
}

type ShippingStep struct{}

func (ShippingStep) Name() string { return "create-shipment" }
func (ShippingStep) Retries() int { return 1 }
func (ShippingStep) Execute(ctx *SagaContext) error {
	if ctx.Data["city"] == "fez" {
		return errors.New("no courier available in fez")
	}
	fmt.Printf("[shipping] shipment created for %v\n", ctx.Data["city"])
	return nil
}
func (ShippingStep) Compensate(*SagaContext) error {
	fmt.Println("[shipping] cancelling shipment")
	return nil
}

func main() {
	orchestrator := NewSagaOrchestrator(PaymentStep{}, InventoryStep{}, ShippingStep{})

	failed := orchestrator.Run(&SagaContext{
		SagaID: "saga-fez-0001",
		Data: map[string]any{
			"customer": "fatima-al-fihri", "city": "fez",
			"sku": "astrolabe-small", "amountDirham": 1200,
		},
	})
	fmt.Println(failed.Status, failed.History)

	ok := orchestrator.Run(&SagaContext{
		SagaID: "saga-cordoba-0002",
		Data: map[string]any{
			"customer": "ibn-rushd", "city": "cordoba",
			"sku": "astrolabe-small", "amountDirham": 1200,
		},
	})
	fmt.Println(ok.Status, ok.History)
}