package server

import (
	"crypto/rand"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/x-grk/longpolling-remote-access/internal/proto"
)

const (
	// statTaskTimeout bounds how long a status request waits for the agent.
	statTaskTimeout = 8 * time.Second
	// statCacheFor is how long one sample answers every open status page, so
	// several tabs cost the agent a single reading.
	statCacheFor = 2 * time.Second
	// speedTaskTimeout covers the agent's ping + down + up windows with slack.
	speedTaskTimeout = 30 * time.Second
	// speedDownCap bounds one download stream; speedDownFor ends it if the
	// agent does not hang up first.
	speedDownCap = 256 << 20
	speedDownFor = 6 * time.Second
	speedUpCap   = 64 << 20
	// speedRelayUpFor is how long the agent's upload is passed on to the
	// browser: the agent's own window with slack.
	speedRelayUpFor = 4 * time.Second
)

// statCache holds the last sample per agent and serializes its refresh.
type statCache struct {
	mu sync.Mutex
	m  map[string]*statEntry
}

type statEntry struct {
	mu  sync.Mutex
	at  time.Time
	res proto.TaskResult
}

func (c *statCache) entry(agentID string) *statEntry {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.m == nil {
		c.m = map[string]*statEntry{}
	}
	e := c.m[agentID]
	if e == nil {
		e = &statEntry{}
		c.m[agentID] = e
	}
	return e
}

func (c *statCache) drop(agentID string) {
	c.mu.Lock()
	delete(c.m, agentID)
	c.mu.Unlock()
}

func (s *Server) handleStatusPage(w http.ResponseWriter, r *http.Request) {
	a, err := s.db.GetAgent(r.PathValue("id"))
	if err != nil || a == nil {
		http.NotFound(w, r)
		return
	}
	s.render(w, "status.html", map[string]any{"Agent": a})
}

// runTask runs a one-shot task on the agent and waits for its result.
func (s *Server) runTask(agentID, typ string, args []byte, timeout time.Duration) (proto.TaskResult, error) {
	taskID, ch := s.hub.CreateTask(agentID, typ, args)
	defer s.hub.DiscardTask(taskID)
	t := time.NewTimer(timeout)
	defer t.Stop()
	select {
	case res := <-ch:
		return res, nil
	case <-t.C:
		return proto.TaskResult{}, errors.New("таймаут ожидания клиента")
	}
}

// onlineAgent answers the request itself and returns false when the agent is
// unknown or offline.
func (s *Server) onlineAgent(w http.ResponseWriter, r *http.Request) (string, bool) {
	id := r.PathValue("id")
	a, err := s.db.GetAgent(id)
	if err != nil || a == nil {
		http.NotFound(w, r)
		return "", false
	}
	if !s.online(a) {
		writeJSON(w, proto.TaskResult{Error: "машина офлайн"})
		return "", false
	}
	return id, true
}

// oldAgentError replaces the "unknown task" answer of an agent that predates
// the task.
func oldAgentError(res proto.TaskResult) proto.TaskResult {
	if !res.OK && strings.Contains(res.Error, "неизвестная операция") {
		res.Error = "клиент не поддерживает эту функцию - обновите его (нужен 0.0.5+)"
	}
	return res
}

// GET /api/panel/agents/{id}/stat - CPU, RAM and disk of the machine right now.
func (s *Server) handleAgentStat(w http.ResponseWriter, r *http.Request) {
	id, ok := s.onlineAgent(w, r)
	if !ok {
		return
	}
	e := s.stats.entry(id)
	e.mu.Lock()
	defer e.mu.Unlock()
	if time.Since(e.at) > statCacheFor {
		res, err := s.runTask(id, proto.TaskSysStat, nil, statTaskTimeout)
		if err != nil {
			// Cached too, so pages queued behind this one don't each wait it out.
			res = proto.TaskResult{Error: err.Error()}
		}
		e.res, e.at = oldAgentError(res), time.Now()
	}
	writeJSON(w, e.res)
}

