alt/services/api/internal/socket/live.go

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
}