Di Chuyển Từ Magento Sang Microservices: Playbook Kiến Trúc 3 Giai Đoạn Zero-Downtime

🇬🇧 Read the English version of this article on tanhdev.com

Tóm tắt cốt lõi (Answer-First): Chuyển đổi một khối monolith thương mại điện tử khổng lồ như Magento 2 sang kiến trúc Microservices phân tán đòi hỏi kỹ thuật chuyển giao không điểm dừng (zero-downtime). Chiến lược khả thi duy nhất trên môi trường production là áp dụng Strangler Fig Pattern 3 giai đoạn:

1. Giai đoạn 1 (Read-Path Strangling): Sử dụng Envoy API Gateway định tuyến toàn bộ traffic đọc (Catalog/Search) sang cụm Go Microservices mới, trong khi dữ liệu được đồng bộ thời gian thực từ MySQL của Magento sang PostgreSQL/OpenSearch thông qua Debezium CDC (Change Data Capture) trên binlog.
2. Giai đoạn 2 (Write-Path & Bidirectional Sync): Tách các domain ghi cốt lõi (Cart, Order, Customer) sang Microservices độc lập, triển khai đồng bộ 2 chiều ngược về Magento thông qua Transactional Outbox Worker bằng Go 1.24 và Apache Kafka để duy trì tương thích ngược với các phân hệ cũ (ERP, WMS, Fulfillment).
3. Giai đoạn 3 (Cutover & 30-Day Hot Standby): Cắt toàn bộ 100% traffic sang Microservices, duy trì Magento ở chế độ bản sao dự phòng nóng (Hot Standby) trong 30 ngày với khả năng rollback dưới 5 phút trước khi decommission máy chủ vật lý/EC2.

1. Cái Bẫy “Big Bang Rewrite” và Thực Tế Khắc Nghiệt Của Magento Monolith

Khi một hệ thống thương mại điện tử Magento 2 đạt mốc hàng triệu SKU, xử lý hàng chục nghìn đơn hàng mỗi ngày trong các chiến dịch Mega Sale, đội ngũ kỹ thuật thường xuyên đối mặt với những cơn ác mộng quen thuộc:

• Thời gian build/deploy kéo dài: Mỗi lần chạy lệnh setup:di:compile và setup:static-content:deploy ngốn từ 25 đến 45 phút, khiến chu kỳ release tính năng mới bị đình trệ.
• Nghẽn cổ chai cơ sở dữ liệu MySQL: Bảng sales_order, quote, và hệ thống bảng EAV (catalog_product_entity_*) chiếm dụng hàng trăm gigabyte, gây lock bảng liên miên khi lượng traffic checkout tăng đột biến.
• Phụ thuộc đan xen (Tight Coupling): Một lỗi nhỏ trong plugin interceptor (aroundSave) của module vận chuyển của bên thứ ba có thể kéo sập toàn bộ luồng thanh toán (checkout flow).

Phản xạ tự nhiên đầu tiên của các kỹ sư là: “Hãy đập bỏ Magento và viết lại từ đầu bằng Microservices Golang/Rust!”

Tuy nhiên, trong kỹ nghệ phần mềm quy mô lớn, “Big Bang Rewrite” (Đập đi xây mới toàn bộ một lần) gần như là một bản án tử hình cho dự án. Một hệ sinh thái Magento đang vận hành mang trên mình hàng trăm logic nghiệp vụ ẩn tàng (hidden business rules), các edge-case tích hợp thuế quan địa phương, cổng thanh toán COD/VNPay/MoMo, và quy trình xử lý đơn hàng nội bộ đã được tinh chỉnh suốt 5–10 năm. Việc phát triển song song một hệ thống mới hoàn toàn trong 12–18 tháng thường dẫn đến:

1. Scope Creep (Phình to phạm vi): Hệ thống cũ liên tục thay đổi tính năng để đáp ứng kinh doanh, khiến hệ thống mới không bao giờ đuổi kịp (Moving Target Problem).
2. Kỳ vọng viển vông về ngày Cutover: Vào ngày chuyển đổi, hàng nghìn dữ liệu lệch pha (data mismatch) giữa hai hệ thống sẽ làm gãy chuỗi cung ứng, gây gián đoạn doanh thu hàng triệu USD.
3. Chi phí cơ hội cạn kiệt: Toàn bộ nguồn lực kỹ thuật bị giam giữ vào việc code lại những thứ đã chạy ổn định thay vì tạo ra giá trị mới cho người dùng.

