// PHẢN MẪU NGUY HIỂM: Nguy cơ bất đồng bộ ghi kép (Dual-Write Hazard)
func (s *PaymentService) HandleDepositUnsafe(ctx context.Context, cmd DepositCommand) error {
    // 1. Lưu vào Database
    if err := s.repo.SaveEvent(ctx, event); err != nil {
        return err
    }
    // 2. Bắn sang Broker -> Nếu server sập hoặc đứt mạng tại đây, Broker KHÔNG BAO GIỜ nhận được event!
    return s.broker.Publish("account-events", event)
}

// Package eventsourcing hiện thực kiến trúc Event Sourcing & CQRS chuẩn Core Banking 2027.
// Sử dụng Go 1.25: iter.Seq range-over-func, NATS JetStream deduplication, và 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"
)

// Khai báo các lỗi miền tài chính chuẩn
var (
	ErrAccountInactive       = errors.New("tài khoản đang ở trạng thái khóa hoặc đóng")
	ErrInsufficientFunds     = errors.New("số dư khả dụng không đủ để thực hiện ghi nợ")
	ErrConcurrencyConflict   = errors.New("xung đột phiên bản sự kiện lạc quan (OCC conflict)")
	ErrInvalidEventSequence  = errors.New("chuỗi sự kiện không liên tục, mất thứ tự phiên bản")
)

// EventType định danh loại biến động nghiệp vụ
type EventType string

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

// DomainEvent biểu diễn cấu trúc sự kiện miền bất biến
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 chứa dữ liệu nghiệp vụ nạp tiền
type DepositPayload struct {
	AmountMinor int64  `json:"amount_minor"`
	ReferenceID string `json:"reference_id"`
}

// WithdrawPayload chứa dữ liệu nghiệp vụ rút tiền
type WithdrawPayload struct {
	AmountMinor int64  `json:"amount_minor"`
	ReferenceID string `json:"reference_id"`
}

// AccountAggregate quản lý trạng thái và thẩm định bất biến tài khoản thanh toán
type AccountAggregate struct {
	ID             string
	Version        int64
	BalanceMinor   int64
	HoldMinor      int64
	IsActive       bool
	uncommittedEvs []DomainEvent
	mu             sync.RWMutex
}

// NewAccountAggregate khởi tạo một aggregate rỗng phục vụ hydrate
func NewAccountAggregate(id string) *AccountAggregate {
	return &AccountAggregate{
		ID: id,
	}
}

// AvailableBalance tính toán số dư thực tế có thể giao dịch
func (a *AccountAggregate) AvailableBalance() int64 {
	return a.BalanceMinor - a.HoldMinor
}

// Apply cập nhật trạng thái aggregate dựa trên sự kiện và tăng version
func (a *AccountAggregate) Apply(evt DomainEvent) error {
	if evt.Version != a.Version+1 {
		return fmt.Errorf("%w: mong đợi phiên bản %d nhưng nhận được %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("loại sự kiện không được hỗ trợ: %s", evt.Type)
	}

	a.Version = evt.Version
	return nil
}

// HydrateFromHistory tái tạo trạng thái aggregate từ chuỗi sự kiện lịch sử
// Sử dụng Go 1.25 iter.Seq range-over-func iterator
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 kiểm tra bất biến nghiệp vụ và tạo sự kiện rút tiền chưa commit
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: khả dụng %d < yêu cầu %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(),
	}

	// Áp dụng tạm thời vào state cục bộ
	if err := a.Apply(evt); err != nil {
		return err
	}
	a.uncommittedEvs = append(a.uncommittedEvs, evt)
	return nil
}

// NatsJetStreamEventStore quản lý lưu trữ sự kiện và tích hợp NATS JetStream Outbox
type NatsJetStreamEventStore struct {
	db     *sql.DB
	js     jetstream.JetStream
	logger *slog.Logger
}

