agentic-coding-workflow/internal/run/handler.go

package run

import (
	"context"
	"encoding/json"
	"net/http"
	"strconv"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/google/uuid"

	ws "github.com/coder/websocket"
	"github.com/coder/websocket/wsjson"

	"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
	httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http"
	"github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware"
	"github.com/yan1h/agent-coding-workflow/internal/workspace"
)

const defaultLogsLimit = 500

// AdminLookup resolves whether a user is an admin（mirrors acp/workspace pattern）。
type AdminLookup interface {
	IsAdmin(ctx context.Context, userID uuid.UUID) (bool, error)
}

// WSConfig controls WebSocket heartbeat and buffering for the log stream.
type WSConfig struct {
	PingInterval time.Duration
	PongTimeout  time.Duration
	SendBuffer   int
}

// Handler 暴露 run profiles 的 HTTP/WS 端点。
type Handler struct {
	svc         *Service
	resolver    middleware.SessionResolver
	adminLookup AdminLookup
	cfg         WSConfig
}

// NewHandler 构造 run.Handler。
func NewHandler(svc *Service, resolver middleware.SessionResolver, al AdminLookup, cfg WSConfig) *Handler {
	return &Handler{svc: svc, resolver: resolver, adminLookup: al, cfg: cfg}
}

// Mount 注册 run profiles 路由。分两组：workspace 作用域（列表/创建）与
// id 作用域（按 id 操作，内部经 profile→workspace 解析做 ACL）。
func (h *Handler) Mount(r chi.Router) {
	r.Route("/api/v1/workspaces/{wsID}/run-profiles", func(r chi.Router) {
		r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{}))
		r.Get("/", h.list)
		r.Post("/", h.create)
	})
	r.Route("/api/v1/run-profiles", func(r chi.Router) {
		r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{}))
		r.Patch("/{id}", h.update)
		r.Delete("/{id}", h.remove)
		r.Post("/{id}/start", h.start)
		r.Post("/{id}/stop", h.stop)
		r.Post("/{id}/restart", h.restart)
		r.Get("/{id}/status", h.status)
		r.Get("/{id}/logs", h.logs)
		r.Get("/{id}/logs/stream", h.logsStream)
	})
}

func (h *Handler) caller(r *http.Request) (workspace.Caller, error) {
	uid, ok := middleware.UserIDFromContext(r.Context())
	if !ok {
		return workspace.Caller{}, errs.New(errs.CodeUnauthorized, "not authenticated")
	}
	admin, err := h.adminLookup.IsAdmin(r.Context(), uid)
	if err != nil {
		return workspace.Caller{}, err
	}
	return workspace.Caller{UserID: uid, IsAdmin: admin}, nil
}

func uuidParam(r *http.Request, key string) (uuid.UUID, error) {
	id, err := uuid.Parse(chi.URLParam(r, key))
	if err != nil {
		return uuid.Nil, errs.New(errs.CodeInvalidInput, "bad "+key)
	}
	return id, nil
}

func (h *Handler) list(w http.ResponseWriter, r *http.Request) {
	c, err := h.caller(r)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	wsID, err := uuidParam(r, "wsID")
	if err != nil {
		writeErr(w, r, err)
		return
	}
	out, err := h.svc.List(r.Context(), c, wsID)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	httpx.WriteJSON(w, http.StatusOK, out)
}

func (h *Handler) create(w http.ResponseWriter, r *http.Request) {
	c, err := h.caller(r)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	wsID, err := uuidParam(r, "wsID")
	if err != nil {
		writeErr(w, r, err)
		return
	}
	var req CreateRunProfileReq
	if derr := json.NewDecoder(r.Body).Decode(&req); derr != nil {
		writeErr(w, r, errs.New(errs.CodeInvalidInput, "bad json"))
		return
	}
	out, err := h.svc.Create(r.Context(), c, wsID, req)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	httpx.WriteJSON(w, http.StatusCreated, out)
}

func (h *Handler) update(w http.ResponseWriter, r *http.Request) {
	c, err := h.caller(r)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	id, err := uuidParam(r, "id")
	if err != nil {
		writeErr(w, r, err)
		return
	}
	var req UpdateRunProfileReq
	if derr := json.NewDecoder(r.Body).Decode(&req); derr != nil {
		writeErr(w, r, errs.New(errs.CodeInvalidInput, "bad json"))
		return
	}
	out, err := h.svc.Update(r.Context(), c, id, req)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	httpx.WriteJSON(w, http.StatusOK, out)
}

