browser/platform/api/job_channel_handlers_test.go

package api

import (
	"net/http"
	"testing"

	"browser.local/platform/domain"
	"browser.local/platform/dto"
)

func TestRunJobChannelAPIWorkflow(t *testing.T) {
	router := newTestRouter()
	hello := decodeBody[dto.RunControlHelloResponse](t, performRunControlHello(t, router, validRunJobControlHelloRequest()))
	heartbeatRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/control/heartbeat", dto.RunControlHeartbeatRequest{
		RunEndpointID:         "run-local",
		SessionToken:          hello.SessionToken,
		Version:               "0.1.0",
		Status:                domain.RunEndpointStatusOnline,
		CapabilityFingerprint: "cap-jobs",
		Capacity:              dto.RunCapacityResponse{MaxJobs: 4},
	})
	assertStatus(t, heartbeatRecorder, http.StatusOK)

	postJSON[dto.JobResponse](t, router, "/api/v1/jobs", dto.JobCreateRequest{
		ID:             "job-1",
		RunEndpointID:  "run-local",
		Capability:     "process.start",
		IdempotencyKey: "idem-1",
	})

	claimRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/claim", dto.RunJobClaimRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		Capabilities:  []string{"process.start"},
		Capacity:      dto.RunCapacityResponse{MaxJobs: 4},
	})
	assertStatus(t, claimRecorder, http.StatusOK)
	claim := decodeBody[dto.RunJobClaimResponse](t, claimRecorder)
	if !claim.HasJob || claim.Job.JobID != "job-1" || claim.Job.State != domain.JobStateAccepted {
		t.Fatalf("expected claimed accepted job, got %+v", claim)
	}

	ack := postRunJobAck(t, router, dto.RunJobAckRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		JobID:         claim.Job.JobID,
		LeaseToken:    claim.Job.LeaseToken,
		Attempt:       claim.Job.Attempt,
		Message:       "started",
	})
	if ack.Job.State != domain.JobStateRunning {
		t.Fatalf("expected running ack, got %+v", ack)
	}

	progressRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/progress", dto.RunJobProgressRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		JobID:         claim.Job.JobID,
		LeaseToken:    claim.Job.LeaseToken,
		Attempt:       claim.Job.Attempt,
		Progress:      dto.JobProgressBody{Percent: 60, Message: "working"},
	})
	assertStatus(t, progressRecorder, http.StatusOK)
	progress := decodeBody[dto.RunJobProgressResponse](t, progressRecorder)
	if progress.Job.Progress.Percent != 60 {
		t.Fatalf("expected progress update, got %+v", progress)
	}

	cancelRecorder := performJSON(t, router, http.MethodPost, "/api/v1/jobs/job-1/cancel", dto.RunJobCancelRequestBody{Reason: "operator requested"})
	assertStatus(t, cancelRecorder, http.StatusOK)
	cancel := decodeBody[dto.RunJobCancelRequestResponse](t, cancelRecorder)
	if !cancel.Accepted || cancel.Reason != "operator requested" {
		t.Fatalf("expected cancel request, got %+v", cancel)
	}

	pollRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/cancel", dto.RunJobCancelPollRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		JobID:         claim.Job.JobID,
		LeaseToken:    claim.Job.LeaseToken,
	})
	assertStatus(t, pollRecorder, http.StatusOK)
	poll := decodeBody[dto.RunJobCancelPollResponse](t, pollRecorder)
	if !poll.HasCancel || poll.JobID != "job-1" {
		t.Fatalf("expected cancel poll result, got %+v", poll)
	}

	resultRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/result", dto.RunJobResultRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		JobID:         claim.Job.JobID,
		LeaseToken:    claim.Job.LeaseToken,
		Attempt:       claim.Job.Attempt,
		State:         domain.JobStateCancelled,
		Progress:      dto.JobProgressBody{Percent: 100, Message: "cancelled"},
		Message:       "cancelled",
	})
	assertStatus(t, resultRecorder, http.StatusOK)
	result := decodeBody[dto.RunJobResultResponse](t, resultRecorder)
	if result.Job.State != domain.JobStateCancelled {
		t.Fatalf("expected cancelled result, got %+v", result)
	}

	reconcileRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/reconcile", dto.RunJobReconcileRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		ActiveJobIDs:  []string{"local-only"},
	})
	assertStatus(t, reconcileRecorder, http.StatusOK)
	reconcile := decodeBody[dto.RunJobReconcileResponse](t, reconcileRecorder)
	if len(reconcile.ActiveJobs) != 0 || len(reconcile.UnknownJobIDs) != 1 || reconcile.UnknownJobIDs[0] != "local-only" {
		t.Fatalf("expected no active platform jobs and one unknown local job, got %+v", reconcile)
	}
}

func TestRunJobChannelAPIErrors(t *testing.T) {
	router := newTestRouter()
	hello := decodeBody[dto.RunControlHelloResponse](t, performRunControlHello(t, router, validRunJobControlHelloRequest()))
	postJSON[dto.JobResponse](t, router, "/api/v1/jobs", dto.JobCreateRequest{ID: "job-1", RunEndpointID: "run-local", Capability: "process.start", IdempotencyKey: "idem-1"})

	invalidClaim := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/claim", dto.RunJobClaimRequest{
		RunEndpointID: "run-local",
		SessionToken:  "stale",
		Capacity:      dto.RunCapacityResponse{MaxJobs: 4},
	})
	assertErrorResponse(t, invalidClaim, http.StatusBadRequest, errorCodeValidation)

	claim := decodeBody[dto.RunJobClaimResponse](t, performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/claim", dto.RunJobClaimRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		Capacity:      dto.RunCapacityResponse{MaxJobs: 4},
	}))

	invalidProgress := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/progress", dto.RunJobProgressRequest{
		RunEndpointID: "run-local",
		SessionToken:  hello.SessionToken,
		JobID:         claim.Job.JobID,
		LeaseToken:    claim.Job.LeaseToken,
		Attempt:       claim.Job.Attempt,
		Progress:      dto.JobProgressBody{Percent: 101},
	})
	assertErrorResponse(t, invalidProgress, http.StatusBadRequest, errorCodeValidation)

	badMethod := performRaw(t, router, http.MethodGet, "/api/v1/run/jobs/claim", "")
	assertErrorResponse(t, badMethod, http.StatusMethodNotAllowed, errorCodeMethodNotAllowed)
}

func postRunJobAck(t *testing.T, router http.Handler, request dto.RunJobAckRequest) dto.RunJobAckResponse {
	t.Helper()
	recorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/ack", request)
	assertStatus(t, recorder, http.StatusOK)
	return decodeBody[dto.RunJobAckResponse](t, recorder)
}

func validRunJobControlHelloRequest() dto.RunControlHelloRequest {
	request := validRunControlHelloRequest()
	request.CapabilityReport.Capabilities = append(request.CapabilityReport.Capabilities, "process.start")
	request.CapabilityReport.Fingerprint = "cap-jobs"
	return request
}