package httpserver

import (
	"context"
	"errors"
	"net/http"
	"strings"
	"sync"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/dispatch"
	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	"git.yuntuo.hk/k2safe/orbit/internal/jsonwire"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/platform/events"
	"github.com/coder/websocket"
)

// EventHub 仅负责已持久化结果的传输；断线不删除租约，消费者按事件ID持久化去重。
// 节点最多128条连接，领取池全局4个、每应用1个在途，队列加在途共128个。
type EventHub struct {
	service  *events.Service
	ctx      context.Context
	cancel   context.CancelFunc
	mu       sync.Mutex
	closing  bool
	slots    chan struct{}
	wg       sync.WaitGroup
	dispatch *dispatch.Pool
	poll     time.Duration
}

func NewEventHub(parent context.Context, service *events.Service, options ...HubOptions) *EventHub {
	ctx, cancel := context.WithCancel(parent)
	limits := resolveHubOptions(options)
	h := &EventHub{service: service, ctx: ctx, cancel: cancel, slots: make(chan struct{}, limits.ConnectionLimit), poll: time.Second}
	h.dispatch = dispatch.New(ctx, limits.DispatchWorkers, limits.DispatchCapacity)
	h.wg.Add(1)
	go func() { defer h.wg.Done(); <-h.dispatch.Done() }()
	return h
}
func RegisterEventConsumers(server *http.Server, hub *EventHub) {
	server.Handler.(*http.ServeMux).Handle("GET /v1/events/connect", hub)
}
func (h *EventHub) Shutdown(ctx context.Context) error {
	h.mu.Lock()
	h.closing = true
	h.cancel()
	h.mu.Unlock()
	done := make(chan struct{})
	go func() { h.wg.Wait(); close(done) }()
	select {
	case <-done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}
func (h *EventHub) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	requestID, err := core.NewID()
	if err != nil {
		writeError(w, "", fault.Unavailable)
		return
	}
	w.Header().Set("X-Request-ID", string(requestID))
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("X-Content-Type-Options", "nosniff")
	if err = noEventQuery(r); err != nil {
		writeError(w, string(requestID), err)
		return
	}
	values := r.Header.Values("Authorization")
	if len(values) != 1 {
		writeError(w, string(requestID), fault.Unauthorized)
		return
	}
	scheme, value, ok := strings.Cut(values[0], " ")
	if !ok || !strings.EqualFold(scheme, "Bearer") {
		writeError(w, string(requestID), fault.Unauthorized)
		return
	}
	token, err := auth.Parse(value)
	if err != nil {
		writeError(w, string(requestID), err)
		return
	}
	supported := false
	for _, header := range r.Header.Values("Sec-WebSocket-Protocol") {
		for _, p := range strings.Split(header, ",") {
			supported = supported || strings.TrimSpace(p) == EventSubprotocol
		}
	}
	if !supported {
		writeJSON(w, http.StatusUpgradeRequired, map[string]any{"request_id": requestID, "error": fault.New("PROTOCOL_REQUIRED", "orbit.events.v1 subprotocol is required")})
		return
	}
	h.mu.Lock()
	if h.closing || h.ctx.Err() != nil {
		h.mu.Unlock()
		writeError(w, string(requestID), fault.Unavailable)
		return
	}
	select {
	case h.slots <- struct{}{}:
		h.wg.Add(1)
	default:
		h.mu.Unlock()
		w.Header().Set("Retry-After", "1")
		writeJSON(w, 429, map[string]any{"request_id": requestID, "error": fault.New("CAPACITY_EXCEEDED", "node event connection capacity exceeded")})
		return
	}
	h.mu.Unlock()
	defer func() { <-h.slots; h.wg.Done() }()
	authorize, stop := context.WithTimeout(r.Context(), 5*time.Second)
	err = h.service.Authorize(authorize, token)
	stop()
	if err != nil {
		writeError(w, string(requestID), err)
		return
	}
	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{EventSubprotocol}, CompressionMode: websocket.CompressionDisabled})
	if err != nil {
		return
	}
	defer conn.CloseNow()
	conn.SetReadLimit(eventMaxFrame)
	ctx, cancel := context.WithCancel(h.ctx)
	defer cancel()
	helloCtx, helloStop := context.WithTimeout(ctx, 5*time.Second)
	kind, raw, err := conn.Read(helloCtx)
	helloStop()
	if err != nil {
		return
	}
	frame, decodeErr := decodeEventRequest(raw, true)
	var input eventHello
	if kind != websocket.MessageText || decodeErr != nil || decodeEventJSON(frame.Body, &input) != nil || input.SubscriptionID.Validate() != nil || input.Capacity < 1 || input.Capacity > 32 {
		_ = conn.Close(websocket.StatusPolicyViolation, "invalid event hello")
		return
	}
	call, callStop := context.WithTimeout(ctx, 8*time.Second)
	binding, err := h.service.Bind(call, token, input.SubscriptionID)
	callStop()
	if err != nil {
		_ = writeEventFrame(ctx, conn, eventFrame{Version: "1", Type: "error", MessageID: requestID, ReplyTo: frame.MessageID, Error: eventFault(err)})
		return
	}
	session, err := core.NewID()
	if err != nil {
		return
	}
	welcome := eventWelcome{SessionID: session, SubscriptionID: input.SubscriptionID, Capacity: input.Capacity, LeaseDurationMS: events.LeaseDuration.Milliseconds(), HeartbeatIntervalMS: 15000, ServerTime: binding.ServerTime, Enabled: binding.Enabled}
	body, err := jsonwire.Marshal(welcome)
	if err != nil {
		return
	}
	if err = writeEventFrame(ctx, conn, eventFrame{Version: "1", Type: "events.welcome", MessageID: requestID, ReplyTo: frame.MessageID, SessionID: session, Body: body}); err != nil {
		return
	}
	h.consume(ctx, cancel, conn, token, binding, session, input.Capacity)
}

