Simple transaction processing E-DSL (vibe coded for demonstration purposes only)

main.go

// E-DSL для параллельной обработки бизнес-транзакций.
//
// Один файл, без внешних зависимостей. Инфраструктура (Redis / PostgreSQL /
// Kafka) вынесена за интерфейсы; здесь даны in-memory реализации, чтобы код
// запускался как есть. Места под реальные адаптеры помечены комментариями
// // [PG], // [Redis], // [Kafka].
//
// Что внутри:
//   - AST транзакции как данные (Step: Op, Seq, Par, Await, Fee, Spawn, Emit)
//   - примитивы hold / capture / void / debit / credit / refund
//   - реестр гейтов с circuit breaker + bulkhead + recover + классификацией ошибок
//   - авто-отключение упавшего гейта и авто-восстановление через health-loop
//   - интерпретатор: сага с компенсацией, параллель, приостановка (Await) и
//     возобновление по событию через детерминированный реплей
//   - конвейер plan -> admit -> execute -> settle (лимиты, комиссии, бесплатная квота)
//   - коррелированные транзакции-сателлиты (взнос Nasia, кешбек) через outbox
//
// go run main.go
package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"
)

// ───────────────────────── базовые типы ─────────────────────────

type Money int64 // в тийинах (1 сум = 100)

func (m Money) String() string {
	neg := ""
	if m < 0 {
		neg, m = "-", -m
	}
	return fmt.Sprintf("%s%d.%02d", neg, m/100, m%100)
}

func sum(v int64) Money { return Money(v * 100) }

type (
	TxID        string
	StepID      string
	GatewayKind string
	AccountRef  string
	TxType      string
)

const (
	KindUzcard   GatewayKind = "uzcard"
	KindHumo     GatewayKind = "humo"
	KindWallet   GatewayKind = "wallet"
	KindBonus    GatewayKind = "bonus"
	KindNasia    GatewayKind = "nasia"
	KindProvider GatewayKind = "provider"
)

const (
	IncomeAccount AccountRef = "sys:income"
	NasiaFunding  AccountRef = "sys:nasia_funding"
	NasiaCollect  AccountRef = "sys:nasia_collection"
)

type Action int

const (
	Hold Action = iota
	Capture
	Void
	Debit
	Credit
	Refund
)

func (a Action) String() string {
	return [...]string{"hold", "capture", "void", "debit", "credit", "refund"}[a]
}

// ───────────────────────── AST транзакции ─────────────────────────

// Step — узел дерева. Транзакция описывается как ДАННЫЕ, а не как вызовы;
// интерпретатор обходит дерево и придаёт ему смысл.
type Step interface{ isStep() }

// Op — лист: одна операция против одного гейта.
type Op struct {
	ID         StepID
	Gateway    GatewayKind
	Action     Action
	Ref        StepID // для Capture/Void/Refund — какую операцию потребляем/реверсим
	Account    AccountRef
	Amount     Money
	Compensate *Op // что выполнить при откате (Void для Hold, Refund для Debit/Capture)
}

func (Op) isStep() {}

// Seq — по очереди, стоп на первой ошибке.
type Seq struct{ Steps []Step }

func (Seq) isStep() {}

// Par — конкурентно.
type Par struct {
	Steps  []Step
	Policy FailurePolicy
}

func (Par) isStep() {}

type FailurePolicy int

const (
	AllOrNothing FailurePolicy = iota
	BestEffort
)

// Await — точка приостановки: ждём подтверждение чужого сервиса.
type Await struct {
	Token   string
	Timeout time.Duration
}

func (Await) isStep() {}

// Fee — комиссия. В фазе plan разворачивается в обычные Op (debit + credit).
type Fee struct {
	Gateway  GatewayKind
	Base     Money // облагаемая сумма
	QuotaKey string
	Policy   FeePolicy
	Payer    AccountRef
	To       AccountRef
}

func (Fee) isStep() {}

// Spawn — создать дочернюю коррелированную транзакцию (через outbox при коммите).
type Spawn struct {
	Type    TxType
	Request Request
}

func (Spawn) isStep() {}

// Emit — доменное событие в outbox (развязанное уведомление).
type Emit struct {
	Event   string
	Amount  Money
	Account AccountRef
}

func (Emit) isStep() {}

// ───────────────────────── комиссии и квота ─────────────────────────

type QuotaView struct{ remaining Money }

func (q QuotaView) Remaining() Money { return q.remaining }

type FeePolicy interface {
	// Compute возвращает комиссию и сколько бесплатной квоты потрачено.
	Compute(base Money, q QuotaView) (fee Money, useQuota Money)
}

// FreeThenPct: до исчерпания квоты — бесплатно, дальше — процент.
type FreeThenPct struct {
	Rate    float64
	OnWhole bool // true: при превышении комиссия на ВСЮ сумму; false: только на превышение
}

func (p FreeThenPct) Compute(base Money, q QuotaView) (Money, Money) {
	free := q.Remaining()
	if base <= free {
		return 0, base
	}
	if p.OnWhole {
		return Money(float64(base) * p.Rate), free
	}
	return Money(float64(base-free) * p.Rate), free
}

// ───────────────────────── ошибки и их классификация ─────────────────────────

type FailureKind int

const (
	Transient   FailureKind = iota // таймаут, 5xx — считается брейкером
	PermanentEr                    // битый запрос — не считается
	BusinessEr                     // нет средств, лимит — не считается
)

type GatewayError struct {
	Kind FailureKind
	msg  string
}

func (e *GatewayError) Error() string { return e.msg }

func classify(err error) FailureKind {
	var ge *GatewayError
	if errors.As(err, &ge) {
		return ge.Kind
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return Transient
	}
	return Transient
}