Giải pháp sống còn duy nhất là Strangler Fig Pattern (Hoa tiêu bóp nghẹt) do Martin Fowler đề xuất: Xây dựng các dịch vụ microservices bao bọc xung quanh monolith cũ, chặn bắt và xử lý từng phần lưu lượng truy cập (traffic) một cách tuần tự, cho đến khi monolith cũ chỉ còn là một cái vỏ rỗng và có thể loại bỏ an toàn.

2. Ba Chướng Ngại Vật Kỹ Thuật Khi Tách Rời Magento

Trước khi định tuyến bất kỳ request nào qua hệ thống mới, kiến trúc sư trưởng phải giải quyết 3 bài toán bất tương thích nền tảng giữa Magento và kiến trúc Microservices hiện đại:

graph TD
    subgraph Monolith ["Magento 2 Legacy Monolith"]
        EAV["EAV Data Model<br/>(400+ Tables, Splitted Attributes)"]
        SeqID["Sequential Integer IDs<br/>(entity_id INT32/64)"]
        MySQLBinlog["MySQL InnoDB Engine<br/>Row-based Binlog (GTID)"]
    end

    subgraph SyncCore ["CDC & Translation Layer"]
        Debezium["Debezium CDC Connector<br/>(Apache Kafka Connect)"]
        IDMap["Bi-directional ID Mapping Service<br/>(Redis + PostgreSQL)"]
        OutboxSync["Go 1.24 Outbox Consumer<br/>(Flattening & Hydration)"]
    end

    subgraph ModernMesh ["Modern Go Microservices"]
        CatalogSvc["Catalog Service<br/>(PostgreSQL Document JSONB)"]
        OrderSvc["Order Service<br/>(Event-Sourced, UUIDv7)"]
        SearchEngine["OpenSearch Cluster<br/>(Flat Product Index)"]
    end

    EAV -->|Streaming Binlog| Debezium
    Debezium --> OutboxSync
    OutboxSync --> IDMap
    OutboxSync --> CatalogSvc
    OutboxSync --> SearchEngine

    OrderSvc -->|Publish order.created| IDMap
    IDMap -->|Translate UUID to INT| Monolith

2.1. Cơn Ác Mộng Schema EAV (Entity-Attribute-Value)

Magento không lưu thông tin sản phẩm trong một bảng phẳng (flat table). Một sản phẩm đơn giản được phân mảnh trên hàng chục bảng:

• catalog_product_entity: Chứa entity_id, sku, created_at.
• catalog_product_entity_varchar: Chứa name, url_key, image.
• catalog_product_entity_int: Chứa status, visibility.
• catalog_product_entity_decimal: Chứa price, special_price, weight.
• catalog_product_entity_text: Chứa description, short_description.

Nếu các Microservices mới cố gắng đọc trực tiếp các bảng này bằng các câu truy vấn đa kết nối (JOIN 15–20 bảng), cơ sở dữ liệu MySQL sẽ nghẽn luồng CPU ngay lập tức.

• Giải pháp: Tầng Streaming ETL qua Debezium CDC sẽ lắng nghe các bảng này, gom nhóm theo entity_id trong một cửa sổ trượt (sliding window 100ms), “làm phẳng” (flatten) toàn bộ thuộc tính thành cấu trúc JSON Document chuẩn hóa, và đẩy trực tiếp vào PostgreSQL (sử dụng kiểu JSONB) cùng cụm OpenSearch phục vụ tìm kiếm.

2.2. Xung Đột Định Danh Khóa Chính: Integer vs. UUIDv7

Magento sử dụng khóa chính dạng số nguyên tự tăng tuần tự (entity_id INT UNSIGNED). Trong khi đó, các Microservices hiện đại sử dụng UUIDv7 (kết hợp timestamp 48-bit với 74-bit entropy ngẫu nhiên) để sinh ID phân tán mà không phụ thuộc vào khóa tập trung của cơ sở dữ liệu.

• Giải pháp: Xây dựng dịch vụ ID Mapping Service trung tâm lưu trữ quan hệ 1-1 giữa (domain, legacy_id) và (domain, uuid). Bảng mapping được index tối ưu trên PostgreSQL và cache toàn bộ trên Redis Cluster để độ trễ tra cứu luôn đạt dưới 0.3ms.

-- Schema bảng ánh xạ định danh hai chiều
CREATE TABLE system_id_mapping (
    domain VARCHAR(32) NOT NULL,          -- 'catalog', 'order', 'customer'
    legacy_id BIGINT NOT NULL,            -- Magento entity_id
    service_uuid UUID NOT NULL,           -- Microservice UUIDv7
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (domain, legacy_id)
);

CREATE UNIQUE INDEX idx_id_mapping_uuid ON system_id_mapping (domain, service_uuid);

