diff --git a/Dockerfile b/Dockerfile index 79e01f52da..088072f55a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -37,6 +37,7 @@ COPY --from=builder /home/node/app/$location/package.json /home/node/app/$locati # Copy Go binary and alias config COPY --from=go-builder /build/streams-adapter /usr/local/bin/streams-adapter COPY --from=builder /home/node/app/packages/streams-adapter/endpoint_aliases.json /home/node/app/endpoint_aliases.json +COPY --from=builder /home/node/app/packages/streams-adapter/adapter_includes.json /home/node/app/adapter_includes.json COPY --from=builder /home/node/app/start-supervisor.sh /usr/local/bin/start-supervisor.sh # Make scripts executable diff --git a/packages/scripts/src/generate-endpoint-aliases/index.ts b/packages/scripts/src/generate-endpoint-aliases/index.ts index 96cf7c7be7..8ee624472e 100644 --- a/packages/scripts/src/generate-endpoint-aliases/index.ts +++ b/packages/scripts/src/generate-endpoint-aliases/index.ts @@ -4,6 +4,7 @@ import path from 'path' import { getWorkspaceAdapters } from '../workspace' const OUTPUT_PATH = 'packages/streams-adapter/endpoint_aliases.json' +const INCLUDES_OUTPUT_PATH = 'packages/streams-adapter/adapter_includes.json' /** * Adapter types that are served by the streams adapter @@ -23,7 +24,18 @@ interface EndpointConfig { } interface AllAdaptersConfig { - adapters: Record }> + adapters: Record< + string, + { + defaultEndpoint?: string + endpoints?: Record + includes?: Record> + } + > +} + +interface AdapterIncludesConfig { + adapters: Record>> } interface LoadResult { @@ -48,6 +60,33 @@ async function loadAdapter(adapterPath: string): Promise { } } +function extractIncludes( + adapter: Adapter, +): Record> | undefined { + const priceAdapter = adapter as Adapter & { + includesMap?: Record> + } + if (!priceAdapter.includesMap) { + return undefined + } + + const includes: Record> = {} + for (const [from, toMap] of Object.entries(priceAdapter.includesMap)) { + if (!toMap || Object.keys(toMap).length === 0) { + continue + } + includes[from] = {} + for (const [to, details] of Object.entries(toMap)) { + if (!details) { + continue + } + includes[from][to] = { inverse: !!details.inverse } + } + } + + return Object.keys(includes).length > 0 ? includes : undefined +} + function extractEndpoints(adapter: Adapter): Record | undefined { const endpoints: Record = {} @@ -91,6 +130,7 @@ async function main(): Promise { result.adapters[adapterKey] = { defaultEndpoint: adapter.defaultEndpoint ?? undefined, endpoints: extractEndpoints(adapter), + includes: extractIncludes(adapter), } } else { skipped.push({ name: meta.descopedName, reason: skipReason || 'unknown' }) @@ -111,6 +151,21 @@ async function main(): Promise { fs.writeFileSync(outPath, JSON.stringify(result, null, 2), 'utf-8') console.log(`Written ${Object.keys(result.adapters).length} EAv3 adapters to ${OUTPUT_PATH}`) + const includesResult: AdapterIncludesConfig = { adapters: {} } + for (const [adapterKey, adapterCfg] of Object.entries(result.adapters)) { + if (adapterCfg.includes) { + includesResult.adapters[adapterKey] = adapterCfg.includes + } + } + + const includesOutPath = path.resolve(process.cwd(), INCLUDES_OUTPUT_PATH) + fs.writeFileSync(includesOutPath, JSON.stringify(includesResult, null, 2), 'utf-8') + console.log( + `Written ${ + Object.keys(includesResult.adapters).length + } EAv3 adapters with includes to ${includesOutPath}`, + ) + if (skipped.length > 0) { console.log(`\nSkipped ${skipped.length} EAv3 adapters:`) for (const { name, reason } of skipped) { diff --git a/packages/streams-adapter/cache/cache.go b/packages/streams-adapter/cache/cache.go index 0ef48ee69a..13ec32e88c 100644 --- a/packages/streams-adapter/cache/cache.go +++ b/packages/streams-adapter/cache/cache.go @@ -11,6 +11,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promauto" types "streams-adapter/common" + "streams-adapter/includes" ) var cacheDataGetCount = promauto.NewCounter( @@ -62,6 +63,7 @@ type Cache struct { items map[string]*types.CacheItem // rawKey → item byTransformedKey map[string]map[string]struct{} // transformedKey → rawKeys (secondary index) pendingObs map[string]*pendingObservation // transformedKey → buffered observation (pre-mapping race) + includes *includes.Index // adapter includes index for inverse flag lookup ttl time.Duration cleanupInterval time.Duration ctx context.Context @@ -69,6 +71,15 @@ type Cache struct { stopOnce sync.Once } +// SetIncludesIndex sets the adapter includes index used to determine the +// inverse flag from the original requested pair. When nil or the pair is not +// present, the cache defaults to not inverting. +func (c *Cache) SetIncludesIndex(idx *includes.Index) { + c.mu.Lock() + defer c.mu.Unlock() + c.includes = idx +} + // New creates a new cache instance func New(cfg Config) *Cache { ctx, cancel := context.WithCancel(context.Background()) @@ -141,7 +152,7 @@ func (c *Cache) SetTransformedKey(rawKey, transformedKey string) { c.removeTransformedKeyMapping(item.TransformedKey, rawKey) } item.TransformedKey = transformedKey - item.RequiresInverse = requiresInverse(item.OriginalRequestData, transformedKey) + item.RequiresInverse = c.requiresInverse(item.OriginalRequestData) item.Status = types.StatusLearned item.Timestamp = time.Now() c.addTransformedKeyMapping(transformedKey, rawKey) @@ -262,7 +273,7 @@ func (c *Cache) RawKeysByTransformed(transformedKey string) ([]string, bool) { return result, true } -func requiresInverse(originalRequestData map[string]interface{}, transformedKey string) bool { +func (c *Cache) requiresInverse(originalRequestData map[string]interface{}) bool { if originalRequestData == nil { return false } @@ -273,14 +284,15 @@ func requiresInverse(originalRequestData map[string]interface{}, transformedKey return false } - transformedParams := parseCacheKey(transformedKey) - transformedBase := strings.ToUpper(transformedParams["base"]) - transformedQuote := strings.ToUpper(transformedParams["quote"]) - if transformedBase == "" || transformedQuote == "" { - return false + // The adapter_includes.json generated from the JS adapter's includes.json is + // the only source of truth for whether an observation must be inverted. + if c.includes != nil { + if inc, ok := c.includes.Lookup(originalBase, originalQuote); ok { + return inc.Inverse + } } - return originalBase == transformedQuote && originalQuote == transformedBase + return false } func getPairValue(data map[string]interface{}, names ...string) string { diff --git a/packages/streams-adapter/cache/cache_test.go b/packages/streams-adapter/cache/cache_test.go index 1bcec4fd0b..efbf7500c0 100644 --- a/packages/streams-adapter/cache/cache_test.go +++ b/packages/streams-adapter/cache/cache_test.go @@ -6,6 +6,7 @@ import ( types "streams-adapter/common" helpers "streams-adapter/helpers" + "streams-adapter/includes" "github.com/goccy/go-json" "github.com/stretchr/testify/assert" @@ -266,6 +267,99 @@ func TestCache_SetObservation_FansOutToSameTransformedKey(t *testing.T) { } } +func TestCache_SetTransformedKey_RequiresInverse_FromIncludes(t *testing.T) { + idx := includes.NewIndex(includes.AdapterIncludes{ + "XAU": {"USD": {Inverse: false}}, + "TRY": {"USD": {Inverse: true}}, + }) + + cases := []struct { + name string + original map[string]interface{} + transformed string + wantInverse bool + }{ + { + name: "metals swapped pair with inverse=false", + original: map[string]interface{}{"base": "XAU", "quote": "USD"}, + transformed: "base=usd:endpoint=forex:quote=xau", + wantInverse: false, + }, + { + name: "fiat swapped pair with inverse=true", + original: map[string]interface{}{"base": "TRY", "quote": "USD"}, + transformed: "base=usd:endpoint=forex:quote=try", + wantInverse: true, + }, + { + name: "direct pair not in includes defaults to false", + original: map[string]interface{}{"base": "USD", "quote": "TRY"}, + transformed: "base=usd:endpoint=forex:quote=try", + wantInverse: false, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c := New(Config{TTL: time.Minute, CleanupInterval: time.Hour}) + defer c.Stop() + c.SetIncludesIndex(idx) + + rawKey, err := helpers.CalculateCacheKey(types.RequestParams{ + "endpoint": "forex", "base": tc.original["base"].(string), "quote": tc.original["quote"].(string), + }) + require.NoError(t, err) + + c.SetNew(rawKey, tc.original, [32]byte{}) + c.SetTransformedKey(rawKey, tc.transformed) + + item := c.Get(rawKey) + require.NotNil(t, item) + require.Equal(t, tc.wantInverse, item.RequiresInverse) + }) + } +} + +func TestCache_SetTransformedKey_RequiresInverse_DefaultFalseWhenNotInIncludes(t *testing.T) { + idx := includes.NewIndex(includes.AdapterIncludes{ + "XAU": {"USD": {Inverse: false}}, + }) + c := New(Config{TTL: time.Minute, CleanupInterval: time.Hour}) + defer c.Stop() + c.SetIncludesIndex(idx) + + rawKey, err := helpers.CalculateCacheKey(types.RequestParams{ + "endpoint": "forex", "base": "EUR", "quote": "USD", + }) + require.NoError(t, err) + transformed := "base=usd:endpoint=forex:quote=eur" + + c.SetNew(rawKey, map[string]interface{}{"base": "EUR", "quote": "USD"}, [32]byte{}) + c.SetTransformedKey(rawKey, transformed) + + item := c.Get(rawKey) + require.NotNil(t, item) + require.False(t, item.RequiresInverse, "pair not in includes must not be inverted") +} + +func TestCache_SetTransformedKey_RequiresInverse_DefaultFalseWithoutIndex(t *testing.T) { + c := New(Config{TTL: time.Minute, CleanupInterval: time.Hour}) + defer c.Stop() + + rawKey, err := helpers.CalculateCacheKey(types.RequestParams{ + "endpoint": "forex", "base": "EUR", "quote": "USD", + }) + require.NoError(t, err) + transformed := "base=usd:endpoint=forex:quote=eur" + + c.SetNew(rawKey, map[string]interface{}{"base": "EUR", "quote": "USD"}, [32]byte{}) + c.SetTransformedKey(rawKey, transformed) + + item := c.Get(rawKey) + require.NotNil(t, item) + require.False(t, item.RequiresInverse, "pair without an includes index must not be inverted") +} + func TestCache_SetTransformedKey_UsesExistingObservationForSharedTransformedKey(t *testing.T) { c := New(Config{TTL: time.Minute, CleanupInterval: time.Hour}) defer c.Stop() diff --git a/packages/streams-adapter/config/config.go b/packages/streams-adapter/config/config.go index f1eface249..a2ad6fcd4d 100644 --- a/packages/streams-adapter/config/config.go +++ b/packages/streams-adapter/config/config.go @@ -33,6 +33,9 @@ type Config struct { // Other configuration LogLevel string AdapterName string + + // Version is populated at runtime by the JS adapter health endpoint. + Version string } // Load reads configuration from environment variables diff --git a/packages/streams-adapter/helpers/inversion.go b/packages/streams-adapter/helpers/inversion.go new file mode 100644 index 0000000000..36c2335e3f --- /dev/null +++ b/packages/streams-adapter/helpers/inversion.go @@ -0,0 +1,82 @@ +package helpers + +import ( + "encoding/json" + "fmt" + "strconv" + + types "streams-adapter/common" +) + +// InvertObservation returns a copy of obs with its numeric result(s) replaced +// by their reciprocal. Used for requests whose original pair is the inverse +// of the transformed key the provider actually publishes. +func InvertObservation(obs *types.Observation) (*types.Observation, error) { + inverted := *obs + + data, err := invertResultInObject(obs.Data) + if err != nil { + return nil, err + } + inverted.Data = data + + if len(obs.Result) > 0 { + result, err := invertRawNumber(obs.Result) + if err != nil { + return nil, err + } + inverted.Result = result + } + + return &inverted, nil +} + +func invertResultInObject(raw json.RawMessage) (json.RawMessage, error) { + var data map[string]interface{} + if err := json.Unmarshal(raw, &data); err != nil { + return nil, fmt.Errorf("unable to invert observation result: %w", err) + } + + result, ok := data["result"] + if !ok { + return nil, fmt.Errorf("unable to invert observation result: missing result") + } + num, err := numberFromInterface(result) + if err != nil { + return nil, err + } + if num == 0 { + return nil, fmt.Errorf("unable to invert observation result: result is zero") + } + + data["result"] = 1 / num + return json.Marshal(data) +} + +func invertRawNumber(raw json.RawMessage) (json.RawMessage, error) { + var num float64 + if err := json.Unmarshal(raw, &num); err != nil { + return nil, fmt.Errorf("unable to invert top-level result: %w", err) + } + if num == 0 { + return nil, fmt.Errorf("unable to invert top-level result: result is zero") + } + return json.Marshal(1 / num) +} + +func numberFromInterface(value interface{}) (float64, error) { + switch v := value.(type) { + case float64: + return v, nil + case json.Number: + return v.Float64() + case string: + num, err := strconv.ParseFloat(v, 64) + if err != nil { + return 0, fmt.Errorf("unable to invert observation result: result is not numeric") + } + return num, nil + default: + return 0, fmt.Errorf("unable to invert observation result: result is not numeric") + } +} diff --git a/packages/streams-adapter/includes/includes.go b/packages/streams-adapter/includes/includes.go new file mode 100644 index 0000000000..d52efa7a7d --- /dev/null +++ b/packages/streams-adapter/includes/includes.go @@ -0,0 +1,82 @@ +package includes + +import ( + "encoding/json" + "fmt" + "os" + "strings" +) + +// IncludeDetails mirrors the include object inside an adapter_includes.json entry. +type IncludeDetails struct { + Inverse bool `json:"inverse"` +} + +// AdapterIncludes maps original pair (from -> to) to include details for one adapter. +type AdapterIncludes map[string]map[string]IncludeDetails + +// Config mirrors the top-level structure of adapter_includes.json. +type Config struct { + Adapters map[string]AdapterIncludes `json:"adapters"` +} + +// Index provides fast lookup of the include details for a single adapter's pairs. +type Index struct { + entries map[string]map[string]IncludeDetails +} + +// Load reads adapter_includes.json and returns the index for the named adapter. +func Load(path, adapterName string) (*Index, error) { + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("open adapter includes config %q: %w", path, err) + } + defer f.Close() + + var cfg Config + if err := json.NewDecoder(f).Decode(&cfg); err != nil { + return nil, fmt.Errorf("decode adapter includes config %q: %w", path, err) + } + + adapterIncludes, ok := cfg.Adapters[adapterName] + if !ok { + return nil, fmt.Errorf("adapter %q not found in adapter includes config %q", adapterName, path) + } + + return NewIndex(adapterIncludes), nil +} + +// NewIndex builds an index from a parsed adapter includes map. +func NewIndex(adapterIncludes AdapterIncludes) *Index { + idx := &Index{entries: make(map[string]map[string]IncludeDetails)} + for from, toMap := range adapterIncludes { + upperFrom := strings.ToUpper(from) + if upperFrom == "" { + continue + } + if idx.entries[upperFrom] == nil { + idx.entries[upperFrom] = make(map[string]IncludeDetails) + } + for to, details := range toMap { + upperTo := strings.ToUpper(to) + if upperTo == "" { + continue + } + idx.entries[upperFrom][upperTo] = details + } + } + return idx +} + +// Lookup returns the include details for a requested pair (case-insensitive). +func (idx *Index) Lookup(from, to string) (IncludeDetails, bool) { + if idx == nil || idx.entries == nil { + return IncludeDetails{}, false + } + m, ok := idx.entries[strings.ToUpper(from)] + if !ok { + return IncludeDetails{}, false + } + d, ok := m[strings.ToUpper(to)] + return d, ok +} diff --git a/packages/streams-adapter/includes/includes_test.go b/packages/streams-adapter/includes/includes_test.go new file mode 100644 index 0000000000..3c53ca373d --- /dev/null +++ b/packages/streams-adapter/includes/includes_test.go @@ -0,0 +1,63 @@ +package includes + +import ( + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNewIndex_Lookup(t *testing.T) { + idx := NewIndex(AdapterIncludes{ + "XAU": {"USD": {Inverse: false}}, + "TRY": {"USD": {Inverse: true}}, + "USD": {"TRY": {Inverse: false}}, + }) + + cases := []struct { + from string + to string + inverse bool + found bool + }{ + {"XAU", "USD", false, true}, + {"xau", "usd", false, true}, // case insensitive + {"TRY", "USD", true, true}, + {"USD", "TRY", false, true}, + {"EUR", "USD", false, false}, + } + + for _, c := range cases { + d, ok := idx.Lookup(c.from, c.to) + require.Equal(t, c.found, ok, "lookup %s/%s", c.from, c.to) + if c.found { + require.Equal(t, c.inverse, d.Inverse, "inverse for %s/%s", c.from, c.to) + } + } +} + +func TestLoad(t *testing.T) { + idx, err := Load(filepath.Join("testdata", "adapter_includes.json"), "test") + require.NoError(t, err) + + xau, ok := idx.Lookup("XAU", "USD") + require.True(t, ok, "XAU/USD must exist") + require.False(t, xau.Inverse) + + try, ok := idx.Lookup("TRY", "USD") + require.True(t, ok, "TRY/USD must exist") + require.True(t, try.Inverse) + + _, ok = idx.Lookup("EUR", "USD") + require.False(t, ok, "EUR/USD must not exist") +} + +func TestLoad_AdapterNotFound(t *testing.T) { + _, err := Load(filepath.Join("testdata", "adapter_includes.json"), "missing") + require.Error(t, err) +} + +func TestLoad_FileNotFound(t *testing.T) { + _, err := Load("does_not_exist.json", "test") + require.Error(t, err) +} diff --git a/packages/streams-adapter/includes/testdata/adapter_includes.json b/packages/streams-adapter/includes/testdata/adapter_includes.json new file mode 100644 index 0000000000..7e0ca54605 --- /dev/null +++ b/packages/streams-adapter/includes/testdata/adapter_includes.json @@ -0,0 +1,43 @@ +{ + "adapters": { + "test": { + "XAU": { + "USD": { "inverse": false } + }, + "TRY": { + "USD": { "inverse": true } + } + }, + "tp": { + "XAU": { + "USD": { "inverse": false } + }, + "XAG": { + "USD": { "inverse": false } + }, + "XPD": { + "USD": { "inverse": false } + }, + "XPT": { + "USD": { "inverse": false } + }, + "TRY": { + "USD": { "inverse": true } + } + }, + "icap": { + "XAU": { + "USD": { "inverse": false } + }, + "XAG": { + "USD": { "inverse": false } + }, + "XPT": { + "USD": { "inverse": false } + }, + "TRY": { + "USD": { "inverse": true } + } + } + } +} diff --git a/packages/streams-adapter/main.go b/packages/streams-adapter/main.go index ed71b28d65..ef3f5887ed 100644 --- a/packages/streams-adapter/main.go +++ b/packages/streams-adapter/main.go @@ -9,6 +9,7 @@ import ( "os" "time" + "github.com/goccy/go-json" "github.com/soheilhy/cmux" "google.golang.org/grpc" @@ -16,13 +17,15 @@ import ( "streams-adapter/config" pb "streams-adapter/gen/streams/v1" "streams-adapter/helpers" + "streams-adapter/includes" "streams-adapter/redcon" "streams-adapter/server" "streams-adapter/transmitter" ) -// waitForEAServer waits for the EA server to be ready before proceeding -func waitForEAServer(cfg *config.Config, logger *slog.Logger) { +// waitForEAServer waits for the EA server to be ready before proceeding. +// It returns the adapter version reported by the EA health endpoint. +func waitForEAServer(cfg *config.Config, logger *slog.Logger) string { eaURL := fmt.Sprintf("http://%s:%s%s/health", cfg.EAHost, cfg.EAPort, cfg.EABaseUrl) maxWaitTime := 60 * time.Second checkInterval := 500 * time.Millisecond @@ -46,9 +49,15 @@ func waitForEAServer(cfg *config.Config, logger *slog.Logger) { // Try to connect to the EA server health endpoint resp, err := client.Get(eaURL) if err == nil && resp.StatusCode == http.StatusOK { + var health struct { + Version string `json:"version"` + } + if err := json.NewDecoder(resp.Body).Decode(&health); err != nil { + logger.Warn("failed to decode EA health response", "error", err) + } resp.Body.Close() - logger.Info("EA server is ready", "elapsed", time.Since(startTime)) - return + logger.Info("EA server is ready", "elapsed", time.Since(startTime), "version", health.Version) + return health.Version } if resp != nil { resp.Body.Close() @@ -72,11 +81,18 @@ func main() { }) defer appCache.Stop() + idx, err := includes.Load("adapter_includes.json", cfg.AdapterName) + if err != nil { + log.Fatalf("Failed to load adapter includes index: path=%s adapter=%s error=%v", + "adapter_includes.json", cfg.AdapterName, err) + } + appCache.SetIncludesIndex(idx) + // Create the gRPC publisher (fanout to subscribed clients) pub := transmitter.NewPublisher() - // Wait for EA server to be ready before starting - waitForEAServer(cfg, logger) + // Wait for EA server to be ready before starting and capture its version + cfg.Version = waitForEAServer(cfg, logger) // Initialize HTTP server httpServer := server.New(cfg, appCache, logger) diff --git a/packages/streams-adapter/redcon/redcon.go b/packages/streams-adapter/redcon/redcon.go index b34382e132..3a2f5f202c 100644 --- a/packages/streams-adapter/redcon/redcon.go +++ b/packages/streams-adapter/redcon/redcon.go @@ -243,9 +243,21 @@ func (s *RedconServer) handleEval(conn redcon.Conn, cmd redcon.Command) { if s.publisher != nil { if rawKeys, ok := s.cache.RawKeysByTransformed(transformedKey); ok { for _, rawKey := range rawKeys { - if payloadHash, ok := s.cache.PayloadHashByRawKey(rawKey); ok { - s.publisher.Publish(payloadHash, obs, ts) + payloadHash, ok := s.cache.PayloadHashByRawKey(rawKey) + if !ok { + continue } + out := obs + if item := s.cache.Get(rawKey); item != nil && item.RequiresInverse && obs.Success { + inverted, err := helpers.InvertObservation(obs) + if err != nil { + s.logger.Error("failed to invert observation for inverse subscriber", + "rawKey", rawKey, "error", err) + continue // do not publish a direct price to an inverse subscriber + } + out = inverted + } + s.publisher.Publish(payloadHash, out, ts) } } } diff --git a/packages/streams-adapter/server/server.go b/packages/streams-adapter/server/server.go index add1219456..07055e7150 100644 --- a/packages/streams-adapter/server/server.go +++ b/packages/streams-adapter/server/server.go @@ -9,7 +9,6 @@ import ( "log/slog" "net" "net/http" - "strconv" "sync" "time" @@ -251,10 +250,14 @@ func (s *Server) Stop() error { // healthHandler handles health check requests func (s *Server) healthHandler(c *gin.Context) { - c.JSON(http.StatusOK, gin.H{ + response := gin.H{ "status": "healthy", "time": time.Now().UTC(), - }) + } + if s.config.Version != "" { + response["version"] = s.config.Version + } + c.JSON(http.StatusOK, response) } // cacheHandler returns all current cache entries for debugging. @@ -404,7 +407,7 @@ func respondWithObservation(c *gin.Context, item *types.CacheItem) { obs := item.Observation if obs.Success { if item.RequiresInverse { - inverted, err := invertObservation(obs) + inverted, err := helpers.InvertObservation(obs) if err != nil { c.JSON(http.StatusBadGateway, ObservationErrorResponse{ ErrorMessage: err.Error(), @@ -425,76 +428,6 @@ func respondWithObservation(c *gin.Context, item *types.CacheItem) { }) } -func invertObservation(obs *types.Observation) (*types.Observation, error) { - inverted := *obs - - data, err := invertResultInObject(obs.Data) - if err != nil { - return nil, err - } - inverted.Data = data - - if len(obs.Result) > 0 { - result, err := invertRawNumber(obs.Result) - if err != nil { - return nil, err - } - inverted.Result = result - } - - return &inverted, nil -} - -func invertResultInObject(raw json.RawMessage) (json.RawMessage, error) { - var data map[string]interface{} - if err := json.Unmarshal(raw, &data); err != nil { - return nil, fmt.Errorf("unable to invert observation result: %w", err) - } - - result, ok := data["result"] - if !ok { - return nil, fmt.Errorf("unable to invert observation result: missing result") - } - num, err := numberFromInterface(result) - if err != nil { - return nil, err - } - if num == 0 { - return nil, fmt.Errorf("unable to invert observation result: result is zero") - } - - data["result"] = 1 / num - return json.Marshal(data) -} - -func invertRawNumber(raw json.RawMessage) (json.RawMessage, error) { - var num float64 - if err := json.Unmarshal(raw, &num); err != nil { - return nil, fmt.Errorf("unable to invert top-level result: %w", err) - } - if num == 0 { - return nil, fmt.Errorf("unable to invert top-level result: result is zero") - } - return json.Marshal(1 / num) -} - -func numberFromInterface(value interface{}) (float64, error) { - switch v := value.(type) { - case float64: - return v, nil - case json.Number: - return v.Float64() - case string: - num, err := strconv.ParseFloat(v, 64) - if err != nil { - return 0, fmt.Errorf("unable to invert observation result: result is not numeric") - } - return num, nil - default: - return 0, fmt.Errorf("unable to invert observation result: result is not numeric") - } -} - // postToAdapter marshals data as {"data": ...} and POSTs it to the JS adapter. // The caller is responsible for closing resp.Body. func (s *Server) postToAdapter(data interface{}) (*http.Response, error) { diff --git a/packages/streams-adapter/server/server_test.go b/packages/streams-adapter/server/server_test.go index 027b91b107..b68f8d4735 100644 --- a/packages/streams-adapter/server/server_test.go +++ b/packages/streams-adapter/server/server_test.go @@ -14,6 +14,7 @@ import ( types "streams-adapter/common" config "streams-adapter/config" "streams-adapter/helpers" + "streams-adapter/includes" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" @@ -71,6 +72,7 @@ func TestMain(m *testing.M) { CacheCleanupIntervalSeconds: 60, LogLevel: "info", AdapterName: "test", + Version: "1.2.3", } testCache = cache.New(cache.Config{ @@ -78,6 +80,12 @@ func TestMain(m *testing.M) { CleanupInterval: 10 * time.Minute, }) + idx, err := includes.Load("../includes/testdata/adapter_includes.json", "test") + if err != nil { + panic(err) + } + testCache.SetIncludesIndex(idx) + testSrv = New(cfg, testCache, slog.Default()) os.Exit(m.Run()) @@ -105,6 +113,7 @@ func TestHealthHandler(t *testing.T) { var body map[string]interface{} require.NoError(t, json.Unmarshal(w.Body.Bytes(), &body)) require.Equal(t, "healthy", body["status"]) + require.Equal(t, "1.2.3", body["version"]) } func TestAdapterHandler_BadRequest(t *testing.T) { @@ -187,6 +196,56 @@ func TestAdapterHandler_CacheHit_InvertsTransformedPairObservation(t *testing.T) require.JSONEq(t, `{"result":46.4407}`, string(item.Observation.Data)) } +func TestAdapterHandler_CacheHit_DoesNotInvertMetalsPairWithIncludes(t *testing.T) { + const goldSpot = 3401.25 + + rawParams := types.RequestParams{"endpoint": "forex", "from": "XAU", "to": "USD"} + rawKey, err := helpers.CalculateCacheKey(rawParams) + require.NoError(t, err) + transformedKey, err := helpers.CalculateCacheKey(types.RequestParams{ + "endpoint": "forex", + "base": "USD", + "quote": "XAU", + }) + require.NoError(t, err) + obs := &types.Observation{ + Data: json.RawMessage(`{"result":3401.25}`), + Result: json.RawMessage(`3401.25`), + Success: true, + } + testCache.SetNew(rawKey, map[string]interface{}{ + "endpoint": "forex", + "from": "XAU", + "to": "USD", + }, [32]byte{}) + testCache.SetTransformedKey(rawKey, transformedKey) + testCache.SetObservation(transformedKey, obs, time.Now(), "test-forex-"+`{"base":"usd","quote":"xau"}`) + + body := `{"data":{"endpoint":"forex","from":"XAU","to":"USD"}}` + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + testSrv.router.ServeHTTP(w, req) + + require.Equal(t, http.StatusOK, w.Code, "body: %s", w.Body.String()) + + var resp types.Observation + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + + var data map[string]float64 + require.NoError(t, json.Unmarshal(resp.Data, &data)) + require.InDelta(t, goldSpot, data["result"], 1e-9, + "XAU/USD must not be inverted when includes declares inverse=false") + + var result float64 + require.NoError(t, json.Unmarshal(resp.Result, &result)) + require.InDelta(t, goldSpot, result, 1e-9) + + item := testCache.Get(rawKey) + require.NotNil(t, item) + require.False(t, item.RequiresInverse, "RequiresInverse must be false for XAU/USD per includes") +} + func TestAdapterHandler_CacheHit_DoesNotInvertDirectTransformedPairObservation(t *testing.T) { params := types.RequestParams{"endpoint": "forex", "base": "USD", "quote": "TRY"} obs := &types.Observation{