┌─────────────────────────────────────────────────────────────────┐
│                    แพลตฟอร์ม AI แบบ EVENT-DRIVEN               │
├─────────────────────────────────────────────────────────────────┤
│                                                                   │
│   ┌──────────────┐      เหตุการณ์     ┌──────────────┐          │
│   │   Producer   │ ──────────────> │ Message      │          │
│   │    Agent     │                   │ Queue        │          │
│   └──────────────┘                   │ (Kafka/      │          │
│                                      │  RabbitMQ)   │          │
│   ┌──────────────┐      เหตุการณ์     └──────┬───────┘          │
│   │   Consumer   │ <──────────────────────┘                     │
│   │    Agent     │                                              │
│   └──────────────┘                                              │
│                                                                   │
│   ┌──────────────┐      เหตุการณ์     ┌──────────────┐          │
│   │  AI Agent    │ <────────────── │   n8n        │          │
│   │ Orchestrator │ ──────────────> │ Workflows    │          │
│   └──────────────┘                   └──────────────┘          │
│                                                                   │
└─────────────────────────────────────────────────────────────────┘

// โครงสร้างเหตุการณ์ AI Agent ตัวอย่าง
interface AIAgentEvent {
  eventId: string;           // UUID v4
  eventType: string;         // "user.message.received"
  timestamp: string;         // ISO 8601 UTC
  correlationId: string;     // เชื่อมโยงเหตุการณ์ที่เกี่ยวข้อง
  causationId?: string;      // เหตุการณ์ก่อนหน้าที่กระตุ้นเหตุการณ์นี้
  payload: {
    agentId: string;
    sessionId: string;
    userId: string;
    intent?: string;
    entities?: Record<string, any>;
    metadata?: Record<string, any>;
  };
  version: number;           // เวอร์ชัน Schema เหตุการณ์
}

┌─────────────────────────────────────────────────────────────┐
│                      EVENT STREAM                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  เวลา ─────────────────────────────────────────────────> │
│                                                             │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐        │
│  │ เหตุ    │  │ เหตุ    │  │ เหตุ    │  │ เหตุ    │  ...  │
│  │การณ์ 1 │  │การณ์ 2 │  │การณ์ 3 │  │การณ์ 4 │        │
│  │ ข้อ    │  │ Agent   │  │ เรียก   │  │ ตอบ     │        │
│  │ความ    │  │ เริ่ม   │  │ เครื่อง │  │ กลับ    │        │
│  └─────────┘  └─────────┘  └─────────┘  └─────────┘        │
│                                                             │
│  พาร์ติชัน 1: agent.session.abc123                        │
│  พาร์ติชัน 2: agent.session.def456                        │
│  พาร์ติชัน 3: agent.session.ghi789                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

UPDATE agent_sessions 
SET status = 'completed', last_response = 'ขอบคุณ!' 
WHERE session_id = 'abc123';

INSERT INTO events (type, payload) VALUES 
('agent.session.started', '{"session_id": "abc123"}'),
('user.message.received', '{"session_id": "abc123", "text": "สวัสดี"}'),
('agent.response.generated', '{"session_id": "abc123", "response": "สวัสดี! ฉันช่วยอะไรได้?"}'),
('user.message.received', '{"session_id": "abc123", "text": "ขอบคุณ ลาก่อน"}'),
('agent.session.completed', '{"session_id": "abc123", "summary": "..."}');

{
  "eventId": "550e8400-e29b-41d4-a716-446655440000",
  "eventType": "agent.tool.invoked",
  "timestamp": "2026-05-25T10:15:30Z",
  "version": 2,
  "schema": "https://api.example.com/schemas/agent-tool-invoked/v2",
  "payload": {
    "agentId": "customer-support-v2.3",
    "sessionId": "sess_abc123",
    "toolName": "vector_search",
    "toolVersion": "1.5.0",
    "input": {
      "query": "นโยบายการคืนสินค้า",
      "topK": 5
    },
    "output": {
      "results": [...],
      "latency_ms": 245
    },
    "metadata": {
      "model": "gpt-4o",
      "temperature": 0.7,
      "_legacy_field": "ล้าสมัยแต่ยังคงเก็บไว้"
    }
  }
}

# docker-compose.yml สำหรับการพัฒนา Kafka
version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

# สร้าง Topic สำหรับเหตุการณ์ Agent ที่มีผลผลิตสูง
kafka-topics.sh --create \
  --topic ai-agent-events \
  --partitions 12 \
  --replication-factor 3 \
  --config retention.ms=604800000 \
  --config compression.type=lz4 \
  --bootstrap-server kafka:9092

# สร้าง Topic แบบ Compacted สำหรับสถานะ Agent (Event Sourcing)
kafka-topics.sh --create \
  --topic ai-agent-state \
  --partitions 6 \
  --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.5 \
  --bootstrap-server kafka:9092

# docker-compose.yml สำหรับ RabbitMQ
version: '3.8'
services:
  rabbitmq:
    image: rabbitmq:3-management-alpine
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: secure_password
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
      - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf

volumes:
  rabbitmq_data:

# rabbitmq.conf
# เปิดใช้งาน Quorum Queues สำหรับความทนทาน
queue_type = quorum

# Stream Queues สำหรับสถานการณ์ผลผลิตสูง
# (RabbitMQ 3.9+)
queue_master_locator = min-masters

# Consumer Prefetch สำหรับ Agent Workers
consumer_timeout = 300000

# Message TTL สำหรับเหตุการณ์ Agent แบบชั่วคราว
# (เซสชั่นระยะยาวอาจต้องการนานกว่านี้)
message_ttl = 3600000

# คำสั่ง Redis CLI สำหรับ AI Agent Streams