// ───────────────────────── гейт-плагин ─────────────────────────

type Result struct {
	Ref string // напр. id резерва, выданный гейтом
}

type Capabilities struct{ External bool }

// Gateway — узкий стабильный контракт. Новый гейт = реализовать это + register.
type Gateway interface {
	Kind() GatewayKind
	Do(ctx context.Context, op Op, prevRef string) (Result, error)
	HealthCheck(ctx context.Context) error
	Capabilities() Capabilities
}

// ledger — универсальный гейт с in-memory балансами.
// Внутренние (wallet, bonus) ходят по балансам; внешние (uzcard, humo, nasia,
// provider) дополнительно умеют падать транзиентно (флаг down) для демонстрации
// брейкера. В бою Do дёргает реальный API сети. // [внешний API]
type ledger struct {
	kind GatewayKind
	ext  bool
	mu   sync.Mutex
	bal  map[AccountRef]Money
	ops  map[string]opRec
	seq  int64
	down int32 // atomic: гейт «лежит»
}

type opRec struct {
	acc  AccountRef
	amt  Money
	kind Action
}

func newLedger(kind GatewayKind, ext bool) *ledger {
	return &ledger{kind: kind, ext: ext, bal: map[AccountRef]Money{}, ops: map[string]opRec{}}
}

func (l *ledger) Kind() GatewayKind          { return l.kind }
func (l *ledger) Capabilities() Capabilities { return Capabilities{External: l.ext} }

func (l *ledger) HealthCheck(ctx context.Context) error {
	if l.ext && atomic.LoadInt32(&l.down) == 1 {
		return &GatewayError{Transient, string(l.kind) + ": health check failed"}
	}
	return nil
}

func (l *ledger) newref() string {
	l.seq++
	return fmt.Sprintf("%s-%d", l.kind, l.seq)
}

func (l *ledger) Do(ctx context.Context, op Op, prev string) (Result, error) {
	if l.ext && atomic.LoadInt32(&l.down) == 1 {
		return Result{}, &GatewayError{Transient, string(l.kind) + ": gateway down"}
	}
	l.mu.Lock()
	defer l.mu.Unlock()

	switch op.Action {
	case Hold:
		if l.bal[op.Account] < op.Amount {
			return Result{}, &GatewayError{BusinessEr, "недостаточно средств для резерва"}
		}
		l.bal[op.Account] -= op.Amount
		ref := l.newref()
		l.ops[ref] = opRec{op.Account, op.Amount, Hold}
		return Result{Ref: ref}, nil

	case Capture:
		r, ok := l.ops[prev]
		if !ok {
			return Result{}, &GatewayError{PermanentEr, "нет резерва для capture"}
		}
		if op.Amount > 0 && op.Amount < r.amt { // частичный capture — вернуть остаток
			l.bal[r.acc] += r.amt - op.Amount
			r.amt = op.Amount
		}
		r.kind = Capture
		l.ops[prev] = r
		return Result{Ref: prev}, nil

	case Void:
		if r, ok := l.ops[prev]; ok && r.kind == Hold {
			l.bal[r.acc] += r.amt
			r.kind = Void
			l.ops[prev] = r
		}
		return Result{Ref: prev}, nil

	case Debit:
		if l.bal[op.Account] < op.Amount {
			return Result{}, &GatewayError{BusinessEr, "недостаточно средств"}
		}
		l.bal[op.Account] -= op.Amount
		ref := l.newref()
		l.ops[ref] = opRec{op.Account, op.Amount, Debit}
		return Result{Ref: ref}, nil

	case Credit:
		l.bal[op.Account] += op.Amount
		ref := l.newref()
		l.ops[ref] = opRec{op.Account, op.Amount, Credit}
		return Result{Ref: ref}, nil

	case Refund:
		if r, ok := l.ops[prev]; ok && (r.kind == Debit || r.kind == Capture) {
			l.bal[r.acc] += r.amt
			r.kind = Refund
			l.ops[prev] = r
		}
		return Result{Ref: prev}, nil
	}
	return Result{}, &GatewayError{PermanentEr, "действие не поддерживается"}
}

func (l *ledger) seed(acc AccountRef, amt Money) {
	l.mu.Lock()
	l.bal[acc] += amt
	l.mu.Unlock()
}

func (l *ledger) balance(acc AccountRef) Money {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.bal[acc]
}

// ───────────────────────── circuit breaker ─────────────────────────

type BreakerState int

const (
	Closed BreakerState = iota
	Open
	HalfOpen
)

type breaker struct {
	mu        sync.Mutex
	state     BreakerState
	failures  int
	threshold int
	cooldown  time.Duration
	openedAt  time.Time
}

func (b *breaker) allow() bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.state == Open && time.Since(b.openedAt) >= b.cooldown {
		b.state = HalfOpen
	}
	return b.state != Open
}

func (b *breaker) onSuccess() {
	b.mu.Lock()
	b.failures, b.state = 0, Closed
	b.mu.Unlock()
}

// onFailure открывает гейт только на ТРАНЗИЕНТНЫХ сбоях. «Нет средств»
// и «превышен лимит» — бизнес-ошибки, брейкер их не считает.
func (b *breaker) onFailure(kind FailureKind) {
	if kind != Transient {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.failures++
	if b.state == HalfOpen || b.failures >= b.threshold {
		b.state, b.openedAt = Open, time.Now()
	}
}

func (b *breaker) tryProbe() bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.state == Open && time.Since(b.openedAt) >= b.cooldown {
		b.state = HalfOpen
		return true
	}
	return false
}

func (b *breaker) closeIt() {
	b.mu.Lock()
	b.state, b.failures = Closed, 0
	b.mu.Unlock()
}

