package xrpc

import (
	"context"
	"fmt"
	"net/http"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/atclient"
	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/gorilla/websocket"
	"tangled.org/core/api/tangled"
)

func (x *Xrpc) HandleCiPipelineSubscribeLogs(w http.ResponseWriter, r *http.Request) {
	var (
		pipelineQuery  = r.URL.Query().Get("pipeline")
		workflows = r.URL.Query()["workflows"]
	)

	pipeline, err := syntax.ParseTID(pipelineQuery)
	if err != nil {
		writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("pipeline parameter invalid: %s", pipelineQuery)})
	}

	x.handleSubscribeLogs(w, r, pipeline, workflows)
}

var wsUpgrader = websocket.Upgrader{
	ReadBufferSize:  10_000,
	WriteBufferSize: 10_000,
}

func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipeline syntax.TID, workflows []string) {
	l := x.Logger.With("pipeline", pipeline, "workflows", workflows)

	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()

	conn, err := wsUpgrader.Upgrade(w, r, w.Header())
	if err != nil {
		// TODO: writeJson(w, )
		return
	}
	defer conn.Close()

	lastWriteLk := sync.Mutex{}
	lastWrite := time.Now()

	// Start a goroutine to ping the client every 30 seconds to check if it's
	// still alive. If the client doesn't respond to a ping within 5 seconds,
	// we'll close the connection and teardown the consumer.
	go func() {
		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-ticker.C:
				lastWriteLk.Lock()
				lw := lastWrite
				lastWriteLk.Unlock()

				if time.Since(lw) < 30*time.Second {
					continue
				}

				if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil {
					l.Warn("failed to ping client", "err", err)
					cancel()
					return
				}
			case <-ctx.Done():
				return
			}
		}
	}()

	conn.SetPingHandler(func(message string) error {
		err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60))
		if err == websocket.ErrCloseSent {
			return nil
		}
		return err
	})

	// Start a goroutine to read messages from the client and discard them.
	go func() {
		for {
			_, _, err := conn.ReadMessage()
			if err != nil {
				l.Warn("failed to read message from client", "err", err)
				cancel()
				return
			}
		}
	}()

	evts := make(chan tangled.CiPipelineSubscribeLogs_Event, 16)
	go mockEvents(ctx, evts)

	for {
		select {
		case <-ctx.Done():
			return
		case evt, ok := <-evts:
			if !ok {
				l.Error("event stream closed unexpectedly")
				return
			}

			wc, err := conn.NextWriter(websocket.BinaryMessage)
			if err != nil {
				l.Error("failed to get next writer", "err", err)
				return // err
			}

			err = evt.Serialize(wc)
			if err != nil {
				return // err
			}

			if err := wc.Close(); err != nil {
				l.Warn("failed to flush-close our event write", "err", err)
				return
			}

			lastWriteLk.Lock()
			lastWrite = time.Now()
			lastWriteLk.Unlock()
		}
	}
}

func strptr(s string) *string { return &s }

// mockEvents streams a fixed, fabricated log sequence onto out, ignoring the
// real pipeline/workflow inputs. Used to exercise the appview log viewer until
// the real event source is wired up.
func mockEvents(ctx context.Context, out chan<- tangled.CiPipelineSubscribeLogs_Event) {
	const wf = "build"
	now := func() string { return time.Now().UTC().Format(time.RFC3339) }

	send := func(ev tangled.CiPipelineSubscribeLogs_Event) bool {
		select {
		case out <- ev:
			return true
		case <-ctx.Done():
			return false
		}
	}
	dataEv := func(step int64, content string) tangled.CiPipelineSubscribeLogs_Event {
		return tangled.CiPipelineSubscribeLogs_Event{Data: &tangled.CiPipelineSubscribeLogs_Data{
			Step: step, Content: content, Stream: "stdout", Time: now(), Workflow: wf,
		}}
	}
	endEv := func(step int64) tangled.CiPipelineSubscribeLogs_Event {
		return tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{
			Status: strptr("end"), Step: step, Time: now(), Workflow: wf,
		}}
	}

	// system step: clone (collapsed in the viewer)
	if !send(tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{
		Status: strptr("start"), Kind: strptr("system"), Step: 1,
		Content: "clone repository", Command: strptr("git clone https://example.com/repo"), Time: now(), Workflow: wf,
	}}) {
		return
	}
	for _, line := range []string{"Cloning into 'repo'...\n", "remote: Enumerating objects\n", "\x1b[32mClone complete\x1b[0m\n"} {
		if !send(dataEv(1, line)) {
			return
		}
	}
	if !send(endEv(1)) {
		return
	}

	// user step: test — stream a counter line every second until the client leaves
	if !send(tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{
		Status: strptr("start"), Kind: strptr("user"), Step: 2,
		Content: "run tests", Command: strptr("go test ./..."), Time: now(), Workflow: wf,
	}}) {
		return
	}
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for i := 1; ; i++ {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if !send(dataEv(2, fmt.Sprintf("test log line %d\n", i))) {
				return
			}
		}
	}
}