# เพิ่มเหตุการณ์ใน Stream
XADD ai-agent-events * \
  eventType user.message.received \
  sessionId sess_abc123 \
  userId user_456 \
  message "สวัสดี ฉันต้องการความช่วยเหลือ"

# สร้าง Consumer Group
XGROUP CREATE ai-agent-events agent-workers $ MKSTREAM

# อ่านเหตุการณ์เป็น Consumer
XREADGROUP GROUP agent-workers worker-1 \
  COUNT 10 \
  BLOCK 5000 \
  STREAMS ai-agent-events >

# ยืนยันการประมวลผล
XACK ai-agent-events agent-workers 1526569495631-0

# ตัดเหตุการณ์เก่า (เก็บ 10000 ล่าสุด)
XTRIM ai-agent-events MAXLEN ~ 10000

┌─────────────────────────────────────────────────────────────┐
│                    สถาปัตยกรรม MQ แบบผสม                     │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────┐                                              │
│  │  Kafka   │ ◄──── Event Store (Event Sourcing)          │
│  │          │       Analytics, Audit Trail                  │
│  └────┬─────┘                                              │
│       │                                                    │
│       │ replicate (เหตุการณ์ที่เลือก)                       │
│       ▼                                                    │
│  ┌──────────┐     ┌──────────┐     ┌──────────┐           │
│  │ RabbitMQ │◄───►│   n8n    │◄───►│   LLM    │           │
│  │          │     │ Workflows │     │  APIs    │           │
│  └────┬─────┘     └──────────┘     └──────────┘           │
│       │                                                    │
│       │ state/session cache                                │
│       ▼                                                    │
│  ┌──────────┐                                              │
│  │  Redis   │ ◄──── Agent Context, Sessions               │
│  │          │       Rate Limiting                          │
│  └──────────┘                                              │
│                                                             │
└─────────────────────────────────────────────────────────────┘

{
  "nodes": [
    {
      "parameters": {
        "topic": "ai-agent-events",
        "groupId": "n8n-agent-workers",
        "options": {
          "fromOffset": "latest"
        }
      },
      "name": "Kafka Trigger",
      "type": "n8n-nodes-base.kafkaTrigger",
      "typeVersion": 1,
      "position": [250, 300]
    }
  ]
}

{
  "nodes": [
    {
      "parameters": {
        "queue": "agent-task-queue",
        "options": {
          "durable": true,
          "acknowledge": true
        }
      },
      "name": "RabbitMQ Trigger",
      "type": "n8n-nodes-base.rabbitmqTrigger",
      "typeVersion": 1,
      "position": [250, 300]
    }
  ]
}

┌─────────────────────────────────────────────────────────────┐
│                    รูปแบบ EVENT ROUTER                     │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌───────────────┐                                          │
│  │ Kafka Trigger │                                          │
│  │ (ทุกเหตุการณ์)│                                          │
│  └───────┬───────┘                                          │
│          │                                                  │
│          ▼                                                  │
│  ┌───────────────┐                                          │
│  │   Switch      │                                          │
│  │ (ประเภทเหตุ) │                                          │
│  └───────┼───────┘                                          │
│          │                                                  │
│    ┌─────┼─────┬────────┐                                   │
│    │     │     │        │                                   │
│    ▼     ▼     ▼        ▼                                   │
│ ┌────┐ ┌────┐ ┌────┐ ┌────┐                                │
│ │Cust│ │Tech│ │Sales│ │Fraud│                                │
│ │Serv│ │Supp│ │Agent│ │Det │                                │
│ │ WF │ │ WF │ │ WF  │ │ WF │                                │
│ └────┘ └────┘ └────┘ └────┘                                │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                  รูปแบบ SCATTER-GATHER                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│                    ┌──────────┐                            │
│                    │  Event   │                            │
│                    │รับแล้ว  │                            │
│                    └────┬─────┘                            │
│                         │                                   │
│            ┌────────────┼────────────┐                    │
│            │            │            │                     │
│            ▼            ▼            ▼                     │
│      ┌──────────┐ ┌──────────┐ ┌──────────┐              │
│      │ Sentiment│ │  Entity  │ │  Intent  │              │
│      │  Agent   │ │  Agent   │ │  Agent   │              │
│      └────┬─────┘ └────┬─────┘ └────┬─────┘              │
│           │            │            │                     │
│           └────────────┼────────────┘                     │
│                        │                                   │
│                        ▼                                   │
│                 ┌──────────┐                              │
│                 │ รวม       │                              │
│                 │ผลลัพธ์  │                              │
│                 └────┬─────┘                              │
│                      │                                     │
│                      ▼                                     │
│                 ┌──────────┐                              │
│                 │ Response │                              │
│                 │  Agent   │                              │
│                 └──────────┘                              │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│              สถาปัตยกรรม AGENT แบบ EVENT SOURCED             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌──────────────┐                                          │
│   │   Event      │          ┌──────────────┐               │
│   │   Store      │◄─────────│   Command    │               │
│   │  (Kafka/     │          │   Handler    │               │
│   │   Database)  │          └──────────────┘               │
│   └──────┬───────┘                                          │
│          │                                                  │
│          │ Append Only                                      │
│          ▼                                                  │
│   ┌──────────────┐          ┌──────────────┐               │
│   │   Event      │─────────►│  Projections │               │
│   │   Stream     │          │  (Read Model)│               │
│   └──────────────┘          └──────┬───────┘               │
│                                    │                        │
│                                    ▼                        │
│                             ┌──────────────┐               │
│                             │   Agent      │               │
│                             │   State      │               │
│                             └──────────────┘               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

