Switch out old APIs (#259)

## Summary
[Ticket](https://app.asana.com/0/0/1209184028451140)

This PR includes both frontend and backend updates. Since @nikhil-zlai
and I reviewed the high-level changes today, it might be helpful for
@sean-zlai and me to go over them during our sync tomorrow. There are
some key points I'd like to highlight, but I wanted to get this up early
for review.

**Reminder:** @sean-zlai will handle merging the new drift/summary
functionality into the observability frontend.

### **Frontend Changes (@sean-zlai)**  
- Added new endpoints for Conf, Confs, search, drift, summary, etc.,
currently called on the `/thrift` route for demonstration. The old
endpoints can be removed once we wire in the new ones.
- Created `LogicalNodeTable.svelte` to display all `ConfTypes`,
including empty states.
- Updated search functionality to work for all `ConfTypes`, though only
`risk.user_transactions.txn_join` is enabled for now.

### **Backend Changes (@nikhil-zlai)**  
- Implemented new endpoints using `RouteHandlerWrapper`.  
- Removed old endpoints (except `/timeseries`, which is still needed for
charts).
- Separated drift and summary into two distinct endpoints.  
- Introduced:  
  - `DriftHandler` (handles drift and summaries).  
  - `ConfHandler` (handles Conf, Confs, and Conf search).  
- Added filtering by column to `DriftStore`.  
- Improved and added various tests.  
- Developed a new serializer that integrates well with
`RouteHandlerWrapper`.

Let me know if you have any questions!

## Checklist
- [x] Added Unit Tests
- [x] Covered by existing CI
- [ ] Integration tested
- [ ] Documentation update

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

## Release Notes

- **New Features**
- Added configuration management capabilities across joins, models,
group bys, and staging queries.
	- Introduced drift analysis and summary tracking for configurations.
	- Enhanced API endpoints for retrieving and searching configurations.
	- New `LogicalNodeTable` component for displaying configuration data.
- Added new structures and enumerations for handling configuration
requests in the API.

- **Improvements**
	- Refactored API response handling to use JSON serialization.
- Improved null value handling in data processing with predefined
constants.
	- Updated type definitions for better type safety.
	- Streamlined configuration retrieval and search mechanisms.
	- Enhanced handling of search results by categorizing entity types.

- **Bug Fixes**
- Resolved issues with null value representation in long and double
iterators.
	- Improved error handling in API request processing.

- **Breaking Changes**
	- Removed previous model and join metadata endpoints.
	- Updated API method signatures for configuration retrieval.
	- Eliminated deprecated response types related to joins and models.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: ken-zlai

api/src/main/scala/ai/chronon/api/Constants.scala

@@ -77,12 +77,7 @@ object Constants {
   val extensionsToIgnore: Array[String] = Array(".class", ".csv", ".java", ".scala", ".py", ".DS_Store")
   val foldersToIgnore: Array[String] = Array(".git")
 
-  // import base64
-  // text_bytes = "chronon".encode('utf-8')
-  // base64_str = base64.b64encode(text_bytes)
-  // int.from_bytes(base64.b64decode(base64_str), "big")
-  //
-  // output: 27980863399423854
-
-  val magicNullDouble: java.lang.Double = -27980863399423854.0
+  // A negative integer within the safe range for both long and double in JavaScript, Java, Scala, Python
+  val magicNullLong: java.lang.Long = -1234567890L
+  val magicNullDouble: java.lang.Double = -1234567890.0
 }

api/src/test/scala/ai/chronon/api/test/TileSeriesSerializationTest.scala

@@ -4,10 +4,13 @@ import ai.chronon.api.Constants
 import ai.chronon.api.ScalaJavaConversions.JListOps
 import ai.chronon.api.ThriftJsonCodec
 import ai.chronon.observability.TileDriftSeries
+import ai.chronon.observability.TileSummarySeries
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
 import java.lang.{Double => JDouble}
+import java.lang.{Long => JLong}
+import scala.jdk.CollectionConverters.asScalaBufferConverter
 
 class TileSeriesSerializationTest extends AnyFlatSpec with Matchers {
 
@@ -27,8 +30,49 @@ class TileSeriesSerializationTest extends AnyFlatSpec with Matchers {
 
     val jsonStr = ThriftJsonCodec.toJsonStr(tileDriftSeries)
 
-    jsonStr should be ("""{"percentileDriftSeries":[0.1,-2.7980863399423856E16,-2.7980863399423856E16,-2.7980863399423856E16,0.5]}""")
+    jsonStr should be (s"""{"percentileDriftSeries":[0.1,${Constants.magicNullDouble},${Constants.magicNullDouble},${Constants.magicNullDouble},0.5]}""")
   }
 
+  it should "deserialize double values correctly" in {
+    val json = s"""{"percentileDriftSeries":[0.1,${Constants.magicNullDouble},${Constants.magicNullDouble},${Constants.magicNullDouble},0.5]}"""
+    
+    val series = ThriftJsonCodec.fromJsonStr[TileDriftSeries](json, true, classOf[TileDriftSeries])(manifest[TileDriftSeries])
+    
+    val drifts = series.getPercentileDriftSeries.asScala.toList
+    drifts.size should be (5)
+    drifts(0) should be (0.1)
+    drifts(1) should be (Constants.magicNullDouble)
+    drifts(2) should be (Constants.magicNullDouble)
+    drifts(3) should be (Constants.magicNullDouble)
+    drifts(4) should be (0.5)
+  }
+
+  "TileSummarySeries" should "serialize with nulls and special long values" in {
+    val tileSummarySeries = new TileSummarySeries()
+
+    val counts: Seq[JLong] = Seq(100L, null, Long.MaxValue, Constants.magicNullLong, 500L)
+      .map(v => if (v == null) Constants.magicNullLong else v.asInstanceOf[JLong])
+
+    val countsList: java.util.List[JLong] = counts.toJava
+    tileSummarySeries.setCount(countsList)
+
+    val jsonStr = ThriftJsonCodec.toJsonStr(tileSummarySeries)
+
+    jsonStr should be (s"""{"count":[100,${Constants.magicNullLong},9223372036854775807,${Constants.magicNullLong},500]}""")
+  }
+
+  it should "deserialize long values correctly" in {
+    val json = s"""{"count":[100,${Constants.magicNullLong},9223372036854775807,${Constants.magicNullLong},500]}"""
+    
+    val series = ThriftJsonCodec.fromJsonStr[TileSummarySeries](json, true, classOf[TileSummarySeries])(manifest[TileSummarySeries])
+    
+    val counts = series.getCount.asScala.toList
+    counts.size should be (5)
+    counts(0) should be (100L)
+    counts(1) should be (Constants.magicNullLong)
+    counts(2) should be (Long.MaxValue)
+    counts(3) should be (Constants.magicNullLong)
+    counts(4) should be (500L)
+  }
 
 }

api/thrift/api.thrift

@@ -432,4 +432,4 @@ struct Model {
     3: optional TDataType outputSchema
     4: optional Source source
     5: optional map<string, string> modelParams
-}
+}

api/thrift/hub.thrift

@@ -115,3 +115,39 @@ struct Submission {
     20: optional i64 finishedTs
     21: optional DateRange dateRange
 }
+
+enum ConfType{
+   STAGING_QUERY = 1
+   GROUP_BY = 2
+   JOIN = 3
+   MODEL = 4
+}
+
+struct ConfRequest {
+   1: optional string confName
+   2: optional ConfType confType
+   
+   // one of either branch or version are set - otherwise we will pull conf for main branch
+   3: optional string branch 
+   4: optional string version
+}
+
+/**
+  * lists all confs of the specified type
+  */
+struct ConfListRequest {
+  1: optional ConfType confType
+  
+  // if not specified we will pull conf list for main branch
+  2: optional string branch
+}
+
+/**
+  * Response for listing configurations of a specific type
+  */
+struct ConfListResponse {
+  1: optional list<api.Join> joins
+  2: optional list<api.GroupBy> groupBys
+  3: optional list<api.Model> models
+  4: optional list<api.StagingQuery> stagingQueries
+}

api/thrift/observability.thrift

@@ -139,4 +139,24 @@ struct DriftSpec {
 
     // default drift metric to use
     6: optional DriftMetric driftMetric = DriftMetric.JENSEN_SHANNON
-}
+}
+
+struct JoinDriftRequest {
+    1: required string name
+    2: required i64 startTs
+    3: required i64 endTs
+    6: optional string offset // Format: "24h" or "7d"
+    7: optional DriftMetric algorithm
+    8: optional string columnName
+}
+
+struct JoinDriftResponse {
+    1: required list<TileDriftSeries> driftSeries
+}
+
+struct JoinSummaryRequest {
+    1: required string name
+    2: required i64 startTs
+    3: required i64 endTs
+    8: required string columnName
+}