// POST /api/panel/agents/{id}/speed - have the agent measure its link to us.
// With relay=1 its streams run through the relay opened for it, to and from
// the browser.
func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) {
	id, ok := s.onlineAgent(w, r)
	if !ok {
		return
	}
	relay := r.URL.Query().Get("relay") != ""
	var args []byte
	if relay {
		rl := s.relays.get(id)
		if rl == nil {
			writeJSON(w, proto.TaskResult{Error: "замер через браузер не открыт"})
			return
		}
		defer s.relays.drop(id, rl)
		args, _ = json.Marshal(proto.SysSpeedArgs{Relay: true})
	}
	res, err := s.runTask(id, proto.TaskSysSpeed, args, speedTaskTimeout)
	if err != nil {
		writeJSON(w, proto.TaskResult{Error: err.Error()})
		return
	}
	res = oldAgentError(res)
	if relay && res.OK {
		// An agent that predates the relay ignores the argument and measures
		// its own link.
		var d proto.SysSpeedResult
		if json.Unmarshal(res.Data, &d) != nil || !d.Relay {
			res = proto.TaskResult{Error: "клиент не умеет замер через браузер - обновите его (нужен 0.0.6+)"}
		}
	}
	writeJSON(w, res)
}

// speedBlock is one chunk of incompressible bytes, made once and sent over and
// over, so a speed test costs the server no CPU beyond the copy.
var speedBlock = sync.OnceValue(func() []byte {
	b := make([]byte, 64<<10)
	rand.Read(b)
	return b
})

func speedStreamHeaders(w http.ResponseWriter) {
	w.Header().Set("Content-Type", "application/octet-stream")
	w.Header().Set("Cache-Control", "no-store")
	// Keep a proxy in front from buffering or compressing the stream.
	w.Header().Set("X-Accel-Buffering", "no")
}

// GET /api/agent/speed/down?bytes=N - stream N bytes (default: as much as fits
// in the time window) for the agent to time. bytes=0 is the ping probe.
func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) {
	if s.authenticateAgent(r) == nil {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	left := speedDownCap
	if v := r.URL.Query().Get("bytes"); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n >= 0 && n < left {
			left = n
		}
	}
	speedStreamHeaders(w)
	block := speedBlock()
	deadline := time.Now().Add(speedDownFor)
	for left > 0 && time.Now().Before(deadline) && r.Context().Err() == nil {
		chunk := block
		if len(chunk) > left {
			chunk = chunk[:left]
		}
		n, err := w.Write(chunk)
		if err != nil {
			return
		}
		left -= n
	}
}

