package orbit

import (
	"context"
	"errors"
	"fmt"
	"math/rand/v2"
	"sync/atomic"
	"time"
)

var ErrEventConsumerAlreadyRun = errors.New("event consumer may only Run once")
var ErrEventShutdownTimeout = errors.New("event consumer shutdown grace period expired")

// EventConsumerOptions 定义自动消费器。Handle必须以事件ID做持久化去重，
// 在去重记录和业务效果同一事务提交后才返回nil；提交结果未知时返回error。
// error或panic不ACK，平台在租约到期后按原预算补发；SDK不在内存中立即重跑。
type EventConsumerOptions struct {
	SubscriptionID    string
	Capacity          uint32                                   // Capacity 最大本地在途事件/处理器，默认4，范围1至32。
	Handle            func(context.Context, ResultEvent) error // Handle不接收租约秘密；必须响应context取消。
	ShutdownTimeout   time.Duration                            // ShutdownTimeout停止时取消处理器，等待返回及ACK，默认30秒，最多5分钟。
	ReconnectMinDelay time.Duration                            // ReconnectMinDelay默认200毫秒，至少10毫秒。
	ReconnectMaxDelay time.Duration                            // ReconnectMaxDelay默认5秒，最多30秒；指数退避并加入抖动。
}

// EventConsumerEvent 只用于本地观察，缓冲满可丢弃；不是可靠结果或审计。
type EventConsumerEvent struct {
	Type    string // Type为connected、reconnecting、handler_started、handler_failed、acknowledged、proof_lost、draining。
	EventID string
	Code    string // Code不含业务异常、数据或秘密。
}
type EventConsumerStats struct {
	Running         int64
	Started         uint64
	HandlerFailures uint64
	Acknowledged    uint64
	Reconnects      uint64
	DroppedEvents   uint64
}
type EventConsumerRunError struct {
	Cause       error
	Unconfirmed int   // Unconfirmed本进程已知且未确认的事件数，不是订阅全局积压。
	Running     int64 // Running仍未退出的处理器；Go不能强杀忽略context的函数。
}

func (e *EventConsumerRunError) Error() string {
	return fmt.Sprintf("event consumer stopped with %d unconfirmed events and %d running handlers: %v", e.Unconfirmed, e.Running, e.Cause)
}
func (e *EventConsumerRunError) Unwrap() error { return e.Cause }

// EventConsumer 自动维护心跳、重连、处理器并发和ACK核对。只能Run一次。
// 服务端交付租约不是业务锁；跨进程重复及并行消费由业务Inbox唯一键和事务处理。
type EventConsumer struct {
	client                                        *Client
	options                                       EventConsumerOptions
	used                                          atomic.Bool
	events                                        chan EventConsumerEvent
	running                                       atomic.Int64
	started, failures, acked, reconnects, dropped atomic.Uint64
}

func NewEventConsumer(client *Client, in EventConsumerOptions) (*EventConsumer, error) {
	if client == nil || client.token == "" || !validID(in.SubscriptionID) || in.Handle == nil {
		return nil, errors.New("authenticated client, subscription and durable event handler are required")
	}
	if in.Capacity == 0 {
		in.Capacity = 4
	}
	if in.ShutdownTimeout == 0 {
		in.ShutdownTimeout = 30 * time.Second
	}
	if in.ReconnectMinDelay == 0 {
		in.ReconnectMinDelay = 200 * time.Millisecond
	}
	if in.ReconnectMaxDelay == 0 {
		in.ReconnectMaxDelay = 5 * time.Second
	}
	if in.Capacity > 32 || in.ShutdownTimeout < time.Millisecond || in.ShutdownTimeout > 5*time.Minute || in.ReconnectMinDelay < 10*time.Millisecond || in.ReconnectMaxDelay < in.ReconnectMinDelay || in.ReconnectMaxDelay > 30*time.Second {
		return nil, errors.New("invalid event consumer capacity or durations")
	}
	return &EventConsumer{client: client, options: in, events: make(chan EventConsumerEvent, 128)}, nil
}
func (c *EventConsumer) Events() <-chan EventConsumerEvent { return c.events }
func (c *EventConsumer) Stats() EventConsumerStats {
	return EventConsumerStats{Running: c.running.Load(), Started: c.started.Load(), HandlerFailures: c.failures.Load(), Acknowledged: c.acked.Load(), Reconnects: c.reconnects.Load(), DroppedEvents: c.dropped.Load()}
}
func (c *EventConsumer) emit(typ, id, code string) {
	select {
	case c.events <- EventConsumerEvent{Type: typ, EventID: id, Code: code}:
	default:
		c.dropped.Add(1)
	}
}
func (c *EventConsumer) String() string {
	return fmt.Sprintf("EventConsumer{SubscriptionID:%s Capacity:%d secrets:redacted}", c.options.SubscriptionID, c.options.Capacity)
}
func (c *EventConsumer) GoString() string { return c.String() }

