package executor

import (
	"encoding/json"
	"sync"

	"tangled.org/core/api/tangled"

	millproto "tangled.org/core/spindle/mill/proto"
	millv1 "tangled.org/core/spindle/mill/proto/gen"
)

// relayLog is the per-session sequential event log the executor keeps so it can
// replay anything the mill missed across a reconnect. Only mill-facing relay
// messages (StatusEvent / LogLine / AttemptResult) get offsets and are buffered;
// control messages do not.
type relayLog struct {
	mu      sync.Mutex
	entries []*millproto.Message
	next    uint64 // next offset to assign
	acked   uint64 // highest offset the mill has acked
}

func newRelayLog() *relayLog {
	return &relayLog{next: 1}
}

func (rl *relayLog) appendStatus(leaseID string, st *tangled.PipelineStatus, pipelineAturi string) *millproto.Message {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	offset := rl.next
	rl.next++
	var exit int64
	if st.ExitCode != nil {
		exit = *st.ExitCode
	}
	var errStr string
	if st.Error != nil {
		errStr = *st.Error
	}
	msg := &millproto.Message{StatusEvent: &millv1.StatusEvent{
		Offset:        offset,
		LeaseId:       leaseID,
		Status:        st.Status,
		Error:         errStr,
		ExitCode:      exit,
		Workflow:      st.Workflow,
		PipelineAturi: pipelineAturi,
	}}
	rl.entries = append(rl.entries, msg)
	return msg
}

func (rl *relayLog) appendLog(leaseID string, raw []byte) *millproto.Message {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	offset := rl.next
	rl.next++
	msg := &millproto.Message{LogLine: &millv1.LogLine{
		Offset:  offset,
		LeaseId: leaseID,
		RawJson: raw,
	}}
	rl.entries = append(rl.entries, msg)
	return msg
}

func (rl *relayLog) appendTerminal(leaseID, status string, st *tangled.PipelineStatus) *millproto.Message {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	offset := rl.next
	rl.next++
	var exit int64
	var errStr string
	if st != nil {
		if st.ExitCode != nil {
			exit = *st.ExitCode
		}
		if st.Error != nil {
			errStr = *st.Error
		}
	}
	msg := &millproto.Message{AttemptResult: &millv1.AttemptResult{
		Offset:         offset,
		LeaseId:        leaseID,
		TerminalStatus: status,
		Error:          errStr,
		ExitCode:       exit,
	}}
	rl.entries = append(rl.entries, msg)
	return msg
}

func (rl *relayLog) ack(upTo uint64) {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	if upTo > rl.acked {
		rl.acked = upTo
	}
	kept := rl.entries[:0]
	for _, e := range rl.entries {
		if offsetOf(e) > rl.acked {
			kept = append(kept, e)
		}
	}
	// nil the drained tail so acked messages can be GC'd (kept aliases the same
	// backing array as entries).
	for i := len(kept); i < len(rl.entries); i++ {
		rl.entries[i] = nil
	}
	rl.entries = kept
}

// since returns buffered entries with offset strictly greater than offset, for
// replay on reconnect.
func (rl *relayLog) since(offset uint64) []*millproto.Message {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	var out []*millproto.Message
	for _, e := range rl.entries {
		if offsetOf(e) > offset {
			out = append(out, e)
		}
	}
	return out
}

func (rl *relayLog) lastOffset() uint64 {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	return rl.next - 1
}

func offsetOf(m *millproto.Message) uint64 {
	switch {
	case m.GetStatusEvent() != nil:
		return m.GetStatusEvent().GetOffset()
	case m.GetLogLine() != nil:
		return m.GetLogLine().GetOffset()
	case m.GetAttemptResult() != nil:
		return m.GetAttemptResult().GetOffset()
	}
	return 0
}

func parseStatus(raw json.RawMessage) (*tangled.PipelineStatus, bool) {
	var st tangled.PipelineStatus
	if err := json.Unmarshal(raw, &st); err != nil {
		return nil, false
	}
	return &st, true
}
