exchange-monitor-go/dashboard.go

package main

import (
	"encoding/json"
	"fmt"
	"io/fs"
	"log"
	"math"
	"net/http"
	"sync"
	"time"

	"exchange-monitor/db"
)

// ============================================================
// SSE Hub — manages connected browser clients
// ============================================================

type sseClient struct {
	ch     chan []byte
	done   chan struct{}
	filter string
}

type SSEHub struct {
	mu      sync.RWMutex
	clients map[*sseClient]bool
	seq     uint64
}

func NewSSEHub() *SSEHub {
	return &SSEHub{
		clients: make(map[*sseClient]bool),
	}
}

func (h *SSEHub) Subscribe(filter string) *sseClient {
	c := &sseClient{
		ch:     make(chan []byte, 64),
		done:   make(chan struct{}),
		filter: filter,
	}
	h.mu.Lock()
	h.clients[c] = true
	h.mu.Unlock()
	log.Printf("[Web] SSE client connected (clients=%d)", len(h.clients))
	return c
}

func (h *SSEHub) Unsubscribe(c *sseClient) {
	h.mu.Lock()
	delete(h.clients, c)
	count := len(h.clients)
	h.mu.Unlock()
	close(c.done)
	log.Printf("[Web] SSE client disconnected (clients=%d)", count)
}

func (h *SSEHub) Broadcast(event string, data interface{}) {
	raw, err := json.Marshal(map[string]interface{}{
		"event": event,
		"data":  data,
		"ts":    time.Now().UnixMilli(),
	})
	if err != nil {
		return
	}

	h.mu.RLock()
	defer h.mu.RUnlock()
	for c := range h.clients {
		select {
		case c.ch <- raw:
		default:
		}
	}
}

// ============================================================
// History Ring Buffers
// ============================================================

const maxHistoryPoints = 500

type pricePoint struct {
	T int64   `json:"t"`
	P float64 `json:"p"`
}

type priceHistory struct {
	mu      sync.RWMutex
	buffers map[string]map[string][]pricePoint
}

func newPriceHistory() *priceHistory {
	return &priceHistory{
		buffers: make(map[string]map[string][]pricePoint),
	}
}

func (ph *priceHistory) Record(coin, exchange string, price float64) {
	ph.mu.Lock()
	defer ph.mu.Unlock()
	if ph.buffers[coin] == nil {
		ph.buffers[coin] = make(map[string][]pricePoint)
	}
	buf := ph.buffers[coin][exchange]
	buf = append(buf, pricePoint{T: time.Now().UnixMilli(), P: price})
	if len(buf) > maxHistoryPoints {
		buf = buf[len(buf)-maxHistoryPoints:]
	}
	ph.buffers[coin][exchange] = buf
}

func (ph *priceHistory) GetHistory(coin, exchange string, limit int) []pricePoint {
	ph.mu.RLock()
	defer ph.mu.RUnlock()
	buf := ph.buffers[coin][exchange]
	if len(buf) == 0 {
		return nil
	}
	if limit <= 0 || limit >= len(buf) {
		r := make([]pricePoint, len(buf))
		copy(r, buf)
		return r
	}
	r := make([]pricePoint, limit)
	copy(r, buf[len(buf)-limit:])
	return r
}

// ============================================================
// Spread History — tracks BG↔HL spread % per coin (P3-2)
// ============================================================

type spreadPoint struct {
	T      int64   `json:"t"`
	Spread float64 `json:"s"` // spread % (positive = BG cheaper than HL for BG->HL direction)
}

type spreadHistory struct {
	mu      sync.RWMutex
	buffers map[string][]spreadPoint // coin -> spread points
}

func newSpreadHistory() *spreadHistory {
	return &spreadHistory{
		buffers: make(map[string][]spreadPoint),
	}
}

func (sh *spreadHistory) Record(coin string, spread float64) {
	sh.mu.Lock()
	defer sh.mu.Unlock()
	sh.buffers[coin] = append(sh.buffers[coin], spreadPoint{T: time.Now().UnixMilli(), Spread: spread})
	if len(sh.buffers[coin]) > maxHistoryPoints {
		sh.buffers[coin] = sh.buffers[coin][len(sh.buffers[coin])-maxHistoryPoints:]
	}
}