2.3. CDC Thời Gian Thực vs. Polling Cronjob

Nhiều đội ngũ cố gắng đồng bộ dữ liệu bằng cách viết các script PHP CLI hoặc Python chạy cron định kỳ mỗi 5 phút quét các bản ghi có updated_at >= NOW() - INTERVAL 5 MINUTE.

• Hậu quả: Polling liên tục tạo các câu lệnh SELECT đè nặng lên disk I/O của MySQL Master, bỏ sót các bản ghi bị xóa cứng (DELETE), và tạo ra độ trễ dữ liệu (data staleness) từ 1–5 phút khiến khách hàng xem phải giá và tồn kho cũ.
• Giải pháp: Triển khai True Change Data Capture (CDC) với Debezium kết nối trực tiếp vào MySQL Binlog ở chế độ Row-based (GTID). Mọi câu lệnh INSERT, UPDATE, DELETE trên Magento đều được chuyển hóa thành sự kiện Kafka trong vòng chưa tới 15 miligiây mà không tiêu tốn tài nguyên tính toán của Magento application server.

3. Kiến Trúc Strangler Fig 3 Giai Đoạn

Quy trình chuyển đổi được chia thành 3 giai đoạn rõ ràng với các ranh giới kiểm soát rủi ro nghiêm ngặt:

sequenceDiagram
    autonumber
    participant Client as Web / Mobile Client
    participant Envoy as Envoy API Gateway
    participant Legacy as Magento Monolith
    participant Mesh as Modern Microservices (Go)
    participant Kafka as Apache Kafka (CDC/Outbox)

    rect rgb(240, 248, 255)
    Note over Client,Kafka: GIAI ĐOẠN 1: Read-Path Strangling
    Client->>Envoy: GET /api/v1/catalog/products
    Envoy->>Mesh: Forward GET (Catalog Service)
    Mesh-->>Client: Trả về dữ liệu phẳng (Postgres/OpenSearch)
    Client->>Envoy: POST /rest/V1/carts/mine/items
    Envoy->>Legacy: Forward POST (Magento Checkout)
    Legacy->>Kafka: Debezium CDC stream MySQL binlog
    Kafka->>Mesh: Go Consumer cập nhật Read Model
    end

    rect rgb(255, 250, 240)
    Note over Client,Kafka: GIAI ĐOẠN 2: Write-Path & Bidirectional Sync
    Client->>Envoy: POST /api/v1/orders (Place Order)
    Envoy->>Mesh: Forward POST (Order Microservice)
    Mesh->>Mesh: Lưu Transactional Outbox (Postgres)
    Mesh->>Kafka: Publish event "order.created"
    Kafka->>Legacy: Legacy Sync Worker ghi vào Magento DB (Reverse Sync)
    Legacy-->>Legacy: ERP/WMS cũ tiếp tục đọc đơn từ Magento
    end

    rect rgb(240, 255, 240)
    Note over Client,Kafka: GIAI ĐOẠN 3: Cutover & Hot Standby (30 Days)
    Client->>Envoy: 100% Traffic (Read & Write)
    Envoy->>Mesh: Toàn bộ xử lý tại Go Microservices
    Mesh->>Kafka: Continuous Reversible Stream
    Kafka->>Legacy: Magento cập nhật như một Passive Replica
    Note over Legacy: Nếu Mesh có sự cố -> Lật Envoy ngược lại Magento trong 5s!
    end

4. Giai Đoạn 1: Định Tuyến Read-Path & Hạ Tầng CDC (Debezium)

Trong giai đoạn này, Magento vẫn là Nguồn Chân Lý Duy Nhất (Single Source of Truth) cho toàn bộ dữ liệu ghi. Hệ thống Microservices mới đóng vai trò là một Read-Replica thông minh với hiệu năng siêu cao.

4.1. Cấu Hình Envoy Proxy Định Tuyến Thông Minh

Envoy Proxy đóng vai trò là tầng ingress bảo vệ, phân tách lưu lượng truy cập dựa trên HTTP Method, URI Path, và Canary Header:

