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

// importDateLayout matches the YYYYMMDD date fields the import command carries,
// mirroring the job payload date contract so the same dates reach the provider.
const importDateLayout = "20060102"

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)
					},
				)
			},
		},
		{
			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(deps, req)
					},
				)
			},
		},
	}
}

// handleImportDailyBars validates the import command, converts it into the
// importer request, and runs the import. The worker owns provider fetch and
// persistence; counts come back so the operator surface can confirm what landed.
func handleImportDailyBars(deps Deps, 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
	}
	mkt, _, err := marketFilterFromProto(req.GetMarket())
	if err != nil {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
	}
	venue, err := venueFromProto(req.GetVenue())
	if err != nil {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
	}
	from, to, err := parseImportDateRange(req.GetFromYyyymmdd(), req.GetToYyyymmdd())
	if err != nil {
		return &altv1.ImportDailyBarsResponse{Error: marketInvalidRequest(err.Error())}, nil
	}
	if deps.DailyBarImporter == nil {
		return &altv1.ImportDailyBarsResponse{Error: marketUnavailableError("daily bar import is not available")}, nil
	}

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

	result, err := deps.DailyBarImporter.ImportDailyBars(ctx, importer.DailyBarRequest{
		Provider: market.Provider(req.GetProvider()),
		Selector: market.UniverseSelector{
			Kind:    market.UniverseSelectorKind(req.GetSelectorKind()),
			Market:  mkt,
			Venue:   venue,
			Symbols: req.GetSymbols(),
			Name:    req.GetName(),
		},
		From: from,
		To:   to,
	})
	if err != nil {
		return &altv1.ImportDailyBarsResponse{Error: marketBackendErrorInfo(err)}, nil
	}
	return &altv1.ImportDailyBarsResponse{
		Provider:        req.GetProvider(),
		InstrumentCount: int32(result.Instruments),
		BarCount:        int32(result.Bars),
	}, nil
}

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

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())
	}
}