import (
    "fmt"
    "github.com/uniyakcom/beat"
)

func main() {
    beat.On("user.created", func(e *beat.Event) error {
        fmt.Printf("User: %s\n", string(e.Data))
        return nil
    })

    beat.Emit(&beat.Event{
        Type: "user.created",
        Data: []byte("alice"),
    })
}

bus, _ := beat.ForAsync()   // Per-P SPSC 高并发
defer bus.Close()

bus.On("order.**", func(e *beat.Event) error {
    fmt.Printf("Order event: %s\n", e.Type)
    return nil
})

bus.Emit(&beat.Event{Type: "order.created", Data: []byte(`{"id":123}`)})

import (
    "context"
    "github.com/uniyakcom/beat"
    "github.com/uniyakcom/beat/message"
    "github.com/uniyakcom/beat/pubsub/local"
    "github.com/uniyakcom/beat/router"
)

func main() {
    bus, _ := beat.ForSync()
    defer bus.Close()

    pub := local.NewPublisher(bus)
    sub := local.NewSubscriber(bus)

    r := router.NewRouter()

    // 纯消费
    r.On("notify", "order.created", sub, func(msg *message.Message) error {
        fmt.Printf("订单: %s\n", string(msg.Payload))
        return nil
    })

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

    go r.Run(ctx)
    <-r.Running()

    pub.Publish(context.Background(), "order.created",
        message.New("", []byte(`{"id":123}`)))
}

// 包级 API（Sync 语义）
beat.On("event", handler)
beat.Emit(event)

// 三核心
bus, _ := beat.ForSync()     // 同步直调
bus, _ := beat.ForAsync()    // Per-P SPSC
bus, _ := beat.ForFlow()     // Pipeline

// 自动检测
bus, _ := beat.New()         // ≥4 核 → Async，<4 核 → Sync

// 字符串配置
bus, _ := beat.Scenario("async")

// 完全控制
bus, _ := beat.Option(&beat.Profile{Name: "async", Conc: 10000, TPS: 50000})

type Message struct {
    UUID      string            // 消息唯一标识（自动生成 UUID v4）
    Key       string            // 分区/路由键（Kafka partition、Redis key）
    Metadata  Metadata          // 元数据 map[string]string
    Payload   []byte            // 消息体
    Timestamp time.Time         // 创建时间戳（自动设置）
}

// 构造
msg := message.New("", payload)           // 通用（含 Ack/Nack channel）
msg := message.NewPub("", payload)        // 发布专用（无 Ack，省分配）
msg.Key = "user-123"                              // 设置路由键

// Publisher 发布者接口
type Publisher interface {
    Publish(ctx context.Context, topic string, messages ...*Message) error
    Close() error
}

// Subscriber 订阅者接口
type Subscriber interface {
    Subscribe(ctx context.Context, topic string) (<-chan *Message, error)
    Close() error
}

r := router.NewRouter()

// 纯消费（不产出）
r.On("notify", "order.created", sub, func(msg *message.Message) error {
    sendEmail(msg.Payload)
    return nil
})

// 完整管道：subscribe → process → publish
r.Handle(
    "transform",
    "raw.data", inSub,
    "processed.data", outPub,
    func(msg *message.Message) ([]*message.Message, error) {
        result := transform(msg.Payload)
        return []*message.Message{message.New("", result)}, nil
    },
)

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
r.Run(ctx)

// 8 个 worker 并行消费，信号量自动背压
r.Handle("fast", "events", sub, "out", pub, handler).
    Workers(8)

// 累积 100 条或 1s 超时后触发批量处理
r.HandleBatch(
    "bulk-insert",
    "log.entries", sub,
    "", nil, // 无产出
    func(msgs []*message.Message) ([]*message.Message, error) {
        rows := make([]Row, len(msgs))
        for i, m := range msgs {
            rows[i] = parseRow(m.Payload)
        }
        return nil, db.BulkInsert(rows)
    },
    100,           // batchSize
    time.Second,   // batchTimeout
)

