exchange-monitor-go/dashboard.go

package main

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

	"exchange-monitor/db"
)

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

type sseClient struct {
	ch     chan []byte
	done   chan struct{}
	filter string // optional coin filter (empty = all)
}

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

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:
			// Client too slow, skip
		}
	}
}

// ============================================================
// Price History — ring buffer for charting
// ============================================================

const maxHistoryPoints = 500

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

type priceHistory struct {
	mu      sync.RWMutex
	buffers map[string]map[string][]pricePoint // coin -> exchange -> points
}

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) {
		result := make([]pricePoint, len(buf))
		copy(result, buf)
		return result
	}
	result := make([]pricePoint, limit)
	copy(result, buf[len(buf)-limit:])
	return result
}

// ============================================================
// Dashboard — main orchestrator
// ============================================================

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

	// cached arb scan results — updated every tick
	mu         sync.RWMutex
	lastScan   []*ArbOpportunity
	scanTime   time.Time
}

func NewDashboard(store *PriceStore, trader *Trader, database *db.DB, addr string) *Dashboard {
	return &Dashboard{
		hub:     NewSSEHub(),
		history: newPriceHistory(),
		store:   store,
		trader:  trader,
		db:      database,
		addr:    addr,
	}
}

// Run starts the HTTP server + SSE broadcaster goroutine.
func (d *Dashboard) Run() {
	// SSE broadcaster — pushes data every ~1s
	go d.broadcastLoop()

	mux := http.NewServeMux()

	// Static files — embed subdirectory
	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))))
	}

	// Main page
	mux.HandleFunc("GET /", d.handleIndex)

	// API endpoints
	mux.HandleFunc("GET /api/status", d.handleStatus)
	mux.HandleFunc("GET /api/history", d.handleHistory)
	mux.HandleFunc("GET /api/trades", d.handleTrades)
	mux.HandleFunc("GET /api/trade/", d.handleTradeDetail)

	// SSE
	mux.HandleFunc("GET /events", d.handleSSE)

	server := &http.Server{
		Addr:         d.addr,
		Handler:      mux,
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 0, // SSE needs no write timeout
	}

	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
		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
			}
			// Add bid-ask spreads
			for ex := range exMap {
				sp := d.store.GetSpread(coin.Name, ex)
				if sp > 0 {
					entry[ex+"_spread"] = sp
				}
			}
			prices = append(prices, entry)
		}
		d.hub.Broadcast("prices", prices)

		// 2. Open positions
		positions := d.trader.GetOpenPositions()
		if len(positions) > 0 {
			posList := make([]map[string]interface{}, 0, len(positions))
			for _, pos := range positions {
				posList = append(posList, 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"),
				})
			}
			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
		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),
		}
		d.hub.Broadcast("stats", stats)
	}
}

// UpdateScan caches the latest arb scan results.
func (d *Dashboard) UpdateScan(opps []*ArbOpportunity) {
	d.mu.Lock()
	d.lastScan = opps
	d.scanTime = time.Now()
	d.mu.Unlock()
}

// RecordPrice adds a price to the history buffer and optionally broadcasts.
func (d *Dashboard) RecordPrice(coin, exchange string, price float64) {
	d.history.Record(coin, exchange, price)
}

// ============================================================
// 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.GetOpenPositions()
	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 == "" {
		// Return available coins/exchanges
		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,
	})
}

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)

	// Send initial heartbeat
	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)
}