func (b *breaker) reopen() {
	b.mu.Lock()
	b.state, b.openedAt = Open, time.Now()
	b.mu.Unlock()
}

// ───────────────────────── реестр гейтов (граница изоляции) ─────────────────────────

type Registry struct {
	mu      sync.Mutex
	gw      map[GatewayKind]Gateway
	brk     map[GatewayKind]*breaker
	sem     map[GatewayKind]chan struct{} // bulkhead: лимит конкурентности на гейт
	enabled map[GatewayKind]bool          // мастер-выключатель из админки // [PG/Redis конфиг]
	timeout time.Duration
}

func newRegistry(timeout time.Duration) *Registry {
	return &Registry{
		gw: map[GatewayKind]Gateway{}, brk: map[GatewayKind]*breaker{},
		sem: map[GatewayKind]chan struct{}{}, enabled: map[GatewayKind]bool{},
		timeout: timeout,
	}
}

func (r *Registry) register(g Gateway, bulkhead, threshold int, cooldown time.Duration) {
	k := g.Kind()
	r.gw[k] = g
	r.brk[k] = &breaker{threshold: threshold, cooldown: cooldown}
	r.sem[k] = make(chan struct{}, bulkhead)
	r.enabled[k] = true
}

func (r *Registry) setEnabled(k GatewayKind, v bool) {
	r.mu.Lock()
	r.enabled[k] = v
	r.mu.Unlock()
}

func (r *Registry) isEnabled(k GatewayKind) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.enabled[k]
}

// Execute — единственный чокпоинт: всякая операция проходит здесь, и здесь же
// живёт вся изоляция. Сбой одного гейта не выходит за эту границу.
func (r *Registry) Execute(ctx context.Context, op Op, prevRef string) (res Result, err error) {
	g, ok := r.gw[op.Gateway]
	if !ok {
		return Result{}, &GatewayError{PermanentEr, "неизвестный гейт " + string(op.Gateway)}
	}
	b := r.brk[op.Gateway]

	// (1) мастер-выключатель + брейкер
	if !r.isEnabled(op.Gateway) || !b.allow() {
		return Result{}, &GatewayError{BusinessEr, "гейт " + string(op.Gateway) + " недоступен"}
	}
	// (2) bulkhead: медленный гейт не выест воркер-пул целиком
	select {
	case r.sem[op.Gateway] <- struct{}{}:
		defer func() { <-r.sem[op.Gateway] }()
	case <-ctx.Done():
		return Result{}, ctx.Err()
	}
	// (3) recover: паника в плагине не роняет движок
	defer func() {
		if p := recover(); p != nil {
			err = &GatewayError{Transient, fmt.Sprintf("паника в %s: %v", op.Gateway, p)}
			b.onFailure(Transient)
		}
	}()
	// (4) таймаут
	cctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()

	res, err = g.Do(cctx, op, prevRef)
	if err != nil {
		b.onFailure(classify(err))
		return res, err
	}
	b.onSuccess()
	return res, nil
}

// healthLoop проверяет упавшие гейты и сам возвращает их в строй.
func (r *Registry) healthLoop(ctx context.Context) {
	t := time.NewTicker(300 * time.Millisecond)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			for kind, b := range r.brk {
				if b.tryProbe() {
					if err := r.gw[kind].HealthCheck(ctx); err == nil {
						b.closeIt()
						log.Printf("[health] %s восстановлен и снова включён", kind)
					} else {
						b.reopen()
					}
				}
			}
		}
	}
}

// ───────────────────────── счётчики: квота и лимиты (резерв/подтверждение/возврат) ─────────────────────────

// counters — общий резервируемый счётчик. Зеркалит дисциплину Hold:
// reserve -> confirm (коммит) | release (откат). Используется и для бесплатной
// квоты, и для дневных лимитов. // [Redis: атомарный INCRBY + durable-журнал]
type counters struct {
	mu   sync.Mutex
	cap  map[string]Money // ёмкость (квота / дневной лимит); нет ключа = безлимит
	used map[string]Money
	resv map[string]resvRec
}

type resvRec struct {
	key       string
	amt       Money
	confirmed bool
}

func newCounters() *counters {
	return &counters{cap: map[string]Money{}, used: map[string]Money{}, resv: map[string]resvRec{}}
}

func (c *counters) setCap(key string, v Money) {
	c.mu.Lock()
	c.cap[key] = v
	c.mu.Unlock()
}

func (c *counters) remaining(key string) Money {
	c.mu.Lock()
	defer c.mu.Unlock()
	if cp, ok := c.cap[key]; ok {
		return cp - c.used[key]
	}
	return Money(1) << 60 // безлимит
}

// reserve идемпотентен по idem; false — если резерв вышел бы за ёмкость.
func (c *counters) reserve(key, idem string, amt Money) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, ok := c.resv[idem]; ok {
		return true
	}
	if cp, ok := c.cap[key]; ok && c.used[key]+amt > cp {
		return false
	}
	c.used[key] += amt
	c.resv[idem] = resvRec{key, amt, false}
	return true
}

func (c *counters) confirm(idem string) {
	c.mu.Lock()
	if r, ok := c.resv[idem]; ok {
		r.confirmed = true
		c.resv[idem] = r
	}
	c.mu.Unlock()
}

func (c *counters) release(idem string) {
	c.mu.Lock()
	if r, ok := c.resv[idem]; ok {
		c.used[r.key] -= r.amt
		delete(c.resv, idem)
	}
	c.mu.Unlock()
}

// ───────────────────────── сага-стор (журнал + приостановка) ─────────────────────────

