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

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"
)

// importDateLayout matches the YYYYMMDD date fields the import command carries.
// The API only validates the shape so a malformed date is rejected before a
// worker connection is opened.
const importDateLayout = "20060102"

const (
	marketErrorInvalidRequest = "invalid_request"
	marketErrorUnavailable    = "unavailable"
	marketErrorTimeout        = "timeout"
	marketErrorInternal       = "internal"
)

// marketHandlers returns the API session handlers for market data reads. The
// API validates query shape and forwards the request unchanged to the worker.
func marketHandlers(worker workerclient.WorkerClient) []sessionHandler {
	return []sessionHandler{
		{
			requestType: protoSocket.TypeNameOf(&altv1.ListInstrumentsRequest{}),
			register: func(client *protoSocket.WsClient) {
				protoSocket.AddRequestListenerTyped[*altv1.ListInstrumentsRequest, *altv1.ListInstrumentsResponse](&client.Communicator, func(req *altv1.ListInstrumentsRequest) (*altv1.ListInstrumentsResponse, error) {
					return handleListInstruments(worker, req)
				})
			},
		},
		{
			requestType: protoSocket.TypeNameOf(&altv1.ListBarsRequest{}),
			register: func(client *protoSocket.WsClient) {
				protoSocket.AddRequestListenerTyped[*altv1.ListBarsRequest, *altv1.ListBarsResponse](&client.Communicator, func(req *altv1.ListBarsRequest) (*altv1.ListBarsResponse, error) {
					return handleListBars(worker, req)
				})
			},
		},
		{
			requestType: protoSocket.TypeNameOf(&altv1.ImportDailyBarsRequest{}),
			register: func(client *protoSocket.WsClient) {
				protoSocket.AddRequestListenerTyped[*altv1.ImportDailyBarsRequest, *altv1.ImportDailyBarsResponse](&client.Communicator, func(req *altv1.ImportDailyBarsRequest) (*altv1.ImportDailyBarsResponse, error) {
					return handleImportDailyBars(worker, req)
				})
			},
		},
	}
}

// handleImportDailyBars validates the import command shape, then forwards it to
// the worker which owns provider fetch and persistence. The API stays a thin
// control plane: it rejects malformed commands before opening a worker
// connection and never runs the import itself.
func handleImportDailyBars(worker workerclient.WorkerClient, req *altv1.ImportDailyBarsRequest) (*altv1.ImportDailyBarsResponse, error) {
	if req.GetProvider() == "" {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("provider is required")}, nil
	}
	if req.GetSelectorKind() == "" {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("selector_kind is required")}, nil
	}
	if len(req.GetSymbols()) == 0 {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("symbols is required")}, nil
	}
	if !isValidMarketFilter(req.GetMarket()) {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("market is unsupported")}, nil
	}
	if !isValidVenueFilter(req.GetVenue()) {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest("venue is unsupported")}, nil
	}
	if err := validateImportDateRange(req.GetFromYyyymmdd(), req.GetToYyyymmdd()); err != nil {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
	}
	if worker == nil {
		return &altv1.ImportDailyBarsResponse{Error: marketWorkerUnavailable()}, nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout)
	defer cancel()

	if err := worker.Connect(ctx); err != nil {
		return &altv1.ImportDailyBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
	}

	res, err := worker.ImportDailyBars(ctx, req)
	if err != nil {
		return &altv1.ImportDailyBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
	}
	if res == nil {
		return &altv1.ImportDailyBarsResponse{Error: marketInternalError("worker returned no import daily bars response")}, nil
	}
	return res, nil
}

func handleListInstruments(worker workerclient.WorkerClient, req *altv1.ListInstrumentsRequest) (*altv1.ListInstrumentsResponse, error) {
	if !isValidMarketFilter(req.GetMarket()) {
		return &altv1.ListInstrumentsResponse{Error: marketInvalidRequest("market filter is unsupported")}, nil
	}
	if worker == nil {
		return &altv1.ListInstrumentsResponse{Error: marketWorkerUnavailable()}, nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout)
	defer cancel()

	if err := worker.Connect(ctx); err != nil {
		return &altv1.ListInstrumentsResponse{Error: marketWorkerErrorInfo(err)}, nil
	}

	res, err := worker.ListInstruments(ctx, req)
	if err != nil {
		return &altv1.ListInstrumentsResponse{Error: marketWorkerErrorInfo(err)}, nil
	}
	if res == nil {
		return &altv1.ListInstrumentsResponse{Error: marketInternalError("worker returned no list instruments response")}, nil
	}
	return res, nil
}