// เหตุการณ์หลักของ Agent
interface AgentSessionStarted {
  type: 'agent.session.started';
  payload: {
    sessionId: string;
    agentId: string;
    userId: string;
    metadata: {
      source: 'web' | 'mobile' | 'api';
      ipAddress: string;
      userAgent: string;
    };
    initialContext?: Record<string, any>;
  };
}

interface UserMessageReceived {
  type: 'user.message.received';
  payload: {
    sessionId: string;
    messageId: string;
    content: string;
    contentType: 'text' | 'image' | 'voice' | 'file';
    attachments?: Attachment[];
    metadata: {
      timestamp: string;
      timezone: string;
      locale: string;
    };
  };
}

interface IntentClassified {
  type: 'intent.classified';
  payload: {
    sessionId: string;
    messageId: string;
    intent: string;
    confidence: number;
    entities: Entity[];
    alternatives: IntentAlternative[];
  };
}

interface ToolInvoked {
  type: 'tool.invoked';
  payload: {
    sessionId: string;
    toolName: string;
    toolVersion: string;
    input: Record<string, any>;
    invocationId: string;
  };
}

interface ToolCompleted {
  type: 'tool.completed';
  payload: {
    sessionId: string;
    invocationId: string;
    output: any;
    durationMs: number;
    success: boolean;
    error?: string;
  };
}

interface LLMRequested {
  type: 'llm.requested';
  payload: {
    sessionId: string;
    requestId: string;
    model: string;
    messages: ChatMessage[];
    parameters: LLMParameters;
    systemPrompt?: string;
  };
}

interface LLMResponded {
  type: 'llm.responded';
  payload: {
    sessionId: string;
    requestId: string;
    response: string;
    tokens: {
      prompt: number;
      completion: number;
      total: number;
    };
    latencyMs: number;
    finishReason: string;
  };
}

interface AgentResponseGenerated {
  type: 'agent.response.generated';
  payload: {
    sessionId: string;
    responseId: string;
    content: string;
    contentType: 'text' | 'structured';
    actions?: AgentAction[];
    confidence: number;
  };
}

interface SessionContextUpdated {
  type: 'session.context.updated';
  payload: {
    sessionId: string;
    updateType: 'memory' | 'preference' | 'state';
    key: string;
    oldValue?: any;
    newValue: any;
  };
}

interface AgentSessionEnded {
  type: 'agent.session.ended';
  payload: {
    sessionId: string;
    reason: 'user_close' | 'timeout' | 'completed' | 'error';
    summary?: string;
    durationSeconds: number;
    messageCount: number;
  };
}

// การจัดการ Snapshot ใน n8n
const SNAPSHOT_FREQUENCY = 100; // ทุก 100 เหตุการณ์

async function maybeCreateSnapshot(sessionId, currentState, eventSequence) {
  if (eventSequence % SNAPSHOT_FREQUENCY === 0) {
    await $getAll('postgres', {
      operation': 'executeQuery',
      query: `
        INSERT INTO agent_session_projections (session_id, state, last_event_sequence, updated_at)
        VALUES ('${sessionId}', '${JSON.stringify(currentState)}', ${eventSequence}, NOW())
        ON CONFLICT (session_id) DO UPDATE SET
          state = EXCLUDED.state,
          last_event_sequence = EXCLUDED.last_event_sequence,
          updated_at = NOW(),
          version = agent_session_projections.version + 1
      `
    });
  }
}

// การเติมสถานะที่ปรับปรุง
async function getStateWithSnapshot(sessionId) {
  // ดึง Snapshot ล่าสุด
  const snapshot = await $getAll('postgres', {
    operation: 'executeQuery',
    query: `SELECT * FROM agent_session_projections WHERE session_id = '${sessionId}'`
  });
  
  if (snapshot.length === 0) {
    // ไม่มี Snapshot - เล่นเหตุการณ์ทั้งหมดซ้ำ
    return replayEvents(sessionId, 0);
  }
  
  // เล่นเฉพาะเหตุการณ์หลัง Snapshot
  const { state, last_event_sequence } = snapshot[0];
  const newEvents = await $getAll('postgres', {
    operation: 'executeQuery',
    query: `
      SELECT * FROM agent_events 
      WHERE session_id = '${sessionId}' 
      AND event_sequence > ${last_event_sequence}
      ORDER BY event_sequence ASC
    `
  });
  
  return applyEvents(state, newEvents);
}