// sagaStore хранит состояние переходов. // [PG: таблицы history + suspend + plan]
// Возобновление после Await — через детерминированный реплей: выполненные Op
// пропускаются, поэтому повторный прогон дерева безопасен.
type sagaStore struct {
	mu      sync.Mutex
	doneMap map[TxID]map[StepID]Result
	refMap  map[TxID]map[StepID]string
	compMap map[TxID][]Op
	sigMap  map[TxID]map[string]bool
	suspMap map[TxID]time.Time
	planMap map[TxID]planRec
	outMap  map[TxID][]outboxEntry
	status  map[TxID]string
}

type planRec struct {
	root Step
	resv []string
	req  Request
}

func newSagaStore() *sagaStore {
	return &sagaStore{
		doneMap: map[TxID]map[StepID]Result{}, refMap: map[TxID]map[StepID]string{},
		compMap: map[TxID][]Op{}, sigMap: map[TxID]map[string]bool{},
		suspMap: map[TxID]time.Time{}, planMap: map[TxID]planRec{},
		outMap: map[TxID][]outboxEntry{}, status: map[TxID]string{},
	}
}

func (s *sagaStore) lookupDone(tx TxID, id StepID) (Result, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	r, ok := s.doneMap[tx][id]
	return r, ok
}

func (s *sagaStore) markDone(tx TxID, id StepID, res Result) {
	s.mu.Lock()
	if s.doneMap[tx] == nil {
		s.doneMap[tx] = map[StepID]Result{}
		s.refMap[tx] = map[StepID]string{}
	}
	s.doneMap[tx][id] = res
	s.refMap[tx][id] = res.Ref
	s.mu.Unlock()
}

func (s *sagaStore) ref(tx TxID, id StepID) string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.refMap[tx][id]
}

func (s *sagaStore) addComp(tx TxID, op Op) {
	s.mu.Lock()
	s.compMap[tx] = append(s.compMap[tx], op)
	s.mu.Unlock()
}

func (s *sagaStore) comps(tx TxID) []Op {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]Op(nil), s.compMap[tx]...)
}

func (s *sagaStore) setSignal(tx TxID, token string) {
	s.mu.Lock()
	if s.sigMap[tx] == nil {
		s.sigMap[tx] = map[string]bool{}
	}
	s.sigMap[tx][token] = true
	s.mu.Unlock()
}

func (s *sagaStore) signaled(tx TxID, token string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.sigMap[tx][token]
}

func (s *sagaStore) suspend(tx TxID, deadline time.Time) {
	s.mu.Lock()
	s.suspMap[tx] = deadline
	s.mu.Unlock()
}

func (s *sagaStore) removeSuspend(tx TxID) {
	s.mu.Lock()
	delete(s.suspMap, tx)
	s.mu.Unlock()
}

func (s *sagaStore) dueSuspended(now time.Time) []TxID {
	s.mu.Lock()
	defer s.mu.Unlock()
	var out []TxID
	for tx, dl := range s.suspMap {
		if now.After(dl) {
			out = append(out, tx)
		}
	}
	return out
}

func (s *sagaStore) savePlan(tx TxID, root Step, resv []string, req Request) {
	s.mu.Lock()
	s.planMap[tx] = planRec{root, resv, req}
	s.mu.Unlock()
}

func (s *sagaStore) getPlan(tx TxID) (planRec, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	p, ok := s.planMap[tx]
	return p, ok
}

func (s *sagaStore) buffer(tx TxID, e outboxEntry) {
	s.mu.Lock()
	s.outMap[tx] = append(s.outMap[tx], e)
	s.mu.Unlock()
}

func (s *sagaStore) flush(tx TxID) []outboxEntry {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := s.outMap[tx]
	delete(s.outMap, tx)
	return out
}

func (s *sagaStore) mark(tx TxID, st string) {
	s.mu.Lock()
	s.status[tx] = st
	s.mu.Unlock()
}

func (s *sagaStore) logf(tx TxID, id StepID, msg string) {
	log.Printf("[saga] %s %s %s", tx, id, msg)
}

// ───────────────────────── идемпотентность входа ─────────────────────────

// [Redis: SETNX] + маленькая durable-таблица как бэкстоп.
type idemStore struct {
	mu   sync.Mutex
	seen map[string]bool
}

func newIdemStore() *idemStore { return &idemStore{seen: map[string]bool{}} }

func (i *idemStore) check(key string) bool {
	i.mu.Lock()
	defer i.mu.Unlock()
	return i.seen[key]
}

func (i *idemStore) mark(key string) {
	i.mu.Lock()
	i.seen[key] = true
	i.mu.Unlock()
}

// ───────────────────────── outbox ─────────────────────────

type outboxEntry struct {
	kind    string // "spawn" | "event"
	event   string
	txType  TxType
	req     Request
	corr    TxID
	amount  Money
	account AccountRef
}

// ───────────────────────── запрос и шаблоны ─────────────────────────

type Request struct {
	Idem     string
	Type     TxType
	Network  GatewayKind
	Source   GatewayKind
	Sender   AccountRef
	Receiver AccountRef
	Payer    AccountRef
	Account  AccountRef
	Merchant AccountRef
	Amount   Money
	Down     Money
	Corr     TxID // correlation_id для сателлитов
}

// Реестр шаблонов: новый сценарий = функция, возвращающая дерево. Движок не меняется.
var templates = map[TxType]func(Request) (Step, error){}

const (
	TxPhoneTopup    TxType = "phone_topup"
	TxP2P           TxType = "p2p"
	TxHoldConfirm   TxType = "hold_confirm"
	TxPayment       TxType = "payment"
	TxNasiaPurchase TxType = "nasia_purchase"
	TxDownPayment   TxType = "down_payment"
	TxCashback      TxType = "cashback"
)