func (sh *spreadHistory) GetHistory(coin string, limit int) []spreadPoint {
	sh.mu.RLock()
	defer sh.mu.RUnlock()
	buf := sh.buffers[coin]
	if len(buf) == 0 {
		return nil
	}
	if limit <= 0 || limit >= len(buf) {
		r := make([]spreadPoint, len(buf))
		copy(r, buf)
		return r
	}
	r := make([]spreadPoint, limit)
	copy(r, buf[len(buf)-limit:])
	return r
}

// ============================================================
// Dashboard
// ============================================================

type Dashboard struct {
	hub     *SSEHub
	history *priceHistory
	spreads *spreadHistory
	store   *PriceStore
	trader  *Trader
	db      *db.DB
	addr    string

	// cached arb scan results
	mu       sync.RWMutex
	lastScan []*ArbOpportunity
	scanTime time.Time

	// P3-5: connection status — exchange -> last update time
	connMu  sync.RWMutex
	connMap map[string]time.Time // exchange name -> last price timestamp
}

func NewDashboard(store *PriceStore, trader *Trader, database *db.DB, addr string) *Dashboard {
	return &Dashboard{
		hub:     NewSSEHub(),
		history: newPriceHistory(),
		spreads: newSpreadHistory(),
		store:   store,
		trader:  trader,
		db:      database,
		addr:    addr,
		connMap: make(map[string]time.Time),
	}
}

func (d *Dashboard) Run() {
	go d.broadcastLoop()

	mux := http.NewServeMux()

	staticSub, err := fs.Sub(staticFS, "web/static")
	if err != nil {
		log.Printf("[Web] Failed to create static sub-fs: %v", err)
	} else {
		mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServer(http.FS(staticSub))))
	}

	mux.HandleFunc("GET /", d.handleIndex)
	mux.HandleFunc("GET /api/status", d.handleStatus)
	mux.HandleFunc("GET /api/history", d.handleHistory)
	mux.HandleFunc("GET /api/spread-history", d.handleSpreadHistory) // P3-2
	mux.HandleFunc("GET /api/trades", d.handleTrades)
	mux.HandleFunc("GET /api/trade/", d.handleTradeDetail)
	mux.HandleFunc("GET /api/connections", d.handleConnStatus) // P3-5
	mux.HandleFunc("GET /events", d.handleSSE)

	server := &http.Server{
		Addr:         d.addr,
		Handler:      mux,
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 0,
	}

	log.Printf("[Web] Dashboard listening on http://%s", d.addr)
	if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
		log.Printf("[Web] Server error: %v", err)
	}
}