// 便捷写法
r.OnBatch("bulk", "log.entries", sub, func(msgs []*message.Message) error {
    return db.BulkInsert(toRows(msgs))
}, 100, time.Second)

// 根据消息内容路由到不同 topic
r.Handle("route", "events", sub, "default.out", pub, handler).
    Route(func(msg *message.Message) string {
        switch msg.Metadata.Get("type") {
        case "urgent":
            return "priority.queue"
        case "bulk":
            return "batch.queue"
        default:
            return "" // 使用默认 publishTopic
        }
    })

dlqPub, _ := beatredis.NewPublisher(redisCfg)

r.Handle("process", "orders", sub, "done", pub, handler).
    Retry(3).
    DLQ(router.DLQConfig{
        Topic:     "orders.dlq",
        Publisher: dlqPub,
    })
// 处理失败 3 次后消息自动发送到 orders.dlq，附带 dlq_reason / dlq_handler / dlq_topic 元数据

// 限制最大在途消息数（与并发配合使用）
r.Handle("controlled", "events", sub, "out", pub, handler).
    Workers(16).
    InFlight(8) // 最多 8 条未确认消息

import (
    "github.com/uniyakcom/beat/middleware/retry"
    "github.com/uniyakcom/beat/middleware/timeout"
    "github.com/uniyakcom/beat/middleware/recoverer"
    "github.com/uniyakcom/beat/middleware/logging"
    "github.com/uniyakcom/beat/middleware/correlation"
)

r := router.NewRouter()

// 全局中间件（洋葱模型：外层先执行）
r.Use(
    recoverer.New(),                          // panic → error 恢复
    correlation.New(),                        // correlation_id 传播
    logging.New(slog.Default()),              // slog 处理日志
    timeout.New(5 * time.Second),             // 消息处理超时
    retry.New(retry.Config{MaxRetries: 3}),   // 指数退避重试
)

// Handler 专属中间件（在全局之后执行）
handler.AddMiddleware(customMiddleware)

import "github.com/uniyakcom/beat/marshal"

m := marshal.JSON{}
data, _ := m.Marshal("topic", msg)          // Message → []byte
restored, _ := m.Unmarshal("topic", data)   // []byte → Message

import beatredis "github.com/uniyakcom/beat-redis"

pub, _ := beatredis.NewPublisher(beatredis.PublisherConfig{
    Client:         rdb,
    UseStreams:      true,
    DirectFields:   true,       // 零封装直存字段
    EnablePipeline: true,       // 批量 Pipeline
})

sub, _ := beatredis.NewSubscriber(beatredis.SubscriberConfig{
    Client:       rdb,
    ConsumerGroup: "my-service",
})

import beatkafka "github.com/uniyakcom/beat-kafka"

pub, _ := beatkafka.NewPublisher(beatkafka.PublisherConfig{
    Brokers: []string{"localhost:9092"},
})
// msg.Key 自动映射为 Kafka partition key
// msg.Metadata 传播为 Kafka headers

import beatnats "github.com/uniyakcom/beat-nats"

pub, _ := beatnats.NewPublisher(beatnats.PublisherConfig{
    Conn:      nc,
    JetStream: js, // 可选：启用持久化
})

import beathttp "github.com/uniyakcom/beat-http"

pub, _ := beathttp.NewPublisher(beathttp.PublisherConfig{
    EndpointURL: "https://api.example.com/webhooks",
})
// POST {url}/{topic}，Header 传播 metadata

import beatsql "github.com/uniyakcom/beat-sql"

pub, _ := beatsql.NewPublisher(beatsql.PublisherConfig{DB: db})

// 事务原子写入
tx, _ := db.Begin()
tx.Exec("INSERT INTO orders ...")
pub.PublishInTx(tx, "order.created", msg)
tx.Commit()

handler := r.Handle(name, subTopic, sub, pubTopic, pub, fn)

handler.
    Workers(8).              // 8 worker 并行
    InFlight(4).             // 最多 4 条在途
    Route(routeFn).          // 动态路由
    Retry(3).                // 路由级重试
    DLQ(dlqCfg).             // 死信队列
    AddMiddleware(myMiddleware) // Handler 专属中间件