func registerTemplates() {
	// Пополнение телефона: дебет источника (с компенсацией) -> кредит провайдеру.
	templates[TxPhoneTopup] = func(r Request) (Step, error) {
		return Seq{Steps: []Step{
			Op{ID: "d", Gateway: r.Source, Action: Debit, Account: r.Payer, Amount: r.Amount,
				Compensate: &Op{ID: "d.c", Gateway: r.Source, Action: Refund, Ref: "d"}},
			Op{ID: "c", Gateway: KindProvider, Action: Credit, Account: r.Account, Amount: r.Amount},
		}}, nil
	}

	// P2P внутри сети + комиссия с учётом бесплатной квоты отправителя.
	templates[TxP2P] = func(r Request) (Step, error) {
		return Seq{Steps: []Step{
			Op{ID: "p", Gateway: r.Network, Action: Debit, Account: r.Sender, Amount: r.Amount,
				Compensate: &Op{ID: "p.c", Gateway: r.Network, Action: Refund, Ref: "p"}},
			Op{ID: "r", Gateway: r.Network, Action: Credit, Account: r.Receiver, Amount: r.Amount},
			Fee{Gateway: r.Network, Base: r.Amount, QuotaKey: "p2p:" + string(r.Sender),
				Policy: FreeThenPct{Rate: 0.01}, Payer: r.Sender, To: IncomeAccount},
		}}, nil
	}

	// Hold -> ждём подтверждение чужого сервиса -> Capture из резерва -> кредит мерчанту.
	templates[TxHoldConfirm] = func(r Request) (Step, error) {
		return Seq{Steps: []Step{
			Op{ID: "h", Gateway: r.Source, Action: Hold, Account: r.Payer, Amount: r.Amount,
				Compensate: &Op{ID: "h.c", Gateway: r.Source, Action: Void, Ref: "h"}},
			Await{Token: "confirm", Timeout: 800 * time.Millisecond},
			Op{ID: "cap", Gateway: r.Source, Action: Capture, Ref: "h", Account: r.Payer, Amount: r.Amount},
			Op{ID: "mc", Gateway: KindWallet, Action: Credit, Account: r.Merchant, Amount: r.Amount},
		}}, nil
	}

	// Платёж + комиссия + уведомление о завершении (на него реагирует кешбек-сервис).
	templates[TxPayment] = func(r Request) (Step, error) {
		return Seq{Steps: []Step{
			Op{ID: "d", Gateway: r.Source, Action: Debit, Account: r.Payer, Amount: r.Amount,
				Compensate: &Op{ID: "d.c", Gateway: r.Source, Action: Refund, Ref: "d"}},
			Op{ID: "m", Gateway: KindWallet, Action: Credit, Account: r.Merchant, Amount: r.Amount},
			Fee{Gateway: r.Source, Base: r.Amount, QuotaKey: "pay:" + string(r.Payer),
				Policy: FreeThenPct{Rate: 0.01}, Payer: r.Payer, To: IncomeAccount},
			Emit{Event: "payment.completed", Amount: r.Amount, Account: r.Account}, // r.Account — бонусный счёт
		}}, nil
	}

	// Nasia: финансирование + кредит мерчанту, а взнос — отдельная коррелированная транзакция.
	templates[TxNasiaPurchase] = func(r Request) (Step, error) {
		return Seq{Steps: []Step{
			Op{ID: "fin", Gateway: KindNasia, Action: Debit, Account: NasiaFunding, Amount: r.Amount,
				Compensate: &Op{ID: "fin.c", Gateway: KindNasia, Action: Refund, Ref: "fin"}},
			Op{ID: "mc", Gateway: KindWallet, Action: Credit, Account: r.Merchant, Amount: r.Amount},
			Spawn{Type: TxDownPayment, Request: Request{Type: TxDownPayment,
				Source: KindWallet, Payer: r.Payer, Account: NasiaCollect, Amount: r.Down}},
		}}, nil
	}

	templates[TxDownPayment] = func(r Request) (Step, error) {
		return Seq{Steps: []Step{
			Op{ID: "dp", Gateway: r.Source, Action: Debit, Account: r.Payer, Amount: r.Amount,
				Compensate: &Op{ID: "dp.c", Gateway: r.Source, Action: Refund, Ref: "dp"}},
			Op{ID: "col", Gateway: KindWallet, Action: Credit, Account: r.Account, Amount: r.Amount},
		}}, nil
	}

	templates[TxCashback] = func(r Request) (Step, error) {
		return Op{ID: "cb", Gateway: KindBonus, Action: Credit, Account: r.Account, Amount: r.Amount}, nil
	}
}

// ───────────────────────── движок ─────────────────────────

var errSuspended = errors.New("suspended")

type Engine struct {
	reg      *Registry
	saga     *sagaStore
	idem     *idemStore
	counters *counters
	outbox   chan outboxEntry // [Kafka]
	reactors map[string]func(outboxEntry)
	perTxMax Money
	seq      int64
	ctx      context.Context
}

func newEngine(ctx context.Context, reg *Registry) *Engine {
	return &Engine{
		reg: reg, saga: newSagaStore(), idem: newIdemStore(), counters: newCounters(),
		outbox: make(chan outboxEntry, 256), reactors: map[string]func(outboxEntry){},
		perTxMax: sum(5000), ctx: ctx,
	}
}