┌─────────────────────────────────────────────────────────────┐
│                    การประสานงาน SAGA                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────────┐                                       │
│   │ Saga            │                                       │
│   │ Orchestrator    │                                       │
│   │ (n8n workflow)  │                                       │
│   └────────┬────────┘                                       │
│            │                                                │
│    ┌───────┼───────┬──────────┬──────────┐                 │
│    │       │       │          │          │                  │
│    ▼       ▼       ▼          ▼          ▼                  │
│ ┌─────┐ ┌─────┐ ┌─────┐   ┌─────┐   ┌─────┐               │
│ │Inv. │ │Pay. │ │Ship │   │Notif│   │Analytics│           │
│ │Agent│ │Agent│ │Agent│   │Agent│   │ Agent  │             │
│ └──┬──┘ └──┬──┘ └──┬──┘   └──┬──┘   └───┬────┘             │
│    │       │       │          │          │                  │
│    └───────┴───────┴──────────┴──────────┘                 │
│                    │                                        │
│                    ▼                                        │
│         ┌─────────────────┐                                │
│         │  Event Store    │                                │
│         │ (Saga State)  │                                │
│         └─────────────────┘                                │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                   SAGA CHOREOGRAPHY                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌────────┐    order.created    ┌────────┐                 │
│  │ Order  │ ──────────────────> │Inv.    │                 │
│  │ Agent  │                     │ Agent  │                 │
│  └────────┘                     └────┬───┘                 │
│                                      │                      │
│                     inv.reserved     │                      │
│  ┌────────┐    <────────────────────┘                      │
│  │ Payment│                                              │
│  │ Agent  │ ──────────────────> ┌────────┐               │
│  └────────┘    payment.processed│ Ship   │               │
│       ▲                         │ Agent  │               │
│       │      ship.reserved      └────┬───┘               │
│       └──────────────────────────────┘                      │
│                                                             │
│  ไม่มีตัวประสานงานกลาง - Agent สื่อสารผ่านเหตุการณ์       │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                   SAGA ORCHESTRATION                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│                    ┌──────────────┐                        │
│                    │ Orchestrator │                        │
│                    └──────┬───────┘                        │
│                           │                                │
│           ┌───────────────┼───────────────┐                │
│           │               │               │                │
│           ▼               ▼               ▼                │
│      ┌────────┐      ┌────────┐      ┌────────┐           │
│      │Inv.    │      │Payment │      │Ship    │           │
│      │ Agent  │      │ Agent  │      │ Agent  │           │
│      └────────┘      └────────┘      └────────┘           │
│                                                             │
│  ตัวประสานงานกลางกำกับดูแลแต่ละขั้นตอน                     │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                      สถาปัตยกรรม CQRS                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────────────────────────────────┐               │
│   │            ฝั่ง COMMAND                  │               │
│   │  ┌──────────────┐    ┌──────────────┐  │               │
│   │  │   Command    │───►│   Event      │  │               │
│   │  │   Handler    │    │   Store      │  │               │
│   │  └──────────────┘    └──────┬───────┘  │               │
│   └─────────────────────────────┼──────────┘               │
│                                 │                           │
│                                 │ เหตุการณ์                │
│                                 ▼                           │
│   ┌─────────────────────────────────────────┐               │
│   │            ฝั่ง PROJECTION               │               │
│   │  ┌──────────────┐    ┌──────────────┐  │               │
│   │  │   Event      │───►│   Read       │  │               │
│   │  │   Projector  │    │   Models     │  │               │
│   │  └──────────────┘    └──────┬───────┘  │               │
│   └─────────────────────────────┼──────────┘               │
│                                 │                           │
│                                 ▼                           │
│   ┌─────────────────────────────────────────┐               │
│   │            ฝั่ง QUERY                    │               │
│   │  ┌──────────────┐    ┌──────────────┐  │               │
│   │  │   Query      │◄───│   Read       │  │               │
│   │  │   Handler    │    │   Database   │  │               │
│   │  └──────────────┘    └──────────────┘  │               │
│   └─────────────────────────────────────────┘               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                การประมวลผล STREAM แบบเรียลไทม์               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   แหล่งที่มา:                                              │
│   ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐             │
│   │Kafka   │ │RabbitMQ│ │Webhooks│ │ IoT    │             │
│   │Events  │ │Queue   │ │       │ │Sensors │             │
│   └───┬────┘ └────┬───┘ └───┬────┘ └───┬────┘             │
│       │           │         │          │                  │
│       └───────────┴─────────┴──────────┘                  │
│                   │                                         │
│                   ▼                                         │
│   ┌───────────────────────────────────────┐                 │
│   │     n8n Workflow แบบเรียลไทม์       │                 │
│   │  ┌─────────┐  ┌─────────┐  ┌────────┐│                 │
│   │  │ Filter  │─►│ Enrich  │─►│ Process││                 │
│   │  │         │  │         │  │        ││                 │
│   │  └─────────┘  └─────────┘  └────────┘│                 │
│   └────────────────┬──────────────────────┘                 │
│                    │                                        │
│                    ▼                                        │
│   ┌───────────────────────────────────────┐                 │
│   │     Sinks / การกระทำ                 │                 │
│   │  ┌────────┐ ┌────────┐ ┌────────┐    │                 │
│   │  │ Alert  │ │ Update │ │ Notify │    │                 │
│   │  │ Trigger│ │Database│ │ User   │    │                 │
│   │  └────────┘ └────────┘ └────────┘    │                 │
│   └─────────────────────────────────────┘                 │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                   กระบวนการจัดการข้อผิดพลาด                   │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────┐                                           │
│   │   เหตุ      │                                           │
│   │  การณ์รับ  │                                           │
│   └──────┬──────┘                                           │
│          │                                                  │
│          ▼                                                  │
│   ┌─────────────┐                                           │
│   │   Try       │◄──────────────────────────────┐           │
│   │  Process    │                               │           │
│   └──────┬──────┘                               │           │
│          │                                      │           │
│    ┌─────┴─────┐                                │           │
│    │           │                                │           │
│ สำเร็จ    ล้มเหลว                              Retry       │
│    │           │                                │           │
│    ▼           ▼                                │           │
│ ┌──────┐   ┌─────────┐      ┌──────────┐        │           │
│ │Ack   │   │ Retry?  │──ไม่──►│  DLQ     │        │           │
│ │Msg   │   │ (count) │      │ (Analyze)│        │           │
│ └──────┘   └────┬────┘      └──────────┘        │           │
│            ใช่  │                               │           │
│                 └───────────────────────────────┘           │
│                   (delay + back-off)                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

// การกำหนดค่า Retry
const MAX_RETRIES = 5;
const BASE_DELAY_MS = 1000;
const MAX_DELAY_MS = 60000;