func (h *Handler) remove(w http.ResponseWriter, r *http.Request) {
	c, err := h.caller(r)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	id, err := uuidParam(r, "id")
	if err != nil {
		writeErr(w, r, err)
		return
	}
	if err := h.svc.Delete(r.Context(), c, id); err != nil {
		writeErr(w, r, err)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

func (h *Handler) start(w http.ResponseWriter, r *http.Request)   { h.lifecycle(w, r, h.svc.Start) }
func (h *Handler) stop(w http.ResponseWriter, r *http.Request)    { h.lifecycle(w, r, h.svc.Stop) }
func (h *Handler) restart(w http.ResponseWriter, r *http.Request) { h.lifecycle(w, r, h.svc.Restart) }
func (h *Handler) status(w http.ResponseWriter, r *http.Request)  { h.lifecycle(w, r, h.svc.Status) }

// lifecycle 复用 start/stop/restart/status 的公共骨架（鉴权 + 解析 id + 调 svc + 写 RunStatusResp）。
func (h *Handler) lifecycle(w http.ResponseWriter, r *http.Request,
	fn func(context.Context, workspace.Caller, uuid.UUID) (RunStatusResp, error)) {
	c, err := h.caller(r)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	id, err := uuidParam(r, "id")
	if err != nil {
		writeErr(w, r, err)
		return
	}
	out, err := fn(r.Context(), c, id)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	httpx.WriteJSON(w, http.StatusOK, out)
}

func (h *Handler) logs(w http.ResponseWriter, r *http.Request) {
	c, err := h.caller(r)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	id, err := uuidParam(r, "id")
	if err != nil {
		writeErr(w, r, err)
		return
	}
	since := parseInt64(r.URL.Query().Get("since"), 0)
	limit := int(parseInt64(r.URL.Query().Get("limit"), defaultLogsLimit))
	out, err := h.svc.Logs(r.Context(), c, id, since, limit)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	httpx.WriteJSON(w, http.StatusOK, out)
}

// logsStream 升级为 WebSocket，推送内存日志（含 since 续传）与实时新行。只读流，
// 不接收客户端消息。鉴权经 middleware.Auth（WS 用 ?token= 传 token）。
func (h *Handler) logsStream(w http.ResponseWriter, r *http.Request) {
	c, err := h.caller(r)
	if err != nil {
		writeErr(w, r, err)
		return
	}
	id, err := uuidParam(r, "id")
	if err != nil {
		writeErr(w, r, err)
		return
	}
	since := parseInt64(r.URL.Query().Get("since"), 0)

	sub, relay, running, err := h.svc.SubscribeLogs(r.Context(), c, id, c.UserID)
	if err != nil {
		writeErr(w, r, err)
		return
	}

	conn, aerr := ws.Accept(w, r, &ws.AcceptOptions{})
	if aerr != nil {
		return
	}
	defer conn.Close(ws.StatusInternalError, "internal")

	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()

	// 初始 state 帧。
	status := StatusStopped
	if running {
		status = StatusRunning
	}
	if werr := wsjson.Write(ctx, conn, LogEnvelope{Kind: "state", Status: string(status), TS: time.Now()}); werr != nil {
		return
	}

	// profile 未运行：无 live relay，推完即关。
	if !running || relay == nil {
		conn.Close(ws.StatusNormalClosure, "not running")
		return
	}
	defer relay.Unsubscribe(sub.ID)

	// 推送历史缓冲（ID > since）。
	lastSent := since
	for _, e := range relay.since(since, 0) {
		if werr := wsjson.Write(ctx, conn, e); werr != nil {
			return
		}
		lastSent = e.ID
	}

	// 心跳。
	pingTicker := time.NewTicker(h.cfg.PingInterval)
	defer pingTicker.Stop()
	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case <-pingTicker.C:
				pingCtx, pcancel := context.WithTimeout(ctx, h.cfg.PongTimeout)
				perr := conn.Ping(pingCtx)
				pcancel()
				if perr != nil {
					cancel()
					return
				}
			}
		}
	}()

	for {
		select {
		case <-ctx.Done():
			return
		case env, ok := <-sub.Send:
			if !ok {
				conn.Close(ws.StatusNormalClosure, "relay closed")
				return
			}
			// 历史已推到 lastSent，避免重复推送同 ID 的日志行（控制帧 ID=0 放行）。
			if env.Kind == "log" && env.ID <= lastSent && env.ID != 0 {
				continue
			}
			if werr := wsjson.Write(ctx, conn, env); werr != nil {
				return
			}
			if env.ID > lastSent {
				lastSent = env.ID
			}
		}
	}
}

func parseInt64(s string, def int64) int64 {
	if s == "" {
		return def
	}
	n, err := strconv.ParseInt(s, 10, 64)
	if err != nil {
		return def
	}
	return n
}

func writeErr(w http.ResponseWriter, r *http.Request, err error) {
	httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err)
}