// Process — конвейер: plan -> admit -> execute -> settle.
func (e *Engine) Process(ctx context.Context, req Request) error {
	if req.Idem == "" {
		req.Idem = fmt.Sprintf("auto-%d", atomic.AddInt64(&e.seq, 1))
	}
	if e.idem.check(req.Idem) {
		log.Printf("[idem] пропуск дубля %s", req.Idem)
		return nil
	}
	tx := TxID(req.Idem)
	build, ok := templates[req.Type]
	if !ok {
		return fmt.Errorf("неизвестный тип транзакции %s", req.Type)
	}
	raw, err := build(req)
	if err != nil {
		return err
	}

	// 1. plan: развернуть Fee в Op, зарезервировать бесплатную квоту.
	root, qresv, err := e.plan(ctx, tx, raw)
	if err != nil {
		return err
	}
	// 2. admit: проверить лимиты против principal+fee.
	lresv, err := e.admit(ctx, tx, root)
	if err != nil {
		e.releaseAll(qresv)
		log.Printf("[admit] %s отклонена: %v", tx, err)
		return err
	}
	resv := append(qresv, lresv...)
	e.idem.mark(req.Idem)
	e.saga.savePlan(tx, root, resv, req)

	return e.runAndSettle(ctx, tx, root, resv)
}

func (e *Engine) runAndSettle(ctx context.Context, tx TxID, root Step, resv []string) error {
	err := e.exec(ctx, tx, root)
	switch {
	case err == nil: // 4a. commit: подтвердить квоту + выпустить outbox
		e.confirmAll(resv)
		for _, en := range e.saga.flush(tx) {
			e.publish(en)
		}
		e.saga.mark(tx, "committed")
		log.Printf("[ok] %s завершена", tx)
		return nil
	case errors.Is(err, errSuspended): // припаркована на Await — НЕ компенсируем
		log.Printf("[wait] %s припаркована, ждёт подтверждения", tx)
		return nil
	default: // 4b. rollback: компенсации + вернуть квоту
		e.compensate(ctx, tx)
		e.releaseAll(resv)
		e.saga.mark(tx, "failed")
		log.Printf("[fail] %s: %v (откат выполнен)", tx, err)
		return err
	}
}

// plan обходит дерево, разворачивает Fee и резервирует квоту.
func (e *Engine) plan(ctx context.Context, tx TxID, s Step) (Step, []string, error) {
	switch n := s.(type) {
	case Op, Await, Spawn, Emit:
		return s, nil, nil
	case Seq:
		var out []Step
		var resv []string
		for _, c := range n.Steps {
			ps, r, err := e.plan(ctx, tx, c)
			if err != nil {
				return nil, nil, err
			}
			if ps != nil {
				out = append(out, ps)
			}
			resv = append(resv, r...)
		}
		return Seq{Steps: out}, resv, nil
	case Par:
		var out []Step
		var resv []string
		for _, c := range n.Steps {
			ps, r, err := e.plan(ctx, tx, c)
			if err != nil {
				return nil, nil, err
			}
			if ps != nil {
				out = append(out, ps)
			}
			resv = append(resv, r...)
		}
		return Par{Steps: out, Policy: n.Policy}, resv, nil
	case Fee:
		rem := e.counters.remaining(n.QuotaKey)
		fee, use := n.Policy.Compute(n.Base, QuotaView{rem})
		idem := string(tx) + ":fee:" + n.QuotaKey
		e.counters.reserve(n.QuotaKey, idem, use)
		if fee == 0 {
			log.Printf("[fee] %s: в бесплатном лимите (квота -%s)", tx, use)
			return nil, []string{idem}, nil
		}
		log.Printf("[fee] %s: комиссия %s (квота -%s)", tx, fee, use)
		d := StepID(idem + ":d")
		return Seq{Steps: []Step{
			Op{ID: d, Gateway: n.Gateway, Action: Debit, Account: n.Payer, Amount: fee,
				Compensate: &Op{ID: StepID(string(d) + ".c"), Gateway: n.Gateway, Action: Refund, Ref: d}},
			Op{ID: StepID(idem + ":c"), Gateway: n.Gateway, Action: Credit, Account: n.To, Amount: fee},
		}}, []string{idem}, nil
	}
	return s, nil, nil
}

// admit — гарды лимитов. Нарушение = бизнес-ошибка, брейкер не трогается.
func (e *Engine) admit(ctx context.Context, tx TxID, root Step) ([]string, error) {
	out := map[AccountRef]Money{}
	collectOutflow(root, out)
	var resv []string
	for acc, amt := range out {
		if amt > e.perTxMax {
			e.releaseAll(resv)
			return nil, &GatewayError{BusinessEr, fmt.Sprintf("превышен лимит на транзакцию для %s (%s)", acc, amt)}
		}
		idem := string(tx) + ":daily:" + string(acc)
		if !e.counters.reserve("daily:"+string(acc), idem, amt) {
			e.releaseAll(resv)
			return nil, &GatewayError{BusinessEr, fmt.Sprintf("превышен дневной лимит для %s", acc)}
		}
		resv = append(resv, idem)
	}
	return resv, nil
}

func collectOutflow(s Step, m map[AccountRef]Money) {
	switch n := s.(type) {
	case Op:
		if n.Action == Debit || n.Action == Hold {
			m[n.Account] += n.Amount
		}
	case Seq:
		for _, c := range n.Steps {
			collectOutflow(c, m)
		}
	case Par:
		for _, c := range n.Steps {
			collectOutflow(c, m)
		}
	}
}