async function processWithRetry(event) {
  const retryCount = event.metadata?.retryCount || 0;
  
  try {
    // พยายามประมวลผล
    const result = await processEvent(event);
    return { success: true, result };
    
  } catch (error) {
    if (retryCount < MAX_RETRIES) {
      // คำนวณความล่าช้าด้วย exponential backoff + jitter
      const delay = Math.min(
        BASE_DELAY_MS * Math.pow(2, retryCount),
        MAX_DELAY_MS
      );
      const jitter = Math.random() * 1000;
      const totalDelay = delay + jitter;
      
      // อัปเดตเหตุการณ์ด้วยข้อมูล retry
      const retryEvent = {
        ...event,
        metadata: {
          ...event.metadata,
          retryCount: retryCount + 1,
          lastError: error.message,
          nextRetryAt: new Date(Date.now() + totalDelay).toISOString()
        }
      };
      
      // เผยแพร่ไปยังคิว retry พร้อมความล่าช้า
      await publishToRetryQueue(retryEvent, totalDelay);
      
      return {
        success: false,
        willRetry: true,
        retryCount: retryCount + 1,
        delayMs: totalDelay
      };
      
    } else {
      // เกิน Max retries - ส่งไปยัง DLQ
      await publishToDLQ(event, error);
      
      return {
        success: false,
        willRetry: false,
        reason: 'MAX_RETRIES_EXCEEDED',
        error: error.message
      };
    }
  }
}

-- Schema ของ Dead Letter Queue
CREATE TABLE dead_letter_queue (
    id SERIAL PRIMARY KEY,
    dlq_id UUID DEFAULT gen_random_uuid(),
    original_event JSONB NOT NULL,
    error_info JSONB NOT NULL,
    retry_count INT DEFAULT 0,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    status VARCHAR(50) DEFAULT 'NEW', -- NEW, REVIEWED, RETRIED, ARCHIVED
    reviewed_by VARCHAR(255),
    reviewed_at TIMESTAMP WITH TIME ZONE,
    resolution_notes TEXT
);

-- มุมมองเมตริก DLQ
CREATE VIEW dlq_metrics AS
SELECT 
    DATE(created_at) as date,
    status,
    COUNT(*) as count,
    error_info->>'code' as error_code
FROM dead_letter_queue
GROUP BY DATE(created_at), status, error_info->>'code';

-- Index สำหรับการค้นหาอย่างรวดเร็ว
CREATE INDEX idx_dlq_status ON dead_letter_queue(status);
CREATE INDEX idx_dlq_created ON dead_letter_queue(created_at);

┌─────────────────────────────────────────────────────────────┐
│              สามเสาหลักของ OBSERVABILITY                      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐        │
│   │   Metrics  │  │    Logs     │  │   Traces    │        │
│   │             │  │             │  │             │        │
│   │ Queue depth │  │ Event flow │  │ End-to-end │        │
│   │ Event rates │  │ Processing │  │ request    │        │
│   │ Latency P99 │  │   errors   │  │   timing   │        │
│   │ Agent perf  │  │ State      │  │ Cross-     │        │
│   │ Error rates │  │ changes    │  │ service    │        │
│   │             │  │            │  │            │        │
│   └─────────────┘  └─────────────┘  └─────────────┘        │
│                                                             │
│   Dashboards: Grafana │  Loki/Kibana │  Jaeger/Tempo       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

// OpenTelemetry tracing สำหรับ n8n workflows
const { trace, context, SpanStatusCode } = require('@opentelemetry/api');
const tracer = trace.getTracer('n8n-agent-workflows');

async function processWithTracing(event) {
  // แยกหรือสร้าง trace context
  const parentSpanContext = extractContext(event.metadata?.traceContext);
  
  const span = tracer.startSpan('process_agent_event', {
    attributes: {
      'event.type': event.eventType,
      'event.id': event.eventId,
      'session.id': event.payload.sessionId,
      'agent.id': event.payload.agentId
    }
  }, parentSpanContext);
  
  try {
    // เพิ่มเหตุการณ์ลงใน context ปัจจุบัน
    const ctx = trace.setSpan(context.active(), span);
    
    await context.with(ctx, async () => {
      // ประมวลผลเหตุการณ์
      await processEvent(event);
    });
    
    span.setStatus({ code: SpanStatusCode.OK });
    
  } catch (error) {
    span.recordException(error);
    span.setStatus({
      code: SpanStatusCode.ERROR,
      message: error.message
    });
    throw error;
    
  } finally {
    span.end();
  }
}

# Prometheus metrics สำหรับการตรวจสอบ AI Agent
# เหล่านี้จะถูกส่งออกโดย n8n workflows

# เมตริกการประมวลผลเหตุการณ์
agent_events_received_total:
  type: counter
  labels: [topic, event_type]
  
agent_events_processed_total:
  type: counter
  labels: [topic, event_type, status]
  
agent_event_processing_duration_seconds:
  type: histogram
  labels: [event_type]
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 2, 5]

# เมตริก Queue
agent_queue_depth:
  type: gauge
  labels: [queue_name]
  
agent_queue_consumer_lag:
  type: gauge
  labels: [topic, partition, consumer_group]

# เมตริก LLM
agent_llm_requests_total:
  type: counter
  labels: [model, status]
  
agent_llm_latency_seconds:
  type: histogram
  labels: [model]
  buckets: [0.1, 0.5, 1, 2, 5, 10, 30]
  
agent_llm_tokens_used_total:
  type: counter
  labels: [model, token_type]

# เมตริก Session
agent_sessions_active:
  type: gauge
  labels: [agent_type]
  
agent_session_duration_seconds:
  type: histogram
  labels: [agent_type, outcome]
  buckets: [30, 60, 300, 600, 1800, 3600]

# เมตริกข้อผิดพลาด
agent_errors_total:
  type: counter
  labels: [error_type, event_type, agent_id]
  
agent_dlq_items_total:
  type: counter
  labels: [reason]