# envoy-strangler-config.yaml
static_resources:
  listeners:
  - name: ingress_listener
    address:
      socket_address:
        address: 0.0.0.0
        port_value: 8080
    filter_chains:
    - filters:
      - name: envoy.filters.network.http_connection_manager
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
          stat_prefix: ingress_http
          route_config:
            name: dynamic_routes
            virtual_hosts:
            - name: ecommerce_vhost
              domains: ["*"]
              routes:
              # 1. Định tuyến luồng đọc Catalog sang Go Catalog Microservice
              - match:
                  prefix: "/api/v1/catalog"
                  headers:
                  - name: ":method"
                    exact_match: "GET"
                route:
                  cluster: go_catalog_service
                  timeout: 1.5s
                  retry_policy:
                    retry_on: "5xx,connect-failure,reset"
                    num_retries: 2
              
              # 2. Canary Testing cho nhóm Beta Users (Header-based)
              - match:
                  prefix: "/api/v1/search"
                  headers:
                  - name: "x-beta-tester"
                    exact_match: "true"
                route:
                  cluster: go_search_service
                  timeout: 1.0s

              # 3. Mặc định toàn bộ Write requests và Admin fallback về Magento Monolith
              - match:
                  prefix: "/"
                route:
                  cluster: legacy_magento_cluster
                  timeout: 30.0s

  clusters:
  - name: go_catalog_service
    connect_timeout: 0.25s
    type: LOGICAL_DNS
    dns_lookup_family: V4_ONLY
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: go_catalog_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: catalog-svc.mesh.internal
                port_value: 8001

  - name: legacy_magento_cluster
    connect_timeout: 1.0s
    type: LOGICAL_DNS
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: legacy_magento_cluster
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: magento.legacy.internal
                port_value: 80

4.2. Cấu Hình Debezium MySQL Connector

Connector theo dõi trực tiếp MySQL binlog của Magento và stream sự kiện thay đổi vào Kafka:

{
  "name": "magento-mysql-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql-magento.internal",
    "database.port": "3306",
    "database.user": "debezium_cdc",
    "database.password": "${file:/secrets/db-credentials.properties:cdc_password}",
    "database.server.id": "184054",
    "database.server.name": "magento_prod",
    "database.include.list": "magento_db",
    "table.include.list": "magento_db.catalog_product_entity,magento_db.catalog_product_entity_varchar,magento_db.catalog_product_entity_int,magento_db.catalog_product_entity_decimal,magento_db.cataloginventory_stock_item",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.magento",
    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",
    "include.schema.changes": "true",
    "snapshot.mode": "when_needed",
    "tombstones.on.delete": "true"
  }
}

4.3. Go 1.24 Catalog CDC Ingestion Worker

Worker Go nhận sự kiện từ Kafka, xử lý flatten dữ liệu EAV và cập nhật PostgreSQL:

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"
	"github.com/segmentio/kafka-go"
)

// DebeziumPayload đại diện cho gói tin CDC từ MySQL binlog
type DebeziumPayload struct {
	Before map[string]interface{} `json:"before"`
	After  map[string]interface{} `json:"after"`
	Op     string                 `json:"op"` // 'c' (create), 'u' (update), 'd' (delete)
	TsMs   int64                  `json:"ts_ms"`
}

type KafkaMessageWrapper struct {
	Payload DebeziumPayload `json:"payload"`
}

type CatalogWorker struct {
	reader *kafka.Reader
	db     *sql.DB
}

func NewCatalogWorker(brokers []string, topic, groupID, dbDSN string) (*CatalogWorker, error) {
	db, err := sql.Open("pgx", dbDSN)
	if err != nil {
		return nil, fmt.Errorf("kết nối postgres thất bại: %w", err)
	}

	db.SetMaxOpenConns(25)
	db.SetMaxIdleConns(10)
	db.SetConnMaxLifetime(5 * time.Minute)

	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:        brokers,
		Topic:          topic,
		GroupID:        groupID,
		MinBytes:       10e3, // 10KB
		MaxBytes:       10e6, // 10MB
		CommitInterval: time.Second,
	})

	return &CatalogWorker{reader: reader, db: db}, nil
}

func (w *CatalogWorker) Start(ctx context.Context) error {
	log.Println("🚀 Catalog CDC Ingestion Worker đang lắng nghe sự kiện...")

	for {
		select {
		case <-ctx.Done():
			log.Println("🛑 Đang đóng Kafka Reader...")
			return w.reader.Close()
		default:
			msg, err := w.reader.FetchMessage(ctx)
			if err != nil {
				if ctx.Err() != nil {
					return nil
				}
				log.Printf("⚠️ Lỗi đọc message: %v\n", err)
				continue
			}

			if err := w.processEvent(ctx, msg.Value); err != nil {
				log.Printf("❌ Lỗi xử lý event offset %d: %v\n", msg.Offset, err)
				// Không commit offset để retry hoặc đưa vào Dead Letter Queue (DLQ)
				continue
			}

			if err := w.reader.CommitMessages(ctx, msg); err != nil {
				log.Printf("⚠️ Lỗi commit offset: %v\n", err)
			}
		}
	}
}

