package server

import (
	"context"
	"encoding/json"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

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

// Sending a file or folder to another machine. The sending agent packs it as
// a tar stream; to another agent the stream goes in pieces through a relay
// here, a few pieces held at most, and the receiving agent unpacks it as it
// comes. To an SSH host the sending agent writes it there itself. The panel
// follows a job: how far the sender is, and the result.

// relay carries one stream between two agents, in numbered pieces.
type relay struct {
	id, from, to string

	mu      sync.Mutex
	changed chan struct{}
	chunks  map[int][]byte
	low     int // pieces below it were taken
	end     int // the last piece's number, -1 until it came
	err     string
}

func (x *relay) notify() {
	close(x.changed)
	x.changed = make(chan struct{})
}

// await waits until ready() holds and returns true with x.mu held.
func (x *relay) await(ctx context.Context, d time.Duration, ready func() bool) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	for {
		x.mu.Lock()
		if ready() {
			return true
		}
		ch := x.changed
		x.mu.Unlock()
		select {
		case <-ch:
		case <-t.C:
			return false
		case <-ctx.Done():
			return false
		}
	}
}

func (x *relay) fail(msg string) {
	x.mu.Lock()
	if x.err == "" {
		x.err = msg
		x.chunks = nil
		x.notify()
	}
	x.mu.Unlock()
}

type shipJob struct {
	id      string
	mu      sync.Mutex
	changed chan struct{}
	ver     int
	prog    proto.ShipProgress
	done    bool
	res     proto.TaskResult
	cancel  context.CancelFunc
	stopped bool
}

func (j *shipJob) update(fn func()) {
	j.mu.Lock()
	fn()
	j.ver++
	close(j.changed)
	j.changed = make(chan struct{})
	j.mu.Unlock()
}

type shipping struct {
	mu     sync.Mutex
	relays map[string]*relay
	jobs   map[string]*shipJob
}

func (c *shipping) relay(id string) *relay {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.relays[id]
}

func (c *shipping) job(id string) *shipJob {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.jobs[id]
}

// shipReq is what the panel asks for: Src on this agent to Dir/Name on agent
// To, or on the SSH host.
type shipReq struct {
	Src      string           `json:"src"`
	Name     string           `json:"name"`
	Dir      string           `json:"dir"`
	Conflict string           `json:"conflict"`
	Move     bool             `json:"move"`
	To       string           `json:"to"`
	SSH      *proto.SSHTarget `json:"ssh"`
	// FromConn, ToConn: that side is an SSH host held by its agent for a
	// file pane (its connection id); To is still the agent holding ToConn.
	FromConn string `json:"from_conn"`
	ToConn   string `json:"to_conn"`
}

// POST /api/panel/agents/{id}/fs/ship - start sending; the answer names the
// job to follow.
func (s *Server) handleFSShip(w http.ResponseWriter, r *http.Request) {
	from := r.PathValue("id")
	var q shipReq
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&q); err != nil {
		writeJSON(w, proto.TaskResult{Error: "bad request"})
		return
	}
	q.Name, q.Dir = strings.TrimSpace(q.Name), strings.TrimSpace(q.Dir)
	if q.Src == "" || q.Name == "" || strings.Contains(q.Name, "/") || q.Dir == "" {
		writeJSON(w, proto.TaskResult{Error: "не указано, что и куда"})
		return
	}
	if !s.fsAgent(w, from, true) {
		return
	}
	where := ""
	if err := s.fillSaved(s.currentAdmin(r).ID, q.SSH); err != nil {
		writeJSON(w, proto.TaskResult{Error: err.Error()})
		return
	}
	if q.SSH != nil {
		t := q.SSH
		t.Host, t.User = strings.TrimSpace(t.Host), strings.TrimSpace(t.User)
		if t.Port == 0 {
			t.Port = 22
		}
		if t.Host == "" || t.User == "" || t.Port < 1 || t.Port > 65535 {
			writeJSON(w, proto.TaskResult{Error: "нужны хост, порт и пользователь"})
			return
		}
		where = "ssh:" + t.User + "@" + t.Host + ":" + strconv.Itoa(t.Port)
	} else {
		if q.To == "" && q.ToConn != "" {
			q.To = from
		}
		if q.To == "" {
			writeJSON(w, proto.TaskResult{Error: "не выбрана машина"})
			return
		}
		if !s.db.CanUseAgent(s.currentAdmin(r), q.To) {
			writeJSON(w, proto.TaskResult{Error: "нет доступа к той машине"})
			return
		}
		if !s.fsAgent(w, q.To, true) {
			return
		}
		where = "agent:" + q.To
	}
	ctx, cancel := context.WithTimeout(context.Background(), xferTimeout)
	j := &shipJob{id: randToken()[:24], changed: make(chan struct{}), cancel: cancel}
	s.ship.mu.Lock()
	if s.ship.jobs == nil {
		s.ship.jobs, s.ship.relays = map[string]*shipJob{}, map[string]*relay{}
	}
	s.ship.jobs[j.id] = j
	s.ship.mu.Unlock()
	if admin := s.currentAdmin(r); admin != nil {
		op := "copy"
		if q.Move {
			op = "move"
		}
		s.db.Audit(admin.ID, from, "fs.ship."+op+":"+q.Src+" -> "+where+":"+strings.TrimSuffix(q.Dir, "/")+"/"+q.Name)
	}
	go s.runShip(ctx, j, from, q)
	data, _ := json.Marshal(map[string]string{"job": j.id})
	writeJSON(w, proto.TaskResult{OK: true, Data: data})
}