┌─────────────────────────────────────────────────────────────┐
│                 การแบ่งพาร์ติชัน EVENT STREAM                 │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   แบ่งตาม Session ID (Session Affinity):                   │
│                                                             │
│   ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │
│   │ พาร์ต 0│ │ พาร์ต 1│ │ พาร์ต 2│ │ พาร์ต 3│          │
│   │ sessA* │ │ sessB* │ │ sessC* │ │ sessD* │          │
│   │ sessA1 │ │ sessB1 │ │ sessC1 │ │ sessD1 │          │
│   │ sessA2 │ │ sessB2 │ │ sessC2 │ │ sessD2 │          │
│   └─────────┘ └─────────┘ └─────────┘ └─────────┘          │
│                                                             │
│   ประโยชน์:                                                  │
│   ✓ Event ordering ต่อ Session                              │
│   ✓ Local state เป็นไปได้                                   │
│   ✓ Natural sharding                                        │
│                                                             │
│   แบ่งตามประเภทเหตุการณ์ (Load Balancing):                 │
│                                                             │
│   ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐          │
│   │ พาร์ต 0│ │ พาร์ต 1│ │ พาร์ต 2│ │ พาร์ต 3│          │
│   │ user.* │ │ agent.*│ │ tool.* │ │system.*│          │
│   └─────────┘ └─────────┘ └─────────┘ └─────────┘          │
│                                                             │
│   ประโยชน์:                                                  │
│   ✓ Consumer เฉพาะทาง                                       │
│   ✓ การปรับขนาดอิสระต่อประเภทเหตุการณ์                     │
│   ✓ นโยบายการเก็บรักษาแยกต่างหาก                           │
│                                                             │
└─────────────────────────────────────────────────────────────┘

# Kubernetes HPA สำหรับ n8n workers
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: n8n-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: n8n-worker
  minReplicas: 3
  maxReplicas: 20
  metrics:
    - type: External
      external:
        metric:
          name: kafka_consumer_lag
          selector:
            matchLabels:
              topic: ai-agent-events
        target:
          type: AverageValue
          averageValue: "100"
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
        - type: Percent
          value: 100
          periodSeconds: 15
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
        - type: Percent
          value: 10
          periodSeconds: 60

┌─────────────────────────────────────────────────────────────┐
│                  การใช้งานในการผลิต                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌────────────────────────────────────────────────────┐    │
│  │                  LOAD BALANCER                      │    │
│  │              (SSL termination, rate limiting)         │    │
│  └──────────────────────┬─────────────────────────────┘    │
│                         │                                   │
│        ┌────────────────┼────────────────┐               │
│        │                │                │                 │
│        ▼                ▼                ▼                 │
│  ┌────────────┐  ┌────────────┐  ┌────────────┐          │
│  │  n8n       │  │  n8n       │  │  n8n       │          │
│  │  Webhook   │  │  Webhook   │  │  Main      │          │
│  │  Instance  │  │  Instance  │  │  Instance  │          │
│  └──────┬─────┘  └──────┬─────┘  └──────┬─────┘          │
│         │               │               │                  │
│         └───────────────┼───────────────┘                  │
│                         │                                  │
│              ┌──────────┴──────────┐                       │
│              ▼                     ▼                        │
│       ┌──────────┐          ┌──────────┐                 │
│       │  Redis   │          │ PostgreSQL│                 │
│       │  Queue   │          │  (State)  │                 │
│       └────┬─────┘          └──────────┘                 │
│            │                                               │
│    ┌───────┴───────┐                                      │
│    │               │                                      │
│    ▼               ▼                                      │
│ ┌──────┐      ┌────────┐                                 │
│ │Worker│      │ Worker │                                 │
│ │ Pod 1│      │ Pod 2  │    ... (auto-scaled)           │
│ └──┬───┘      └───┬────┘                                 │
│    │              │                                       │
│    └──────────────┘                                       │
│           │                                               │
│           ▼                                               │
│    ┌──────────────┐                                       │
│    │ Apache Kafka │                                       │
│    │  Cluster     │                                       │
│    │  (3 brokers) │                                       │
│    └──────────────┘                                       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│            PIPELINE การประมวลผลคำสั่งซื้อ E-COMMERCE        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   เหตุการณ์คำสั่งซื้อ → Kafka → n8n Workflows → บริการ    │
│                                                             │
│   ┌──────────────────────────────────────────────────┐     │
│   │  Event Flow:                                      │     │
│   │                                                   │     │
│   │  order.created                                    │     │
│   │       │                                           │     │
│   │       ▼                                           │     │
│   │  ┌─────────────┐  inventory.check                 │     │
│   │  │ Validation  │───────────────┐                 │     │
│   │  │   Agent     │               │                 │     │
│   │  └─────────────┘               ▼                   │     │
│   │                         ┌─────────┐              │     │
│   │                    ┌────│Inventory│              │     │
│   │                    │     │ Service │              │     │
│   │                    │     └────┬────┘              │     │
│   │                    │          │                   │     │
│   │                    │     inventory.checked       │     │
│   │                    │          │                   │     │
│   │                    │          ▼                   │     │
│   │                    │     ┌─────────┐            │     │
│   │                    └────►│ Fraud   │            │     │
│   │                          │Detection│            │     │
│   │                          │  Agent  │            │     │
│   │                          └────┬────┘            │     │
│   │                               │                 │     │
│   │                               ▼                 │     │
│   │                    ┌──────────────────────┐       │     │
│   │                    │   Decision Point   │       │     │
│   │                    │  (risk_score > 0.7)│       │     │
│   │                    └─────────┬──────────┘       │     │
│   │                               │                 │     │
│   │            ┌─────────────────┴────────────────┐│     │
│   │            │                                  ││     │
│   │            ▼                                  ▼│     │
│   │     ┌──────────┐                    ┌──────────┐      │
│   │     │  Auto    │                    │  Manual  │      │
│   │     │ Approve  │                    │ Review   │      │
│   │     └────┬─────┘                    └────┬─────┘      │
│   │          │                               │             │
│   │          └─────────────────┬───────────────┘             │
│   │                            │                           │
│   │                            ▼                           │
│   │                     order.processed                     │
│   │                            │                           │
│   │                            ▼                           │
│   │                     ┌──────────────┐                   │
│   │                     │ Fulfillment  │                   │
│   │                     │    Agent     │                   │
│   │                     └──────────────┘                   │
│   │                                                        │
│   └──────────────────────────────────────────────────┘     │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│            ระบบตรวจจับการฉ้อโกงแบบเรียลไทม์                 │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌────────────────────────────────────────────────────┐    │
│   │  กระแสธุรกรรม (Kafka - 100 partitions)              │    │
│   └────────────────────────┬───────────────────────────┘    │
│                            │                                │
│              ┌───────────────┼───────────────┐               │
│              │               │               │                │
│              ▼               ▼               ▼               │
│   ┌────────────────┐ ┌──────────────┐ ┌──────────────┐     │
│   │ Feature        │ │ Rule Engine  │ │ ML Inference │     │
│   │ Engineering    │ │ (Fast Path)  │ │ (Deep Check) │     │
│   └───────┬────────┘ └──────┬───────┘ └──────┬───────┘     │
│           │                 │                │             │
│           └─────────────────┼────────────────┘             │
│                             │                                │
│                             ▼                                │
│                   ┌──────────────────┐                       │
│                   │  Risk Scoring    │                       │
│                   │    & ตัดสินใจ   │                       │
│                   └────────┬─────────┘                       │
│                            │                                 │
│              ┌─────────────┼─────────────┐                  │
│              │             │             │                   │
│              ▼             ▼             ▼                   │
│        ┌────────┐  ┌──────────┐  ┌──────────┐            │
│        │ ALLOW  │  │ CHALLENGE│  │  BLOCK   │            │
│        │        │  │ (2FA)    │  │          │            │
│        └────────┘  └──────────┘  └──────────┘            │
│                                                             │
│   ┌────────────────────────────────────────────────────┐    │
│   │  ลูปข้อเตือนกลับ: การฉ้อโกงที่ยืนยัน → Model     │    │
│   │  Retrain                                          │    │
│   └────────────────────────────────────────────────────┘    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