func (w *CatalogWorker) processEvent(ctx context.Context, data []byte) error {
	var env KafkaMessageWrapper
	if err := json.Unmarshal(data, &env); err != nil {
		return fmt.Errorf("unmarshal error: %w", err)
	}

	payload := env.Payload
	switch payload.Op {
	case "c", "u": // Create hoặc Update
		entityID, ok := payload.After["entity_id"].(float64)
		if !ok {
			return nil // Bỏ qua nếu không có entity_id
		}
		sku, _ := payload.After["sku"].(string)

		// Lưu trữ/Cập nhật vào bảng catalog phẳng trên PostgreSQL
		query := `
			INSERT INTO catalog_products (legacy_id, sku, updated_at, attributes)
			VALUES ($1, $2, NOW(), $3)
			ON CONFLICT (legacy_id) DO UPDATE SET
				sku = EXCLUDED.sku,
				updated_at = NOW(),
				attributes = catalog_products.attributes || EXCLUDED.attributes;
		`
		attrJSON, _ := json.Marshal(payload.After)
		_, err := w.db.ExecContext(ctx, query, int64(entityID), sku, attrJSON)
		if err != nil {
			return fmt.Errorf("upsert pg thất bại: %w", err)
		}
		log.Printf("✅ Đã cập nhật sản phẩm ID=%d, SKU=%s\n", int64(entityID), sku)

	case "d": // Delete
		entityID, ok := payload.Before["entity_id"].(float64)
		if !ok {
			return nil
		}
		query := `UPDATE catalog_products SET is_active = false, deleted_at = NOW() WHERE legacy_id = $1;`
		_, err := w.db.ExecContext(ctx, query, int64(entityID))
		if err != nil {
			return fmt.Errorf("soft delete pg thất bại: %w", err)
		}
		log.Printf("🗑️ Đã đánh dấu xóa sản phẩm ID=%d\n", int64(entityID))
	}

	return nil
}

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

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	worker, err := NewCatalogWorker(
		[]string{"localhost:9092"},
		"magento_prod.magento_db.catalog_product_entity",
		"catalog-ingestion-group",
		"postgres://postgres:secret@localhost:5432/mesh_catalog?sslmode=disable",
	)
	if err != nil {
		log.Fatalf("Khởi tạo worker thất bại: %v", err)
	}

	go func() {
		<-sigChan
		log.Println("Nhận tín hiệu dừng, tiến hành tắt gracefully...")
		cancel()
	}()

	if err := worker.Start(ctx); err != nil {
		log.Fatalf("Worker gặp sự cố: %v", err)
	}
}

5. Giai Đoạn 2: Tách Write-Path & Đồng Bộ Hai Chiều (Bidirectional Sync)

Giai đoạn 2 là giai đoạn phức tạp và tiềm ẩn rủi ro lớn nhất. Khi các thao tác ghi (ví dụ: tạo đơn hàng Place Order) được chuyển giao cho Order Service mới xử lý, hệ thống Magento cũ vẫn cần dữ liệu này để các phân hệ quản lý kho (WMS), hậu cần vận chuyển (Fulfillment), và kế toán nội bộ tiếp tục hoạt động bình thường.

5.1. Mô Hình Transactional Outbox Ngược Về Magento

Để đảm bảo tính nhất quán dữ liệu không bị phá vỡ nếu tiến trình bị sập giữa chừng, Order Microservice áp dụng mẫu thiết kế Transactional Outbox:

sequenceDiagram
    autonumber
    participant Client as Mobile / Web Client
    participant OrderSvc as Order Microservice (Go)
    participant OrderDB as PostgreSQL (Order & Outbox)
    participant Relay as Debezium Outbox Relay
    participant Kafka as Apache Kafka
    participant Worker as Legacy Sync Worker (Go)
    participant Magento as Magento 2 Database / API

    Client->>OrderSvc: POST /api/v1/orders (Tạo đơn hàng)
    activate OrderSvc
    OrderSvc->>OrderDB: BEGIN TRANSACTION
    OrderSvc->>OrderDB: INSERT INTO orders (id, total, items...)
    OrderSvc->>OrderDB: INSERT INTO outbox_events (aggregate_type, payload...)
    OrderSvc->>OrderDB: COMMIT
    OrderSvc-->>Client: Trả về HTTP 201 Created (Order UUID)
    deactivate OrderSvc

    Note over OrderDB,Relay: CDC đọc outbox table không lock
    Relay->>Kafka: Publish "order.created"
    Kafka->>Worker: Consume Event
    activate Worker
    Worker->>Worker: Tra cứu/Sinh Legacy ID (system_id_mapping)
    Worker->>Magento: Ghi đơn hàng vào sales_order (REST Bulk API)
    alt Thành công
        Worker->>Worker: Cập nhật status synced = true
    else Thất bại / Timeout
        Worker->>Kafka: Gửi vào Dead Letter Queue (DLQ) sau 5 lần retry
    end
    deactivate Worker