// exec — интерпретатор. Три исхода: nil (успех), errSuspended (приостановка), ошибка.
func (e *Engine) exec(ctx context.Context, tx TxID, s Step) error {
	switch n := s.(type) {
	case nil:
		return nil
	case Op:
		if _, ok := e.saga.lookupDone(tx, n.ID); ok { // реплей: уже выполнено — пропускаем
			return nil
		}
		var prev string
		if n.Ref != "" {
			prev = e.saga.ref(tx, n.Ref)
		}
		res, err := e.reg.Execute(ctx, n, prev)
		if err != nil {
			return err
		}
		e.saga.markDone(tx, n.ID, res)
		e.saga.logf(tx, n.ID, n.Action.String()+" ok")
		if n.Compensate != nil {
			e.saga.addComp(tx, *n.Compensate)
		}
		return nil
	case Seq:
		for _, c := range n.Steps {
			if err := e.exec(ctx, tx, c); err != nil {
				return err
			}
		}
		return nil
	case Par:
		errs := make([]error, len(n.Steps))
		var wg sync.WaitGroup
		for i := range n.Steps {
			wg.Add(1)
			go func(i int) { defer wg.Done(); errs[i] = e.exec(ctx, tx, n.Steps[i]) }(i)
		}
		wg.Wait()
		for _, err := range errs {
			if err != nil && n.Policy == AllOrNothing {
				return err
			}
		}
		return nil
	case Await:
		if e.saga.signaled(tx, n.Token) {
			return nil // подтверждение пришло — идём дальше
		}
		e.saga.suspend(tx, time.Now().Add(n.Timeout))
		e.saga.logf(tx, "", "suspended на '"+n.Token+"'")
		return errSuspended
	case Spawn:
		e.saga.buffer(tx, outboxEntry{kind: "spawn", txType: n.Type, req: n.Request, corr: tx})
		return nil
	case Emit:
		e.saga.buffer(tx, outboxEntry{kind: "event", event: n.Event, amount: n.Amount, account: n.Account, corr: tx})
		return nil
	case Fee:
		return fmt.Errorf("Fee должен быть развёрнут в plan")
	}
	return fmt.Errorf("неизвестный узел %T", s)
}

// compensate выполняет накопленные компенсации в обратном порядке.
func (e *Engine) compensate(ctx context.Context, tx TxID) {
	comps := e.saga.comps(tx)
	for i := len(comps) - 1; i >= 0; i-- {
		c := comps[i]
		prev := ""
		if c.Ref != "" {
			prev = e.saga.ref(tx, c.Ref)
		}
		if _, err := e.reg.Execute(ctx, c, prev); err != nil {
			log.Printf("[comp] %s %s ошибка: %v", tx, c.Action, err)
		} else {
			e.saga.logf(tx, c.ID, c.Action.String()+" (компенсация)")
		}
	}
}

// Signal — подтверждение чужого сервиса: будит припаркованную транзакцию. // [Kafka consumer]
func (e *Engine) Signal(tx TxID, token string) {
	e.saga.setSignal(tx, token)
	e.saga.removeSuspend(tx)
	if p, ok := e.saga.getPlan(tx); ok {
		go e.runAndSettle(e.ctx, tx, p.root, p.resv) // реплей: done-шаги пропустятся, Await пройдёт
	}
}

// timeoutSweeper добивает просроченные приостановки в ветку компенсации (Void резерва).
func (e *Engine) timeoutSweeper(ctx context.Context) {
	t := time.NewTicker(150 * time.Millisecond)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			for _, tx := range e.saga.dueSuspended(time.Now()) {
				e.saga.removeSuspend(tx)
				p, _ := e.saga.getPlan(tx)
				e.compensate(ctx, tx)
				e.releaseAll(p.resv)
				e.saga.mark(tx, "timed_out")
				log.Printf("[timeout] %s: подтверждение не пришло, резерв снят", tx)
			}
		}
	}
}

func (e *Engine) publish(en outboxEntry) {
	select {
	case e.outbox <- en:
	default:
		log.Printf("[outbox] переполнен, потеряно: %v", en.kind)
	}
}

// outboxConsumer исполняет сателлиты и реакции на события. // [Kafka consumer group]
func (e *Engine) outboxConsumer(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case en := <-e.outbox:
			switch en.kind {
			case "spawn":
				req := en.req
				req.Corr = en.corr
				if req.Idem == "" {
					req.Idem = string(en.corr) + ":spawn"
				}
				log.Printf("[spawn] сателлит %s (corr=%s)", req.Type, en.corr)
				_ = e.Process(e.ctx, req)
			case "event":
				log.Printf("[event] %s (corr=%s)", en.event, en.corr)
				if r := e.reactors[en.event]; r != nil {
					r(en)
				}
			}
		}
	}
}

func (e *Engine) confirmAll(resv []string) {
	for _, id := range resv {
		e.counters.confirm(id)
	}
}

func (e *Engine) releaseAll(resv []string) {
	for _, id := range resv {
		e.counters.release(id)
	}
}

// ───────────────────────── демонстрация ─────────────────────────