// broadcastLoop pushes data to SSE clients every 1 second.
func (d *Dashboard) broadcastLoop() {
	tick := time.NewTicker(1 * time.Second)
	defer tick.Stop()

	for range tick.C {
		snap := d.store.GetAll()
		if len(snap) == 0 {
			continue
		}

		// 1. Prices + spreads + connection status
		var prices []map[string]interface{}
		for _, coin := range TrackedCoins {
			exMap := snap[coin.Name]
			if exMap == nil {
				continue
			}
			entry := map[string]interface{}{
				"coin": coin.Name,
			}
			for ex, p := range exMap {
				entry[ex] = p
			}
			for ex := range exMap {
				sp := d.store.GetSpread(coin.Name, ex)
				if sp > 0 {
					entry[ex+"_spread"] = sp
				}
			}

			// P3-2: Calculate BG↔HL spread and record
			bgP := exMap[ExBitget]
			hlP := exMap[ExHyperLiquid]
			if bgP > 0 && hlP > 0 {
				spreadPct := (hlP - bgP) / bgP * 100
				entry["bg_hl_spread"] = spreadPct
				d.spreads.Record(coin.Name, spreadPct)
			}

			prices = append(prices, entry)
		}
		d.hub.Broadcast("prices", prices)

		// 2. Open positions with live PnL (P3-3) — read from decoupled snapshot, never blocks trader
		positions := d.trader.ReadSnapshot()
		posList := make([]map[string]interface{}, 0, len(positions))
		for _, pos := range positions {
			posEntry := map[string]interface{}{
				"coin":         pos.Coin,
				"direction":    pos.Direction,
				"amount_usd":   pos.AmountUSD,
				"entry_spread": pos.EntrySpread,
				"scales":       pos.ScaleLevels,
				"duration":     time.Since(pos.StartedAt).Round(time.Second).String(),
				"started_at":   pos.StartedAt.Format("15:04:05"),
			}

			// Calculate live PnL from current prices — use weighted average for scale-ins
			if exMap := snap[pos.Coin]; exMap != nil {
				bgP := exMap[ExBitget]
				hlP := exMap[ExHyperLiquid]
				if bgP > 0 && hlP > 0 {
					var longCurrent, shortCurrent float64
					if pos.LongLeg.Exchange == ExBitget {
						longCurrent, shortCurrent = bgP, hlP
					} else {
						longCurrent, shortCurrent = hlP, bgP
					}
					longAvg := weightedAvgPrice(pos.LongEntryPrices, pos.AmountUSD/float64(max(1, len(pos.LongEntryPrices))))
					shortAvg := weightedAvgPrice(pos.ShortEntryPrices, pos.AmountUSD/float64(max(1, len(pos.ShortEntryPrices))))
					longPnl := (longCurrent - longAvg) / longAvg * 100
					shortPnl := (shortAvg - shortCurrent) / shortAvg * 100
					totalFees := 2 * (makerFees[ExBitget] + makerFees[ExHyperLiquid])
					netPnl := longPnl + shortPnl - totalFees

					currentSpread := (hlP - bgP) / bgP * 100
					posEntry["current_spread"] = math.Round(currentSpread*10000) / 10000
					posEntry["pnl_est"] = math.Round(netPnl*10000) / 10000
				}
			}

			posList = append(posList, posEntry)
		}
		d.hub.Broadcast("positions", posList)

		// 3. Arb scan results
		d.mu.RLock()
		scanCopy := d.lastScan
		d.mu.RUnlock()

		if len(scanCopy) > 0 {
			scanList := make([]map[string]interface{}, 0, len(scanCopy))
			for _, opp := range scanCopy {
				scanList = append(scanList, map[string]interface{}{
					"coin":       opp.Coin,
					"direction":  opp.Direction,
					"buy_ex":     opp.BuyEx,
					"sell_ex":    opp.SellEx,
					"buy_price":  opp.BuyPrice,
					"sell_price": opp.SellPrice,
					"net_profit": opp.NetProfit,
					"gross":      opp.GrossBasis,
				})
			}
			d.hub.Broadcast("arb", scanList)
		}

		// 4. Stats + connection status (P3-5)
		converged, diverged, flat, total := d.trader.GetClosedStats()
		stats := map[string]interface{}{
			"total_trades":   total,
			"converged":      converged,
			"diverged":       diverged,
			"flat":           flat,
			"open_positions": len(positions),
			"coins":          len(prices),
		}

		// Connection status
		d.connMu.RLock()
		connInfo := make(map[string]string)
		for ex, lastTime := range d.connMap {
			age := time.Since(lastTime)
			if age < 10*time.Second {
				connInfo[ex] = "online"
			} else if age < 30*time.Second {
				connInfo[ex] = "stale"
			} else {
				connInfo[ex] = "offline"
			}
		}
		d.connMu.RUnlock()
		stats["connections"] = connInfo

		d.hub.Broadcast("stats", stats)
	}
}

// ============================================================
// Public methods called from main.go / trader
// ============================================================

func (d *Dashboard) UpdateScan(opps []*ArbOpportunity) {
	d.mu.Lock()
	d.lastScan = opps
	d.scanTime = time.Now()
	d.mu.Unlock()
}

func (d *Dashboard) RecordPrice(coin, exchange string, price float64) {
	d.history.Record(coin, exchange, price)
}

// RecordConnStatus updates the last-seen time for an exchange (P3-5).
func (d *Dashboard) RecordConnStatus(exchange string) {
	d.connMu.Lock()
	d.connMap[exchange] = time.Now()
	d.connMu.Unlock()
}

// BroadcastEvent sends an immediate SSE event (P3-4).
func (d *Dashboard) BroadcastEvent(event string, data interface{}) {
	d.hub.Broadcast(event, data)
}

// ============================================================
// HTTP Handlers
// ============================================================

func (d *Dashboard) handleIndex(w http.ResponseWriter, r *http.Request) {
	data, err := staticFS.ReadFile("web/static/index.html")
	if err != nil {
		http.Error(w, "Not found", 404)
		return
	}
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	w.Write(data)
}