5.2. Thuật Toán Giải Quyết Xung Đột Dữ Liệu (Conflict Resolution)

Khi cả hai hệ thống cùng có khả năng ghi, nguy cơ xung đột trạng thái (ví dụ: khách hủy đơn trên Microservice nhưng nhân viên CSKH lại bấm xác nhận đơn trên Magento Admin) là điều không thể tránh khỏi. Chúng tôi áp dụng ma trận phân giải trạng thái hữu hạn (State Machine Transition Matrix) kết hợp với Monotonic Microsecond Timestamp:

5.3. Production Go 1.24 Legacy Sync Worker (Đồng Bộ Đơn Hàng)

Dưới đây là mã nguồn của Legacy Sync Worker chuyên trách tiêu thụ sự kiện từ Kafka, chuyển dịch cấu trúc Order hiện đại sang payload tương thích với API của Magento:

package main

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"
	"github.com/segmentio/kafka-go"
)

type OrderCreatedEvent struct {
	OrderID     string    `json:"order_id"` // UUIDv7
	CustomerEmail string  `json:"customer_email"`
	GrandTotal  float64   `json:"grand_total"`
	Currency    string    `json:"currency"`
	CreatedAt   time.Time `json:"created_at"`
	Items       []OrderItem `json:"items"`
}

type OrderItem struct {
	SKU      string  `json:"sku"`
	Qty      int     `json:"qty"`
	Price    float64 `json:"price"`
}

// MagentoOrderPayload tương thích Magento 2 REST API format
type MagentoOrderPayload struct {
	Entity struct {
		CustomerEmail string                   `json:"customer_email"`
		BaseGrandTotal float64                 `json:"base_grand_total"`
		GrandTotal    float64                  `json:"grand_total"`
		OrderCurrencyCode string               `json:"order_currency_code"`
		Items         []MagentoOrderItemPayload `json:"items"`
	} `json:"entity"`
}

type MagentoOrderItemPayload struct {
	SKU       string  `json:"sku"`
	QtyOrdered int    `json:"qty_ordered"`
	Price     float64 `json:"price"`
	BasePrice float64 `json:"base_price"`
}

type LegacySyncWorker struct {
	kafkaReader *kafka.Reader
	dlqWriter   *kafka.Writer
	db          *sql.DB
	httpClient  *http.Client
	magentoURL  string
	apiToken    string
}

func NewLegacySyncWorker(brokers []string, dbDSN, magentoURL, token string) (*LegacySyncWorker, error) {
	db, err := sql.Open("pgx", dbDSN)
	if err != nil {
		return nil, fmt.Errorf("postgres conn err: %w", err)
	}

	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers: brokers,
		Topic:   "order.events.created",
		GroupID: "legacy-magento-sync-worker",
	})

	dlqWriter := &kafka.Writer{
		Addr:     kafka.TCP(brokers...),
		Topic:    "order.events.created.dlq",
		Balancer: &kafka.LeastBytes{},
	}

	client := &http.Client{
		Timeout: 10 * time.Second,
	}

	return &LegacySyncWorker{
		kafkaReader: reader,
		dlqWriter:   dlqWriter,
		db:          db,
		httpClient:  client,
		magentoURL:  magentoURL,
		apiToken:    token,
	}, nil
}

func (s *LegacySyncWorker) Run(ctx context.Context) {
	log.Println("🔄 Legacy Sync Worker sẵn sàng đồng bộ sang Magento...")

	for {
		msg, err := s.kafkaReader.ReadMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			log.Printf("Lỗi đọc message: %v\n", err)
			continue
		}

		var event OrderCreatedEvent
		if err := json.Unmarshal(msg.Value, &event); err != nil {
			log.Printf("Unmarshal event lỗi: %v, đẩy sang DLQ\n", err)
			s.sendToDLQ(ctx, msg.Value, err.Error())
			continue
		}

		if err := s.syncToMagentoWithRetry(ctx, event, 3); err != nil {
			log.Printf("❌ Đồng bộ Order %s thất bại sau retry: %v. Đẩy sang DLQ.\n", event.OrderID, err)
			s.sendToDLQ(ctx, msg.Value, err.Error())
		} else {
			log.Printf("🎉 Đồng bộ thành công Order %s sang Magento\n", event.OrderID)
		}
	}
}