func (s *Server) runShip(ctx context.Context, j *shipJob, from string, q shipReq) {
	defer j.cancel()
	res := s.ship1(ctx, j, from, q)
	if !res.OK && j.stopped {
		res.Error = "отменено"
	}
	j.update(func() { j.done, j.res = true, res })
	// The result waits a while for the panel to come for it.
	time.AfterFunc(xferIdle, func() {
		s.ship.mu.Lock()
		delete(s.ship.jobs, j.id)
		s.ship.mu.Unlock()
	})
}

func oldAgent(res proto.TaskResult) proto.TaskResult {
	if unknownOp(res) {
		res.Error = "агент не умеет отправлять на другие машины - обновите его"
	}
	return res
}

func (s *Server) ship1(ctx context.Context, j *shipJob, from string, q shipReq) proto.TaskResult {
	args := proto.FSShipArgs{Src: q.Src, Name: q.Name, Dir: q.Dir, Conflict: q.Conflict, Job: j.id}
	if q.SSH != nil {
		args.SSH, args.Move = q.SSH, q.Move
		res, err := s.waitTask(ctx, from, proto.TaskFSShipOut, args, xferTimeout)
		if err != nil {
			return proto.TaskResult{Error: err.Error()}
		}
		return oldAgent(res)
	}

	x := &relay{id: randToken()[:24], from: from, to: q.To, changed: make(chan struct{}), chunks: map[int][]byte{}, end: -1}
	s.ship.mu.Lock()
	s.ship.relays[x.id] = x
	s.ship.mu.Unlock()
	defer func() {
		s.ship.mu.Lock()
		delete(s.ship.relays, x.id)
		s.ship.mu.Unlock()
	}()
	go func() { <-ctx.Done(); x.fail("отменено") }()
	args.Relay, args.Chunk = x.id, xferChunk
	inArgs := args
	args.Conn, inArgs.Conn = q.FromConn, q.ToConn

	var in proto.TaskResult
	inDone := make(chan struct{})
	go func() {
		defer close(inDone)
		res, err := s.waitTask(ctx, q.To, proto.TaskFSShipIn, inArgs, xferTimeout)
		if err != nil {
			res = proto.TaskResult{Error: err.Error()}
		}
		in = oldAgent(res)
		if !in.OK {
			x.fail(in.Error)
		}
	}()
	out, err := s.waitTask(ctx, from, proto.TaskFSShipOut, args, xferTimeout)
	if err != nil {
		out = proto.TaskResult{Error: err.Error()}
	}
	out = oldAgent(out)
	if !out.OK {
		x.fail(out.Error)
	}
	<-inDone
	if !in.OK {
		return proto.TaskResult{Error: "на той машине: " + in.Error}
	}
	if !out.OK {
		return out
	}
	if q.Move {
		task, del := proto.TaskFSDelTree, any(proto.FSPathArgs{Path: q.Src})
		if q.FromConn != "" {
			raw, _ := json.Marshal(proto.FSPathArgs{Path: q.Src})
			task, del = proto.TaskSSHFSOp, proto.SSHFSArgs{Conn: q.FromConn, Op: "deltree", Args: raw}
		}
		res, err := s.waitTask(ctx, from, task, del, fsTaskTimeout)
		if err == nil && !res.OK {
			err = errText(res.Error)
		}
		if err != nil {
			return proto.TaskResult{Error: "отправлено, но здесь не удалилось: " + err.Error()}
		}
	}
	// The receiver's count: what it put down and what it kept as it was.
	return in
}

type errText string

func (e errText) Error() string { return string(e) }