// โครงสร้างทดสอบประสิทธิภาพสำหรับ n8n workflows
class WorkflowBenchmark {
  constructor(config) {
    this.targetThroughput = config.targetThroughput; // เหตุการณ์/วินาที
    this.durationSeconds = config.durationSeconds;
    this.warmupSeconds = config.warmupSeconds || 30;
    this.metrics = [];
  }

  async run() {
    console.log(`เริ่ม Benchmark: ${this.targetThroughput} เหตุการณ์/วินาที`);
    
    // ช่วง Warmup
    await this.warmup();
    
    // ช่วงการวัด
    const startTime = Date.now();
    const promises = [];
    
    const intervalMs = 1000 / this.targetThroughput;
    
    while (Date.now() - startTime < this.durationSeconds * 1000) {
      const event = this.generateEvent();
      promises.push(this.measureEvent(event));
      await sleep(intervalMs);
    }
    
    await Promise.all(promises);
    
    return this.generateReport();
  }
  
  async measureEvent(event) {
    const start = process.hrtime.bigint();
    
    try {
      await this.sendEvent(event);
      const end = process.hrtime.bigint();
      
      this.metrics.push({
        success: true,
        latencyMs: Number(end - start) / 1000000,
        timestamp: Date.now()
      });
    } catch (error) {
      this.metrics.push({
        success: false,
        error: error.message,
        timestamp: Date.now()
      });
    }
  }
  
  generateReport() {
    const latencies = this.metrics.filter(m => m.success).map(m => m.latencyMs);
    const errors = this.metrics.filter(m => !m.success);
    
    return {
      throughput: this.metrics.length / this.durationSeconds,
      avgLatencyMs: latencies.reduce((a, b) => a + b, 0) / latencies.length,
      p50LatencyMs: percentile(latencies, 50),
      p95LatencyMs: percentile(latencies, 95),
      p99LatencyMs: percentile(latencies, 99),
      errorRate: errors.length / this.metrics.length,
      totalEvents: this.metrics.length
    };
  }
}

// ก่อน: เหตุการณ์ที่บวม
const largeEvent = {
  eventId: uuid(),
  timestamp: new Date().toISOString(),
  payload: {
    user: { /* 50KB ข้อมูลผู้ใช้ */ },
    session: { /* 30KB ข้อมูล Session */ },
    message: "สวัสดี"
  }
};

// หลัง: อ้างอิงข้อมูลหนัก
const compactEvent = {
  eventId: uuid(),
  timestamp: new Date().toISOString(),
  payload: {
    userId: "user_123",       // เฉพาะอ้างอิง
    sessionId: "sess_456",   // เฉพาะอ้างอิง
    message: "สวัสดี"
  },
  // รวมการอ้างอิงไปยังข้อมูลเต็มใน Event Store ถ้าจำเป็น
  references: {
    userSnapshot: "snapshot:user_123:1704067200"
  }
};

# การกำหนดค่า n8n สำหรับ Connection Pooling
DB_POSTGRESDB_POOL_SIZE: 20
DB_POSTGRESDB_POOL_MIN: 5

# Redis Connection Pool
QUEUE_BULL_REDIS_POOL_SIZE: 10

// ประมวลผลเหตุการณ์หลายรายการเป็น Batch
const BATCH_SIZE = 100;
const FLUSH_INTERVAL = 1000;

class BatchedProcessor {
  constructor() {
    this.buffer = [];
    this.timer = null;
  }