online/src/main/scala/ai/chronon/online/stats/DriftStore.scala

@@ -84,11 +84,14 @@ class DriftStore(kvStore: KVStore,
     val tileKeyMap = tileKeysForJoin(joinConf, None, columnPrefix)
     val requestContextMap: Map[GetRequest, SummaryRequestContext] = tileKeyMap.flatMap {
       case (group, keys) =>
-        keys.map { key =>
-          val keyBytes = serializer.serialize(key)
-          val get = GetRequest(keyBytes, summaryDataset, startTsMillis = startMs, endTsMillis = endMs)
-          get -> SummaryRequestContext(get, key, group)
-        }
+        // Only create requests for keys that match our column prefix
+        keys
+          .filter(key => columnPrefix.forall(prefix => key.getColumn == prefix))
+          .map { key =>
+            val keyBytes = serializer.serialize(key)
+            val get = GetRequest(keyBytes, summaryDataset, startTsMillis = startMs, endTsMillis = endMs)
+            get -> SummaryRequestContext(get, key, group)
+          }
     }
 
     val responseFuture = kvStore.multiGet(requestContextMap.keys.toSeq)

online/src/main/scala/ai/chronon/online/stats/PivotUtils.scala

@@ -95,7 +95,7 @@ object PivotUtils {
         if (isSetFunc(summary)) {
           JLong.valueOf(extract(summary))
         } else {
-          null
+          Constants.magicNullLong
         }
       }
     }

