// DANGEROUS ANTI-PATTERN: Dual-Write Inconsistency Hazard
func (s *PaymentService) HandleDepositUnsafe(ctx context.Context, cmd DepositCommand) error {
    // 1. Write to database
    if err := s.repo.SaveEvent(ctx, event); err != nil {
        return err
    }
    // 2. Publish to broker -> If node crashes or network drops here, event is lost forever!
    return s.broker.Publish("account-events", event)
}

// Package main implements a production-grade Event Sourcing and CQRS aggregate engine for 2027 SOTA architectures.
// Leverages Go 1.25 iter.Seq range-over-func iterators, NATS JetStream message deduplication, and optimistic concurrency.
package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"iter"
	"log/slog"
	"os"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// Standard core banking domain error definitions
var (
	ErrAccountInactive      = errors.New("target banking account is closed or frozen")
	ErrInsufficientFunds    = errors.New("insufficient available balance for withdrawal")
	ErrConcurrencyConflict  = errors.New("optimistic concurrency check failed (version mismatch)")
	ErrInvalidEventSequence = errors.New("non-sequential event version stream detected")
)

// EventType defines discrete financial domain occurrences
type EventType string

const (
	EventAccountOpened  EventType = "AccountOpened"
	EventFundsDeposited EventType = "FundsDeposited"
	EventFundsWithdrawn EventType = "FundsWithdrawn"
	EventHoldPlaced     EventType = "HoldPlaced"
	EventHoldReleased   EventType = "HoldReleased"
)

// DomainEvent encapsulates an immutable historical business fact
type DomainEvent struct {
	EventID     string          `json:"event_id"`
	AggregateID string          `json:"aggregate_id"`
	Type        EventType       `json:"type"`
	Version     int64           `json:"version"`
	Payload     json.RawMessage `json:"payload"`
	OccurredAt  time.Time       `json:"occurred_at"`
}

// DepositPayload conveys credit parameters
type DepositPayload struct {
	AmountMinor int64  `json:"amount_minor"`
	ReferenceID string `json:"reference_id"`
}

// WithdrawPayload conveys debit parameters
type WithdrawPayload struct {
	AmountMinor int64  `json:"amount_minor"`
	ReferenceID string `json:"reference_id"`
}

// AccountAggregate represents the write-side aggregate root enforcing transactional invariants.
type AccountAggregate struct {
	ID             string
	Version        int64
	BalanceMinor   int64
	HoldMinor      int64
	IsActive       bool
	uncommittedEvs []DomainEvent
	mu             sync.RWMutex
}

// NewAccountAggregate constructs an empty aggregate ready for state hydration.
func NewAccountAggregate(id string) *AccountAggregate {
	return &AccountAggregate{
		ID: id,
	}
}

// AvailableBalance computes spendable funds taking reservation holds into account.
func (a *AccountAggregate) AvailableBalance() int64 {
	return a.BalanceMinor - a.HoldMinor
}

// Apply mutates internal aggregate state deterministically based on an event.
func (a *AccountAggregate) Apply(evt DomainEvent) error {
	if evt.Version != a.Version+1 {
		return fmt.Errorf("%w: expected version %d, got %d", ErrInvalidEventSequence, a.Version+1, evt.Version)
	}

	switch evt.Type {
	case EventAccountOpened:
		a.IsActive = true
		a.BalanceMinor = 0
		a.HoldMinor = 0

	case EventFundsDeposited:
		var p DepositPayload
		if err := json.Unmarshal(evt.Payload, &p); err != nil {
			return err
		}
		a.BalanceMinor += p.AmountMinor

	case EventFundsWithdrawn:
		var p WithdrawPayload
		if err := json.Unmarshal(evt.Payload, &p); err != nil {
			return err
		}
		a.BalanceMinor -= p.AmountMinor

	default:
		return fmt.Errorf("unsupported domain event type: %s", evt.Type)
	}

	a.Version = evt.Version
	return nil
}

// HydrateFromHistory reconstructs aggregate state by replaying an event sequence.
// Utilizes Go 1.25 iter.Seq range-over-func iterator for zero-allocation stream processing.
func (a *AccountAggregate) HydrateFromHistory(events iter.Seq[DomainEvent]) error {
	a.mu.Lock()
	defer a.mu.Unlock()

	for evt := range events {
		if err := a.Apply(evt); err != nil {
			return err
		}
	}
	return nil
}

// Withdraw validates banking invariants and stages an uncommitted FundsWithdrawn event.
func (a *AccountAggregate) Withdraw(amountMinor int64, refID string) error {
	a.mu.Lock()
	defer a.mu.Unlock()

	if !a.IsActive {
		return ErrAccountInactive
	}
	if a.AvailableBalance() < amountMinor {
		return fmt.Errorf("%w: spendable %d < requested %d", ErrInsufficientFunds, a.AvailableBalance(), amountMinor)
	}

	payload, _ := json.Marshal(WithdrawPayload{
		AmountMinor: amountMinor,
		ReferenceID: refID,
	})

	evt := DomainEvent{
		EventID:     fmt.Sprintf("EVT-%d-%s", time.Now().UnixNano(), a.ID),
		AggregateID: a.ID,
		Type:        EventFundsWithdrawn,
		Version:     a.Version + 1,
		Payload:     payload,
		OccurredAt:  time.Now().UTC(),
	}

	if err := a.Apply(evt); err != nil {
		return err
	}
	a.uncommittedEvs = append(a.uncommittedEvs, evt)
	return nil
}

// NatsJetStreamEventStore coordinates database persistence and outbox delivery via NATS JetStream.
type NatsJetStreamEventStore struct {
	db     *sql.DB
	js     jetstream.JetStream
	logger *slog.Logger
}