// NewNatsJetStreamEventStore khởi tạo kho lưu trữ sự kiện tích hợp NATS JetStream
func NewNatsJetStreamEventStore(db *sql.DB, natsURL string, logger *slog.Logger) (*NatsJetStreamEventStore, error) {
	nc, err := nats.Connect(natsURL)
	if err != nil {
		return nil, fmt.Errorf("không thể kết nối NATS: %w", err)
	}

	js, err := jetstream.New(nc)
	if err != nil {
		return nil, fmt.Errorf("không thể khởi tạo JetStream context: %w", err)
	}

	// Đảm bảo Stream cho sự kiện ngân hàng tồn tại
	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("lỗi khởi tạo Stream JetStream: %w", err)
	}

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

// CommitEvents thực thi lưu trữ nguyên tử vào Event Store và Transactional Outbox
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. Ghi sự kiện vào Event Store
		_, err := eventStmt.ExecContext(ctx,
			evt.EventID, evt.AggregateID, string(evt.Type), evt.Version, evt.Payload, evt.OccurredAt,
		)
		if err != nil {
			// Bắt lỗi vi phạm khóa duy nhất (OCC Version Conflict)
			return fmt.Errorf("%w: %v", ErrConcurrencyConflict, err)
		}

		// 2. Ghi sự kiện vào bảng Outbox cùng transaction ACID
		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. Đẩy thông điệp sang NATS JetStream với cơ chế khử trùng lặp (Deduplication)
	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) // Khử trùng lặp chuẩn NATS

		_, err := s.js.PublishMsg(ctx, msg)
		if err != nil {
			// Lưu ý: Nếu NATS publish gặp lỗi ở đây, Worker Outbox nền sẽ quét lại bảng transactional_outbox để gửi lại
			s.logger.Warn("Publish trực tiếp tới JetStream bị trễ, Outbox Worker sẽ xử lý lại", "event_id", evt.EventID, "err", err)
		}
	}

	agg.uncommittedEvs = nil
	s.logger.Info("Sự kiện tài chính đã commit nguyên tử thành công", "aggregate_id", agg.ID, "version", agg.Version)
	return nil
}

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
	logger.Info("Khởi động mô-đun Event Sourcing & CQRS Go 1.25")

	// Minh họa khởi tạo và áp dụng sự kiện trên AccountAggregate
	acc := NewAccountAggregate("ACC-VN-998822")
	_ = acc.Apply(DomainEvent{
		EventID:     "EVT-001",
		AggregateID: "ACC-VN-998822",
		Type:        EventAccountOpened,
		Version:     1,
		OccurredAt:  time.Now().UTC(),
	})

	depositPayload, _ := json.Marshal(DepositPayload{AmountMinor: 10000000, ReferenceID: "SALARY-AUG"})
	_ = acc.Apply(DomainEvent{
		EventID:     "EVT-002",
		AggregateID: "ACC-VN-998822",
		Type:        EventFundsDeposited,
		Version:     2,
		Payload:     depositPayload,
		OccurredAt:  time.Now().UTC(),
	})

	logger.Info("Trạng thái tài khoản sau nạp tiền", "balance_minor", acc.BalanceMinor, "version", acc.Version)

	if err := acc.Withdraw(2500000, "ATM-WITHDRAW-01"); err != nil {
		logger.Error("Không thể rút tiền", "err", err)
	} else {
		logger.Info("Rút tiền thành công, số dư khả dụng mới", "balance_minor", acc.BalanceMinor, "version", acc.Version)
	}
}

-- Lược đồ DDL Bảng Snapshot Tài Khoản
CREATE TABLE account_snapshots (
    account_id UUID NOT NULL,
    version BIGINT NOT NULL,
    balance BIGINT NOT NULL, -- Đơn vị số nguyên minor (đồng hoặc xu)
    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)
);

-- Truy vấn khôi phục trạng thái: Lấy bản Snapshot mới nhất
SELECT * FROM account_snapshots 
WHERE account_id = 'c4b8b4b2-2975-4c07-9b2f-7c152a5c5a01' 
ORDER BY version DESC LIMIT 1;

-- Sau đó chỉ nạp các sự kiện phát sinh sau thời điểm Snapshot:
SELECT * FROM domain_events 
WHERE aggregate_id = 'c4b8b4b2-2975-4c07-9b2f-7c152a5c5a01' 
  AND version > 45000 
ORDER BY version ASC;