package api

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"browser.local/run/protocol"
)

func TestPlatformClientLightweightChannelsCompleteWhileArtifactChunkIsBlocked(t *testing.T) {
	artifactStarted := make(chan struct{})
	releaseArtifact := make(chan struct{})
	artifactDone := make(chan struct{})

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/v1/run/artifacts/chunks":
			body, err := io.ReadAll(r.Body)
			if err != nil {
				t.Fatalf("read artifact payload: %v", err)
			}
			if r.Header.Get("X-Artifact-Chunk-Index") != "0" || len(body) == 0 || r.Header.Get("Content-Type") != "application/octet-stream" {
				t.Fatalf("unexpected artifact payload headers=%+v body=%q", r.Header, string(body))
			}
			close(artifactStarted)
			<-releaseArtifact
			writeTestJSON(t, w, protocol.ArtifactChunkUploadResponse{Accepted: true, TransferID: r.Header.Get("X-Artifact-Transfer-Id"), ArtifactID: r.Header.Get("X-Artifact-Id"), ChunkIndex: 0, ReceivedChunkIndexes: []int{0}, NextMissingChunkIndex: 1, ServerTime: fixedClientTestTime()})
			close(artifactDone)
		case "/api/v1/run/control/heartbeat":
			var request protocol.RunHeartbeatRequest
			decodeTestRequest(t, r, &request)
			writeTestJSON(t, w, protocol.RunHeartbeatResponse{Accepted: true, RunEndpointID: request.RunEndpointID, NextHeartbeatSeconds: 15, ServerTime: fixedClientTestTime()})
		case "/api/v1/run/jobs/result":
			var request protocol.RunJobResultRequest
			decodeTestRequest(t, r, &request)
			encoded, _ := json.Marshal(request)
			for _, forbidden := range []string{"payload", "entries", "/Users/", "unix://", "tcp://", "Bearer ", "sk-", "password="} {
				if strings.Contains(string(encoded), forbidden) {
					t.Fatalf("job result carried forbidden transfer content %q: %s", forbidden, string(encoded))
				}
			}
			job := validRunJobAssignment()
			job.State = request.State
			job.ResultRef = request.ResultRef
			writeTestJSON(t, w, protocol.RunJobResultResponse{Accepted: true, Job: job, ServerTime: fixedClientTestTime()})
		case "/api/v1/run/logs/batches":
			var request protocol.LogBatchIngestRequest
			decodeTestRequest(t, r, &request)
			if len(request.Entries) != 1 || request.FirstSeq != 1 || request.LastSeq != 1 {
				t.Fatalf("unexpected log batch: %+v", request)
			}
			writeTestJSON(t, w, protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: request.LogStreamID, AcceptedFrom: 1, AcceptedTo: 1, LatestSeq: 1, ServerTime: fixedClientTestTime()})
		default:
			t.Fatalf("unexpected request path %s", r.URL.Path)
		}
	}))
	defer server.Close()

	client, err := NewPlatformClient(server.URL)
	if err != nil {
		t.Fatalf("new client: %v", err)
	}

	errCh := make(chan error, 1)
	go func() {
		_, err := client.UploadArtifactChunk(context.Background(), validClientArtifactChunk())
		errCh <- err
	}()

	select {
	case <-artifactStarted:
	case <-time.After(time.Second):
		t.Fatal("artifact request did not start")
	}

	lightCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if _, err := client.Heartbeat(lightCtx, validRunHeartbeatRequest("session-token")); err != nil {
		t.Fatalf("heartbeat should not wait for artifact chunk: %v", err)
	}
	if _, err := client.CompleteJob(lightCtx, validRunJobResultRequest()); err != nil {
		t.Fatalf("job result should not wait for artifact chunk: %v", err)
	}
	if _, err := client.IngestLogBatch(lightCtx, validClientLogBatch()); err != nil {
		t.Fatalf("log ingest should not wait for artifact chunk: %v", err)
	}

	select {
	case <-artifactDone:
		t.Fatal("artifact chunk completed before release")
	default:
	}
	close(releaseArtifact)
	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("artifact chunk upload: %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("artifact chunk did not finish after release")
	}
}

