alt/services/worker/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/packages/domain/market"
	"git.toki-labs.com/toki/alt/services/worker/internal/storage"
)

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

func marketHandlers(deps Deps) []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(deps, 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(deps, req)
					},
				)
			},
		},
	}
}

func handleListInstruments(deps Deps, req *altv1.ListInstrumentsRequest) (*altv1.ListInstrumentsResponse, error) {
	marketFilter, hasMarketFilter, err := marketFilterFromProto(req.GetMarket())
	if err != nil {
		return &altv1.ListInstrumentsResponse{Error: marketInvalidRequest(err.Error())}, nil
	}
	if deps.Instruments == nil {
		return &altv1.ListInstrumentsResponse{Error: marketUnavailableError("instrument store is not available")}, nil
	}

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

	insts, err := deps.Instruments.ListInstruments(ctx)
	if err != nil {
		return &altv1.ListInstrumentsResponse{Error: marketBackendErrorInfo(err)}, nil
	}
	filtered := filterInstruments(insts, marketFilter, hasMarketFilter, req.GetProvider())
	return &altv1.ListInstrumentsResponse{Instruments: instrumentsToProto(filtered)}, nil
}

func handleListBars(deps Deps, req *altv1.ListBarsRequest) (*altv1.ListBarsResponse, error) {
	if req.GetInstrumentId() == "" {
		return &altv1.ListBarsResponse{Error: marketInvalidRequest("instrument_id is required")}, nil
	}
	timeframe, err := timeframeFromProto(req.GetTimeframe())
	if err != nil {
		return &altv1.ListBarsResponse{Error: marketInvalidRequest(err.Error())}, 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 deps.Bars == nil {
		return &altv1.ListBarsResponse{Error: marketUnavailableError("bar store is not available")}, nil
	}

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

	from := time.UnixMilli(req.GetFromUnixMs()).UTC()
	to := time.UnixMilli(req.GetToUnixMs()).UTC()
	bars, err := deps.Bars.GetBars(ctx, market.InstrumentID(req.GetInstrumentId()), timeframe, from, to)
	if err != nil {
		return &altv1.ListBarsResponse{Error: marketBackendErrorInfo(err)}, nil
	}
	return &altv1.ListBarsResponse{Bars: barsToProto(bars)}, nil
}

func marketFilterFromProto(m altv1.Market) (market.Market, bool, error) {
	switch m {
	case altv1.Market_MARKET_UNSPECIFIED:
		return "", false, nil
	case altv1.Market_MARKET_KR, altv1.Market_MARKET_US:
		converted, err := marketFromProto(m)
		return converted, true, err
	default:
		return "", false, errors.New("market filter is unsupported")
	}
}

func filterInstruments(insts []market.Instrument, marketFilter market.Market, hasMarketFilter bool, provider string) []market.Instrument {
	if !hasMarketFilter && provider == "" {
		return insts
	}
	out := make([]market.Instrument, 0, len(insts))
	for _, inst := range insts {
		if hasMarketFilter && inst.Market != marketFilter {
			continue
		}
		if provider != "" && inst.ProviderSymbols[provider] == "" {
			continue
		}
		out = append(out, inst)
	}
	return out
}

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

func marketUnavailableError(message string) *altv1.ErrorInfo {
	return errorInfo(marketErrorUnavailable, message)
}

func marketBackendErrorInfo(err error) *altv1.ErrorInfo {
	if err == nil {
		return nil
	}
	switch {
	case errors.Is(err, storage.ErrInstrumentNotFound):
		return errorInfo(marketErrorNotFound, err.Error())
	case errors.Is(err, context.DeadlineExceeded):
		return errorInfo(marketErrorTimeout, err.Error())
	default:
		return errorInfo(marketErrorInternal, err.Error())
	}
}