package socket import ( "context" "errors" "time" protoSocket "git.toki-labs.com/toki/proto-socket/go" altv1 "git.toki-labs.com/toki/alt/packages/contracts/gen/go/alt/v1" "git.toki-labs.com/toki/alt/services/api/internal/workerclient" ) // workerRequestTimeout bounds a single forwarded worker call. The API is a thin // control plane: it validates the request shape and relays to the worker, so the // timeout only protects the client session from a wedged worker. const workerRequestTimeout = 10 * time.Second const ( backtestErrorInvalidRequest = "invalid_request" backtestErrorUnavailable = "unavailable" backtestErrorTimeout = "timeout" backtestErrorInternal = "internal" ) // backtestHandlers returns the API session handlers for the backtest surface. // Each handler validates the request shape and forwards it to the worker through // the injected Worker client; execution and persistence stay worker-owned. func backtestHandlers(worker workerclient.WorkerClient) []sessionHandler { return []sessionHandler{ { requestType: protoSocket.TypeNameOf(&altv1.StartBacktestRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.StartBacktestRequest, *altv1.StartBacktestResponse](&client.Communicator, func(req *altv1.StartBacktestRequest) (*altv1.StartBacktestResponse, error) { return handleStartBacktest(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.GetBacktestRunRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetBacktestRunRequest, *altv1.GetBacktestRunResponse](&client.Communicator, func(req *altv1.GetBacktestRunRequest) (*altv1.GetBacktestRunResponse, error) { return handleGetBacktestRun(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.ListBacktestRunsRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.ListBacktestRunsRequest, *altv1.ListBacktestRunsResponse](&client.Communicator, func(req *altv1.ListBacktestRunsRequest) (*altv1.ListBacktestRunsResponse, error) { return handleListBacktestRuns(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.GetBacktestRunDetailRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetBacktestRunDetailRequest, *altv1.GetBacktestRunDetailResponse](&client.Communicator, func(req *altv1.GetBacktestRunDetailRequest) (*altv1.GetBacktestRunDetailResponse, error) { return handleGetBacktestRunDetail(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.GetBacktestResultRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetBacktestResultRequest, *altv1.GetBacktestResultResponse](&client.Communicator, func(req *altv1.GetBacktestResultRequest) (*altv1.GetBacktestResultResponse, error) { return handleGetBacktestResult(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.CompareBacktestRunsRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.CompareBacktestRunsRequest, *altv1.CompareBacktestRunsResponse](&client.Communicator, func(req *altv1.CompareBacktestRunsRequest) (*altv1.CompareBacktestRunsResponse, error) { return handleCompareBacktestRuns(worker, req) }) }, }, } } func handleStartBacktest(worker workerclient.WorkerClient, req *altv1.StartBacktestRequest) (*altv1.StartBacktestResponse, error) { if req.GetSpec() == nil { return &altv1.StartBacktestResponse{Error: invalidRequest("spec is required")}, nil } if worker == nil { return &altv1.StartBacktestResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.StartBacktestResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.StartBacktest(ctx, req) if err != nil { return &altv1.StartBacktestResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.StartBacktestResponse{Error: internalError("worker returned no start response")}, nil } return res, nil } func handleGetBacktestRun(worker workerclient.WorkerClient, req *altv1.GetBacktestRunRequest) (*altv1.GetBacktestRunResponse, error) { if req.GetRunId() == "" { return &altv1.GetBacktestRunResponse{Error: invalidRequest("run_id is required")}, nil } if worker == nil { return &altv1.GetBacktestRunResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetBacktestRunResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetBacktestRun(ctx, req) if err != nil { return &altv1.GetBacktestRunResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetBacktestRunResponse{Error: internalError("worker returned no run response")}, nil } return res, nil } func handleListBacktestRuns(worker workerclient.WorkerClient, req *altv1.ListBacktestRunsRequest) (*altv1.ListBacktestRunsResponse, error) { if worker == nil { return &altv1.ListBacktestRunsResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.ListBacktestRunsResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.ListBacktestRuns(ctx, req) if err != nil { return &altv1.ListBacktestRunsResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.ListBacktestRunsResponse{Error: internalError("worker returned no list response")}, nil } return res, nil } func handleGetBacktestRunDetail(worker workerclient.WorkerClient, req *altv1.GetBacktestRunDetailRequest) (*altv1.GetBacktestRunDetailResponse, error) { if req.GetRunId() == "" { return &altv1.GetBacktestRunDetailResponse{Error: invalidRequest("run_id is required")}, nil } if worker == nil { return &altv1.GetBacktestRunDetailResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetBacktestRunDetailResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetBacktestRunDetail(ctx, req) if err != nil { return &altv1.GetBacktestRunDetailResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetBacktestRunDetailResponse{Error: internalError("worker returned no detail response")}, nil } return res, nil } func handleGetBacktestResult(worker workerclient.WorkerClient, req *altv1.GetBacktestResultRequest) (*altv1.GetBacktestResultResponse, error) { if req.GetRunId() == "" { return &altv1.GetBacktestResultResponse{Error: invalidRequest("run_id is required")}, nil } if worker == nil { return &altv1.GetBacktestResultResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetBacktestResultResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetBacktestResult(ctx, req) if err != nil { return &altv1.GetBacktestResultResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetBacktestResultResponse{Error: internalError("worker returned no result response")}, nil } return res, nil } func handleCompareBacktestRuns(worker workerclient.WorkerClient, req *altv1.CompareBacktestRunsRequest) (*altv1.CompareBacktestRunsResponse, error) { for _, id := range req.GetRunIds() { if id == "" { return &altv1.CompareBacktestRunsResponse{Error: invalidRequest("run_ids must not contain empty values")}, nil } } if len(req.GetRunIds()) == 0 { return &altv1.CompareBacktestRunsResponse{}, nil } if worker == nil { return &altv1.CompareBacktestRunsResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.CompareBacktestRunsResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.CompareBacktestRuns(ctx, req) if err != nil { return &altv1.CompareBacktestRunsResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.CompareBacktestRunsResponse{Error: internalError("worker returned no compare response")}, nil } return res, nil } // invalidRequest builds a consistent client-facing validation payload. func invalidRequest(reason string) *altv1.ErrorInfo { return errorInfo(backtestErrorInvalidRequest, "invalid backtest request: "+reason) } func workerUnavailable() *altv1.ErrorInfo { return errorInfo(backtestErrorUnavailable, "backtest worker unavailable") } func internalError(message string) *altv1.ErrorInfo { return errorInfo(backtestErrorInternal, message) } // workerErrorInfo normalises worker transport failures into stable, // client-facing response payloads. Returning ErrorInfo instead of Go errors is // required because proto-socket request listeners do not serialize handler // errors as responses. func workerErrorInfo(err error) *altv1.ErrorInfo { if err == nil { return nil } switch { case errors.Is(err, workerclient.ErrUnavailable): return errorInfo(backtestErrorUnavailable, "backtest worker unavailable") case errors.Is(err, workerclient.ErrTimeout): return errorInfo(backtestErrorTimeout, "backtest worker timeout") default: return errorInfo(backtestErrorInternal, err.Error()) } } func errorInfo(code, message string) *altv1.ErrorInfo { return &altv1.ErrorInfo{Code: code, Message: message} }