  async add(event) {
    this.buffer.push(event);
    
    if (this.buffer.length >= BATCH_SIZE) {
      await this.flush();
    } else if (!this.timer) {
      this.timer = setTimeout(() => this.flush(), FLUSH_INTERVAL);
    }
  }

  async flush() {
    if (this.buffer.length === 0) return;
    
    const batch = this.buffer.splice(0, BATCH_SIZE);
    clearTimeout(this.timer);
    this.timer = null;
    
    // ประมวลผล Batch
    await $getAll('postgres', {
      operation: 'insert',
      table: 'agent_events',
      data: batch
    });
  }
}

┌─────────────────────────────────────────────────────────────┐
│              สถาปัตยกรรม EVENT แบบ SERVERLESS              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   แหล่งที่มา Event:                                        │
│   ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐           │
│   │ API    │ │ Kafka  │ │ S3     │ │ DynamoDB│          │
│   │ Gateway│ │ Events │ │ Events │ │ Streams │          │
│   └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘          │
│       │          │          │          │                 │
│       └──────────┴──────────┴──────────┘                 │
│                   │                                        │
│                   ▼                                        │
│   ┌────────────────────────────────────────────┐          │
│   │     แพลตฟอร์ม Serverless Function        │          │
│   │  ┌─────────┐ ┌─────────┐ ┌─────────┐      │          │
│   │  │ Lambda  │ │ Cloud   │ │ Azure   │      │          │
│   │  │ Function│ │ Function│ │ Function│      │          │
│   │  │ (AWS)   │ │ (GCP)   │ │ (Azure) │      │          │
│   │  └─────────┘ └─────────┘ └─────────┘      │          │
│   └────────────────────────────────────────────┘          │
│                   │                                        │
│                   ▼                                        │
│   ┌────────────────────────────────────────────┐          │
│   │           Event Sinks                      │          │
│   │  ┌────────┐ ┌────────┐ ┌────────┐        │          │
│   │  │ Kafka  │ │ SQS    │ │Webhook │        │          │
│   │  │ Topic  │ │ Queue  │ │        │        │          │
│   │  └────────┘ └────────┘ └────────┘        │          │
│   └────────────────────────────────────────────┘          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

// Edge deployment pattern
// n8n เบา ๆ ที่ Edge, ประมวลผล Core ในคลาวด์

// การกำหนดค่า Edge node (Raspberry Pi / Edge server)
const edgeConfig = {
  mode: 'edge',
  upstream: 'https://core-n8n.example.com',
  localProcessing: [
    'intent-classification',
    'entity-extraction',
    'basic-response'
  ],
  forwardToCloud: [
    'complex-reasoning',
    'model-training',
    'analytics'
  ]
};

// Edge workflow
async function edgeProcess(event) {
  // พยายามประมวลผลท้องถิ่นก่อน
  if (canProcessLocally(event, edgeConfig)) {
    const result = await localModelInference(event);
    return { processed: 'local', result };
  }
  
  // ส่งต่อไปยังคลาวด์พร้อม Trace context
  return await forwardToCloud(event, edgeConfig.upstream);
}

# Kubernetes event-driven model serving
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: fraud-detection-model
spec:
  template:
    metadata:
      annotations:
        autoscaling.knative.dev/minScale: "1"
        autoscaling.knative.dev/maxScale: "100"
        autoscaling.knative.dev/targetConcurrency: "10"
    spec:
      containers:
        - image: gcr.io/project/fraud-model:v2.3
          ports:
            - containerPort: 8080
          resources:
            limits:
              nvidia.com/gpu: 1
              memory: "8Gi"
            requests:
              memory: "4Gi"
---
# Event source กระตุ้น model inference
apiVersion: sources.knative.dev/v1
kind: KafkaSource
metadata:
  name: fraud-detection-source
spec:
  consumerGroup: fraud-detection
  bootstrapServers:
    - kafka-cluster:9092
  topics:
    - transactions.features
  sink:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: fraud-detection-model

// Chain of thought เป็น Event stream
const chainEvents = [
  { type: 'chain.step', step: 1, thought: 'เข้าใจคำถาม...' },
  { type: 'chain.tool_call', step: 2, tool: 'search', input: '...' },
  { type: 'chain.observation', step: 3, result: '...' },
  { type: 'chain.step', step: 4, thought: 'วิเคราะห์ผลลัพธ์...' },
  { type: 'chain.complete', step: 5, response: 'คำตอบสุดท้าย' }
];

// แต่ละขั้นตอนเป็นเหตุการณ์ที่สามารถ:
// - ตรวจสอบแบบเรียลไทม์
// - ลองใหม่ได้อิสระ
// - กำหนดเส้นทางไปยังตัวประมวลผลเฉพาะทาง

{
  "eventType": "multimodal.message.received",
  "payload": {
    "sessionId": "sess_abc",
    "content": [
      { "type": "text", "content": "รูปภาพนี้มีอะไร?" },
      { "type": "image", "url": "s3://bucket/image.jpg", "format": "jpeg" }
    ],
    "processing": {
      "vision_model": "gpt-4o-vision",
      "text_model": "gpt-4o",
      "routing": "parallel"
    }
  }
}

// ฝึกโมเดลบน Event streams โดยไม่ต้องรวมศูนย์ข้อมูล
const federatedUpdate = {
  clientId: 'client_a',
  globalModelVersion: 'v2.3',
  localUpdate: {
    gradientUpdate: 'เวกเตอร์_gradient_เข้ารหัส',
    sampleCount: 1000,
    loss: 0.23
  },
  timestamp: '2026-05-25T10:00:00Z'
};

// รวมที่ตัวประสานงานกลาง
// อัปเดตโมเดล global
// แจกจ่ายเวอร์ชันใหม่