func main() {
	log.SetFlags(0)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// гейты: (bulkhead, порог брейкера, кулдаун)
	reg := newRegistry(2 * time.Second)
	uz := newLedger(KindUzcard, true)
	humo := newLedger(KindHumo, true)
	wallet := newLedger(KindWallet, false)
	bonus := newLedger(KindBonus, false)
	nasia := newLedger(KindNasia, true)
	provider := newLedger(KindProvider, true)
	reg.register(uz, 8, 3, 1*time.Second)
	reg.register(humo, 8, 3, 1*time.Second)
	reg.register(wallet, 16, 5, 1*time.Second)
	reg.register(bonus, 16, 5, 1*time.Second)
	reg.register(nasia, 4, 3, 1*time.Second)
	reg.register(provider, 8, 3, 1*time.Second)

	// стартовые балансы
	uz.seed("acc:alice", sum(10000))
	uz.seed("acc:dave", sum(10000))
	wallet.seed("acc:bob", sum(10000))
	wallet.seed("acc:erin", sum(10000))
	nasia.seed(NasiaFunding, sum(10_000_000))

	registerTemplates()
	eng := newEngine(ctx, reg)

	// конфиг лимитов/квот из «админки» // [PG -> Redis hot-reload]
	eng.counters.setCap("p2p:acc:alice", sum(200)) // бесплатно до 200.00 сум P2P

	// кешбек-сервис: на payment.completed возвращает кешбек -> отдельная коррелированная транзакция
	eng.reactors["payment.completed"] = func(en outboxEntry) {
		cb := Money(float64(en.amount) * 0.01)
		if cb <= 0 {
			return
		}
		_ = eng.Process(eng.ctx, Request{Idem: string(en.corr) + ":cb", Type: TxCashback,
			Account: en.account, Amount: cb, Corr: en.corr})
	}

	// фоновые воркеры
	go reg.healthLoop(ctx)
	go eng.outboxConsumer(ctx)
	go eng.timeoutSweeper(ctx)

	sect := func(s string) { log.Printf("\n══════ %s ══════", s) }

	sect("1. Пополнение телефона с кошелька")
	eng.Process(ctx, Request{Idem: "t1", Type: TxPhoneTopup, Source: KindWallet,
		Payer: "acc:bob", Account: "phone:998901112233", Amount: sum(50)})
	log.Printf("  bob кошелёк=%s  телефон=%s", wallet.balance("acc:bob"), provider.balance("phone:998901112233"))

	sect("2. P2P Uzcard→Uzcard: бесплатный лимит, затем комиссия")
	eng.Process(ctx, Request{Idem: "p1", Type: TxP2P, Network: KindUzcard, Sender: "acc:alice", Receiver: "acc:carol", Amount: sum(150)})
	eng.Process(ctx, Request{Idem: "p2", Type: TxP2P, Network: KindUzcard, Sender: "acc:alice", Receiver: "acc:carol", Amount: sum(150)})
	log.Printf("  alice uzcard=%s  carol uzcard=%s  доход(комиссия)=%s",
		uz.balance("acc:alice"), uz.balance("acc:carol"), uz.balance(IncomeAccount))

	sect("3. Лимит на транзакцию")
	eng.Process(ctx, Request{Idem: "big", Type: TxP2P, Network: KindUzcard, Sender: "acc:alice", Receiver: "acc:carol", Amount: sum(6000)})

	sect("4a. Hold → подтверждение → Capture")
	eng.Process(ctx, Request{Idem: "h1", Type: TxHoldConfirm, Source: KindUzcard, Payer: "acc:alice", Merchant: "m:shop", Amount: sum(80)})
	time.Sleep(120 * time.Millisecond)
	log.Printf("  → приходит подтверждение от чужого сервиса")
	eng.Signal("h1", "confirm")
	time.Sleep(120 * time.Millisecond)
	log.Printf("  alice uzcard=%s  shop кошелёк=%s", uz.balance("acc:alice"), wallet.balance("m:shop"))

	sect("4b. Hold без подтверждения → таймаут → Void")
	before := uz.balance("acc:alice")
	eng.Process(ctx, Request{Idem: "h2", Type: TxHoldConfirm, Source: KindUzcard, Payer: "acc:alice", Merchant: "m:shop", Amount: sum(80)})
	log.Printf("  alice uzcard=%s (зарезервировано), ждём таймаут...", uz.balance("acc:alice"))
	time.Sleep(1100 * time.Millisecond)
	log.Printf("  alice uzcard=%s (резерв снят, было %s)", uz.balance("acc:alice"), before)

	sect("5. Авто-отключение упавшего гейта и авто-восстановление")
	atomic.StoreInt32(&uz.down, 1)
	log.Printf("  uzcard «упал»; шлём переводы — после 3 сбоев гейт отключится")
	for i := 0; i < 5; i++ {
		eng.Process(ctx, Request{Idem: fmt.Sprintf("f%d", i), Type: TxP2P, Network: KindUzcard,
			Sender: "acc:alice", Receiver: "acc:carol", Amount: sum(10)})
	}
	atomic.StoreInt32(&uz.down, 0)
	log.Printf("  uzcard починили; ждём, пока health-loop вернёт его в строй...")
	time.Sleep(1500 * time.Millisecond)
	eng.Process(ctx, Request{Idem: "f-ok", Type: TxP2P, Network: KindUzcard, Sender: "acc:alice", Receiver: "acc:carol", Amount: sum(10)})

	sect("6. Платёж + кешбек (сателлит по correlation_id)")
	eng.Process(ctx, Request{Idem: "pay1", Type: TxPayment, Source: KindUzcard,
		Payer: "acc:dave", Merchant: "m:shop", Account: "bonus:dave", Amount: sum(200)})
	time.Sleep(200 * time.Millisecond)
	log.Printf("  dave uzcard=%s  shop кошелёк=%s  bonus:dave=%s",
		uz.balance("acc:dave"), wallet.balance("m:shop"), bonus.balance("bonus:dave"))

	sect("7. Nasia: покупка + взнос (сателлит по correlation_id)")
	eng.Process(ctx, Request{Idem: "nas1", Type: TxNasiaPurchase,
		Payer: "acc:erin", Merchant: "m:shop", Amount: sum(300), Down: sum(60)})
	time.Sleep(200 * time.Millisecond)
	log.Printf("  erin кошелёк=%s  nasia_collection=%s  shop кошелёк=%s",
		wallet.balance("acc:erin"), wallet.balance(NasiaCollect), wallet.balance("m:shop"))

	sect("8. Идемпотентность входа")
	eng.Process(ctx, Request{Idem: "t1", Type: TxPhoneTopup, Source: KindWallet,
		Payer: "acc:bob", Account: "phone:998901112233", Amount: sum(50)})

	time.Sleep(150 * time.Millisecond)
	log.Printf("\nГотово.")
	cancel()
	time.Sleep(50 * time.Millisecond)
}