type consumerJob struct {
	proof    EventProof
	complete bool
}
type consumerResult struct {
	id     string
	failed bool
}

// Run在ctx取消后停止启动新处理器，取消在途处理器，在收尾时限内确认已提交结果。
// 收到认证/协议等永久错误时也取消并等待在途处理器；未确认结果仍保留在服务端。
// 重连先核对已完成处理器的原证明；证明被替代时等待新投递，由业务去重命中后确认。
func (c *EventConsumer) Run(parent context.Context) error {
	if !c.used.CompareAndSwap(false, true) {
		return ErrEventConsumerAlreadyRun
	}
	defer close(c.events)
	if parent.Err() != nil {
		return nil
	}
	ctx, cancel := context.WithCancel(context.WithoutCancel(parent))
	defer cancel()
	handlerCtx, stopHandlers := context.WithCancel(ctx)
	defer stopHandlers()
	abort := make(chan struct{})
	watchDone := make(chan struct{})
	go func() {
		defer close(watchDone)
		select {
		case <-ctx.Done():
			return
		case <-parent.Done():
		case <-abort:
		}
		stopHandlers()
		timer := time.NewTimer(c.options.ShutdownTimeout)
		defer timer.Stop()
		select {
		case <-ctx.Done():
		case <-timer.C:
			cancel()
		}
	}()
	defer func() { cancel(); <-watchDone }()
	var conn *EventConnection
	var heartbeatDone <-chan struct{}
	closeConnection := func() {
		if conn != nil {
			_ = conn.Close()
			<-heartbeatDone
			conn = nil
		}
	}
	defer closeConnection()
	jobs := make(map[string]*consumerJob)
	results := make(chan consumerResult, c.options.Capacity)
	draining := false
	var cause error
	beginDrain := func(err error) {
		if !draining {
			draining = true
			stopHandlers()
			close(abort)
			c.emit("draining", "", "")
		}
		if cause == nil {
			cause = err
		}
	}
	finish := func(err error) error {
		if err == nil {
			return nil
		}
		return &EventConsumerRunError{Cause: err, Unconfirmed: len(jobs), Running: c.running.Load()}
	}
	backoff := c.options.ReconnectMinDelay
	var dialAt time.Time
	everConnected := false
	disconnect := func(err error) {
		closeConnection()
		if !eventRetryable(err) {
			beginDrain(err)
			return
		}
		floor := backoff / 2
		dialAt = time.Now().Add(floor + time.Duration(rand.Int64N(int64(backoff-floor)+1)))
		backoff = min(backoff*2, c.options.ReconnectMaxDelay)
		c.emit("reconnecting", "", eventErrorCode(err))
	}
	for {
		if parent.Err() != nil {
			beginDrain(nil)
		}
		if ctx.Err() != nil {
			return finish(ErrEventShutdownTimeout)
		}
		if draining && (len(jobs) == 0 || cause != nil && c.running.Load() == 0) {
			return finish(cause)
		}
		if conn == nil && cause == nil && !time.Now().Before(dialAt) {
			if everConnected {
				c.reconnects.Add(1)
			}
			var err error
			conn, err = c.client.DialEvents(ctx, EventRegistration{SubscriptionID: c.options.SubscriptionID, Capacity: c.options.Capacity})
			if err != nil {
				disconnect(err)
				continue
			}
			everConnected = true
			heartbeatDone = startEventHeartbeat(ctx, conn)
			c.emit("connected", "", "")
			continue
		}
		if conn != nil {
			if err := conn.Err(); err != nil {
				disconnect(err)
				continue
			}
			// 一个ACK在途；读取与自动心跳独立，不需要接收方不断调用Next维持连接。
			acknowledged := false
			for id, j := range jobs {
				if !j.complete {
					continue
				}
				messageID, err := NewEventMessageID()
				if err == nil {
					_, err = conn.Ack(ctx, messageID, j.proof)
				}
				if err == nil {
					delete(jobs, id)
					c.acked.Add(1)
					c.emit("acknowledged", id, "")
					backoff = c.options.ReconnectMinDelay
				} else {
					var remote *EventError
					if errors.As(err, &remote) && remote.Code == "DELIVERY_LEASE_LOST" {
						delete(jobs, id)
						c.emit("proof_lost", id, remote.Code)
					} else {
						disconnect(err)
					}
				}
				acknowledged = true
				break
			}
			if acknowledged {
				continue
			}
		}
		var deliveries <-chan EventGrant
		var disconnected <-chan struct{}
		if conn != nil {
			disconnected = conn.Done()
			if !draining && len(jobs) < int(c.options.Capacity) {
				deliveries = conn.deliveries
			}
		}
		var parentDone <-chan struct{}
		if !draining {
			parentDone = parent.Done()
		}
		// 定时器只负责重连；活跃连接没有每轮轮询。
		var wake <-chan time.Time
		var timer *time.Timer
		if conn == nil && cause == nil {
			timer = time.NewTimer(max(time.Millisecond, time.Until(dialAt)))
			wake = timer.C
		}
		select {
		case <-ctx.Done():
		case <-parentDone:
			beginDrain(nil)
		case <-wake:
		case <-disconnected:
			disconnect(conn.Err())
		case result := <-results:
			j := jobs[result.id]
			if result.failed {
				delete(jobs, result.id)
				c.failures.Add(1)
				c.emit("handler_failed", result.id, "HANDLER_FAILED")
			} else {
				j.complete = true
			}
		case grant := <-deliveries:
			if parent.Err() != nil || conn.Err() != nil {
				break
			}
			id := grant.Event.ID
			if j := jobs[id]; j != nil {
				if j.proof.DeliveryID != grant.Delivery.ID {
					disconnect(ErrEventProtocol)
					break
				}
				if grant.Delivery.LeaseEpoch > j.proof.Epoch {
					j.proof = grant.Proof()
				}
				break
			}
			jobs[id] = &consumerJob{proof: grant.Proof()}
			c.running.Add(1)
			c.started.Add(1)
			c.emit("handler_started", id, "")
			go func(event ResultEvent) {
				failed := true
				defer func() {
					_ = recover()
					c.running.Add(-1)
					select {
					case results <- consumerResult{id: event.ID, failed: failed}:
					case <-ctx.Done():
					}
				}()
				failed = c.options.Handle(handlerCtx, event) != nil
			}(grant.Event)
		}
		if timer != nil {
			timer.Stop()
		}
	}
}
func startEventHeartbeat(ctx context.Context, conn *EventConnection) <-chan struct{} {
	done := make(chan struct{})
	go func() {
		defer close(done)
		ticker := time.NewTicker(time.Duration(conn.Welcome().HeartbeatIntervalMS) * time.Millisecond)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-conn.Done():
				return
			case <-ticker.C:
				id, err := NewEventMessageID()
				if err == nil {
					_, err = conn.Ping(ctx, id)
				}
				if err != nil {
					conn.fail(err)
					return
				}
			}
		}
	}()
	return done
}
func eventRetryable(err error) bool {
	if errors.Is(err, ErrEventDisconnected) || errors.Is(err, ErrEventBackpressure) || errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	var remote *EventError
	if errors.As(err, &remote) {
		return remote.Code == "UNAVAILABLE"
	}
	var httpErr *HTTPError
	if errors.As(err, &httpErr) {
		return httpErr.StatusCode == 408 || httpErr.StatusCode == 429 || httpErr.StatusCode >= 500 && httpErr.StatusCode <= 599
	}
	return false
}
func eventErrorCode(err error) string {
	var remote *EventError
	if errors.As(err, &remote) {
		return remote.Code
	}
	return "TRANSPORT_INTERRUPTED"
}