// GET /api/panel/fs/job/{jid}?v=N - the job's state once it differs from
// version N, or after a while anyway.
func (s *Server) handleShipJob(w http.ResponseWriter, r *http.Request) {
	j := s.ship.job(r.PathValue("jid"))
	if j == nil {
		writeJSON(w, map[string]any{"done": true, "result": proto.TaskResult{Error: "задание не найдено"}})
		return
	}
	seen, _ := strconv.Atoi(r.URL.Query().Get("v"))
	t := time.NewTimer(25 * time.Second)
	defer t.Stop()
	for {
		j.mu.Lock()
		if j.ver != seen || j.done {
			out := map[string]any{"v": j.ver, "progress": j.prog, "done": j.done}
			if j.done {
				out["result"] = j.res
			}
			j.mu.Unlock()
			writeJSON(w, out)
			return
		}
		ch := j.changed
		j.mu.Unlock()
		select {
		case <-ch:
		case <-t.C:
			seen = -1 // answer with what there is
		case <-r.Context().Done():
			return
		}
	}
}

// POST /api/panel/fs/job/{jid}/cancel
func (s *Server) handleShipCancel(w http.ResponseWriter, r *http.Request) {
	if j := s.ship.job(r.PathValue("jid")); j != nil {
		j.update(func() { j.stopped = true })
		j.cancel()
	}
	w.WriteHeader(http.StatusNoContent)
}

// POST /api/agent/fs/job/{jid}/progress - the sender's progress. 410: the
// job is over or cancelled, stop.
func (s *Server) handleShipProgress(w http.ResponseWriter, r *http.Request) {
	if s.authenticateAgent(r) == nil {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	j := s.ship.job(r.PathValue("jid"))
	var p proto.ShipProgress
	if j == nil || json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&p) != nil {
		http.Error(w, "gone", http.StatusGone)
		return
	}
	stop := false
	j.update(func() { j.prog, stop = p, j.stopped || j.done })
	if stop {
		http.Error(w, "gone", http.StatusGone)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// POST /api/agent/fs/relay/{rid}/{i} - piece i of the stream from the
// sender, X-Last on the last. 503: no room yet, again; 410: it is over.
func (s *Server) handleRelayPost(w http.ResponseWriter, r *http.Request) {
	a := s.authenticateAgent(r)
	if a == nil {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	x := s.ship.relay(r.PathValue("rid"))
	i, err := strconv.Atoi(r.PathValue("i"))
	if x == nil || x.from != a.ID || err != nil || i < 0 {
		http.Error(w, "передача не найдена", http.StatusGone)
		return
	}
	data, err := readPiece(w, r)
	if err != nil {
		http.Error(w, "кусок не дошёл", http.StatusBadRequest)
		return
	}
	if !x.await(r.Context(), xferWait, func() bool { return x.err != "" || i < x.low+xferBuffer }) {
		http.Error(w, "нет места, повторите", http.StatusServiceUnavailable)
		return
	}
	defer x.mu.Unlock()
	if x.err != "" {
		http.Error(w, x.err, http.StatusGone)
		return
	}
	if i >= x.low && x.chunks[i] == nil {
		x.chunks[i] = data
		if r.Header.Get("X-Last") != "" {
			x.end = i
		}
		x.notify()
	}
	w.WriteHeader(http.StatusNoContent)
}

// GET /api/agent/fs/relay/{rid}/{i} - piece i for the receiver, X-Last on
// the last. Asking for i says every piece below it is written, so those are
// let go. 503: not here yet, ask again; 410: it is over.
func (s *Server) handleRelayGet(w http.ResponseWriter, r *http.Request) {
	a := s.authenticateAgent(r)
	if a == nil {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	x := s.ship.relay(r.PathValue("rid"))
	i, err := strconv.Atoi(r.PathValue("i"))
	if x == nil || x.to != a.ID || err != nil || i < 0 {
		http.Error(w, "передача не найдена", http.StatusGone)
		return
	}
	x.mu.Lock()
	if i > x.low {
		for k := x.low; k < i; k++ {
			delete(x.chunks, k)
		}
		x.low = i
		x.notify()
	}
	x.mu.Unlock()
	if !x.await(r.Context(), xferWait, func() bool { return x.err != "" || x.chunks[i] != nil }) {
		http.Error(w, "кусок ещё не пришёл", http.StatusServiceUnavailable)
		return
	}
	if x.err != "" {
		msg := x.err
		x.mu.Unlock()
		http.Error(w, msg, http.StatusGone)
		return
	}
	data, last := x.chunks[i], i == x.end
	x.mu.Unlock()
	w.Header().Set("Content-Type", "application/octet-stream")
	w.Header().Set("Content-Length", strconv.Itoa(len(data)))
	if last {
		w.Header().Set("X-Last", "1")
	}
	_, _ = w.Write(data)
}