func (s *LegacySyncWorker) syncToMagentoWithRetry(ctx context.Context, order OrderCreatedEvent, maxRetries int) error {
	var lastErr error

	for attempt := 1; attempt <= maxRetries; attempt++ {
		// 1. Kiểm tra xem Order đã được sync chưa (Idempotency Guard)
		var existingLegacyID sql.NullInt64
		err := s.db.QueryRowContext(ctx, "SELECT legacy_id FROM system_id_mapping WHERE service_uuid = $1 AND domain = 'order'", order.OrderID).Scan(&existingLegacyID)
		if err == nil && existingLegacyID.Valid {
			log.Printf("ℹ️ Order %s đã tồn tại trên Magento với ID=%d (Bỏ qua sync)\n", order.OrderID, existingLegacyID.Int64)
			return nil
		}

		// 2. Chuyển đổi sang Magento Payload
		var mgItems []MagentoOrderItemPayload
		for _, it := range order.Items {
			mgItems = append(mgItems, MagentoOrderItemPayload{
				SKU:        it.SKU,
				QtyOrdered: it.Qty,
				Price:      it.Price,
				BasePrice:  it.Price,
			})
		}

		mgPayload := MagentoOrderPayload{}
		mgPayload.Entity.CustomerEmail = order.CustomerEmail
		mgPayload.Entity.GrandTotal = order.GrandTotal
		mgPayload.Entity.BaseGrandTotal = order.GrandTotal
		mgPayload.Entity.OrderCurrencyCode = order.Currency
		mgPayload.Entity.Items = mgItems

		payloadBytes, _ := json.Marshal(mgPayload)
		req, _ := http.NewRequestWithContext(ctx, "POST", s.magentoURL+"/rest/V1/orders", bytes.NewReader(payloadBytes))
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Authorization", "Bearer "+s.apiToken)

		resp, err := s.httpClient.Do(req)
		if err == nil && resp.StatusCode >= 200 && resp.StatusCode < 300 {
			// Đọc response Magento để lấy entity_id mới tạo
			var mgResponse struct {
				EntityID int64 `json:"entity_id"`
			}
			_ = json.NewDecoder(resp.Body).Decode(&mgResponse)
			_ = resp.Body.Close()

			// 3. Ghi quan hệ ánh xạ vào DB
			_, _ = s.db.ExecContext(ctx,
				"INSERT INTO system_id_mapping (domain, legacy_id, service_uuid) VALUES ('order', $1, $2) ON CONFLICT DO NOTHING",
				mgResponse.EntityID, order.OrderID,
			)
			return nil
		}

		if resp != nil {
			_ = resp.Body.Close()
		}
		lastErr = err
		log.Printf("⚠️ Sync attempt %d/%d thất bại: %v. Đợi retry...\n", attempt, maxRetries, err)
		time.Sleep(time.Duration(attempt) * 500 * time.Millisecond)
	}

	return lastErr
}

func (s *LegacySyncWorker) sendToDLQ(ctx context.Context, data []byte, reason string) {
	_ = s.dlqWriter.WriteMessages(ctx, kafka.Message{
		Key:   []byte(reason),
		Value: data,
		Time:  time.Now(),
	})
}

6. Giai Đoạn 3: Cắt Cầu (Cutover), 30-Day Hot Standby & Rollback Runbook

Giai đoạn cuối cùng đánh dấu sự độc lập hoàn toàn của hệ sinh thái Microservices mới. Tuy nhiên, thay vì xóa bỏ máy chủ Magento ngay khi đạt 100% traffic, chúng tôi thiết lập cơ chế Hot Standby đảo ngược (Reverse Hot Standby) kéo dài 30 ngày.

stateDiagram-v2
    [*] --> Phase1_ReadStrangled: Tuần 1-3 (Read qua Mesh)
    Phase1_ReadStrangled --> Phase2_Bidirectional: Tuần 4-7 (Write qua Mesh + Reverse Sync)
    Phase2_Bidirectional --> Phase3_HotStandby: Tuần 8 (100% Traffic, Magento làm Standby)
    
    state Phase3_HotStandby {
        [*] --> ContinuousChecksum: Data Integrity Verification Job (Mỗi 1h)
        ContinuousChecksum --> NormalOperation: Dữ liệu khớp 100%
        ContinuousChecksum --> DisasterDetected: Phát hiện lỗi kiến trúc nghiêm trọng
        DisasterDetected --> RollbackExecution: Kích hoạt Rollback trong 5 phút
    }
    
    NormalOperation --> Decommission: Sau 30 ngày an toàn
    RollbackExecution --> Phase2_Bidirectional: Lật Envoy về lại Magento Monolith
    Decommission --> [*]: Terminate EC2 / DB Dropped

