diff --git a/agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md b/agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md index cffcc2d..78de4a4 100644 --- a/agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md +++ b/agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md @@ -1,4 +1,3 @@ -# Milestone: core server 모듈 경계 정리 ## 상태 diff --git a/agent-task/m-core-server-boundary-cleanup/03+01,02_handlers/CODE_REVIEW-local-G07.md b/agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/code_review_local_G07_0.log similarity index 71% rename from agent-task/m-core-server-boundary-cleanup/03+01,02_handlers/CODE_REVIEW-local-G07.md rename to agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/code_review_local_G07_0.log index dfb6526..ed4f7aa 100644 --- a/agent-task/m-core-server-boundary-cleanup/03+01,02_handlers/CODE_REVIEW-local-G07.md +++ b/agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/code_review_local_G07_0.log @@ -42,43 +42,47 @@ task=m-core-server-boundary-cleanup/03+01,02_handlers, plan=1, tag=HANDLERS | 항목 | 완료 여부 | |------|---------| -| [HANDLERS-1] Handler File Boundary | [ ] | +| [HANDLERS-1] Handler File Boundary | [x] | ## 구현 체크리스트 -- [ ] predecessor `01_routes`와 `02+01_dto`의 `complete.log` 존재를 확인한다. -- [ ] runner registry/bootstrap handlers를 `runner_handlers.go` 또는 역할이 더 명확한 파일로 이동한다. -- [ ] job/execution/log/artifact handlers를 `job_handlers.go`와 `execution_handlers.go`로 나눈다. -- [ ] runner claim/report/cancel/status/self-update handlers를 `runner_cicd_handlers.go`로 이동한다. -- [ ] `server.go`는 Server type, constructors, Start/Shutdown 같은 server lifecycle만 소유하게 정리한다. -- [ ] `gofmt`를 실행하고 `cd services/core && go test -count=1 ./internal/httpserver`를 통과시킨다. -- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다. +- [x] predecessor `01_routes`와 `02+01_dto`의 archived `complete.log` 존재를 확인한다. +- [x] runner registry/bootstrap handlers를 `runner_handlers.go`로 이동한다. +- [x] job/execution/log/artifact handlers를 `job_handlers.go`와 `execution_handlers.go`로 나눈다. +- [x] runner claim/report/cancel/status/self-update handlers를 `runner_cicd_handlers.go`로 이동한다. +- [x] `server.go`는 Server type, constructors, Start/Shutdown 같은 server lifecycle만 소유하게 정리한다. +- [x] `gofmt`를 실행하고 `cd services/core && go test -count=1 ./internal/httpserver`를 통과시킨다. +- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. ## 코드리뷰 전용 체크리스트 > **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다. > 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다. -- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. -- [ ] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. -- [ ] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G07_N.log`로 아카이브한다. -- [ ] active `PLAN-*-G??.md`를 `plan_local_G07_M.log`로 아카이브한다. -- [ ] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. -- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. -- [ ] PASS이면 active task 디렉터리 `agent-task/m-core-server-boundary-cleanup/03+01,02_handlers/`를 `agent-task/archive/YYYY/MM/m-core-server-boundary-cleanup/03+01,02_handlers/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. -- [ ] PASS이고 task group이 `m-core-server-boundary-cleanup`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. -- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-core-server-boundary-cleanup/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. +- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다. +- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다. +- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_local_G07_N.log`로 아카이브한다. +- [x] active `PLAN-*-G??.md`를 `plan_local_G07_M.log`로 아카이브한다. +- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다. +- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다. +- [x] PASS이면 active task 디렉터리 `agent-task/m-core-server-boundary-cleanup/03+01,02_handlers/`를 `agent-task/archive/YYYY/MM/m-core-server-boundary-cleanup/03+01,02_handlers/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다. +- [x] PASS이고 task group이 `m-core-server-boundary-cleanup`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다. +- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-core-server-boundary-cleanup/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다. - [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-local-G07.md`와 `CODE_REVIEW-local-G07.md`를 작성하고 `complete.log`를 작성하지 않는다. - [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다. - [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다. ## 계획 대비 변경 사항 -_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._ +- predecessor `complete.log` 확인: `agent-task/archive/2026/06/m-core-server-boundary-cleanup/01_routes/complete.log`와 `agent-task/archive/2026/06/m-core-server-boundary-cleanup/02+01_dto/complete.log` 존재를 확인했다. +- 계획의 파일 분할 구조(`runner_handlers.go`, `job_handlers.go`, `execution_handlers.go`, `runner_cicd_handlers.go`)는 그대로 따랐다. ## 주요 설계 결정 -_구현 에이전트가 주요 설계 결정 사항을 기록한다._ +- 같은 package 유지: 모든 신규 파일은 `package httpserver`로 두었다. 새 package를 만들지 않았으므로 import cycle 없음. +- helper 함수 배치: `writeRunnerRegisterResponse`, `writeHeartbeatResponse`, `shellEscape`는 runner handler와 함께 `runner_handlers.go`에 두었다. `writeResponse`는 기존대로 `dto.go`에 유지했다. +- `normalizeRunnerExecutionRequest`, `ensureRunnerKnown`은 runner cicd handler에서만 사용하므로 `runner_cicd_handlers.go`에 배치했다. +- `server.go`는 Server struct, 3개 constructor, Start/StartListener/Shutdown만 남겼다. ## 사용자 리뷰 요청 @@ -113,14 +117,18 @@ _구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 ```bash $ cd services/core && go test -count=1 ./internal/httpserver -(output) +ok github.com/toki/oto/services/core/internal/httpserver 0.005s ``` ### 최종 검증 ```bash $ cd services/core && go test -count=1 ./... -(output) +? github.com/toki/oto/services/core/cmd/oto-core [no test files] +ok github.com/toki/oto/services/core/internal/cicdstate 0.002s +ok github.com/toki/oto/services/core/internal/httpserver 0.005s +ok github.com/toki/oto/services/core/internal/runnerregistry 0.003s +? github.com/toki/oto/services/core/oto [no test files] ``` --- @@ -128,3 +136,17 @@ $ cd services/core && go test -count=1 ./... > **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?** > If anything is blank, go back and fill it in before saving this file. > Leave review-agent-only sections unchanged. + +## 코드리뷰 결과 + +- 종합 판정: PASS +- 차원별 평가: + - correctness: Pass + - completeness: Pass + - test coverage: Pass + - API contract: Pass + - code quality: Pass + - plan deviation: Pass + - verification trust: Pass +- 발견된 문제: 없음 +- 다음 단계: PASS이므로 `complete.log` 작성 후 active task directory를 archive로 이동한다. diff --git a/agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/complete.log b/agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/complete.log new file mode 100644 index 0000000..46e7a18 --- /dev/null +++ b/agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/complete.log @@ -0,0 +1,43 @@ +# Complete - m-core-server-boundary-cleanup/03+01,02_handlers + +## 완료 일시 + +2026-06-08 + +## 요약 + +HTTP handler file boundary cleanup completed in 1 review loop; final verdict PASS. + +## 루프 이력 + +| Plan | Review | Verdict | 메모 | +|------|--------|---------|------| +| `plan_local_G07_0.log` | `code_review_local_G07_0.log` | PASS | handler body를 기능별 파일로 이동했고 API behavior와 services/core 검증이 유지됨 | + +## 구현/정리 내용 + +- `server.go`를 `Server` type, constructors, Start/Shutdown lifecycle 중심으로 정리했다. +- runner registry/bootstrap handlers를 `runner_handlers.go`로 이동했다. +- job/execution/log/artifact handlers를 `job_handlers.go`와 `execution_handlers.go`로 분리했다. +- runner claim/report/cancel/status/self-update handlers와 runner execution helper를 `runner_cicd_handlers.go`로 이동했다. + +## 최종 검증 + +- `gofmt -l services/core/internal/httpserver/server.go services/core/internal/httpserver/runner_handlers.go services/core/internal/httpserver/job_handlers.go services/core/internal/httpserver/execution_handlers.go services/core/internal/httpserver/runner_cicd_handlers.go services/core/internal/httpserver/routes.go services/core/internal/httpserver/dto.go services/core/internal/httpserver/server_test.go` - PASS; output empty. +- `cd services/core && go test -count=1 ./internal/httpserver` - PASS; `ok github.com/toki/oto/services/core/internal/httpserver 0.006s`. +- `cd services/core && go test -count=1 ./...` - PASS; core service packages passed, no-test packages reported as expected. + +## Roadmap Completion + +- Milestone: `agent-roadmap/phase/independent-control-plane/milestones/core-server-boundary-cleanup.md` +- Completed task ids: + - `handlers`: PASS; evidence=`plan_local_G07_0.log`, `code_review_local_G07_0.log`; verification=`cd services/core && go test -count=1 ./...` +- Not completed task ids: 없음 + +## 잔여 Nit + +- 없음 + +## 후속 작업 + +- 없음 diff --git a/agent-task/m-core-server-boundary-cleanup/03+01,02_handlers/PLAN-local-G07.md b/agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/plan_local_G07_0.log similarity index 100% rename from agent-task/m-core-server-boundary-cleanup/03+01,02_handlers/PLAN-local-G07.md rename to agent-task/archive/2026/06/m-core-server-boundary-cleanup/03+01,02_handlers/plan_local_G07_0.log diff --git a/services/core/internal/httpserver/execution_handlers.go b/services/core/internal/httpserver/execution_handlers.go new file mode 100644 index 0000000..c045fe0 --- /dev/null +++ b/services/core/internal/httpserver/execution_handlers.go @@ -0,0 +1,140 @@ +package httpserver + +import ( + "encoding/json" + "net/http" + "strings" + + "github.com/toki/oto/services/core/internal/cicdstate" +) + +func handleGetExecution(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + execID := r.PathValue("id") + if execID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + execID = parts[4] + } + } + if execID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) + return + } + exec, err := store.GetExecution(execID) + if err != nil { + writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) + return + } + writeResponse(w, http.StatusOK, execToJSON(exec)) + } +} + +func handleAppendLog(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeResponse(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"}) + return + } + execID := r.PathValue("id") + if execID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + execID = parts[4] + } + } + if execID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) + return + } + var req struct { + Line string `json:"line"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Line == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "line is required"}) + return + } + if err := store.AppendLog(execID, req.Line); err != nil { + writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) + return + } + writeResponse(w, http.StatusCreated, map[string]string{"status": "accepted"}) + } +} + +func handleGetLogs(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + execID := r.PathValue("id") + if execID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + execID = parts[4] + } + } + if execID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) + return + } + logs, err := store.GetLogs(execID) + if err != nil { + writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) + return + } + writeResponse(w, http.StatusOK, logsToJSON(logs)) + } +} + +func handleAppendArtifact(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeResponse(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"}) + return + } + execID := r.PathValue("id") + if execID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + execID = parts[4] + } + } + if execID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) + return + } + var req struct { + Name string `json:"name"` + Path string `json:"path"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Name == "" || req.Path == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "name and path are required"}) + return + } + if err := store.AppendArtifact(execID, req.Name, req.Path); err != nil { + writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) + return + } + writeResponse(w, http.StatusCreated, map[string]string{"status": "accepted"}) + } +} + +func handleGetArtifacts(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + execID := r.PathValue("id") + if execID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + execID = parts[4] + } + } + if execID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) + return + } + artifacts, err := store.GetArtifacts(execID) + if err != nil { + writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) + return + } + writeResponse(w, http.StatusOK, artifactsToJSON(artifacts)) + } +} diff --git a/services/core/internal/httpserver/job_handlers.go b/services/core/internal/httpserver/job_handlers.go new file mode 100644 index 0000000..495ea48 --- /dev/null +++ b/services/core/internal/httpserver/job_handlers.go @@ -0,0 +1,92 @@ +package httpserver + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + + "github.com/toki/oto/services/core/internal/cicdstate" + otopb "github.com/toki/oto/services/core/oto" +) + +func handleCreateJob(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req struct { + ID string `json:"id"` + Name string `json:"name"` + RunRequest *otopb.RunRequest `json:"run_request"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"}) + return + } + if req.ID == "" || req.Name == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "id and name are required"}) + return + } + job, err := store.CreateJob(req.ID, req.Name, runRequestFromJSON(req.RunRequest)) + if err != nil { + writeResponse(w, http.StatusConflict, map[string]string{"error": err.Error()}) + return + } + writeResponse(w, http.StatusCreated, jobToJSON(job)) + } +} + +func handleGetJob(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + jobID := r.PathValue("id") + if jobID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + jobID = parts[4] + } + } + if jobID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing job id"}) + return + } + job, err := store.GetJob(jobID) + if err != nil { + if errors.Is(err, fmt.Errorf("job not found: %s", jobID)) || err.Error() == "job not found: "+jobID { + writeResponse(w, http.StatusNotFound, map[string]string{"error": "job not found"}) + return + } + writeResponse(w, http.StatusInternalServerError, map[string]string{"error": "internal error"}) + return + } + writeResponse(w, http.StatusOK, jobToJSON(job)) + } +} + +func handleCreateExecution(store *cicdstate.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + jobID := r.PathValue("id") + if jobID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + jobID = parts[4] + } + } + if jobID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing job id"}) + return + } + var req struct { + ID string `json:"id"` + } + _ = json.NewDecoder(r.Body).Decode(&req) + if req.ID == "" { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": "execution id is required"}) + return + } + exec, err := store.CreateExecution(jobID, req.ID) + if err != nil { + writeResponse(w, http.StatusBadRequest, map[string]string{"error": err.Error()}) + return + } + writeResponse(w, http.StatusCreated, execToJSON(exec)) + } +} diff --git a/services/core/internal/httpserver/runner_cicd_handlers.go b/services/core/internal/httpserver/runner_cicd_handlers.go new file mode 100644 index 0000000..0dc81c7 --- /dev/null +++ b/services/core/internal/httpserver/runner_cicd_handlers.go @@ -0,0 +1,494 @@ +package httpserver + +import ( + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "github.com/toki/oto/services/core/internal/cicdstate" + "github.com/toki/oto/services/core/internal/runnerregistry" + otopb "github.com/toki/oto/services/core/oto" +) + +func handleRunnerClaimJob(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req otopb.JobClaimRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ + ErrorMessage: "invalid claim request", + }) + return + } + + if req.RunnerId == "" { + req.RunnerId = runnerID + } else if req.RunnerId != runnerID { + writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ + ErrorMessage: "runner id mismatch between path and body", + }) + return + } + if err := ensureRunnerKnown(registry, runnerID); err != nil { + writeResponse(w, http.StatusNotFound, &otopb.JobClaimResponse{ + RunnerId: runnerID, + ErrorMessage: err.Error(), + }) + return + } + + jobID := strings.TrimSpace(req.GetJobId()) + execID := strings.TrimSpace(req.GetExecutionId()) + if jobID == "" && execID == "" { + var foundJob *cicdstate.Job + for _, j := range store.SnapshotJobs() { + if j.State == cicdstate.StateQueued { + if foundJob == nil || j.CreatedAt.Before(foundJob.CreatedAt) { + copyJ := j + foundJob = ©J + } + } + } + if foundJob == nil { + writeResponse(w, http.StatusOK, &otopb.JobClaimResponse{ + Accepted: false, + RunnerId: runnerID, + }) + return + } + jobID = foundJob.ID + execID = fmt.Sprintf("exec-%d", time.Now().UnixNano()) + } else if jobID == "" || execID == "" { + writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ + RunnerId: runnerID, + ErrorMessage: "job id and execution id are required", + }) + return + } + job, err := store.GetJob(jobID) + if err != nil { + writeResponse(w, http.StatusNotFound, &otopb.JobClaimResponse{ + RunnerId: runnerID, + JobId: jobID, + ErrorMessage: "job not found", + }) + return + } + if job.State != cicdstate.StateQueued { + writeResponse(w, http.StatusConflict, &otopb.JobClaimResponse{ + RunnerId: runnerID, + JobId: jobID, + ErrorMessage: "job is not queued", + State: job.State, + }) + return + } + if job.RunInput == nil { + writeResponse(w, http.StatusConflict, &otopb.JobClaimResponse{ + RunnerId: runnerID, + JobId: jobID, + ErrorMessage: "job has no run request", + State: job.State, + }) + return + } + if err := validateRunInput(job.RunInput); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ + RunnerId: runnerID, + JobId: jobID, + ErrorMessage: err.Error(), + }) + return + } + if _, err := store.CreateExecution(jobID, execID); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + ErrorMessage: err.Error(), + }) + return + } + _ = store.SetExecutionRunnerID(execID, runnerID) + if err := store.TransitionJob(jobID, cicdstate.StateRunning); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + ErrorMessage: err.Error(), + }) + return + } + if err := store.TransitionExecution(execID, cicdstate.StateRunning); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + ErrorMessage: err.Error(), + }) + return + } + + writeResponse(w, http.StatusOK, &otopb.JobClaimResponse{ + Accepted: true, + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + State: cicdstate.StateRunning, + RunRequest: runRequestToProto(job.RunInput, runnerID, jobID, execID), + }) + } +} + +func handleRunnerReportExecution(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req otopb.ExecutionReportRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ + ErrorMessage: "invalid execution report request", + }) + return + } + + if req.RunnerId == "" { + req.RunnerId = runnerID + } else if req.RunnerId != runnerID { + writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ + ErrorMessage: "runner id mismatch between path and body", + }) + return + } + if req.ExecutionId == "" { + req.ExecutionId = execID + } else if req.ExecutionId != execID { + writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ + ErrorMessage: "execution id mismatch between path and body", + }) + return + } + if err := ensureRunnerKnown(registry, runnerID); err != nil { + writeResponse(w, http.StatusNotFound, &otopb.ExecutionReportResponse{ + RunnerId: runnerID, + ExecutionId: execID, + ErrorMessage: err.Error(), + }) + return + } + + exec, err := store.GetExecution(execID) + if err != nil { + writeResponse(w, http.StatusNotFound, &otopb.ExecutionReportResponse{ + RunnerId: runnerID, + ExecutionId: execID, + ErrorMessage: "execution not found", + }) + return + } + jobID := strings.TrimSpace(req.GetJobId()) + if jobID == "" { + jobID = exec.JobID + } else if jobID != exec.JobID { + writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + ErrorMessage: "job id mismatch for execution", + }) + return + } + + targetState := cicdstate.StateFailed + if req.GetSuccess() { + targetState = cicdstate.StateSucceeded + } + if err := store.TransitionExecution(execID, targetState); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + ErrorMessage: err.Error(), + }) + return + } + if err := store.TransitionJob(jobID, targetState); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + ErrorMessage: err.Error(), + }) + return + } + if strings.TrimSpace(req.GetMessage()) != "" { + _ = store.AppendLog(execID, req.GetMessage()) + } + + writeResponse(w, http.StatusOK, &otopb.ExecutionReportResponse{ + Accepted: true, + RunnerId: runnerID, + JobId: jobID, + ExecutionId: execID, + State: targetState, + }) + } +} + +func handleRunnerAppendLog(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req otopb.LogAppendRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.LogAppendResponse{ + ErrorMessage: "invalid log append request", + }) + return + } + if err := normalizeRunnerExecutionRequest(registry, runnerID, execID, req.GetRunnerId(), req.GetExecutionId()); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.LogAppendResponse{ + ErrorMessage: err.Error(), + }) + return + } + line := strings.TrimSpace(req.GetLine()) + if line == "" { + writeResponse(w, http.StatusBadRequest, &otopb.LogAppendResponse{ + ErrorMessage: "line is required", + }) + return + } + if err := store.AppendLog(execID, line); err != nil { + writeResponse(w, http.StatusNotFound, &otopb.LogAppendResponse{ + ErrorMessage: "execution not found", + }) + return + } + writeResponse(w, http.StatusCreated, &otopb.LogAppendResponse{ + Accepted: true, + }) + } +} + +func handleRunnerAppendArtifact(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req otopb.ArtifactReportRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.ArtifactReportResponse{ + ErrorMessage: "invalid artifact report request", + }) + return + } + if err := normalizeRunnerExecutionRequest(registry, runnerID, execID, req.GetRunnerId(), req.GetExecutionId()); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.ArtifactReportResponse{ + ErrorMessage: err.Error(), + }) + return + } + name := strings.TrimSpace(req.GetName()) + path := strings.TrimSpace(req.GetPath()) + if name == "" || path == "" { + writeResponse(w, http.StatusBadRequest, &otopb.ArtifactReportResponse{ + ErrorMessage: "name and path are required", + }) + return + } + if err := store.AppendArtifact(execID, name, path); err != nil { + writeResponse(w, http.StatusNotFound, &otopb.ArtifactReportResponse{ + ErrorMessage: "execution not found", + }) + return + } + writeResponse(w, http.StatusCreated, &otopb.ArtifactReportResponse{ + Accepted: true, + }) + } +} + +func normalizeRunnerExecutionRequest(registry *runnerregistry.Registry, runnerID string, execID string, bodyRunnerID string, bodyExecID string) error { + if bodyRunnerID != "" && bodyRunnerID != runnerID { + return fmt.Errorf("runner id mismatch between path and body") + } + if bodyExecID != "" && bodyExecID != execID { + return fmt.Errorf("execution id mismatch between path and body") + } + return ensureRunnerKnown(registry, runnerID) +} + +func ensureRunnerKnown(registry *runnerregistry.Registry, runnerID string) error { + if strings.TrimSpace(runnerID) == "" { + return fmt.Errorf("missing runner id") + } + if registry == nil { + return nil + } + if _, ok := registry.Snapshot(runnerID); !ok { + return fmt.Errorf("runner not found") + } + return nil +} + +func handleRunnerCancelExecution(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req otopb.CancelRunRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ + Success: false, + ErrorMessage: "invalid cancel request", + }) + return + } + + if err := normalizeRunnerExecutionRequest(registry, runnerID, execID, req.GetRunnerId(), req.GetExecutionId()); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ + Success: false, + ErrorMessage: err.Error(), + }) + return + } + + exec, err := store.GetExecution(execID) + if err != nil { + writeResponse(w, http.StatusNotFound, &otopb.CancelRunResponse{ + Success: false, + ErrorMessage: "execution not found", + }) + return + } + + if exec.RunnerID == "" || exec.RunnerID != runnerID { + writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ + Success: false, + ErrorMessage: "execution owner mismatch", + }) + return + } + + if err := store.CancelJobExecution(exec.JobID, execID); err != nil { + writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ + Success: false, + ErrorMessage: err.Error(), + }) + return + } + + writeResponse(w, http.StatusOK, &otopb.CancelRunResponse{ + Success: true, + }) + } +} + +func handleRunnerStatus(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + record, ok := registry.Snapshot(runnerID) + if !ok { + writeResponse(w, http.StatusNotFound, map[string]interface{}{ + "accepted": false, + "message": "runner not found", + }) + return + } + + var lastExec *cicdstate.Execution + for _, exec := range store.SnapshotExecutions() { + if exec.RunnerID == runnerID { + if lastExec == nil || exec.CreatedAt.After(lastExec.CreatedAt) { + copyExec := exec + lastExec = ©Exec + } + } + } + + currentJobID := "" + currentExecID := "" + if lastExec != nil { + currentJobID = lastExec.JobID + currentExecID = lastExec.ID + } + + writeResponse(w, http.StatusOK, map[string]interface{}{ + "accepted": true, + "runner_id": runnerID, + "status": record.Status, + "current_job_id": currentJobID, + "current_execution_id": currentExecID, + "message": record.FailureReason, + }) + } +} + +func handleRunnerSelfUpdate(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req otopb.SelfUpdateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeResponse(w, http.StatusBadRequest, map[string]interface{}{ + "success": false, + "accepted": false, + "deferred": false, + "error_message": "invalid self-update request", + "message": "invalid self-update request", + }) + return + } + + if req.RunnerId == "" { + req.RunnerId = runnerID + } else if req.RunnerId != runnerID { + writeResponse(w, http.StatusBadRequest, map[string]interface{}{ + "success": false, + "accepted": false, + "deferred": false, + "error_message": "runner id mismatch between path and body", + "message": "runner id mismatch between path and body", + }) + return + } + + if err := ensureRunnerKnown(registry, runnerID); err != nil { + writeResponse(w, http.StatusNotFound, map[string]interface{}{ + "success": false, + "accepted": false, + "deferred": false, + "error_message": err.Error(), + "message": err.Error(), + }) + return + } + + if !strings.HasPrefix(req.GetDownloadUrl(), "https://") { + writeResponse(w, http.StatusBadRequest, map[string]interface{}{ + "success": false, + "accepted": false, + "deferred": false, + "error_message": "download_url must use HTTPS scheme", + "message": "download_url must use HTTPS scheme", + }) + return + } + + hasActiveExecution := false + for _, exec := range store.SnapshotExecutions() { + if exec.RunnerID == runnerID && exec.State == cicdstate.StateRunning { + hasActiveExecution = true + break + } + } + + if hasActiveExecution { + writeResponse(w, http.StatusOK, map[string]interface{}{ + "success": false, + "accepted": false, + "deferred": true, + "target_version": req.GetVersion(), + "message": "deferred: runner is currently running a job", + }) + return + } + + writeResponse(w, http.StatusOK, map[string]interface{}{ + "success": true, + "accepted": true, + "deferred": false, + "target_version": req.GetVersion(), + "message": "self-update request accepted", + }) + } +} diff --git a/services/core/internal/httpserver/runner_handlers.go b/services/core/internal/httpserver/runner_handlers.go new file mode 100644 index 0000000..ac61cf0 --- /dev/null +++ b/services/core/internal/httpserver/runner_handlers.go @@ -0,0 +1,343 @@ +package httpserver + +import ( + "encoding/json" + "fmt" + "net/http" + "net/url" + "os" + "strings" + "time" + + "github.com/toki/oto/services/core/internal/runnerregistry" + otopb "github.com/toki/oto/services/core/oto" +) + +func handleHealthz(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + w.WriteHeader(http.StatusOK) + w.Write([]byte("OK")) +} + +func handleReadyz(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + w.WriteHeader(http.StatusOK) + w.Write([]byte("OK")) +} + +func handleRunnerRegister(registry *runnerregistry.Registry) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + + var request otopb.RegisterRunnerRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + writeRunnerRegisterResponse(w, http.StatusBadRequest, &otopb.RegisterRunnerResponse{ + Accepted: false, + RejectReason: "invalid registration request", + Error: &otopb.ProtocolError{ + Code: "invalid_registration_request", + Message: "invalid registration request", + }, + }) + return + } + + response := registry.Register(&request) + writeRunnerRegisterResponse(w, http.StatusOK, response) + } +} + +func handleRunnerHeartbeat(registry *runnerregistry.Registry) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + + runnerID := r.PathValue("id") + if runnerID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + runnerID = parts[4] + } + } + + if runnerID == "" { + writeHeartbeatResponse(w, http.StatusBadRequest, &otopb.HeartbeatResponse{ + Success: false, + ErrorMessage: "missing runner id", + }) + return + } + + var req otopb.HeartbeatRequest + if r.ContentLength > 0 || r.Body != nil { + if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err.Error() != "EOF" { + writeHeartbeatResponse(w, http.StatusBadRequest, &otopb.HeartbeatResponse{ + Success: false, + ErrorMessage: "invalid heartbeat request", + }) + return + } + } + + if req.RunnerId == "" { + req.RunnerId = runnerID + } else if req.RunnerId != runnerID { + writeHeartbeatResponse(w, http.StatusBadRequest, &otopb.HeartbeatResponse{ + Success: false, + ErrorMessage: "runner id mismatch between path and body", + }) + return + } + + response := registry.Heartbeat(req.RunnerId, req.Status) + if !response.Success { + status := http.StatusNotFound + if response.ErrorMessage != "unknown runner" { + status = http.StatusBadRequest + } + writeHeartbeatResponse(w, status, response) + return + } + + writeHeartbeatResponse(w, http.StatusOK, response) + } +} + +func handleRunnerDisconnect(registry *runnerregistry.Registry) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + + runnerID := r.PathValue("id") + if runnerID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + runnerID = parts[4] + } + } + + if runnerID == "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "success": false, + "error": "missing runner id", + }) + return + } + + success := registry.Disconnect(runnerID) + w.Header().Set("Content-Type", "application/json") + if !success { + w.WriteHeader(http.StatusNotFound) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "success": false, + "error": "runner not found", + }) + return + } + + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "success": true, + }) + } +} + +func handleGetRunner(registry *runnerregistry.Registry) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + + runnerID := r.PathValue("id") + if runnerID == "" { + parts := strings.Split(r.URL.Path, "/") + if len(parts) >= 5 { + runnerID = parts[4] + } + } + + if runnerID == "" { + http.Error(w, "missing runner id", http.StatusBadRequest) + return + } + + record, ok := registry.Snapshot(runnerID) + if !ok { + http.Error(w, "runner not found", http.StatusNotFound) + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "runner_id": record.RunnerID, + "alias": record.Alias, + "protocol_version": record.ProtocolVersion, + "status": record.Status, + "accepted_at": record.AcceptedAt.Format(time.RFC3339), + "first_heartbeat_at": record.FirstHeartbeatAt.Format(time.RFC3339), + "last_heartbeat_at": record.LastHeartbeatAt.Format(time.RFC3339), + "failure_reason": record.FailureReason, + }) + } +} + +func shellEscape(s string) string { + return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" +} + +func handleRunnerBootstrapCommand(registry *runnerregistry.Registry) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + + var request otopb.BootstrapCommandRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "error": "invalid bootstrap command request", + }) + return + } + + runnerID := strings.TrimSpace(request.GetRunnerId()) + token := strings.TrimSpace(request.GetEnrollmentToken()) + if runnerID == "" || token == "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "error": "missing runner id or enrollment token", + }) + return + } + + scheme := "http" + if r.TLS != nil { + scheme = "https" + } + serverURL := scheme + "://" + r.Host + + u, err := url.Parse(serverURL) + if err != nil || u.Host == "" || u.Scheme == "" { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "error": "invalid server URL", + }) + return + } + + // Host validation to prevent command injection and host header spoofing + for _, char := range u.Host { + if !((char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || + char == '.' || char == '-' || char == ':' || char == '[' || char == ']') { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "error": "invalid characters in server Host", + }) + return + } + } + + releaseBaseURL := os.Getenv("OTO_RUNNER_RELEASE_BASE_URL") + if releaseBaseURL == "" { + if scheme == "https" { + releaseBaseURL = serverURL + "/releases" + } else { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "error": "release base URL must use HTTPS (set OTO_RUNNER_RELEASE_BASE_URL environment variable)", + }) + return + } + } else { + if !strings.HasPrefix(releaseBaseURL, "https://") { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "error": "OTO_RUNNER_RELEASE_BASE_URL must use https:// scheme", + }) + return + } + } + + escapedScriptURL := shellEscape(serverURL + "/bootstrap/oto-agent.sh") + escapedServerURL := shellEscape(serverURL) + escapedRunnerID := shellEscape(runnerID) + escapedToken := shellEscape(token) + escapedReleaseURL := shellEscape(releaseBaseURL) + + bootstrapCmd := fmt.Sprintf( + "curl -fsSL %s | bash -s -- --server-url %s --agent-id %s --enrollment-token %s --release-base-url %s", + escapedScriptURL, escapedServerURL, escapedRunnerID, escapedToken, escapedReleaseURL, + ) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(&otopb.BootstrapCommandResponse{ + BootstrapCommand: bootstrapCmd, + }) + } +} + +func handleServeBootstrapScript() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + return + } + paths := []string{ + "../../apps/runner/assets/script/shell/oto_agent_bootstrap.sh", + "../../../../apps/runner/assets/script/shell/oto_agent_bootstrap.sh", + "apps/runner/assets/script/shell/oto_agent_bootstrap.sh", + } + var content []byte + var err error + for _, p := range paths { + content, err = os.ReadFile(p) + if err == nil { + break + } + } + if err != nil { + http.Error(w, "Bootstrap script not found: "+err.Error(), http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/x-sh") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(content) + } +} + +func writeRunnerRegisterResponse(w http.ResponseWriter, status int, response *otopb.RegisterRunnerResponse) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(response) +} + +func writeHeartbeatResponse(w http.ResponseWriter, status int, response *otopb.HeartbeatResponse) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(response) +} diff --git a/services/core/internal/httpserver/server.go b/services/core/internal/httpserver/server.go index fa12f96..837dd5e 100644 --- a/services/core/internal/httpserver/server.go +++ b/services/core/internal/httpserver/server.go @@ -2,19 +2,11 @@ package httpserver import ( "context" - "encoding/json" - "errors" - "fmt" "net" "net/http" - "net/url" - "os" - "strings" - "time" "github.com/toki/oto/services/core/internal/cicdstate" "github.com/toki/oto/services/core/internal/runnerregistry" - otopb "github.com/toki/oto/services/core/oto" ) // Server wraps the HTTP server for the OTO Core service. @@ -60,1028 +52,3 @@ func (s *Server) StartListener(ln net.Listener) error { func (s *Server) Shutdown(ctx context.Context) error { return s.httpServer.Shutdown(ctx) } - -func handleHealthz(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - w.WriteHeader(http.StatusOK) - w.Write([]byte("OK")) -} - -func handleReadyz(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - w.WriteHeader(http.StatusOK) - w.Write([]byte("OK")) -} - -func handleRunnerRegister(registry *runnerregistry.Registry) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - - var request otopb.RegisterRunnerRequest - if err := json.NewDecoder(r.Body).Decode(&request); err != nil { - writeRunnerRegisterResponse(w, http.StatusBadRequest, &otopb.RegisterRunnerResponse{ - Accepted: false, - RejectReason: "invalid registration request", - Error: &otopb.ProtocolError{ - Code: "invalid_registration_request", - Message: "invalid registration request", - }, - }) - return - } - - response := registry.Register(&request) - writeRunnerRegisterResponse(w, http.StatusOK, response) - } -} - -func handleRunnerHeartbeat(registry *runnerregistry.Registry) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - - runnerID := r.PathValue("id") - if runnerID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - runnerID = parts[4] - } - } - - if runnerID == "" { - writeHeartbeatResponse(w, http.StatusBadRequest, &otopb.HeartbeatResponse{ - Success: false, - ErrorMessage: "missing runner id", - }) - return - } - - var req otopb.HeartbeatRequest - // Decode request if body is not empty - if r.ContentLength > 0 || r.Body != nil { - if err := json.NewDecoder(r.Body).Decode(&req); err != nil && err.Error() != "EOF" { - writeHeartbeatResponse(w, http.StatusBadRequest, &otopb.HeartbeatResponse{ - Success: false, - ErrorMessage: "invalid heartbeat request", - }) - return - } - } - - if req.RunnerId == "" { - req.RunnerId = runnerID - } else if req.RunnerId != runnerID { - writeHeartbeatResponse(w, http.StatusBadRequest, &otopb.HeartbeatResponse{ - Success: false, - ErrorMessage: "runner id mismatch between path and body", - }) - return - } - - response := registry.Heartbeat(req.RunnerId, req.Status) - if !response.Success { - status := http.StatusNotFound - if response.ErrorMessage != "unknown runner" { - status = http.StatusBadRequest - } - writeHeartbeatResponse(w, status, response) - return - } - - writeHeartbeatResponse(w, http.StatusOK, response) - } -} - -func handleRunnerDisconnect(registry *runnerregistry.Registry) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - - runnerID := r.PathValue("id") - if runnerID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - runnerID = parts[4] - } - } - - if runnerID == "" { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "success": false, - "error": "missing runner id", - }) - return - } - - success := registry.Disconnect(runnerID) - w.Header().Set("Content-Type", "application/json") - if !success { - w.WriteHeader(http.StatusNotFound) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "success": false, - "error": "runner not found", - }) - return - } - - w.WriteHeader(http.StatusOK) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "success": true, - }) - } -} - -func writeRunnerRegisterResponse(w http.ResponseWriter, status int, response *otopb.RegisterRunnerResponse) { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(status) - _ = json.NewEncoder(w).Encode(response) -} - -func writeHeartbeatResponse(w http.ResponseWriter, status int, response *otopb.HeartbeatResponse) { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(status) - _ = json.NewEncoder(w).Encode(response) -} - -func handleGetRunner(registry *runnerregistry.Registry) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - - runnerID := r.PathValue("id") - if runnerID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - runnerID = parts[4] - } - } - - if runnerID == "" { - http.Error(w, "missing runner id", http.StatusBadRequest) - return - } - - record, ok := registry.Snapshot(runnerID) - if !ok { - http.Error(w, "runner not found", http.StatusNotFound) - return - } - - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusOK) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "runner_id": record.RunnerID, - "alias": record.Alias, - "protocol_version": record.ProtocolVersion, - "status": record.Status, - "accepted_at": record.AcceptedAt.Format(time.RFC3339), - "first_heartbeat_at": record.FirstHeartbeatAt.Format(time.RFC3339), - "last_heartbeat_at": record.LastHeartbeatAt.Format(time.RFC3339), - "failure_reason": record.FailureReason, - }) - } -} - -func shellEscape(s string) string { - return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" -} - -func handleRunnerBootstrapCommand(registry *runnerregistry.Registry) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - - var request otopb.BootstrapCommandRequest - if err := json.NewDecoder(r.Body).Decode(&request); err != nil { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "invalid bootstrap command request", - }) - return - } - - runnerID := strings.TrimSpace(request.GetRunnerId()) - token := strings.TrimSpace(request.GetEnrollmentToken()) - if runnerID == "" || token == "" { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "missing runner id or enrollment token", - }) - return - } - - scheme := "http" - if r.TLS != nil { - scheme = "https" - } - serverURL := scheme + "://" + r.Host - - u, err := url.Parse(serverURL) - if err != nil || u.Host == "" || u.Scheme == "" { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "invalid server URL", - }) - return - } - - // Host validation to prevent command injection and host header spoofing - for _, char := range u.Host { - if !((char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || - char == '.' || char == '-' || char == ':' || char == '[' || char == ']') { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "invalid characters in server Host", - }) - return - } - } - - releaseBaseURL := os.Getenv("OTO_RUNNER_RELEASE_BASE_URL") - if releaseBaseURL == "" { - if scheme == "https" { - releaseBaseURL = serverURL + "/releases" - } else { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "release base URL must use HTTPS (set OTO_RUNNER_RELEASE_BASE_URL environment variable)", - }) - return - } - } else { - if !strings.HasPrefix(releaseBaseURL, "https://") { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusBadRequest) - _ = json.NewEncoder(w).Encode(map[string]interface{}{ - "error": "OTO_RUNNER_RELEASE_BASE_URL must use https:// scheme", - }) - return - } - } - - escapedScriptURL := shellEscape(serverURL + "/bootstrap/oto-agent.sh") - escapedServerURL := shellEscape(serverURL) - escapedRunnerID := shellEscape(runnerID) - escapedToken := shellEscape(token) - escapedReleaseURL := shellEscape(releaseBaseURL) - - bootstrapCmd := fmt.Sprintf( - "curl -fsSL %s | bash -s -- --server-url %s --agent-id %s --enrollment-token %s --release-base-url %s", - escapedScriptURL, escapedServerURL, escapedRunnerID, escapedToken, escapedReleaseURL, - ) - - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusOK) - _ = json.NewEncoder(w).Encode(&otopb.BootstrapCommandResponse{ - BootstrapCommand: bootstrapCmd, - }) - } -} - -func handleServeBootstrapScript() http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) - return - } - paths := []string{ - "../../apps/runner/assets/script/shell/oto_agent_bootstrap.sh", - "../../../../apps/runner/assets/script/shell/oto_agent_bootstrap.sh", - "apps/runner/assets/script/shell/oto_agent_bootstrap.sh", - } - var content []byte - var err error - for _, p := range paths { - content, err = os.ReadFile(p) - if err == nil { - break - } - } - if err != nil { - http.Error(w, "Bootstrap script not found: "+err.Error(), http.StatusNotFound) - return - } - w.Header().Set("Content-Type", "application/x-sh") - w.WriteHeader(http.StatusOK) - _, _ = w.Write(content) - } -} - -func handleCreateJob(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - var req struct { - ID string `json:"id"` - Name string `json:"name"` - RunRequest *otopb.RunRequest `json:"run_request"` - } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"}) - return - } - if req.ID == "" || req.Name == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "id and name are required"}) - return - } - job, err := store.CreateJob(req.ID, req.Name, runRequestFromJSON(req.RunRequest)) - if err != nil { - writeResponse(w, http.StatusConflict, map[string]string{"error": err.Error()}) - return - } - writeResponse(w, http.StatusCreated, jobToJSON(job)) - } -} - -func handleGetJob(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - jobID := r.PathValue("id") - if jobID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - jobID = parts[4] - } - } - if jobID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing job id"}) - return - } - job, err := store.GetJob(jobID) - if err != nil { - if errors.Is(err, fmt.Errorf("job not found: %s", jobID)) || err.Error() == "job not found: "+jobID { - writeResponse(w, http.StatusNotFound, map[string]string{"error": "job not found"}) - return - } - writeResponse(w, http.StatusInternalServerError, map[string]string{"error": "internal error"}) - return - } - writeResponse(w, http.StatusOK, jobToJSON(job)) - } -} - -func handleCreateExecution(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - jobID := r.PathValue("id") - if jobID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - jobID = parts[4] - } - } - if jobID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing job id"}) - return - } - var req struct { - ID string `json:"id"` - } - _ = json.NewDecoder(r.Body).Decode(&req) - if req.ID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "execution id is required"}) - return - } - exec, err := store.CreateExecution(jobID, req.ID) - if err != nil { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": err.Error()}) - return - } - writeResponse(w, http.StatusCreated, execToJSON(exec)) - } -} - -func handleGetExecution(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - execID := r.PathValue("id") - if execID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - execID = parts[4] - } - } - if execID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) - return - } - exec, err := store.GetExecution(execID) - if err != nil { - writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) - return - } - writeResponse(w, http.StatusOK, execToJSON(exec)) - } -} - -func handleAppendLog(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - writeResponse(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"}) - return - } - execID := r.PathValue("id") - if execID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - execID = parts[4] - } - } - if execID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) - return - } - var req struct { - Line string `json:"line"` - } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Line == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "line is required"}) - return - } - if err := store.AppendLog(execID, req.Line); err != nil { - writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) - return - } - writeResponse(w, http.StatusCreated, map[string]string{"status": "accepted"}) - } -} - -func handleGetLogs(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - execID := r.PathValue("id") - if execID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - execID = parts[4] - } - } - if execID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) - return - } - logs, err := store.GetLogs(execID) - if err != nil { - writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) - return - } - writeResponse(w, http.StatusOK, logsToJSON(logs)) - } -} - -func handleAppendArtifact(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - writeResponse(w, http.StatusMethodNotAllowed, map[string]string{"error": "method not allowed"}) - return - } - execID := r.PathValue("id") - if execID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - execID = parts[4] - } - } - if execID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) - return - } - var req struct { - Name string `json:"name"` - Path string `json:"path"` - } - if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Name == "" || req.Path == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "name and path are required"}) - return - } - if err := store.AppendArtifact(execID, req.Name, req.Path); err != nil { - writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) - return - } - writeResponse(w, http.StatusCreated, map[string]string{"status": "accepted"}) - } -} - -func handleGetArtifacts(store *cicdstate.Store) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - execID := r.PathValue("id") - if execID == "" { - parts := strings.Split(r.URL.Path, "/") - if len(parts) >= 5 { - execID = parts[4] - } - } - if execID == "" { - writeResponse(w, http.StatusBadRequest, map[string]string{"error": "missing execution id"}) - return - } - artifacts, err := store.GetArtifacts(execID) - if err != nil { - writeResponse(w, http.StatusNotFound, map[string]string{"error": "execution not found"}) - return - } - writeResponse(w, http.StatusOK, artifactsToJSON(artifacts)) - } -} - -func handleRunnerClaimJob(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - var req otopb.JobClaimRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ - ErrorMessage: "invalid claim request", - }) - return - } - - if req.RunnerId == "" { - req.RunnerId = runnerID - } else if req.RunnerId != runnerID { - writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ - ErrorMessage: "runner id mismatch between path and body", - }) - return - } - if err := ensureRunnerKnown(registry, runnerID); err != nil { - writeResponse(w, http.StatusNotFound, &otopb.JobClaimResponse{ - RunnerId: runnerID, - ErrorMessage: err.Error(), - }) - return - } - - jobID := strings.TrimSpace(req.GetJobId()) - execID := strings.TrimSpace(req.GetExecutionId()) - if jobID == "" && execID == "" { - var foundJob *cicdstate.Job - for _, j := range store.SnapshotJobs() { - if j.State == cicdstate.StateQueued { - if foundJob == nil || j.CreatedAt.Before(foundJob.CreatedAt) { - copyJ := j - foundJob = ©J - } - } - } - if foundJob == nil { - writeResponse(w, http.StatusOK, &otopb.JobClaimResponse{ - Accepted: false, - RunnerId: runnerID, - }) - return - } - jobID = foundJob.ID - execID = fmt.Sprintf("exec-%d", time.Now().UnixNano()) - } else if jobID == "" || execID == "" { - writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ - RunnerId: runnerID, - ErrorMessage: "job id and execution id are required", - }) - return - } - job, err := store.GetJob(jobID) - if err != nil { - writeResponse(w, http.StatusNotFound, &otopb.JobClaimResponse{ - RunnerId: runnerID, - JobId: jobID, - ErrorMessage: "job not found", - }) - return - } - if job.State != cicdstate.StateQueued { - writeResponse(w, http.StatusConflict, &otopb.JobClaimResponse{ - RunnerId: runnerID, - JobId: jobID, - ErrorMessage: "job is not queued", - State: job.State, - }) - return - } - if job.RunInput == nil { - writeResponse(w, http.StatusConflict, &otopb.JobClaimResponse{ - RunnerId: runnerID, - JobId: jobID, - ErrorMessage: "job has no run request", - State: job.State, - }) - return - } - if err := validateRunInput(job.RunInput); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ - RunnerId: runnerID, - JobId: jobID, - ErrorMessage: err.Error(), - }) - return - } - if _, err := store.CreateExecution(jobID, execID); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - ErrorMessage: err.Error(), - }) - return - } - _ = store.SetExecutionRunnerID(execID, runnerID) - if err := store.TransitionJob(jobID, cicdstate.StateRunning); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - ErrorMessage: err.Error(), - }) - return - } - if err := store.TransitionExecution(execID, cicdstate.StateRunning); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.JobClaimResponse{ - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - ErrorMessage: err.Error(), - }) - return - } - - writeResponse(w, http.StatusOK, &otopb.JobClaimResponse{ - Accepted: true, - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - State: cicdstate.StateRunning, - RunRequest: runRequestToProto(job.RunInput, runnerID, jobID, execID), - }) - } -} - -func handleRunnerReportExecution(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - var req otopb.ExecutionReportRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ - ErrorMessage: "invalid execution report request", - }) - return - } - - if req.RunnerId == "" { - req.RunnerId = runnerID - } else if req.RunnerId != runnerID { - writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ - ErrorMessage: "runner id mismatch between path and body", - }) - return - } - if req.ExecutionId == "" { - req.ExecutionId = execID - } else if req.ExecutionId != execID { - writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ - ErrorMessage: "execution id mismatch between path and body", - }) - return - } - if err := ensureRunnerKnown(registry, runnerID); err != nil { - writeResponse(w, http.StatusNotFound, &otopb.ExecutionReportResponse{ - RunnerId: runnerID, - ExecutionId: execID, - ErrorMessage: err.Error(), - }) - return - } - - exec, err := store.GetExecution(execID) - if err != nil { - writeResponse(w, http.StatusNotFound, &otopb.ExecutionReportResponse{ - RunnerId: runnerID, - ExecutionId: execID, - ErrorMessage: "execution not found", - }) - return - } - jobID := strings.TrimSpace(req.GetJobId()) - if jobID == "" { - jobID = exec.JobID - } else if jobID != exec.JobID { - writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - ErrorMessage: "job id mismatch for execution", - }) - return - } - - targetState := cicdstate.StateFailed - if req.GetSuccess() { - targetState = cicdstate.StateSucceeded - } - if err := store.TransitionExecution(execID, targetState); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - ErrorMessage: err.Error(), - }) - return - } - if err := store.TransitionJob(jobID, targetState); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.ExecutionReportResponse{ - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - ErrorMessage: err.Error(), - }) - return - } - if strings.TrimSpace(req.GetMessage()) != "" { - _ = store.AppendLog(execID, req.GetMessage()) - } - - writeResponse(w, http.StatusOK, &otopb.ExecutionReportResponse{ - Accepted: true, - RunnerId: runnerID, - JobId: jobID, - ExecutionId: execID, - State: targetState, - }) - } -} - -func handleRunnerAppendLog(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - var req otopb.LogAppendRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.LogAppendResponse{ - ErrorMessage: "invalid log append request", - }) - return - } - if err := normalizeRunnerExecutionRequest(registry, runnerID, execID, req.GetRunnerId(), req.GetExecutionId()); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.LogAppendResponse{ - ErrorMessage: err.Error(), - }) - return - } - line := strings.TrimSpace(req.GetLine()) - if line == "" { - writeResponse(w, http.StatusBadRequest, &otopb.LogAppendResponse{ - ErrorMessage: "line is required", - }) - return - } - if err := store.AppendLog(execID, line); err != nil { - writeResponse(w, http.StatusNotFound, &otopb.LogAppendResponse{ - ErrorMessage: "execution not found", - }) - return - } - writeResponse(w, http.StatusCreated, &otopb.LogAppendResponse{ - Accepted: true, - }) - } -} - -func handleRunnerAppendArtifact(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - var req otopb.ArtifactReportRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.ArtifactReportResponse{ - ErrorMessage: "invalid artifact report request", - }) - return - } - if err := normalizeRunnerExecutionRequest(registry, runnerID, execID, req.GetRunnerId(), req.GetExecutionId()); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.ArtifactReportResponse{ - ErrorMessage: err.Error(), - }) - return - } - name := strings.TrimSpace(req.GetName()) - path := strings.TrimSpace(req.GetPath()) - if name == "" || path == "" { - writeResponse(w, http.StatusBadRequest, &otopb.ArtifactReportResponse{ - ErrorMessage: "name and path are required", - }) - return - } - if err := store.AppendArtifact(execID, name, path); err != nil { - writeResponse(w, http.StatusNotFound, &otopb.ArtifactReportResponse{ - ErrorMessage: "execution not found", - }) - return - } - writeResponse(w, http.StatusCreated, &otopb.ArtifactReportResponse{ - Accepted: true, - }) - } -} - -func normalizeRunnerExecutionRequest(registry *runnerregistry.Registry, runnerID string, execID string, bodyRunnerID string, bodyExecID string) error { - if bodyRunnerID != "" && bodyRunnerID != runnerID { - return fmt.Errorf("runner id mismatch between path and body") - } - if bodyExecID != "" && bodyExecID != execID { - return fmt.Errorf("execution id mismatch between path and body") - } - return ensureRunnerKnown(registry, runnerID) -} - -func ensureRunnerKnown(registry *runnerregistry.Registry, runnerID string) error { - if strings.TrimSpace(runnerID) == "" { - return fmt.Errorf("missing runner id") - } - if registry == nil { - return nil - } - if _, ok := registry.Snapshot(runnerID); !ok { - return fmt.Errorf("runner not found") - } - return nil -} - -func handleRunnerCancelExecution(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string, execID string) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - var req otopb.CancelRunRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ - Success: false, - ErrorMessage: "invalid cancel request", - }) - return - } - - if err := normalizeRunnerExecutionRequest(registry, runnerID, execID, req.GetRunnerId(), req.GetExecutionId()); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ - Success: false, - ErrorMessage: err.Error(), - }) - return - } - - exec, err := store.GetExecution(execID) - if err != nil { - writeResponse(w, http.StatusNotFound, &otopb.CancelRunResponse{ - Success: false, - ErrorMessage: "execution not found", - }) - return - } - - if exec.RunnerID == "" || exec.RunnerID != runnerID { - writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ - Success: false, - ErrorMessage: "execution owner mismatch", - }) - return - } - - if err := store.CancelJobExecution(exec.JobID, execID); err != nil { - writeResponse(w, http.StatusBadRequest, &otopb.CancelRunResponse{ - Success: false, - ErrorMessage: err.Error(), - }) - return - } - - writeResponse(w, http.StatusOK, &otopb.CancelRunResponse{ - Success: true, - }) - } -} - -func handleRunnerStatus(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - record, ok := registry.Snapshot(runnerID) - if !ok { - writeResponse(w, http.StatusNotFound, map[string]interface{}{ - "accepted": false, - "message": "runner not found", - }) - return - } - - // Find currently running or last execution - var lastExec *cicdstate.Execution - for _, exec := range store.SnapshotExecutions() { - if exec.RunnerID == runnerID { - if lastExec == nil || exec.CreatedAt.After(lastExec.CreatedAt) { - copyExec := exec - lastExec = ©Exec - } - } - } - - currentJobID := "" - currentExecID := "" - if lastExec != nil { - currentJobID = lastExec.JobID - currentExecID = lastExec.ID - } - - writeResponse(w, http.StatusOK, map[string]interface{}{ - "accepted": true, - "runner_id": runnerID, - "status": record.Status, - "current_job_id": currentJobID, - "current_execution_id": currentExecID, - "message": record.FailureReason, - }) - } -} - -func handleRunnerSelfUpdate(store *cicdstate.Store, registry *runnerregistry.Registry, runnerID string) http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - var req otopb.SelfUpdateRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeResponse(w, http.StatusBadRequest, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": "invalid self-update request", - "message": "invalid self-update request", - }) - return - } - - // Normalize runner ID - if req.RunnerId == "" { - req.RunnerId = runnerID - } else if req.RunnerId != runnerID { - writeResponse(w, http.StatusBadRequest, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": "runner id mismatch between path and body", - "message": "runner id mismatch between path and body", - }) - return - } - - if err := ensureRunnerKnown(registry, runnerID); err != nil { - writeResponse(w, http.StatusNotFound, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": err.Error(), - "message": err.Error(), - }) - return - } - - if !strings.HasPrefix(req.GetDownloadUrl(), "https://") { - writeResponse(w, http.StatusBadRequest, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": false, - "error_message": "download_url must use HTTPS scheme", - "message": "download_url must use HTTPS scheme", - }) - return - } - - // Check active executions - hasActiveExecution := false - for _, exec := range store.SnapshotExecutions() { - if exec.RunnerID == runnerID && exec.State == cicdstate.StateRunning { - hasActiveExecution = true - break - } - } - - if hasActiveExecution { - writeResponse(w, http.StatusOK, map[string]interface{}{ - "success": false, - "accepted": false, - "deferred": true, - "target_version": req.GetVersion(), - "message": "deferred: runner is currently running a job", - }) - return - } - - writeResponse(w, http.StatusOK, map[string]interface{}{ - "success": true, - "accepted": true, - "deferred": false, - "target_version": req.GetVersion(), - "message": "self-update request accepted", - }) - } -}