type eventClaimResult struct {
	grant *events.Grant
	err   error
}

func (h *EventHub) claim(ctx context.Context, t auth.Token, sub, connection core.ID, capacity uint32) (out eventClaimResult) {
	defer func() {
		if recover() != nil {
			out = eventClaimResult{err: fault.Unavailable}
		}
	}()
	call, stop := context.WithTimeout(ctx, 8*time.Second)
	defer stop()
	out.grant, out.err = h.service.Claim(call, t, sub, events.ClaimInput{ConnectionID: connection, Capacity: capacity})
	return out
}
func (h *EventHub) consume(ctx context.Context, cancel context.CancelFunc, conn *websocket.Conn, t auth.Token, b events.ConsumerBinding, session core.ID, capacity uint32) {
	incoming := make(chan eventFrame, 4)
	outgoing := make(chan outboundFrame, 8)
	var loops sync.WaitGroup
	loops.Add(2)
	go func() {
		defer loops.Done()
		defer cancel()
		for {
			readCtx, stop := context.WithTimeout(ctx, 60*time.Second)
			kind, raw, err := conn.Read(readCtx)
			stop()
			if err != nil {
				return
			}
			frame, err := decodeEventRequest(raw, false)
			if kind != websocket.MessageText || err != nil || frame.SessionID != session {
				_ = conn.Close(websocket.StatusPolicyViolation, "invalid event frame")
				return
			}
			select {
			case incoming <- frame:
			case <-ctx.Done():
				return
			}
		}
	}()
	go func() {
		defer loops.Done()
		defer cancel()
		for {
			select {
			case <-ctx.Done():
				return
			case frame := <-outgoing:
				writeCtx, stop := context.WithTimeout(ctx, 5*time.Second)
				err := conn.Write(writeCtx, websocket.MessageText, frame.data)
				stop()
				if frame.done != nil {
					frame.done <- err
				}
				if err != nil {
					return
				}
			}
		}
	}()
	defer func() { cancel(); _ = conn.CloseNow(); loops.Wait() }()
	send := func(frame eventFrame, wait bool) bool {
		frame.Version = "1"
		frame.SessionID = session
		if frame.MessageID == "" {
			var err error
			frame.MessageID, err = core.NewID()
			if err != nil {
				return false
			}
		}
		raw, err := jsonwire.Marshal(frame)
		if err != nil || len(raw) > eventMaxFrame {
			return false
		}
		out := outboundFrame{data: raw}
		if wait {
			out.done = make(chan error, 1)
		}
		select {
		case outgoing <- out:
		default:
			return false
		}
		if !wait {
			return true
		}
		select {
		case err := <-out.done:
			return err == nil
		case <-ctx.Done():
			return false
		}
	}
	results := make(chan eventClaimResult, 1)
	pending, burst := false, 0
	request := func() {
		if pending || burst == 0 || len(outgoing) >= 4 {
			return
		}
		pending = h.dispatch.Submit(dispatch.Job{Context: ctx, Application: string(b.ApplicationID), Queue: string(t.NamespaceID) + "/" + string(b.SubscriptionID), Run: func(job context.Context) { results <- h.claim(job, t, b.SubscriptionID, session, capacity) }})
		if pending {
			burst--
		}
	}
	ticker := time.NewTicker(h.poll)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case result := <-results:
			pending = false
			if result.err != nil {
				send(eventFrame{Type: "error", Error: eventFault(result.err)}, true)
				return
			}
			if result.grant == nil {
				burst = 0
				continue
			}
			raw, err := jsonwire.Marshal(result.grant)
			if err != nil || !send(eventFrame{Type: "event.deliver", Body: raw}, false) {
				return
			}
			request()
		case frame := <-incoming:
			call, stop := context.WithTimeout(ctx, 8*time.Second)
			var body any
			var err error
			switch frame.Type {
			case "event.ack":
				var in events.AckInput
				if err = decodeEventJSON(frame.Body, &in); err == nil {
					body, err = h.service.Ack(call, t, b.SubscriptionID, in)
				}
			case "events.ping":
				var in struct{}
				if err = decodeEventJSON(frame.Body, &in); err == nil {
					var current events.ConsumerBinding
					current, err = h.service.Bind(call, t, b.SubscriptionID)
					if err == nil {
						body = eventPong{ServerTime: current.ServerTime, Enabled: current.Enabled}
					}
				}
			default:
				err = fault.Invalid("unsupported event message type")
			}
			stop()
			reply := eventFrame{Type: "ack", ReplyTo: frame.MessageID}
			if err != nil {
				reply.Type = "error"
				reply.Error = eventFault(err)
			} else {
				var marshalErr error
				reply.Body, marshalErr = jsonwire.Marshal(body)
				if marshalErr != nil {
					return
				}
			}
			fatal := err != nil && (fault.Code(err) == "UNAUTHENTICATED" || fault.Code(err) == "PERMISSION_DENIED" || fault.Code(err) == "NOT_FOUND" || fault.Code(err) == "UNAVAILABLE")
			if !send(reply, fatal) || fatal {
				return
			}
		case <-ticker.C:
			if !pending {
				burst = 4
				request()
			}
		}
	}
}
func eventFault(err error) *fault.Error {
	var known *fault.Error
	if errors.As(err, &known) {
		return known
	}
	return fault.Unavailable
}
func writeEventFrame(ctx context.Context, conn *websocket.Conn, frame eventFrame) error {
	raw, err := jsonwire.Marshal(frame)
	if err != nil {
		return err
	}
	if len(raw) > eventMaxFrame {
		return fault.Invalid("event frame too large")
	}
	bounded, stop := context.WithTimeout(ctx, 5*time.Second)
	defer stop()
	return conn.Write(bounded, websocket.MessageText, raw)
}