func (d *Dashboard) handleStatus(w http.ResponseWriter, r *http.Request) {
	snap := d.store.GetAll()
	positions := d.trader.ReadSnapshot()
	converged, diverged, flat, total := d.trader.GetClosedStats()

	resp := map[string]interface{}{
		"prices":    snap,
		"positions": len(positions),
		"stats":     map[string]int{"total": total, "converged": converged, "diverged": diverged, "flat": flat},
	}
	writeJSON(w, resp)
}

func (d *Dashboard) handleHistory(w http.ResponseWriter, r *http.Request) {
	coin := r.URL.Query().Get("coin")
	exchange := r.URL.Query().Get("exchange")
	if coin == "" || exchange == "" {
		snap := d.store.GetAll()
		coins := make([]string, 0, len(snap))
		for c := range snap {
			coins = append(coins, c)
		}
		writeJSON(w, map[string]interface{}{"coins": coins, "exchanges": []string{"Binance", "HyperLiquid", "Bitget", "dYdX"}})
		return
	}
	points := d.history.GetHistory(coin, exchange, 300)
	writeJSON(w, map[string]interface{}{
		"coin":     coin,
		"exchange": exchange,
		"points":   points,
	})
}

// handleSpreadHistory returns BG↔HL spread history for a coin (P3-2).
func (d *Dashboard) handleSpreadHistory(w http.ResponseWriter, r *http.Request) {
	coin := r.URL.Query().Get("coin")
	if coin == "" {
		writeJSON(w, map[string]interface{}{"coins": trackedCoinNames()})
		return
	}
	points := d.spreads.GetHistory(coin, 300)
	writeJSON(w, map[string]interface{}{
		"coin":   coin,
		"points": points,
	})
}

// handleConnStatus returns connection health for all exchanges (P3-5).
func (d *Dashboard) handleConnStatus(w http.ResponseWriter, r *http.Request) {
	d.connMu.RLock()
	conns := make(map[string]string)
	for ex, t := range d.connMap {
		age := time.Since(t)
		switch {
		case age < 10*time.Second:
			conns[ex] = "online"
		case age < 30*time.Second:
			conns[ex] = "stale"
		default:
			conns[ex] = "offline"
		}
	}
	d.connMu.RUnlock()
	writeJSON(w, conns)
}

func (d *Dashboard) handleTrades(w http.ResponseWriter, r *http.Request) {
	if d.db == nil {
		writeJSON(w, map[string]interface{}{"trades": []interface{}{}, "total": 0})
		return
	}
	page := 1
	limit := 20
	coin := r.URL.Query().Get("coin")
	trades, total, err := d.db.GetTrades(page, limit, coin)
	if err != nil {
		http.Error(w, err.Error(), 500)
		return
	}
	writeJSON(w, map[string]interface{}{
		"trades": trades,
		"total":  total,
		"page":   page,
		"limit":  limit,
	})
}

func (d *Dashboard) handleTradeDetail(w http.ResponseWriter, r *http.Request) {
	if d.db == nil {
		http.Error(w, "DB not available", 503)
		return
	}
	var id int64
	if _, err := fmt.Sscanf(r.URL.Path, "/api/trade/%d", &id); err != nil {
		http.Error(w, "Invalid trade ID", 400)
		return
	}
	trade, orders, err := d.db.GetTradeByID(id)
	if err != nil {
		http.Error(w, err.Error(), 404)
		return
	}
	writeJSON(w, map[string]interface{}{
		"trade":  trade,
		"orders": orders,
	})
}

func (d *Dashboard) handleSSE(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "Streaming not supported", 500)
		return
	}

	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("Access-Control-Allow-Origin", "*")

	client := d.hub.Subscribe("")
	defer d.hub.Unsubscribe(client)

	fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"ok\"}\n\n")
	flusher.Flush()

	for {
		select {
		case <-r.Context().Done():
			return
		case msg, ok := <-client.ch:
			if !ok {
				return
			}
			fmt.Fprintf(w, "data: %s\n\n", msg)
			flusher.Flush()
		}
	}
}

func writeJSON(w http.ResponseWriter, v interface{}) {
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(v)
}

func trackedCoinNames() []string {
	names := make([]string, len(TrackedCoins))
	for i, c := range TrackedCoins {
		names[i] = c.Name
	}
	return names
}