package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"strconv"
	"time"

	"nhooyr.io/websocket"
)

// Ingestor connects to the Bybit V5 WebSocket, parses trade messages,
// sends ticks to the writer channel, and feeds ticks to the aggregator.
type Ingestor struct {
	cfg        Config
	tickCh     chan<- Tick
	aggregator *Aggregator
}

// NewIngestor creates an Ingestor wired to the tick channel and aggregator.
func NewIngestor(cfg Config, tickCh chan<- Tick, agg *Aggregator) *Ingestor {
	return &Ingestor{
		cfg:        cfg,
		tickCh:     tickCh,
		aggregator: agg,
	}
}

// Run connects to the WebSocket and processes messages until ctx is cancelled.
// It automatically reconnects on connection failures.
func (ing *Ingestor) Run(ctx context.Context) {
	log.Println("[ingestor] Starting WebSocket ingestor...")
	defer log.Println("[ingestor] Ingestor stopped.")

	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

		if err := ing.connectAndConsume(ctx); err != nil {
			if ctx.Err() != nil {
				return // Context cancelled, clean exit
			}
			log.Printf("[ingestor] Connection error: %v. Reconnecting in 5s...", err)
			select {
			case <-time.After(5 * time.Second):
			case <-ctx.Done():
				return
			}
		}
	}
}

func (ing *Ingestor) connectAndConsume(ctx context.Context) error {
	url := ing.cfg.WebSocketURL
	log.Printf("[ingestor] Connecting to %s ...", url)

	conn, _, err := websocket.Dial(ctx, url, nil)
	if err != nil {
		return fmt.Errorf("dial: %w", err)
	}
	defer conn.CloseNow()

	// Set a generous read limit for large trade batches (up to 1024 trades per message)
	conn.SetReadLimit(1 << 20) // 1 MB

	// Subscribe to public trades
	topic := fmt.Sprintf("publicTrade.%s", ing.cfg.Symbol)
	subMsg := map[string]interface{}{
		"op":   "subscribe",
		"args": []string{topic},
	}
	subJSON, _ := json.Marshal(subMsg)

	if err := conn.Write(ctx, websocket.MessageText, subJSON); err != nil {
		return fmt.Errorf("subscribe: %w", err)
	}
	log.Printf("[ingestor] Subscribed to %s", topic)

	// Read loop
	for {
		select {
		case <-ctx.Done():
			conn.Close(websocket.StatusNormalClosure, "shutting down")
			return nil
		default:
		}

		_, data, err := conn.Read(ctx)
		if err != nil {
			return fmt.Errorf("read: %w", err)
		}

		ing.handleMessage(data)
	}
}

func (ing *Ingestor) handleMessage(data []byte) {
	var msg BybitWSMessage

	if err := json.Unmarshal(data, &msg); err != nil {
		// subscription confirmations, pings, etc.
		return
	}

	if len(msg.Data) == 0 {
		return
	}

	recvTS := time.Now().UnixMilli()

	for _, raw := range msg.Data {

		price, err := strconv.ParseFloat(raw.P, 64)
		if err != nil {
			log.Printf("[ingestor] bad price %q: %v", raw.P, err)
			continue
		}

		volume, err := strconv.ParseFloat(raw.V, 64)
		if err != nil {
			log.Printf("[ingestor] bad volume %q: %v", raw.V, err)
			continue
		}

		tick := Tick{
			// IDs
			TradeID: raw.I,
			Seq:     raw.Seq,

			// Timing
			TradeTS:   raw.T,
			MessageTS: msg.TS,
			RecvTS:    recvTS,

			// Trade data
			Symbol: raw.S,
			Side:   raw.SD,
			Price:  price,
			Volume: volume,

			// Metadata
			TickDir:    raw.L,
			BlockTrade: raw.BT,
			RPI:        raw.RPI,
		}

		// Feed feature generator
		ing.aggregator.ProcessTick(tick)

		// Feed writer
		select {
		case ing.tickCh <- tick:
		default:
			log.Println("[ingestor] WARNING: tick channel full, dropping tick")
		}
	}
}