online/src/test/scala/ai/chronon/online/test/stats/PivotUtilsTest.scala

@@ -113,7 +113,7 @@ class PivotUtilsTest extends AnyFlatSpec with Matchers {
       (ts3, 3000L)
     ))
 
-    result.getCount.asScala shouldEqual List(100L, null, 300L)
+    result.getCount.asScala.toList shouldEqual List(100L, Constants.magicNullLong, 300L)
   }
 
   it should "preserve timestamp order" in {
@@ -313,4 +313,62 @@ class PivotUtilsTest extends AnyFlatSpec with Matchers {
     series(0) shouldBe Constants.magicNullDouble
     series(1) shouldBe 0.5
   }
+
+  it should "handle Long.MAX_VALUE and magicNullLong values" in {
+    val ts1 = new TileSummary()
+    ts1.setCount(Long.MaxValue)
+
+    val ts2 = new TileSummary()
+    // count is not set, should become magicNullLong
+
+    val ts3 = new TileSummary()
+    ts3.setCount(100L)
+
+    val result = pivot(Array(
+      (ts1, 1000L),
+      (ts2, 2000L),
+      (ts3, 3000L)
+    ))
+
+    result.getCount.asScala shouldEqual List(Long.MaxValue, Constants.magicNullLong, 100L)
+  }
+
+  it should "handle all null Long values" in {
+    val ts1 = new TileSummary()
+    val ts2 = new TileSummary()
+    val ts3 = new TileSummary()
+    // no counts set for any summary
+
+    val result = pivot(Array(
+      (ts1, 1000L),
+      (ts2, 2000L),
+      (ts3, 3000L)
+    ))
+
+    // Since all values are unset, they should all be magicNullLong rather than null
+    result.getCount.asScala.toList shouldEqual List.fill(3)(Constants.magicNullLong)
+  }
+
+  it should "handle mixed null and non-null Long fields" in {
+    val ts1 = new TileSummary()
+    ts1.setCount(100L)
+    ts1.setNullCount(10L)
+
+    val ts2 = new TileSummary()
+    // count not set
+    ts2.setNullCount(20L)
+
+    val ts3 = new TileSummary()
+    ts3.setCount(300L)
+    // nullCount not set
+
+    val result = pivot(Array(
+      (ts1, 1000L),
+      (ts2, 2000L),
+      (ts3, 3000L)
+    ))
+
+    result.getCount.asScala shouldEqual List(100L, Constants.magicNullLong, 300L)
+    result.getNullCount.asScala shouldEqual List(10L, 20L, Constants.magicNullLong)
+  }
 }