// NewNatsJetStreamEventStore connects to NATS and ensures the target stream exists.
func NewNatsJetStreamEventStore(db *sql.DB, natsURL string, logger *slog.Logger) (*NatsJetStreamEventStore, error) {
	nc, err := nats.Connect(natsURL)
	if err != nil {
		return nil, fmt.Errorf("failed to establish NATS connection: %w", err)
	}

	js, err := jetstream.New(nc)
	if err != nil {
		return nil, fmt.Errorf("failed to create JetStream context: %w", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	_, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
		Name:      "BANKING_EVENTS",
		Subjects:  []string{"banking.account.>"},
		Retention: jetstream.InterestPolicy,
		Storage:   jetstream.FileStorage,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to provision banking JetStream stream: %w", err)
	}

	return &NatsJetStreamEventStore{
		db:     db,
		js:     js,
		logger: logger,
	}, nil
}

// CommitEvents writes staged events to the Event Store and Outbox atomically, then emits to JetStream.
func (s *NatsJetStreamEventStore) CommitEvents(ctx context.Context, agg *AccountAggregate) error {
	agg.mu.Lock()
	defer agg.mu.Unlock()

	if len(agg.uncommittedEvs) == 0 {
		return nil
	}

	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return err
	}
	defer tx.Rollback()

	eventStmt, err := tx.PrepareContext(ctx, `
		INSERT INTO domain_events (event_id, aggregate_id, event_type, version, payload, occurred_at)
		VALUES ($1, $2, $3, $4, $5, $6)
	`)
	if err != nil {
		return err
	}
	defer eventStmt.Close()

	outboxStmt, err := tx.PrepareContext(ctx, `
		INSERT INTO transactional_outbox (id, topic, payload, aggregate_id, created_at)
		VALUES ($1, $2, $3, $4, $5)
	`)
	if err != nil {
		return err
	}
	defer outboxStmt.Close()

	for _, evt := range agg.uncommittedEvs {
		// 1. Insert into immutable event store
		_, err := eventStmt.ExecContext(ctx,
			evt.EventID, evt.AggregateID, string(evt.Type), evt.Version, evt.Payload, evt.OccurredAt,
		)
		if err != nil {
			return fmt.Errorf("%w: %v", ErrConcurrencyConflict, err)
		}

		// 2. Insert into transactional outbox within the same ACID boundary
		subject := fmt.Sprintf("banking.account.%s", evt.AggregateID)
		evtBytes, _ := json.Marshal(evt)
		_, err = outboxStmt.ExecContext(ctx,
			evt.EventID, subject, evtBytes, evt.AggregateID, evt.OccurredAt,
		)
		if err != nil {
			return err
		}
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	// 3. Publish to NATS JetStream with native deduplication token
	for _, evt := range agg.uncommittedEvs {
		subject := fmt.Sprintf("banking.account.%s", evt.AggregateID)
		evtBytes, _ := json.Marshal(evt)

		msg := nats.NewMsg(subject)
		msg.Data = evtBytes
		msg.Header.Set(jetstream.MsgIDHeader, evt.EventID) // JetStream deduplication window

		_, err := s.js.PublishMsg(ctx, msg)
		if err != nil {
			s.logger.Warn("Synchronous JetStream publish failed; fallback to background outbox worker",
				"event_id", evt.EventID,
				"error", err,
			)
		}
	}

	agg.uncommittedEvs = nil
	s.logger.Info("Events committed atomically to store and published to JetStream",
		"aggregate_id", agg.ID,
		"committed_version", agg.Version,
	)
	return nil
}

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
	logger.Info("Core Banking Event Sourcing & CQRS Engine initialized successfully.")

	// Demonstrate aggregate initialization and event handling
	acc := NewAccountAggregate("ACC-US-882201")
	_ = acc.Apply(DomainEvent{
		EventID:     "EVT-INIT-01",
		AggregateID: "ACC-US-882201",
		Type:        EventAccountOpened,
		Version:     1,
		OccurredAt:  time.Now().UTC(),
	})

	deposit, _ := json.Marshal(DepositPayload{AmountMinor: 5000000, ReferenceID: "WIRE-DEPOSIT-01"})
	_ = acc.Apply(DomainEvent{
		EventID:     "EVT-DEP-02",
		AggregateID: "ACC-US-882201",
		Type:        EventFundsDeposited,
		Version:     2,
		Payload:     deposit,
		OccurredAt:  time.Now().UTC(),
	})

	logger.Info("Hydrated account status", "balance_minor", acc.BalanceMinor, "version", acc.Version)

	if err := acc.Withdraw(1500000, "POS-PURCHASE-01"); err != nil {
		logger.Error("Withdrawal rejected", "error", err)
	} else {
		logger.Info("Withdrawal staged successfully", "new_balance", acc.BalanceMinor, "version", acc.Version)
	}
}

-- Account Snapshots Master Table
CREATE TABLE account_snapshots (
    account_id UUID NOT NULL,
    version BIGINT NOT NULL,
    balance BIGINT NOT NULL, -- Fixed-point minor currency unit
    reserved_holds BIGINT NOT NULL,
    status VARCHAR(16) NOT NULL,
    snapshot_data JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
    PRIMARY KEY (account_id, version)
);

-- Step 1: Fetch the most recent snapshot for the target account
SELECT * FROM account_snapshots 
WHERE account_id = 'c4b8b4b2-2975-4c07-9b2f-7c152a5c5a01' 
ORDER BY version DESC LIMIT 1;

-- Step 2: Stream only incremental events generated since the snapshot version
SELECT * FROM domain_events 
WHERE aggregate_id = 'c4b8b4b2-2975-4c07-9b2f-7c152a5c5a01' 
  AND version > 45000 
ORDER BY version ASC;