6.1. Tiêu Chí Go / No-Go Cắt Cầu (Cutover Criteria)

Trước khi chuyển 100% traffic sang Go Microservices, ban kỹ thuật phải nghiệm thu bảng kiểm tra (checklist) sau:

1. P99 Latency SLA: Độ trễ P99 của toàn bộ API Read < 80ms, Write < 250ms trong đợt giả lập tải 300% đỉnh điểm.
2. Zero Ingestion Lag: Độ trễ của Debezium CDC và Outbox Consumer duy trì < 200ms trong 72 giờ liên tục.
3. Data Integrity Checksum: Script kiểm tra đối soát tự động chạy mỗi giờ giữa PostgreSQL và Magento MySQL đạt tỷ lệ khớp 100% về doanh số và số lượng hàng tồn kho.

6.2. Runbook Rollback Trong 5 Phút (Disaster Recovery)

Nếu xảy ra thảm họa phần mềm chưa từng lường trước trong tuần đầu sau cutover, quy trình rollback về Magento được thực thi chuẩn xác theo 4 bước:

1. Bước 1: Chuyển cờ Envoy Routing về Magento Monolith (Thời gian: 10 giây)

   # Cập nhật Envoy ConfigMap hoặc trigger Consul Service Mesh
   kubectl apply -f /deploy/envoy/rollback-to-magento.yaml
   kubectl rollout restart deployment/envoy-gateway -n ingress
   

2. Bước 2: Dừng (Pause) các Consumer của Microservices (Thời gian: 30 giây) Tạm dừng các Go worker để ngăn chặn race condition ghi đè trạng thái.

   kubectl scale deployment/order-service-worker --replicas=0 -n core-services
   

3. Bước 3: Xả hàng đợi Outbox ngược vào Magento (Thời gian: 2 phút) Kích hoạt script kiểm tra các đơn hàng phát sinh trong Microservice những phút cuối cùng chưa kịp sync sang Magento để bơm trực tiếp vào database MySQL.

   go run /tools/emergency_drain/main.go --mode=flush-to-legacy
   

4. Bước 4: Kiểm tra trạng thái Checkout trên Magento (Thời gian: 1 phút) Bắn synthetic transaction tạo 1 đơn hàng test trên Magento frontend để xác nhận luồng thanh toán hoạt động ổn định 100%.

Sau 30 ngày duy trì Hot Standby với zero incident, toàn bộ hạ tầng EC2 của Magento được tắt nguồn (powered down), snapshot cơ sở dữ liệu MySQL được lưu trữ lạnh trên AWS S3 Glacier, chính thức hoàn tất quá trình Strangler Fig.

7. So Sánh Hiệu Năng & Chi Phí Hạ Tầng

Sau khi di chuyển hoàn toàn 10 core domain từ Magento sang Golang Microservices, kết quả đo lường thực tế trên hệ thống sản xuất (production) thu được:

8. Tổng Kết & Bài Học Thực Chiến

Việc di chuyển một monolith đồ sộ như Magento sang Microservices không phải là bài toán viết lại code, mà là bài toán bảo toàn trạng thái dữ liệu trong trạng thái chuyển động.

1. Đừng bao giờ cắt cầu kiểu Big Bang: Rủi ro kinh doanh luôn lớn hơn bất kỳ vẻ đẹp kỹ thuật thuần túy nào.
2. Debezium CDC là vũ khí tối thượng: Giúp tách dữ liệu khỏi Magento mà không cần can thiệp hay làm chậm mã nguồn PHP cũ.
3. Idempotency là chìa khóa của đồng bộ 2 chiều: Bất kỳ message nào trên mạng phân tán đều có thể bị gửi trùng lặp (At-least-once delivery). Mọi worker đều phải có cơ chế guard idempotent nghiêm ngặt.
4. Luôn có đường lui (Hot Standby): Một kiến trúc sư giỏi không phải là người tự tin hệ thống mới không bao giờ lỗi, mà là người thiết kế sẵn đường thoát hiểm an toàn khi tình huống xấu nhất xảy ra.

🔗 Tài Liệu & Chuyên Đề Chuyên Sâu Liên Quan:

• Kiến trúc Microservices Golang & DDD: Thiết kế 21 Service E-Commerce
• Mastering Event-Driven Architecture với Dapr & Golang
• Ảo Tưởng Microservices: Vì Sao Modular Monolith Golang Là Điểm Đến Lý Tưởng
• MySQL Horizontal Scaling: Sharding, Vitess & Distributed Transactions

❓ Câu Hỏi Thường Gặp (FAQ)