func TestPlatformClientControlAndJobsCompleteWhileLogIngestIsBlocked(t *testing.T) {
	logStarted := make(chan struct{})
	releaseLog := make(chan struct{})
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/api/v1/run/logs/batches":
			var request protocol.LogBatchIngestRequest
			decodeTestRequest(t, r, &request)
			close(logStarted)
			<-releaseLog
			writeTestJSON(t, w, protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: request.LogStreamID, AcceptedFrom: request.FirstSeq, AcceptedTo: request.LastSeq, LatestSeq: request.LastSeq, ServerTime: fixedClientTestTime()})
		case "/api/v1/run/control/heartbeat":
			var request protocol.RunHeartbeatRequest
			decodeTestRequest(t, r, &request)
			writeTestJSON(t, w, protocol.RunHeartbeatResponse{Accepted: true, RunEndpointID: request.RunEndpointID, NextHeartbeatSeconds: 15, ServerTime: fixedClientTestTime()})
		case "/api/v1/run/jobs/result":
			var request protocol.RunJobResultRequest
			decodeTestRequest(t, r, &request)
			job := validRunJobAssignment()
			job.State = request.State
			writeTestJSON(t, w, protocol.RunJobResultResponse{Accepted: true, Job: job, ServerTime: fixedClientTestTime()})
		case "/api/v1/run/jobs/reconcile":
			var request protocol.RunJobReconcileRequest
			decodeTestRequest(t, r, &request)
			writeTestJSON(t, w, protocol.RunJobReconcileResponse{Accepted: true, RunEndpointID: request.RunEndpointID, ConfirmedJobs: []protocol.RunJobAssignment{validRunJobAssignment()}, ServerTime: fixedClientTestTime()})
		default:
			t.Fatalf("unexpected request path %s", r.URL.Path)
		}
	}))
	defer server.Close()
	client, err := NewPlatformClient(server.URL)
	if err != nil {
		t.Fatalf("new client: %v", err)
	}
	logErr := make(chan error, 1)
	go func() {
		_, callErr := client.IngestLogBatch(context.Background(), validClientLogBatch())
		logErr <- callErr
	}()
	select {
	case <-logStarted:
	case <-time.After(time.Second):
		t.Fatal("log request did not start")
	}

	lightCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if _, err := client.Heartbeat(lightCtx, validRunHeartbeatRequest("session-token")); err != nil {
		t.Fatalf("heartbeat should not wait for log ingest: %v", err)
	}
	if _, err := client.CompleteJob(lightCtx, validRunJobResultRequest()); err != nil {
		t.Fatalf("job result should not wait for log ingest: %v", err)
	}
	if _, err := client.ReconcileJobs(lightCtx, validRunJobReconcileRequest()); err != nil {
		t.Fatalf("job reconcile should not wait for log ingest: %v", err)
	}
	close(releaseLog)
	select {
	case err := <-logErr:
		if err != nil {
			t.Fatalf("log ingest: %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("log ingest did not finish after release")
	}
}

func TestPlatformClientControlJobsAndLogsCompleteWhileDependencyOrUpdateInputIsBlocked(t *testing.T) {
	for _, scenario := range []struct {
		name string
		path string
		call func(context.Context, PlatformClient) error
	}{
		{
			name: "dependency adapter input",
			path: "/api/v1/run/jobs/dependency-input",
			call: func(ctx context.Context, client PlatformClient) error {
				_, err := client.GetDependencyExecutionInput(ctx, protocol.DependencyExecutionInputRequest{RunEndpointID: "run-test", SessionToken: "session-token", JobID: "job-dependency", LeaseToken: "lease-dependency", Attempt: 1})
				return err
			},
		},
		{
			name: "self-update chunk",
			path: "/api/v1/run/jobs/update-chunk",
			call: func(ctx context.Context, client PlatformClient) error {
				_, err := client.ReadRunUpdateChunk(ctx, protocol.RunUpdateChunkRequest{RunEndpointID: "run-test", SessionToken: "session-token", JobID: "job-update", LeaseToken: "lease-update", Attempt: 1, Offset: 0, Length: 8})
				return err
			},
		},
	} {
		t.Run(scenario.name, func(t *testing.T) {
			blockedStarted := make(chan struct{})
			releaseBlocked := make(chan struct{})
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				switch r.URL.Path {
				case scenario.path:
					close(blockedStarted)
					<-releaseBlocked
					if scenario.path == "/api/v1/run/jobs/dependency-input" {
						writeTestJSON(t, w, protocol.DependencyExecutionInputResponse{JobID: "job-dependency", ServerInstanceID: "server-1", RunEndpointID: "run-test", PluginID: "game.runtime", PluginVersion: "1.0.0", ProfileKey: "local", TargetOS: "linux", TargetArch: "amd64", PlanDigest: "sha256:" + strings.Repeat("a", 64), Bindings: map[string]string{}})
						return
					}
					writeTestJSON(t, w, protocol.RunUpdateChunkResponse{JobID: "job-update", ArtifactID: "artifact-update", Offset: 0, TotalBytes: 8, Checksum: "sha256:" + strings.Repeat("b", 64), Payload: []byte("12345678"), Complete: true})
				case "/api/v1/run/control/heartbeat":
					var request protocol.RunHeartbeatRequest
					decodeTestRequest(t, r, &request)
					writeTestJSON(t, w, protocol.RunHeartbeatResponse{Accepted: true, RunEndpointID: request.RunEndpointID, NextHeartbeatSeconds: 15, ServerTime: fixedClientTestTime()})
				case "/api/v1/run/jobs/result":
					var request protocol.RunJobResultRequest
					decodeTestRequest(t, r, &request)
					job := validRunJobAssignment()
					job.State = request.State
					writeTestJSON(t, w, protocol.RunJobResultResponse{Accepted: true, Job: job, ServerTime: fixedClientTestTime()})
				case "/api/v1/run/logs/batches":
					var request protocol.LogBatchIngestRequest
					decodeTestRequest(t, r, &request)
					writeTestJSON(t, w, protocol.LogBatchIngestResponse{Accepted: true, LogStreamID: request.LogStreamID, AcceptedFrom: request.FirstSeq, AcceptedTo: request.LastSeq, LatestSeq: request.LastSeq, ServerTime: fixedClientTestTime()})
				default:
					t.Fatalf("unexpected request path %s", r.URL.Path)
				}
			}))
			defer server.Close()
			client, err := NewPlatformClient(server.URL)
			if err != nil {
				t.Fatalf("new client: %v", err)
			}
			blockedDone := make(chan error, 1)
			go func() { blockedDone <- scenario.call(context.Background(), client) }()
			select {
			case <-blockedStarted:
			case <-time.After(time.Second):
				t.Fatal("blocked dependency/update request did not start")
			}

			lightCtx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			if _, err := client.Heartbeat(lightCtx, validRunHeartbeatRequest("session-token")); err != nil {
				t.Fatalf("heartbeat should not wait for dependency/update input: %v", err)
			}
			if _, err := client.CompleteJob(lightCtx, validRunJobResultRequest()); err != nil {
				t.Fatalf("job result should not wait for dependency/update input: %v", err)
			}
			if _, err := client.IngestLogBatch(lightCtx, validClientLogBatch()); err != nil {
				t.Fatalf("log ingest should not wait for dependency/update input: %v", err)
			}
			close(releaseBlocked)
			select {
			case err := <-blockedDone:
				if err != nil {
					t.Fatalf("blocked request completion: %v", err)
				}
			case <-time.After(time.Second):
				t.Fatal("blocked request did not complete after release")
			}
		})
	}
}