service_commons/src/main/java/ai/chronon/service/RouteHandlerWrapper.java

@@ -2,6 +2,7 @@
 
 import ai.chronon.api.thrift.*;
 import ai.chronon.api.thrift.protocol.TBinaryProtocol;
+import ai.chronon.api.thrift.protocol.TSimpleJSONProtocol;
 import ai.chronon.api.thrift.transport.TTransportException;
 import io.vertx.core.Handler;
 import io.vertx.core.json.JsonObject;
@@ -50,7 +51,6 @@ public class RouteHandlerWrapper {
     private static final ThreadLocal<Base64.Encoder> base64Encoder = ThreadLocal.withInitial(Base64::getEncoder);
     private static final ThreadLocal<Base64.Decoder> base64Decoder = ThreadLocal.withInitial(Base64::getDecoder);
 
-
     public static <T extends TBase> T deserializeTBinaryBase64(String base64Data, Class<? extends TBase> clazz) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, TException {
         byte[] binaryData = base64Decoder.get().decode(base64Data);
         T tb = (T) clazz.getDeclaredConstructor().newInstance();
@@ -87,8 +87,20 @@ public static <I, O> Handler<RoutingContext> createHandler(Function<I, O> transf
 
                 String responseFormat = ctx.request().getHeader(RESPONSE_CONTENT_TYPE_HEADER);
                 if (responseFormat == null || responseFormat.equals("application/json")) {
-                    // Send json response
-                    ctx.response().setStatusCode(200).putHeader("content-type", JSON_TYPE_VALUE).end(JsonObject.mapFrom(output).encode());
+                    try {
+                        TSerializer serializer = new TSerializer(new TSimpleJSONProtocol.Factory());
+                        String jsonString = serializer.toString((TBase)output);
+                        ctx.response()
+                           .setStatusCode(200)
+                           .putHeader("content-type", JSON_TYPE_VALUE)
+                           .end(jsonString);
+                    } catch (TException e) {
+                        LOGGER.error("Failed to serialize response", e);
+                        throw new RuntimeException(e);
+                    } catch (Exception e) {
+                        LOGGER.error("Unexpected error during serialization", e);
+                        throw new RuntimeException(e);
+                    }
                 } else {
                     if (!responseFormat.equals(TBINARY_B64_TYPE_VALUE)) {
                         throw new IllegalArgumentException(String.format("Unsupported response-content-type: %s. Supported values are: %s and %s", responseFormat, JSON_TYPE_VALUE, TBINARY_B64_TYPE_VALUE));