// POST /api/agent/speed/up - swallow the agent's upload.
func (s *Server) handleSpeedUp(w http.ResponseWriter, r *http.Request) {
	if s.authenticateAgent(r) == nil {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	_, _ = io.Copy(io.Discard, http.MaxBytesReader(w, r.Body, speedUpCap))
	w.WriteHeader(http.StatusNoContent)
}

// speedRelay joins the agent's speed test streams to the browser's, so the
// bytes travel browser - panel - client. Both pipes are unbuffered: a slow
// reader holds its writer back, and the rate is that of the whole path.
type speedRelay struct {
	downR *io.PipeReader // browser -> agent
	downW *io.PipeWriter
	upR   *io.PipeReader // agent -> browser
	upW   *io.PipeWriter
}

func (rl *speedRelay) close() {
	rl.downR.CloseWithError(io.ErrClosedPipe)
	rl.downW.CloseWithError(io.ErrClosedPipe)
	rl.upR.CloseWithError(io.ErrClosedPipe)
	rl.upW.CloseWithError(io.ErrClosedPipe)
}

// speedRelays holds the relay of each agent under test, one per agent.
type speedRelays struct {
	mu sync.Mutex
	m  map[string]*speedRelay
}

// open starts a fresh relay for the agent, ending the one it had.
func (c *speedRelays) open(agentID string) {
	rl := &speedRelay{}
	rl.downR, rl.downW = io.Pipe()
	rl.upR, rl.upW = io.Pipe()
	c.mu.Lock()
	if c.m == nil {
		c.m = map[string]*speedRelay{}
	}
	old := c.m[agentID]
	c.m[agentID] = rl
	c.mu.Unlock()
	if old != nil {
		old.close()
	}
}

func (c *speedRelays) get(agentID string) *speedRelay {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.m[agentID]
}

// drop ends the agent's relay; with rl given, only if it is still that one.
func (c *speedRelays) drop(agentID string, rl *speedRelay) {
	c.mu.Lock()
	cur := c.m[agentID]
	if cur != nil && (rl == nil || rl == cur) {
		delete(c.m, agentID)
	} else {
		cur = nil
	}
	c.mu.Unlock()
	if cur != nil {
		cur.close()
	}
}

// GET /api/panel/ping - an empty answer for the browser to time.
func (s *Server) handlePanelPing(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(http.StatusNoContent)
}

// POST /api/panel/agents/{id}/speed/relay - open the relay of a speed test
// that runs through the browser. The test itself ends it.
func (s *Server) handleSpeedRelayOpen(w http.ResponseWriter, r *http.Request) {
	id, ok := s.onlineAgent(w, r)
	if !ok {
		return
	}
	s.relays.open(id)
	writeJSON(w, proto.TaskResult{OK: true})
}

// POST /api/panel/agents/{id}/speed/down - a piece of the browser's upload,
// passed on to the agent. 204 asks for the next piece, 200 says the agent has
// stopped reading.
func (s *Server) handleSpeedRelayFeed(w http.ResponseWriter, r *http.Request) {
	body := http.MaxBytesReader(w, r.Body, speedUpCap)
	rl := s.relays.get(r.PathValue("id"))
	if rl != nil {
		if _, err := io.Copy(rl.downW, body); err == nil {
			w.WriteHeader(http.StatusNoContent)
			return
		}
	}
	// Take the rest of the piece before answering: an answer sent over an
	// unread body makes Go drop the connection, which a proxy in front turns
	// into a 502.
	_, _ = io.Copy(io.Discard, body)
	w.Header().Set("Cache-Control", "no-store")
	_, _ = io.WriteString(w, "done")
}

// GET /api/panel/agents/{id}/speed/up - the agent's upload, for the browser to
// time. Ends when the agent is done sending.
func (s *Server) handleSpeedRelayDrain(w http.ResponseWriter, r *http.Request) {
	rl := s.relays.get(r.PathValue("id"))
	if rl == nil {
		http.Error(w, "gone", http.StatusGone)
		return
	}
	speedStreamHeaders(w)
	_, _ = io.Copy(w, rl.upR)
	// The browser may have left: don't let the agent keep sending to nobody.
	rl.upR.CloseWithError(io.ErrClosedPipe)
}

// GET /api/agent/speed/relay/down - what the browser uploads, for the agent to
// time.
func (s *Server) handleSpeedRelayDown(w http.ResponseWriter, r *http.Request) {
	a := s.authenticateAgent(r)
	if a == nil {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	rl := s.relays.get(a.ID)
	if rl == nil {
		http.NotFound(w, r)
		return
	}
	speedStreamHeaders(w)
	// The agent waits on the headers, the browser may not have sent a byte yet.
	_ = http.NewResponseController(w).Flush()
	_, _ = io.Copy(w, rl.downR)
	// The agent hung up at the end of its window: stop the browser's upload.
	rl.downR.CloseWithError(io.ErrClosedPipe)
}

// POST /api/agent/speed/relay/up - the agent's upload, passed on to the browser.
func (s *Server) handleSpeedRelayUp(w http.ResponseWriter, r *http.Request) {
	a := s.authenticateAgent(r)
	if a == nil {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	rl := s.relays.get(a.ID)
	if rl == nil {
		http.NotFound(w, r)
		return
	}
	body := http.MaxBytesReader(w, r.Body, speedUpCap)
	// A proxy in front may hand over the whole upload at once: pass on only
	// what the browser takes in the window, or the agent would wait for the
	// rest to trickle through a slower link.
	t := time.AfterFunc(speedRelayUpFor, func() { rl.upW.Close() })
	_, _ = io.Copy(rl.upW, body)
	t.Stop()
	rl.upW.Close()
	_, _ = io.Copy(io.Discard, body)
	w.WriteHeader(http.StatusNoContent)
}