For the "transactions" table, insert rather than upsert

We generally always upsert, since that matches best the Datomic semantics, but
transactions are never updated, so a simple INSERT will do, and it ensures that
when importing a transaction twice that the process throws (the primary key
constraint will be violated), which acts as a consistency guarantee (the caller
can fetch the latest t value and retry.)

Author: plexus

.gitignore

@@ -12,3 +12,4 @@ resources/public/ui
 .store
 out
 .#*
+deps.local.edn

CHANGELOG.md

@@ -2,10 +2,12 @@
 
 ## Added
 
-## Fixed
+- Added a `find-max-t` helper function, for picking up work where it was left off
 
 ## Changed
 
+- Throw when trying to reprocess an earlier transaction
+
 # 0.2.37 (2022-09-14 / 5b770a2)
 
 ## Added
@@ -20,4 +22,4 @@
 
 ## Fixed
 
-## Changed
+## Changed

deps.local.edn

@@ -20,3 +20,17 @@
   (d/log conn)
   1000
   9010))
+
+;; Given we are currently processing a certain transaction (say t=1012), can we
+;; find the t value of the previous transaction? This could be useful to add an
+;; additional guarantee that transactions are processed exactly in order, by
+;; adding a postgresql trigger that validates that transactions form an unbroken
+;; chain.
+
+(let [max-t 1012
+      db (d/as-of (d/db conn) (dec max-t))]
+  (d/q '[:find (max ?t) .
+         :where
+         [?i :db/txInstant]
+         [(datomic.api/tx->t ?i) ?t]]
+       db))

repl_sessions/find_prev_tx.clj

@@ -269,21 +269,24 @@
                 [:delete
                  {:table  (table-name ctx mem-attr)
                   :values {"db__id" eid}}]
-                [:upsert
-                 {:table  (table-name ctx mem-attr)
-                  :by #{"db__id"}
-                  :values (into (cond-> {"db__id" eid}
-                                  ;; Bit of manual fudgery to also get the "t"
-                                  ;; value of each transaction into
-                                  ;; our "transactions" table.
-                                  (= :db/txInstant mem-attr)
-                                  (assoc "t" (d/tx->t (-t (first datoms)))))
-                                (map (juxt #(column-name ctx mem-attr (ctx-ident ctx (-a %)))
-                                           #(when (-added? %)
-                                              (encode-value ctx
-                                                            (ctx-valueType ctx (-a %))
-                                                            (-v %)))))
-                                datoms)}])))))
+                (let [table (table-name ctx mem-attr)]
+                  [(if (= "transactions" table)
+                     :insert
+                     :upsert)
+                   {:table table
+                    :by #{"db__id"}
+                    :values (into (cond-> {"db__id" eid}
+                                    ;; Bit of manual fudgery to also get the "t"
+                                    ;; value of each transaction into
+                                    ;; our "transactions" table.
+                                    (= :db/txInstant mem-attr)
+                                    (assoc "t" (d/tx->t (-t (first datoms)))))
+                                  (map (juxt #(column-name ctx mem-attr (ctx-ident ctx (-a %)))
+                                             #(when (-added? %)
+                                                (encode-value ctx
+                                                              (ctx-valueType ctx (-a %))
+                                                              (-v %)))))
+                                  datoms)}]))))))
 
 (defn card-many-entity-ops
   "Add operations `:ops` to the context for all the `cardinality/many` datoms in a
@@ -435,6 +438,10 @@
                         :if-not-exists]}))
    columns))
 