func handleListBars(worker workerclient.WorkerClient, req *altv1.ListBarsRequest) (*altv1.ListBarsResponse, error) {
	if req.GetInstrumentId() == "" {
		return &altv1.ListBarsResponse{Error: marketInvalidRequest("instrument_id is required")}, nil
	}
	if !isValidTimeframe(req.GetTimeframe()) {
		return &altv1.ListBarsResponse{Error: marketInvalidRequest("timeframe is unsupported")}, nil
	}
	if req.GetFromUnixMs() == 0 {
		return &altv1.ListBarsResponse{Error: marketInvalidRequest("from_unix_ms is required")}, nil
	}
	if req.GetToUnixMs() == 0 {
		return &altv1.ListBarsResponse{Error: marketInvalidRequest("to_unix_ms is required")}, nil
	}
	if req.GetFromUnixMs() > req.GetToUnixMs() {
		return &altv1.ListBarsResponse{Error: marketInvalidRequest("from_unix_ms cannot be after to_unix_ms")}, nil
	}
	if worker == nil {
		return &altv1.ListBarsResponse{Error: marketWorkerUnavailable()}, nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), workerRequestTimeout)
	defer cancel()

	if err := worker.Connect(ctx); err != nil {
		return &altv1.ListBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
	}

	res, err := worker.ListBars(ctx, req)
	if err != nil {
		return &altv1.ListBarsResponse{Error: marketWorkerErrorInfo(err)}, nil
	}
	if res == nil {
		return &altv1.ListBarsResponse{Error: marketInternalError("worker returned no list bars response")}, nil
	}
	return res, nil
}

func isValidMarketFilter(m altv1.Market) bool {
	switch m {
	case altv1.Market_MARKET_UNSPECIFIED, altv1.Market_MARKET_KR, altv1.Market_MARKET_US:
		return true
	default:
		return false
	}
}

// isValidVenueFilter accepts the unspecified venue (the import selector may not
// pin a venue) and the known venues, rejecting only out-of-range enum values.
func isValidVenueFilter(v altv1.Venue) bool {
	switch v {
	case altv1.Venue_VENUE_UNSPECIFIED, altv1.Venue_VENUE_KRX, altv1.Venue_VENUE_NASDAQ, altv1.Venue_VENUE_NYSE:
		return true
	default:
		return false
	}
}

// validateImportDateRange requires both endpoints of the import date range. The
// live KIS importer requires a bounded from/to range and rejects an inverted
// range, so the API closes those operator-input errors here as typed
// invalid_request instead of letting them surface as worker-internal failures.
func validateImportDateRange(fromValue, toValue string) error {
	if fromValue == "" {
		return errors.New("from_yyyymmdd is required")
	}
	if toValue == "" {
		return errors.New("to_yyyymmdd is required")
	}
	from, err := time.Parse(importDateLayout, fromValue)
	if err != nil {
		return errors.New("from_yyyymmdd must be YYYYMMDD")
	}
	to, err := time.Parse(importDateLayout, toValue)
	if err != nil {
		return errors.New("to_yyyymmdd must be YYYYMMDD")
	}
	if from.After(to) {
		return errors.New("from_yyyymmdd cannot be after to_yyyymmdd")
	}
	return nil
}

func isValidTimeframe(t altv1.Timeframe) bool {
	switch t {
	case altv1.Timeframe_TIMEFRAME_DAILY, altv1.Timeframe_TIMEFRAME_MINUTE_1, altv1.Timeframe_TIMEFRAME_MINUTE_5:
		return true
	default:
		return false
	}
}

func marketInvalidRequest(reason string) *altv1.ErrorInfo {
	return errorInfo(marketErrorInvalidRequest, "invalid market request: "+reason)
}

func marketWorkerUnavailable() *altv1.ErrorInfo {
	return errorInfo(marketErrorUnavailable, "market worker unavailable")
}

func marketInternalError(message string) *altv1.ErrorInfo {
	return errorInfo(marketErrorInternal, message)
}

func marketWorkerErrorInfo(err error) *altv1.ErrorInfo {
	if err == nil {
		return nil
	}
	switch {
	case errors.Is(err, workerclient.ErrUnavailable):
		return errorInfo(marketErrorUnavailable, "market worker unavailable")
	case errors.Is(err, workerclient.ErrTimeout):
		return errorInfo(marketErrorTimeout, "market worker timeout")
	default:
		return errorInfo(marketErrorInternal, err.Error())
	}
}