fix datastream redis keys doubling the schematic prefix

lib/schematic/datastream/resource_cache.rb

@@ -3,8 +3,6 @@
 module Schematic
   module DataStream
     class ResourceCache
-      CACHE_KEY_PREFIX = "schematic"
-
       def initialize(primary_cache:, lookup_cache:, key_prefix:, get_id:, get_keys:, cache_version: nil)
         @primary_cache = primary_cache
         @lookup_cache = lookup_cache
@@ -65,11 +63,11 @@ def delete_entity(entity)
       private
 
       def build_id_key(id)
-        "#{CACHE_KEY_PREFIX}:#{@key_prefix}:#{@cache_version}:#{id}"
+        "#{@key_prefix}:#{@cache_version}:#{id}"
       end
 
       def build_lookup_key(key_name, key_value)
-        "#{CACHE_KEY_PREFIX}:#{@key_prefix}:#{@cache_version}:#{key_name.to_s.downcase}:#{key_value}"
+        "#{@key_prefix}:#{@cache_version}:#{key_name.to_s.downcase}:#{key_value.to_s.downcase}"
       end
     end
   end

test/custom.test.rb

@@ -1002,6 +1002,33 @@ def build_client(api_key: "test_api_key")
     assert_nil @cache.get_by_id("comp_1")
     assert_nil @cache.get_by_keys({ "org_id" => "abc" })
   end
+
+  it "writes replicator-layout keys through a prefixing Redis provider" do
+    # The replicator writes schematic:company:<version>:<id> and
+    # schematic:company:<version>:<key>:<value> (key and value lowercased).
+    # RedisCacheProvider supplies the "schematic:" prefix, so ResourceCache
+    # must not add another one or lookups against a replicator miss.
+    store = {}
+    redis = Object.new
+    redis.define_singleton_method(:get) { |key| store[key] }
+    redis.define_singleton_method(:set) { |key, value| store[key] = value }
+    redis.define_singleton_method(:setex) { |key, _ttl, value| store[key] = value }
+    redis.define_singleton_method(:del) { |*keys| keys.flatten.each { |k| store.delete(k) } }
+    provider = Schematic::RedisCacheProvider.new(client: redis, ttl: 300, key_prefix: "schematic:")
+
+    cache = Schematic::DataStream::ResourceCache.new(
+      primary_cache: provider,
+      lookup_cache: provider,
+      key_prefix: "company",
+      get_id: ->(c) { c[:id] },
+      get_keys: ->(c) { c[:keys] || {} },
+      cache_version: "v1"
+    )
+    cache.cache_entity({ id: "comp_1", keys: { "externalId" => "Acme-Co" } })
+
+    assert_equal ["schematic:company:v1:comp_1", "schematic:company:v1:externalid:acme-co"], store.keys.sort
+    assert_equal "comp_1", cache.get_by_keys({ "externalId" => "ACME-CO" })[:id]
+  end
 end
 
 # =============================================================================

testapp/app.rb

@@ -114,7 +114,11 @@ def handle_configure(request, response)
   if config["useDataStream"]
     opts[:use_data_stream] = true
 
-    datastream_opts = { cache_ttl: CACHE_TTL }
+    # Entity caches keep the SDK default TTL, matching the Go testapp. The short
+    # CACHE_TTL is only for the flag-check cache above; in replicator mode the
+    # replicator owns the Redis entries and a short TTL on the SDK's write-back
+    # (track -> update_company_metrics) would expire them.
+    datastream_opts = {}
     datastream_opts[:redis_client] = redis_client if redis_client
 
     if config["replicatorUrl"]