+(defmethod op->sql :insert [[_ {:keys [table by values]}]]
+  [{:insert-into [(keyword table)]
+    :values      [values]}])
+
 (defmethod op->sql :upsert [[_ {:keys [table by values]}]]
   (let [op {:insert-into   [(keyword table)]
             :values        [values]
@@ -510,27 +517,27 @@
   configuration regarding tables and target db, and eventually `:ops` that need
   to be processed."
   ([conn metaschema]
-   (initial-ctx conn metaschema 999))
+   (initial-ctx conn metaschema nil))
   ([conn metaschema t]
-  ;; Bootstrap, make sure we have info about idents that datomic creates itself
-  ;; at db creation time. d/as-of t=999 is basically an empty database with only
-  ;; metaschema attributes (:db/txInstant etc), since the first "real"
-  ;; transaction is given t=1000. Interesting to note that Datomic seems to
-  ;; bootstrap in pieces: t=0 most basic idents, t=57 add double, t=63 add
-  ;; docstrings, ...
-   (let [idents (pull-idents (d/as-of (d/db conn) t))]
+   ;; Bootstrap, make sure we have info about idents that datomic creates itself
+   ;; at db creation time. d/as-of t=999 is basically an empty database with only
+   ;; metaschema attributes (:db/txInstant etc), since the first "real"
+   ;; transaction is given t=1000. Interesting to note that Datomic seems to
+   ;; bootstrap in pieces: t=0 most basic idents, t=57 add double, t=63 add
+   ;; docstrings, ...
+   (let [idents (pull-idents (d/as-of (d/db conn) (or t 999)))]
      {;; Track datomic schema
       :entids (into {} (map (juxt :db/ident :db/id)) idents)
       :idents (into {} (map (juxt :db/id identity)) idents)
-     ;; Configure/track relational schema
+      ;; Configure/track relational schema
       :tables (-> metaschema
                   :tables
                   (update :db/txInstant assoc :name "transactions")
                   (update :db/ident assoc :name "idents"))
-     ;; Mapping from datomic to relational type
+      ;; Mapping from datomic to relational type
       :db-types pg-type
-     ;; Create two columns that don't have a attribute as such in datomic, but
-     ;; which we still want to track
+      ;; Create two columns that don't have a attribute as such in datomic, but
+      ;; which we still want to track
       :ops [[:ensure-columns
              {:table   "idents"
               :columns {:db/id {:name "db__id"
@@ -565,3 +572,16 @@
                      (jdbc/execute! jdbc-tx %)) queries))
         (recur (dissoc ctx :ops) txs))
       ctx)))
+
+(defn find-max-t
+  "Find the highest value in the transactions table in postgresql. The sync should
+  continue from `(inc (find-max-t ds))`"
+  [ds]
+  (:max
+   (first
+    (try
+      (jdbc/execute! ds ["SELECT max(t) FROM transactions"])
+      (catch Exception e
+        ;; If the transactions table doesn't yet exist, return `nil`, so we start
+        ;; from the beginning of the log
+        nil)))))

src/lambdaisland/plenish.clj

@@ -26,11 +26,14 @@
       @(d/transact *conn* factories/schema)
       (f))))
 
-(defn import! [metaschema]
-  (let [ctx (plenish/initial-ctx *conn* metaschema)]
-    (plenish/import-tx-range
-     ctx *conn* *ds*
-     (d/tx-range (d/log *conn*) nil nil))))
+(defn import!
+  ([metaschema]
+   (import! metaschema nil))
+  ([metaschema t]
+   (let [ctx (plenish/initial-ctx *conn* metaschema t)]
+     (plenish/import-tx-range
+      ctx *conn* *ds*
+      (d/tx-range (d/log *conn*) t nil)))))
 
 (defn transact! [tx]
   @(d/transact *conn* tx))
@@ -110,7 +113,7 @@
                :idents/ident "fruit/orange"}]
              (jdbc/execute! *ds* ["SELECT fruit.db__id, idents.ident FROM fruit, idents WHERE fruit.type = idents.db__id;"]))))))
 
-(deftest update-cardinality-one-attribute
+(deftest update-cardinality-one-attribute--membership
   (testing "membership attribute"
     (transact! [{:db/ident :fruit/type
                  :db/valueType :db.type/string
@@ -125,8 +128,9 @@
       (import! {:tables {:fruit/type {}}})
       (is (= [{:fruit/db__id apple-id
                :fruit/type "orange"}]
-             (jdbc/execute! *ds* ["SELECT * FROM fruit"])))))
+             (jdbc/execute! *ds* ["SELECT * FROM fruit"]))))))
 
+(deftest update-cardinality-one-attribute--regular
   (testing "regular attribute"
     (transact! [{:db/ident :veggie/type
                  :db/valueType :db.type/string
@@ -174,6 +178,15 @@
                :veggie_x_rating/rating 5}]
              (jdbc/execute! *ds* ["SELECT * FROM veggie_x_rating"]))))))
 
+(deftest duplicate-import-throws
+  (testing "Trying to import a transaction that was already processed should throw"
+    (fd/create! *conn* factories/cart)
+    (import! factories/metaschema)
+
+    (let [max-t (plenish/find-max-t *ds*)]
+      (is (thrown? com.impossibl.postgres.jdbc.PGSQLIntegrityConstraintViolationException
+                   (import! factories/metaschema max-t))))))
+
 (comment
   ;; REPL alternative to fixture
   (recreate-replica!)

test/lambdaisland/plenish_test.clj

@@ -1,2 +1,3 @@
 #kaocha/v1
-{:plugins [:notifier :print-invocations :profiling]}
+{:plugins [:notifier :print-invocations :profiling]
+ :capture-output? false}