WIP: speculative-tx returns a strigified edn with the document state …

…after the transaction

gxtdb-rs/tests/mock.rs

@@ -22,6 +22,13 @@ impl GrpcApi for ServerMock {
         todo!()
     }
 
+    async fn speculative_tx(
+        &self,
+        _request: tonic::Request<proto_api::SpeculativeTxRequest>,
+    ) -> Result<tonic::Response<proto_api::SpeculativeTxResponse>, tonic::Status> {
+        todo!()
+    }
+
     async fn entity_tx(
         &self,
         _request: tonic::Request<proto_api::EntityTxRequest>,

kreya/grpc/com/xtdb/protos/GrpcApi/SpeculativeTx-request.json

@@ -4,13 +4,15 @@
       "put": {
         "xtId": "id1",
         "idType":"Keyword",
-        "document": {"key": "ARROZ"}
-      }
-    },
-    {
-      "delete": {
-        "documentId": "id1",
-        "idType":"Keyword",
+        "document": {
+          "key": "ARROZ",
+          "listing": [1, 2, 3, 4],
+          "number": 3,
+          "bool?": true,
+          "map": {
+            "k2": "v2"
+          }
+        }
       }
     }
   ]

resources/transactions.proto

@@ -60,4 +60,5 @@ message SpeculativeTxResponse {
   int64 tx_id = 3;
   int32 entity_cache_size = 4;
   int32 batch_size = 5;
+  string edn_document = 6;
 }

src/com/xtdb/protos.cljc

@@ -422,14 +422,15 @@
 ;-----------------------------------------------------------------------------
 ; SpeculativeTxResponse
 ;-----------------------------------------------------------------------------
-(defrecord SpeculativeTxResponse-record [valid-time tx-time tx-id entity-cache-size batch-size]
+(defrecord SpeculativeTxResponse-record [valid-time tx-time tx-id entity-cache-size batch-size edn-document]
   pb/Writer
   (serialize [this os]
     (serdes.core/write-String 1  {:optimize true} (:valid-time this) os)
     (serdes.core/write-String 2  {:optimize true} (:tx-time this) os)
     (serdes.core/write-Int64 3  {:optimize true} (:tx-id this) os)
     (serdes.core/write-Int32 4  {:optimize true} (:entity-cache-size this) os)
-    (serdes.core/write-Int32 5  {:optimize true} (:batch-size this) os))
+    (serdes.core/write-Int32 5  {:optimize true} (:batch-size this) os)
+    (serdes.core/write-String 6  {:optimize true} (:edn-document this) os))
   pb/TypeReflection
   (gettype [this]
     "com.xtdb.protos.SpeculativeTxResponse"))
@@ -439,13 +440,14 @@
 (s/def :com.xtdb.protos.SpeculativeTxResponse/tx-id int?)
 (s/def :com.xtdb.protos.SpeculativeTxResponse/entity-cache-size int?)
 (s/def :com.xtdb.protos.SpeculativeTxResponse/batch-size int?)
