bc47e426b0
- Add ListPipelines/GetPipeline to CIProvider port with Woodpecker adapter
- Add DNS alias endpoints: GET/POST/DELETE /projects/{id}/domains
- Implement worker executor daemon, build executor, and git operations
- Add build service, worker service, and build audit tracking
- Add worker registry with PostgreSQL adapter and migration
- Add multi-provider code agent interface (Claude Code + OpenCode)
- Add create-and-build combo endpoint
- Update landing-page cookbook to reflect all gaps closed
- Fix tech debt: unified validation, auth scopes, error wrapping, slog patterns
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
456 lines
12 KiB
Go
456 lines
12 KiB
Go
// Package handlers provides HTTP handlers for the rdev API.
|
|
package handlers
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
"github.com/orchard9/rdev/internal/domain"
|
|
"github.com/orchard9/rdev/internal/service"
|
|
"github.com/orchard9/rdev/internal/validate"
|
|
"github.com/orchard9/rdev/pkg/api"
|
|
)
|
|
|
|
// WorkHandler handles work queue endpoints.
|
|
type WorkHandler struct {
|
|
workService *service.WorkService
|
|
}
|
|
|
|
// NewWorkHandler creates a new work handler.
|
|
func NewWorkHandler(workService *service.WorkService) *WorkHandler {
|
|
return &WorkHandler{
|
|
workService: workService,
|
|
}
|
|
}
|
|
|
|
// Mount registers the work queue routes.
|
|
func (h *WorkHandler) Mount(r api.Router) {
|
|
r.Route("/work", func(r chi.Router) {
|
|
// Task submission
|
|
r.Post("/enqueue", h.Enqueue)
|
|
|
|
// Worker endpoints (for workers polling for tasks)
|
|
r.Post("/dequeue", h.Dequeue)
|
|
|
|
// Task management
|
|
r.Get("/{taskId}", h.GetTask)
|
|
r.Get("/{taskId}/status", h.GetStatus)
|
|
r.Post("/{taskId}/complete", h.Complete)
|
|
r.Post("/{taskId}/fail", h.Fail)
|
|
r.Post("/{taskId}/cancel", h.Cancel)
|
|
|
|
// Project-scoped list
|
|
r.Get("/projects/{projectId}", h.ListByProject)
|
|
|
|
// Queue stats
|
|
r.Get("/stats", h.Stats)
|
|
})
|
|
}
|
|
|
|
// EnqueueWorkRequest is the request body for POST /work/enqueue.
|
|
type EnqueueWorkRequest struct {
|
|
ProjectID string `json:"project_id"`
|
|
TaskType string `json:"task_type"`
|
|
Spec map[string]any `json:"task_spec"`
|
|
Priority int `json:"priority,omitempty"`
|
|
CallbackURL string `json:"callback_url,omitempty"`
|
|
MaxRetries int `json:"max_retries,omitempty"`
|
|
}
|
|
|
|
// EnqueueWorkResponse is the response for POST /work/enqueue.
|
|
type EnqueueWorkResponse struct {
|
|
TaskID string `json:"task_id"`
|
|
StatusURL string `json:"status_url"`
|
|
}
|
|
|
|
// Enqueue adds a task to the work queue.
|
|
// POST /work/enqueue
|
|
func (h *WorkHandler) Enqueue(w http.ResponseWriter, r *http.Request) {
|
|
var req EnqueueWorkRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
api.WriteBadRequest(w, r, "invalid request body")
|
|
return
|
|
}
|
|
|
|
// Validate required fields
|
|
if req.ProjectID == "" {
|
|
api.WriteBadRequest(w, r, "project_id is required")
|
|
return
|
|
}
|
|
if req.TaskType == "" {
|
|
api.WriteBadRequest(w, r, "task_type is required")
|
|
return
|
|
}
|
|
|
|
// Validate task type
|
|
taskType := domain.WorkTaskType(req.TaskType)
|
|
if !taskType.IsValid() {
|
|
api.WriteBadRequest(w, r, "task_type must be one of: build, test, deploy, custom")
|
|
return
|
|
}
|
|
|
|
// Validate callback URL if provided
|
|
if err := validate.HTTPURL(req.CallbackURL, "callback_url"); err != nil {
|
|
api.WriteBadRequest(w, r, "callback_url must be a valid HTTP/HTTPS URL")
|
|
return
|
|
}
|
|
|
|
result, err := h.workService.EnqueueTask(r.Context(), service.EnqueueTaskRequest{
|
|
ProjectID: req.ProjectID,
|
|
Type: taskType,
|
|
Spec: req.Spec,
|
|
Priority: req.Priority,
|
|
CallbackURL: req.CallbackURL,
|
|
MaxRetries: req.MaxRetries,
|
|
})
|
|
if err != nil {
|
|
api.WriteInternalError(w, r, "failed to enqueue task")
|
|
return
|
|
}
|
|
|
|
api.WriteCreated(w, r, EnqueueWorkResponse{
|
|
TaskID: result.TaskID,
|
|
StatusURL: result.StatusURL,
|
|
})
|
|
}
|
|
|
|
// DequeueWorkRequest is the request body for POST /work/dequeue.
|
|
type DequeueWorkRequest struct {
|
|
WorkerID string `json:"worker_id"`
|
|
}
|
|
|
|
// DequeueWorkResponse is the response for POST /work/dequeue.
|
|
type DequeueWorkResponse struct {
|
|
Task *WorkTaskDTO `json:"task,omitempty"`
|
|
}
|
|
|
|
// WorkTaskDTO is the data transfer object for work tasks.
|
|
type WorkTaskDTO struct {
|
|
ID string `json:"id"`
|
|
ProjectID string `json:"project_id"`
|
|
Type string `json:"type"`
|
|
Spec map[string]any `json:"spec"`
|
|
Status string `json:"status"`
|
|
Priority int `json:"priority"`
|
|
WorkerID string `json:"worker_id,omitempty"`
|
|
CallbackURL string `json:"callback_url,omitempty"`
|
|
CreatedAt string `json:"created_at"`
|
|
StartedAt string `json:"started_at,omitempty"`
|
|
CompletedAt string `json:"completed_at,omitempty"`
|
|
Result *WorkResultDTO `json:"result,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
RetryCount int `json:"retry_count"`
|
|
MaxRetries int `json:"max_retries"`
|
|
}
|
|
|
|
// WorkResultDTO is the data transfer object for work results.
|
|
type WorkResultDTO struct {
|
|
Output string `json:"output,omitempty"`
|
|
Artifacts map[string]string `json:"artifacts,omitempty"`
|
|
}
|
|
|
|
// toWorkTaskDTO converts a domain.WorkTask to a WorkTaskDTO.
|
|
func toWorkTaskDTO(t *domain.WorkTask) *WorkTaskDTO {
|
|
if t == nil {
|
|
return nil
|
|
}
|
|
dto := &WorkTaskDTO{
|
|
ID: t.ID,
|
|
ProjectID: t.ProjectID,
|
|
Type: string(t.Type),
|
|
Spec: t.Spec,
|
|
Status: string(t.Status),
|
|
Priority: t.Priority,
|
|
WorkerID: t.WorkerID,
|
|
CallbackURL: t.CallbackURL,
|
|
CreatedAt: t.CreatedAt.Format("2006-01-02T15:04:05Z07:00"),
|
|
Error: t.Error,
|
|
RetryCount: t.RetryCount,
|
|
MaxRetries: t.MaxRetries,
|
|
}
|
|
if t.StartedAt != nil {
|
|
dto.StartedAt = t.StartedAt.Format("2006-01-02T15:04:05Z07:00")
|
|
}
|
|
if t.CompletedAt != nil {
|
|
dto.CompletedAt = t.CompletedAt.Format("2006-01-02T15:04:05Z07:00")
|
|
}
|
|
if t.Result != nil {
|
|
dto.Result = &WorkResultDTO{
|
|
Output: t.Result.Output,
|
|
Artifacts: t.Result.Artifacts,
|
|
}
|
|
}
|
|
return dto
|
|
}
|
|
|
|
// Dequeue claims the next available task for a worker.
|
|
// POST /work/dequeue
|
|
func (h *WorkHandler) Dequeue(w http.ResponseWriter, r *http.Request) {
|
|
var req DequeueWorkRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
api.WriteBadRequest(w, r, "invalid request body")
|
|
return
|
|
}
|
|
|
|
if req.WorkerID == "" {
|
|
api.WriteBadRequest(w, r, "worker_id is required")
|
|
return
|
|
}
|
|
|
|
task, err := h.workService.DequeueTask(r.Context(), req.WorkerID)
|
|
if err != nil {
|
|
api.WriteInternalError(w, r, "failed to dequeue task")
|
|
return
|
|
}
|
|
|
|
api.WriteSuccess(w, r, DequeueWorkResponse{
|
|
Task: toWorkTaskDTO(task),
|
|
})
|
|
}
|
|
|
|
// GetTask retrieves a task by ID.
|
|
// GET /work/{taskId}
|
|
func (h *WorkHandler) GetTask(w http.ResponseWriter, r *http.Request) {
|
|
taskID := chi.URLParam(r, "taskId")
|
|
|
|
task, err := h.workService.GetTask(r.Context(), taskID)
|
|
if err != nil {
|
|
if errors.Is(err, domain.ErrWorkTaskNotFound) {
|
|
api.WriteNotFound(w, r, fmt.Sprintf("task not found: %s", taskID))
|
|
return
|
|
}
|
|
api.WriteInternalError(w, r, "failed to get task")
|
|
return
|
|
}
|
|
|
|
api.WriteSuccess(w, r, toWorkTaskDTO(task))
|
|
}
|
|
|
|
// GetStatus returns the status of a task.
|
|
// GET /work/{taskId}/status
|
|
func (h *WorkHandler) GetStatus(w http.ResponseWriter, r *http.Request) {
|
|
taskID := chi.URLParam(r, "taskId")
|
|
|
|
task, err := h.workService.GetTask(r.Context(), taskID)
|
|
if err != nil {
|
|
if errors.Is(err, domain.ErrWorkTaskNotFound) {
|
|
api.WriteNotFound(w, r, fmt.Sprintf("task not found: %s", taskID))
|
|
return
|
|
}
|
|
api.WriteInternalError(w, r, "failed to get task")
|
|
return
|
|
}
|
|
|
|
api.WriteSuccess(w, r, map[string]any{
|
|
"task_id": task.ID,
|
|
"status": string(task.Status),
|
|
"error": task.Error,
|
|
})
|
|
}
|
|
|
|
// CompleteWorkRequest is the request body for POST /work/{taskId}/complete.
|
|
type CompleteWorkRequest struct {
|
|
Output string `json:"output,omitempty"`
|
|
Artifacts map[string]string `json:"artifacts,omitempty"`
|
|
}
|
|
|
|
// Complete marks a task as successfully completed.
|
|
// POST /work/{taskId}/complete
|
|
func (h *WorkHandler) Complete(w http.ResponseWriter, r *http.Request) {
|
|
taskID := chi.URLParam(r, "taskId")
|
|
|
|
var req CompleteWorkRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
api.WriteBadRequest(w, r, "invalid request body")
|
|
return
|
|
}
|
|
|
|
result := &domain.WorkResult{
|
|
Output: req.Output,
|
|
Artifacts: req.Artifacts,
|
|
}
|
|
|
|
if err := h.workService.CompleteTask(r.Context(), taskID, result); err != nil {
|
|
if errors.Is(err, domain.ErrWorkTaskNotFound) {
|
|
api.WriteNotFound(w, r, fmt.Sprintf("task not found: %s", taskID))
|
|
return
|
|
}
|
|
api.WriteInternalError(w, r, "failed to complete task")
|
|
return
|
|
}
|
|
|
|
api.WriteSuccess(w, r, map[string]any{
|
|
"task_id": taskID,
|
|
"status": "completed",
|
|
"message": "task completed successfully",
|
|
})
|
|
}
|
|
|
|
// FailWorkRequest is the request body for POST /work/{taskId}/fail.
|
|
type FailWorkRequest struct {
|
|
Error string `json:"error"`
|
|
}
|
|
|
|
// Fail marks a task as failed.
|
|
// POST /work/{taskId}/fail
|
|
func (h *WorkHandler) Fail(w http.ResponseWriter, r *http.Request) {
|
|
taskID := chi.URLParam(r, "taskId")
|
|
|
|
var req FailWorkRequest
|
|
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
|
api.WriteBadRequest(w, r, "invalid request body")
|
|
return
|
|
}
|
|
|
|
if req.Error == "" {
|
|
api.WriteBadRequest(w, r, "error message is required")
|
|
return
|
|
}
|
|
|
|
if err := h.workService.FailTask(r.Context(), taskID, req.Error); err != nil {
|
|
if errors.Is(err, domain.ErrWorkTaskNotFound) {
|
|
api.WriteNotFound(w, r, fmt.Sprintf("task not found: %s", taskID))
|
|
return
|
|
}
|
|
api.WriteInternalError(w, r, "failed to fail task")
|
|
return
|
|
}
|
|
|
|
api.WriteSuccess(w, r, map[string]any{
|
|
"task_id": taskID,
|
|
"status": "failed",
|
|
"message": "task marked as failed",
|
|
})
|
|
}
|
|
|
|
// Cancel cancels a pending task.
|
|
// POST /work/{taskId}/cancel
|
|
func (h *WorkHandler) Cancel(w http.ResponseWriter, r *http.Request) {
|
|
taskID := chi.URLParam(r, "taskId")
|
|
|
|
if err := h.workService.CancelTask(r.Context(), taskID); err != nil {
|
|
if errors.Is(err, domain.ErrWorkTaskNotFound) {
|
|
api.WriteNotFound(w, r, fmt.Sprintf("task not found: %s", taskID))
|
|
return
|
|
}
|
|
api.WriteBadRequest(w, r, err.Error())
|
|
return
|
|
}
|
|
|
|
api.WriteSuccess(w, r, map[string]any{
|
|
"task_id": taskID,
|
|
"status": "cancelled",
|
|
"message": "task cancelled successfully",
|
|
})
|
|
}
|
|
|
|
// ListByProject returns tasks for a project with pagination.
|
|
// GET /work/projects/{projectId}?status=pending&limit=50&offset=0
|
|
func (h *WorkHandler) ListByProject(w http.ResponseWriter, r *http.Request) {
|
|
projectID := chi.URLParam(r, "projectId")
|
|
|
|
// Parse and validate optional status filter
|
|
var status *domain.WorkTaskStatus
|
|
if s := r.URL.Query().Get("status"); s != "" {
|
|
st := domain.WorkTaskStatus(s)
|
|
if !st.IsValid() {
|
|
api.WriteBadRequest(w, r, "invalid status filter: must be pending, running, completed, failed, or cancelled")
|
|
return
|
|
}
|
|
status = &st
|
|
}
|
|
|
|
// Parse pagination options
|
|
opts := domain.DefaultWorkListOptions()
|
|
if limitStr := r.URL.Query().Get("limit"); limitStr != "" {
|
|
limit, err := strconv.Atoi(limitStr)
|
|
if err != nil {
|
|
api.WriteBadRequest(w, r, "limit must be a valid integer")
|
|
return
|
|
}
|
|
opts.Limit = limit
|
|
}
|
|
if offsetStr := r.URL.Query().Get("offset"); offsetStr != "" {
|
|
offset, err := strconv.Atoi(offsetStr)
|
|
if err != nil {
|
|
api.WriteBadRequest(w, r, "offset must be a valid integer")
|
|
return
|
|
}
|
|
opts.Offset = offset
|
|
}
|
|
|
|
result, err := h.workService.ListByProject(r.Context(), projectID, status, opts)
|
|
if err != nil {
|
|
api.WriteInternalError(w, r, "failed to list tasks")
|
|
return
|
|
}
|
|
|
|
dtos := make([]*WorkTaskDTO, len(result.Tasks))
|
|
for i, t := range result.Tasks {
|
|
dtos[i] = toWorkTaskDTO(t)
|
|
}
|
|
|
|
api.WriteSuccess(w, r, map[string]any{
|
|
"tasks": dtos,
|
|
"total": result.Total,
|
|
"limit": result.Limit,
|
|
"offset": result.Offset,
|
|
})
|
|
}
|
|
|
|
// WorkStatsResponse is the response for GET /work/stats.
|
|
type WorkStatsResponse struct {
|
|
Pending int64 `json:"pending"`
|
|
Running int64 `json:"running"`
|
|
Completed int64 `json:"completed"`
|
|
Failed int64 `json:"failed"`
|
|
Cancelled int64 `json:"cancelled"`
|
|
OldestPending string `json:"oldest_pending,omitempty"`
|
|
}
|
|
|
|
// Stats returns queue statistics.
|
|
// GET /work/stats
|
|
func (h *WorkHandler) Stats(w http.ResponseWriter, r *http.Request) {
|
|
stats, err := h.workService.GetStats(r.Context())
|
|
if err != nil {
|
|
api.WriteInternalError(w, r, "failed to get queue stats")
|
|
return
|
|
}
|
|
|
|
resp := WorkStatsResponse{
|
|
Pending: stats.Pending,
|
|
Running: stats.Running,
|
|
Completed: stats.Completed,
|
|
Failed: stats.Failed,
|
|
Cancelled: stats.Cancelled,
|
|
}
|
|
if stats.OldestPending != nil {
|
|
// Convert to human-readable duration
|
|
resp.OldestPending = formatDuration(*stats.OldestPending)
|
|
}
|
|
|
|
api.WriteSuccess(w, r, resp)
|
|
}
|
|
|
|
// formatDuration formats a duration in a human-readable way.
|
|
func formatDuration(d interface{ Seconds() float64 }) string {
|
|
secs := d.Seconds()
|
|
if secs < 60 {
|
|
return fmt.Sprintf("%.0fs", secs)
|
|
}
|
|
mins := secs / 60
|
|
if mins < 60 {
|
|
return fmt.Sprintf("%.1fm", mins)
|
|
}
|
|
hours := mins / 60
|
|
if hours < 24 {
|
|
return fmt.Sprintf("%.1fh", hours)
|
|
}
|
|
days := hours / 24
|
|
return fmt.Sprintf("%.1fd", days)
|
|
}
|