4142c53fa6
Replace SSE /desktop/events with priority AMQP dispatch for monitoring
runs, and add phase1/phase2 infrastructure so desktop workers can lease,
resume, report, skip, and cancel monitoring tasks over the existing
dispatch WebSocket.
- Add monitoring collect outbox worker and phase2 desktop task fields
- Add /desktop/monitoring/tasks/{lease,resume,result,skip,cancel} routes
- Introduce execution-devtools and network-observer on desktop runtime
- Refactor runtime-controller, LoginView, and doubao adapter for the new flow
- Add channelName column to tracking views
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
221 lines
5.9 KiB
Go
221 lines
5.9 KiB
Go
package transport
|
|
|
|
import (
|
|
"errors"
|
|
"io"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/geo-platform/tenant-api/internal/bootstrap"
|
|
"github.com/geo-platform/tenant-api/internal/shared/auth"
|
|
"github.com/geo-platform/tenant-api/internal/shared/response"
|
|
"github.com/geo-platform/tenant-api/internal/tenant/app"
|
|
)
|
|
|
|
type DesktopTaskHandler struct {
|
|
svc *app.DesktopTaskService
|
|
publishSvc *app.PublishJobService
|
|
}
|
|
|
|
func NewDesktopTaskHandler(a *bootstrap.App) *DesktopTaskHandler {
|
|
return &DesktopTaskHandler{
|
|
svc: app.NewDesktopTaskService(
|
|
a.DB,
|
|
a.MonitoringDB,
|
|
a.RabbitMQ,
|
|
a.Logger,
|
|
).WithCache(a.Cache),
|
|
publishSvc: app.NewPublishJobService(
|
|
a.DB,
|
|
a.RabbitMQ,
|
|
a.Redis,
|
|
a.Logger,
|
|
).WithCache(a.Cache),
|
|
}
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) Lease(c *gin.Context) {
|
|
var req app.LeaseDesktopTaskRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
|
|
return
|
|
}
|
|
|
|
var routeTaskID *uuid.UUID
|
|
if rawID := c.Param("id"); rawID != "" {
|
|
parsed, err := uuid.Parse(rawID)
|
|
if err != nil {
|
|
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
|
|
return
|
|
}
|
|
routeTaskID = &parsed
|
|
}
|
|
|
|
data, err := h.svc.Lease(c.Request.Context(), MustDesktopClient(c.Request.Context()), req, routeTaskID)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) ListPublish(c *gin.Context) {
|
|
req := app.ListPublishTasksRequest{
|
|
Page: 1,
|
|
PageSize: 10,
|
|
Title: strings.TrimSpace(c.Query("title")),
|
|
}
|
|
|
|
if rawPage := c.Query("page"); rawPage != "" {
|
|
parsed, err := strconv.Atoi(rawPage)
|
|
if err != nil || parsed <= 0 {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", "page must be a positive integer"))
|
|
return
|
|
}
|
|
req.Page = parsed
|
|
}
|
|
|
|
rawPageSize := c.Query("page_size")
|
|
if rawPageSize == "" {
|
|
rawPageSize = c.Query("limit")
|
|
}
|
|
if rawPageSize != "" {
|
|
parsed, err := strconv.Atoi(rawPageSize)
|
|
if err != nil || parsed <= 0 {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", "page_size must be a positive integer"))
|
|
return
|
|
}
|
|
req.PageSize = parsed
|
|
}
|
|
|
|
data, err := h.svc.ListPublishTasks(c.Request.Context(), MustDesktopClient(c.Request.Context()), req)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) Extend(c *gin.Context) {
|
|
desktopID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
|
|
return
|
|
}
|
|
|
|
var req app.ExtendDesktopTaskRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
|
|
return
|
|
}
|
|
|
|
data, err := h.svc.Extend(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID, req)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) Result(c *gin.Context) {
|
|
desktopID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
|
|
return
|
|
}
|
|
|
|
var req app.CompleteDesktopTaskRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
|
|
return
|
|
}
|
|
|
|
data, err := h.svc.Complete(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID, req)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) RetryPublish(c *gin.Context) {
|
|
desktopID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
|
|
return
|
|
}
|
|
|
|
data, err := h.publishSvc.RetryByDesktopTask(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) Cancel(c *gin.Context) {
|
|
desktopID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
|
|
return
|
|
}
|
|
|
|
var req app.CancelDesktopTaskRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
|
|
return
|
|
}
|
|
|
|
data, err := h.svc.Cancel(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID, req)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) TenantCancel(c *gin.Context) {
|
|
desktopID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
|
|
return
|
|
}
|
|
|
|
var req app.CancelDesktopTaskRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
|
|
return
|
|
}
|
|
|
|
data, err := h.svc.TenantCancel(c.Request.Context(), auth.MustActor(c.Request.Context()), desktopID, req)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|
|
|
|
func (h *DesktopTaskHandler) Reconcile(c *gin.Context) {
|
|
desktopID, err := uuid.Parse(c.Param("id"))
|
|
if err != nil {
|
|
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
|
|
return
|
|
}
|
|
|
|
var req app.ReconcileDesktopTaskRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
|
|
return
|
|
}
|
|
|
|
data, err := h.svc.Reconcile(c.Request.Context(), auth.MustActor(c.Request.Context()), desktopID, req)
|
|
if err != nil {
|
|
response.Error(c, err)
|
|
return
|
|
}
|
|
response.Success(c, data)
|
|
}
|