-(s/def ::SpeculativeTxResponse-spec (s/keys :opt-un [:com.xtdb.protos.SpeculativeTxResponse/valid-time :com.xtdb.protos.SpeculativeTxResponse/tx-time :com.xtdb.protos.SpeculativeTxResponse/tx-id :com.xtdb.protos.SpeculativeTxResponse/entity-cache-size :com.xtdb.protos.SpeculativeTxResponse/batch-size]))
-(def SpeculativeTxResponse-defaults {:valid-time "" :tx-time "" :tx-id 0 :entity-cache-size 0 :batch-size 0})
+(s/def :com.xtdb.protos.SpeculativeTxResponse/edn-document string?)
+(s/def ::SpeculativeTxResponse-spec (s/keys :opt-un [:com.xtdb.protos.SpeculativeTxResponse/valid-time :com.xtdb.protos.SpeculativeTxResponse/tx-time :com.xtdb.protos.SpeculativeTxResponse/tx-id :com.xtdb.protos.SpeculativeTxResponse/entity-cache-size :com.xtdb.protos.SpeculativeTxResponse/batch-size :com.xtdb.protos.SpeculativeTxResponse/edn-document]))
+(def SpeculativeTxResponse-defaults {:valid-time "" :tx-time "" :tx-id 0 :entity-cache-size 0 :batch-size 0 :edn-document ""})
 
 (defn cis->SpeculativeTxResponse
   "CodedInputStream to SpeculativeTxResponse"
   [is]
-  (map->SpeculativeTxResponse-record (tag-map SpeculativeTxResponse-defaults (fn [tag index] (case index 1 [:valid-time (serdes.core/cis->String is)] 2 [:tx-time (serdes.core/cis->String is)] 3 [:tx-id (serdes.core/cis->Int64 is)] 4 [:entity-cache-size (serdes.core/cis->Int32 is)] 5 [:batch-size (serdes.core/cis->Int32 is)] [index (serdes.core/cis->undefined tag is)])) is)))
+  (map->SpeculativeTxResponse-record (tag-map SpeculativeTxResponse-defaults (fn [tag index] (case index 1 [:valid-time (serdes.core/cis->String is)] 2 [:tx-time (serdes.core/cis->String is)] 3 [:tx-id (serdes.core/cis->Int64 is)] 4 [:entity-cache-size (serdes.core/cis->Int32 is)] 5 [:batch-size (serdes.core/cis->Int32 is)] 6 [:edn-document (serdes.core/cis->String is)] [index (serdes.core/cis->undefined tag is)])) is)))
 
 (defn ecis->SpeculativeTxResponse
   "Embedded CodedInputStream to SpeculativeTxResponse"

src/gxtdb/adapters/json.clj

@@ -10,4 +10,14 @@
       :string-value (-> kind :string-value str)
       :list-value (->> kind :list-value :values (mapv value-record->edn))
       :struct-value (->> kind :struct-value :fields (map (fn [[k v]] {(utils/str->keyword k) (value-record->edn v)})) (reduce into {}))
-      "")))
+      "")))
+
+(defn edn->proto-struct [edn]
+  (cond
+    (number? edn) {:kind {:number-value edn}}
+    (string? edn) {:kind {:string-value edn}}
+    (keyword? edn) {:kind {:string-value (name edn)}}
+    (boolean? edn) {:kind {:bool-value edn}}
+    (map? edn) {:kind {:struct-value {:fields (->> edn (map (fn [[k v]] {(keyword k) (edn->proto-struct v)})) (reduce into {}))}}}
+    (vector? edn) {:kind {:list-value {:values (mapv edn->proto-struct edn)}}}
+    :else {:hello "world"}))

src/gxtdb/adapters/tx_log.clj

@@ -66,4 +66,6 @@
    :valid-time (-> edn :valid-time utils/->inst-str)
    :tx-id (:tx-id edn)
    :entity-cache-size (:entity-cache-size edn)
-   :batch-size (:batch-size edn)})
+   :batch-size (:batch-size edn)
+   :edn-document  (-> edn :document-store :!docs deref str)})
+

src/gxtdb/controllers.clj

@@ -15,7 +15,6 @@
 (defn with-tx [xtdb-node tx-ops]
   (let [tx-log (tx-log-adapter/proto->tx-log tx-ops)
         xt-response (xt/with-tx (xt/db xtdb-node) tx-log)]
-    (pprint/pprint  (-> xt-response record->map :document-store :!docs deref))
     (tx-log-adapter/speculative-tx->proto (record->map xt-response))))
 
 (defn entity-tx [xtdb-node params]

src/gxtdb/utils.clj

@@ -1,7 +1,8 @@
 (ns gxtdb.utils
   (:require [clojure.instant :refer [read-instant-date]]
             [clojure.string :as str]
-            [xtdb.api :as xt])
+            [xtdb.api :as xt]
+            [clojure.pprint :as pprint])
   (:import java.text.SimpleDateFormat))
 
 #_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
@@ -63,4 +64,11 @@
     [true true] (xt/open-db xtdb-node)
     [false true] (xt/db xtdb-node)
     [true false] (xt/open-db xtdb-node db-basis)
-    [false false] (xt/db xtdb-node db-basis)))
+    [false false] (xt/db xtdb-node db-basis)))
+
+#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
+(defn inspect [value]
+  (println)
+  (pprint/pprint value)
+  (println)
+  value)

test/gxtdb/service_test.clj

@@ -6,7 +6,8 @@
             [io.pedestal.http :as pedestal]
             [protojure.grpc.client.providers.http2 :refer [connect]]
             [protojure.pedestal.core :as protojure.pedestal]
-            [xtdb.api :as xt]))
+            [xtdb.api :as xt]
+            [clojure.string :as str]))
 
 ;; Setup.
 (def ^:dynamic *opts* {})
@@ -105,4 +106,5 @@
                            connected
                            {:tx-ops [{:transaction-type
                                       {:put {:id-type :string, :xt-id "id1", :document {:fields {"key" {:kind {:string-value "Hellow"}}}}}}}]})]
-      (is (= 2 (:tx-id with-tx))))))
+      (is (= 2 (:tx-id with-tx)))
+      (is (str/includes? (:edn-document with-tx) ":crux.db/id \"id1\"")))))