package socket import ( "context" 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" ) // apiLiveHandlers returns the API session handlers for the live trading surface. // Each handler validates the request shape and forwards it to the worker through // the injected WorkerClient; live execution stays worker-owned. func apiLiveHandlers(worker workerclient.WorkerClient) []sessionHandler { return []sessionHandler{ { requestType: protoSocket.TypeNameOf(&altv1.GetLiveBrokerCapabilitiesRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetLiveBrokerCapabilitiesRequest, *altv1.GetLiveBrokerCapabilitiesResponse](&client.Communicator, func(req *altv1.GetLiveBrokerCapabilitiesRequest) (*altv1.GetLiveBrokerCapabilitiesResponse, error) { return handleGetLiveBrokerCapabilities(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.SubmitLiveOrderRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.SubmitLiveOrderRequest, *altv1.SubmitLiveOrderResponse](&client.Communicator, func(req *altv1.SubmitLiveOrderRequest) (*altv1.SubmitLiveOrderResponse, error) { return handleSubmitLiveOrder(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.CancelLiveOrderRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.CancelLiveOrderRequest, *altv1.CancelLiveOrderResponse](&client.Communicator, func(req *altv1.CancelLiveOrderRequest) (*altv1.CancelLiveOrderResponse, error) { return handleCancelLiveOrder(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.GetLiveOrderRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetLiveOrderRequest, *altv1.GetLiveOrderResponse](&client.Communicator, func(req *altv1.GetLiveOrderRequest) (*altv1.GetLiveOrderResponse, error) { return handleGetLiveOrder(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.GetLiveRiskPolicyRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetLiveRiskPolicyRequest, *altv1.GetLiveRiskPolicyResponse](&client.Communicator, func(req *altv1.GetLiveRiskPolicyRequest) (*altv1.GetLiveRiskPolicyResponse, error) { return handleGetLiveRiskPolicy(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.GetLiveKillSwitchRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetLiveKillSwitchRequest, *altv1.GetLiveKillSwitchResponse](&client.Communicator, func(req *altv1.GetLiveKillSwitchRequest) (*altv1.GetLiveKillSwitchResponse, error) { return handleGetLiveKillSwitch(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.SetLiveKillSwitchRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.SetLiveKillSwitchRequest, *altv1.SetLiveKillSwitchResponse](&client.Communicator, func(req *altv1.SetLiveKillSwitchRequest) (*altv1.SetLiveKillSwitchResponse, error) { return handleSetLiveKillSwitch(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.SyncLiveAccountRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.SyncLiveAccountRequest, *altv1.SyncLiveAccountResponse](&client.Communicator, func(req *altv1.SyncLiveAccountRequest) (*altv1.SyncLiveAccountResponse, error) { return handleSyncLiveAccount(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.GetLiveAccountSnapshotRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.GetLiveAccountSnapshotRequest, *altv1.GetLiveAccountSnapshotResponse](&client.Communicator, func(req *altv1.GetLiveAccountSnapshotRequest) (*altv1.GetLiveAccountSnapshotResponse, error) { return handleGetLiveAccountSnapshot(worker, req) }) }, }, { requestType: protoSocket.TypeNameOf(&altv1.ListLiveAuditEventsRequest{}), register: func(client *protoSocket.WsClient) { protoSocket.AddRequestListenerTyped[*altv1.ListLiveAuditEventsRequest, *altv1.ListLiveAuditEventsResponse](&client.Communicator, func(req *altv1.ListLiveAuditEventsRequest) (*altv1.ListLiveAuditEventsResponse, error) { return handleListLiveAuditEvents(worker, req) }) }, }, } } func handleListLiveAuditEvents(worker workerclient.WorkerClient, req *altv1.ListLiveAuditEventsRequest) (*altv1.ListLiveAuditEventsResponse, error) { if req.GetAccountId() == "" { return &altv1.ListLiveAuditEventsResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "list_live_audit_events: account_id is required"}, }, nil } if worker == nil { return &altv1.ListLiveAuditEventsResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.ListLiveAuditEventsResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.ListLiveAuditEvents(ctx, req) if err != nil { return &altv1.ListLiveAuditEventsResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.ListLiveAuditEventsResponse{Error: internalError("worker returned no list live audit events response")}, nil } return res, nil } func handleGetLiveRiskPolicy(worker workerclient.WorkerClient, req *altv1.GetLiveRiskPolicyRequest) (*altv1.GetLiveRiskPolicyResponse, error) { if req.GetAccountId() == "" { return &altv1.GetLiveRiskPolicyResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "get_live_risk_policy: account_id is required"}, }, nil } if worker == nil { return &altv1.GetLiveRiskPolicyResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetLiveRiskPolicyResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetLiveRiskPolicy(ctx, req) if err != nil { return &altv1.GetLiveRiskPolicyResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetLiveRiskPolicyResponse{Error: internalError("worker returned no live risk policy response")}, nil } return res, nil } func handleGetLiveKillSwitch(worker workerclient.WorkerClient, req *altv1.GetLiveKillSwitchRequest) (*altv1.GetLiveKillSwitchResponse, error) { if req.GetAccountId() == "" { return &altv1.GetLiveKillSwitchResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "get_live_kill_switch: account_id is required"}, }, nil } if worker == nil { return &altv1.GetLiveKillSwitchResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetLiveKillSwitchResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetLiveKillSwitch(ctx, req) if err != nil { return &altv1.GetLiveKillSwitchResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetLiveKillSwitchResponse{Error: internalError("worker returned no live kill switch response")}, nil } return res, nil } func handleSetLiveKillSwitch(worker workerclient.WorkerClient, req *altv1.SetLiveKillSwitchRequest) (*altv1.SetLiveKillSwitchResponse, error) { if req.GetAccountId() == "" { return &altv1.SetLiveKillSwitchResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "set_live_kill_switch: account_id is required"}, }, nil } if worker == nil { return &altv1.SetLiveKillSwitchResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.SetLiveKillSwitchResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.SetLiveKillSwitch(ctx, req) if err != nil { return &altv1.SetLiveKillSwitchResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.SetLiveKillSwitchResponse{Error: internalError("worker returned no set live kill switch response")}, nil } return res, nil } func handleGetLiveBrokerCapabilities(worker workerclient.WorkerClient, req *altv1.GetLiveBrokerCapabilitiesRequest) (*altv1.GetLiveBrokerCapabilitiesResponse, error) { if req.GetAccountId() == "" { return &altv1.GetLiveBrokerCapabilitiesResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "invalid live trading request: account_id is required"}, }, nil } if worker == nil { return &altv1.GetLiveBrokerCapabilitiesResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetLiveBrokerCapabilitiesResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetLiveBrokerCapabilities(ctx, req) if err != nil { return &altv1.GetLiveBrokerCapabilitiesResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetLiveBrokerCapabilitiesResponse{Error: internalError("worker returned no live capability response")}, nil } return res, nil } func handleSubmitLiveOrder(worker workerclient.WorkerClient, req *altv1.SubmitLiveOrderRequest) (*altv1.SubmitLiveOrderResponse, error) { if req.GetAccountId() == "" { return &altv1.SubmitLiveOrderResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "submit_live_order: account_id is required"}, }, nil } if req.GetIntent() == nil { return &altv1.SubmitLiveOrderResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "submit_live_order: intent is required"}, }, nil } if worker == nil { return &altv1.SubmitLiveOrderResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.SubmitLiveOrderResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.SubmitLiveOrder(ctx, req) if err != nil { return &altv1.SubmitLiveOrderResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.SubmitLiveOrderResponse{Error: internalError("worker returned no submit live order response")}, nil } return res, nil } func handleCancelLiveOrder(worker workerclient.WorkerClient, req *altv1.CancelLiveOrderRequest) (*altv1.CancelLiveOrderResponse, error) { if req.GetAccountId() == "" { return &altv1.CancelLiveOrderResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "cancel_live_order: account_id is required"}, }, nil } if req.GetLiveOrderId() == "" { return &altv1.CancelLiveOrderResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "cancel_live_order: live_order_id is required"}, }, nil } if worker == nil { return &altv1.CancelLiveOrderResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.CancelLiveOrderResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.CancelLiveOrder(ctx, req) if err != nil { return &altv1.CancelLiveOrderResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.CancelLiveOrderResponse{Error: internalError("worker returned no cancel live order response")}, nil } return res, nil } func handleSyncLiveAccount(worker workerclient.WorkerClient, req *altv1.SyncLiveAccountRequest) (*altv1.SyncLiveAccountResponse, error) { if req.GetAccountId() == "" { return &altv1.SyncLiveAccountResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "sync_live_account: account_id is required"}, }, nil } if worker == nil { return &altv1.SyncLiveAccountResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.SyncLiveAccountResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.SyncLiveAccount(ctx, req) if err != nil { return &altv1.SyncLiveAccountResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.SyncLiveAccountResponse{Error: internalError("worker returned no sync live account response")}, nil } return res, nil } func handleGetLiveAccountSnapshot(worker workerclient.WorkerClient, req *altv1.GetLiveAccountSnapshotRequest) (*altv1.GetLiveAccountSnapshotResponse, error) { if req.GetAccountId() == "" { return &altv1.GetLiveAccountSnapshotResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "get_live_account_snapshot: account_id is required"}, }, nil } if worker == nil { return &altv1.GetLiveAccountSnapshotResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetLiveAccountSnapshotResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetLiveAccountSnapshot(ctx, req) if err != nil { return &altv1.GetLiveAccountSnapshotResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetLiveAccountSnapshotResponse{Error: internalError("worker returned no get live account snapshot response")}, nil } return res, nil } func handleGetLiveOrder(worker workerclient.WorkerClient, req *altv1.GetLiveOrderRequest) (*altv1.GetLiveOrderResponse, error) { if req.GetAccountId() == "" { return &altv1.GetLiveOrderResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "get_live_order: account_id is required"}, }, nil } if req.GetLiveOrderId() == "" { return &altv1.GetLiveOrderResponse{ Error: &altv1.ErrorInfo{Code: backtestErrorInvalidRequest, Message: "get_live_order: live_order_id is required"}, }, nil } if worker == nil { return &altv1.GetLiveOrderResponse{Error: workerUnavailable()}, nil } ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout) defer cancel() if err := worker.Connect(ctx); err != nil { return &altv1.GetLiveOrderResponse{Error: workerErrorInfo(err)}, nil } res, err := worker.GetLiveOrder(ctx, req) if err != nil { return &altv1.GetLiveOrderResponse{Error: workerErrorInfo(err)}, nil } if res == nil { return &altv1.GetLiveOrderResponse{Error: internalError("worker returned no get live order response")}, nil } return res, nil }