package agent

import (
	"context"
	"errors"
	"path/filepath"
	"sync"
	"testing"
	"time"

	rvboxv1 "github.com/rvbox/rvbox/gen/go/rvbox/v1"
	"github.com/rvbox/rvbox/internal/agentproto"
	"github.com/rvbox/rvbox/internal/client/spool"
	"github.com/rvbox/rvbox/internal/domain"
	"github.com/rvbox/rvbox/internal/observability"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func TestRunOnceReconcilesReplaysAndAdvertisesCapacity_HP_RUNTIME_01(t *testing.T) {
	ctx := context.Background()
	store, err := spool.Open(ctx, spool.Options{DataDir: filepath.Join(t.TempDir(), "spool"), BusyTimeout: time.Second})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	issue, err := domain.ParseUUIDv7("019c46f1-1d02-7000-8000-0000000000b1")
	if err != nil {
		t.Fatal(err)
	}
	hash := [32]byte{1}
	if _, err := store.AcceptCommand(ctx, spool.Command{IssueUUID: issue, ImmutableSHA256: hash, Revision: 1, Phase: uint32(rvboxv1.CommandLifecycle_COMMAND_ACCEPTED), ExecutionSpec: []byte("spec")}, time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	eventPayload, err := proto.Marshal(&rvboxv1.CommandEvent{Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.AppendEvent(ctx, issue, spool.EventInput{Kind: 4, Compression: 1, RawBytes: uint64(len(eventPayload)), Payload: eventPayload, CreatedAt: time.Now().UTC()}); err != nil {
		t.Fatal(err)
	}
	if _, err := store.AssignSendWindow(ctx, issue, 1, 1<<20); err != nil {
		t.Fatal(err)
	}
	welcome, _ := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_ServerWelcome{ServerWelcome: &rvboxv1.ServerWelcome{SelectedProtocol: &rvboxv1.ProtocolVersion{Major: 1, Minor: 0}, ServerTime: timestamppb.Now()}}})
	reconcileRequest, _ := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_ReconcileRequest{ReconcileRequest: &rvboxv1.ReconcileRequest{}}})
	reconcileResult, _ := proto.Marshal(&rvboxv1.AgentEnvelope{SessionId: "session", SessionGeneration: 1, Payload: &rvboxv1.AgentEnvelope_ReconcileResult{ReconcileResult: &rvboxv1.ReconcileResult{}}})
	transport := &runnerTransport{reads: [][]byte{welcome, reconcileRequest, reconcileResult}, terminal: errors.New("transport closed")}
	hello := &rvboxv1.ClientHello{ClientId: "runner-client", SupportedProtocol: &rvboxv1.ProtocolRange{Major: 1, MinMinor: 0, MaxMinor: 0}, DaemonVersion: "test", Platform: rvboxv1.Platform_PLATFORM_WINDOWS, Architecture: "amd64", DaemonCwd: `C:\`, SupportedShells: []rvboxv1.ShellType{rvboxv1.ShellType_SHELL_POWERSHELL}, ClientInstanceId: store.ClientInstanceID().String(), MaxRunningCommands: 1, MaxQueuedCommands: 1, SentAt: timestamppb.Now()}
	metrics := observability.New()
	var active []bool
	err = RunOnce(ctx, RunnerOptions{Store: store, Dial: func(context.Context) (Transport, error) { return transport, nil }, Hello: hello, Limits: agentproto.DefaultLimits(), Backoff: BackoffOptions{Initial: time.Millisecond, Maximum: time.Millisecond, StableReset: time.Second}, Jitter: func(value time.Duration) time.Duration { return 0 }, Now: func() time.Time { return time.Now().UTC() }, Metrics: metrics, OnSessionActive: func(value bool) { active = append(active, value) }})
	if !errors.Is(err, transport.terminal) {
		t.Fatalf("RunOnce error = %v, want transport close", err)
	}
	transport.mu.Lock()
	writes := append([][]byte(nil), transport.writes...)
	transport.mu.Unlock()
	if len(writes) != 4 {
		t.Fatalf("wire write count = %d, want hello/snapshot/event/capacity", len(writes))
	}
	last, err := agentproto.DecodeEnvelope(writes[len(writes)-1], agentproto.DefaultLimits(), rvboxv1.Platform_PLATFORM_WINDOWS)
	if err != nil || last.GetClientCapacity() == nil || last.GetClientCapacity().GetMaxRunningCommands() != 1 {
		t.Fatalf("capacity advertisement = %#v, %v", last, err)
	}
	eventEnvelope, err := agentproto.DecodeEnvelope(writes[2], agentproto.DefaultLimits(), rvboxv1.Platform_PLATFORM_WINDOWS)
	if err != nil || eventEnvelope.GetCommandEvent() == nil || eventEnvelope.GetCommandEvent().GetEventSeq() != 1 || eventEnvelope.GetCommandEvent().GetIssueUuid() != issue.String() {
		t.Fatalf("replayed event = %#v, %v", eventEnvelope, err)
	}
	_, _, counters := metrics.Snapshot()
	if counters["session_established"] != 1 {
		t.Fatalf("session metrics = %#v", counters)
	}
	if len(active) != 2 || !active[0] || active[1] {
		t.Fatalf("active transitions = %#v", active)
	}
}

func TestRunnerOptionsRejectMissingJitter_HP_RUNTIME_02(t *testing.T) {
	if err := (RunnerOptions{}).validate(); err == nil {
		t.Fatal("empty runner options unexpectedly validated")
	}
}

func TestRunReportsRetryableSessionError_HP_RUNTIME_04(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	store, err := spool.Open(ctx, spool.Options{DataDir: filepath.Join(t.TempDir(), "spool"), BusyTimeout: time.Second})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	want := errors.New("dial refused")
	reported := make(chan error, 1)
	hello := &rvboxv1.ClientHello{ClientId: "runner-client", SupportedProtocol: &rvboxv1.ProtocolRange{Major: 1, MinMinor: 0, MaxMinor: 0}, DaemonVersion: "test", Platform: rvboxv1.Platform_PLATFORM_WINDOWS, Architecture: "amd64", DaemonCwd: `C:\\`, SupportedShells: []rvboxv1.ShellType{rvboxv1.ShellType_SHELL_POWERSHELL}, ClientInstanceId: store.ClientInstanceID().String(), MaxRunningCommands: 1, MaxQueuedCommands: 1, SentAt: timestamppb.Now()}
	err = Run(ctx, RunnerOptions{
		Store: store, Dial: func(context.Context) (Transport, error) { return nil, want }, Hello: hello,
		Limits: agentproto.DefaultLimits(), Backoff: BackoffOptions{Initial: time.Millisecond, Maximum: time.Millisecond, StableReset: time.Second},
		Jitter: func(time.Duration) time.Duration { return 0 }, Now: func() time.Time { return time.Now().UTC() },
		OnSessionError: func(got error) { reported <- got; cancel() },
	})
	if err != nil {
		t.Fatalf("Run error = %v, want graceful cancellation", err)
	}
	select {
	case got := <-reported:
		if !errors.Is(got, want) {
			t.Fatalf("reported error = %v, want %v", got, want)
		}
	default:
		t.Fatal("retryable session error was not reported")
	}
}

func TestFlushEventsAssignsAndSendsOnlyUnacknowledgedRows_HP_RUNTIME_03(t *testing.T) {
	ctx := context.Background()
	store, err := spool.Open(ctx, spool.Options{DataDir: filepath.Join(t.TempDir(), "spool"), BusyTimeout: time.Second})
	if err != nil {
		t.Fatal(err)
	}
	defer store.Close()
	issue, err := domain.ParseUUIDv7("019c46f1-1d02-7000-8000-0000000000b2")
	if err != nil {
		t.Fatal(err)
	}
	hash := [32]byte{2}
	if _, err := store.AcceptCommand(ctx, spool.Command{IssueUUID: issue, ImmutableSHA256: hash, Revision: 1, Phase: uint32(rvboxv1.CommandLifecycle_COMMAND_ACCEPTED)}, time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	payload, err := proto.Marshal(&rvboxv1.CommandEvent{Payload: &rvboxv1.CommandEvent_Lifecycle{Lifecycle: &rvboxv1.LifecycleChange{Lifecycle: rvboxv1.CommandLifecycle_COMMAND_RUNNING, CommandRevision: 1}}})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.AppendEvent(ctx, issue, spool.EventInput{Kind: 4, Compression: 1, RawBytes: uint64(len(payload)), Payload: payload, CreatedAt: time.Now().UTC()}); err != nil {
		t.Fatal(err)
	}
	transport := &runnerTransport{}
	sent := map[domain.UUID]uint64{}
	session := Session{ID: "session", Generation: 1}
	if err := flushEvents(ctx, transport, store, session, issue, agentproto.DefaultLimits(), sent); err != nil {
		t.Fatal(err)
	}
	if sent[issue] != 1 || len(transport.writes) != 1 {
		t.Fatalf("sent cursor/writes = %d/%d", sent[issue], len(transport.writes))
	}
	if err := flushEvents(ctx, transport, store, session, issue, agentproto.DefaultLimits(), sent); err != nil {
		t.Fatal(err)
	}
	if len(transport.writes) != 1 {
		t.Fatalf("already-sent event was duplicated: %d writes", len(transport.writes))
	}
}

type runnerTransport struct {
	mu       sync.Mutex
	reads    [][]byte
	writes   [][]byte
	terminal error
}

func (transport *runnerTransport) Write(_ context.Context, payload []byte) error {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	transport.writes = append(transport.writes, append([]byte(nil), payload...))
	return nil
}

func (transport *runnerTransport) Read(context.Context) ([]byte, error) {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	if len(transport.reads) == 0 {
		return nil, transport.terminal
	}
	payload := transport.reads[0]
	transport.reads = transport.reads[1:]
	return append([]byte(nil), payload...), nil
}

func (transport *runnerTransport) Close() error { return nil }