From 9497d69b44a8163857663007a7ecd272aa120dad Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Thu, 28 Jul 2016 15:30:46 -0700
Subject: [PATCH] Respect :db/unique constraints; test upserts.

This version includes SQLite-level unique indexes; these should never be
needed.  I've included them as a fail-safe while testing; they'll help
us catch errors in the transaction layer above.
---
 src/datomish/db.cljc            | 226 ++++++++++++++++++++------------
 src/datomish/schema.cljc        |   4 +
 src/datomish/sqlite_schema.cljc |  10 +-
 test/datomish/db_test.cljc      | 198 ++++++++++++++++++++++++++--
 4 files changed, 343 insertions(+), 95 deletions(-)

diff --git a/src/datomish/db.cljc b/src/datomish/db.cljc
index d9dc9ee6..806660db 100644
--- a/src/datomish/db.cljc
+++ b/src/datomish/db.cljc
@@ -77,7 +77,7 @@
   (let [e (:e row)
         a (:a row)
         v (:v row)]
-    (Datom. e a (ds/<-SQLite schema a v) (:tx row) (:added row))))
+    (Datom. e a (ds/<-SQLite schema a v) (:tx row) (and (some? (:added row)) (not= 0 (:added row))))))
 
 (defrecord DB [sqlite-connection schema idents current-tx]
   ;; idents is map {ident -> entid} of known idents.  See http://docs.datomic.com/identity.html#idents.
@@ -98,7 +98,7 @@
           v (and v (ds/->SQLite schema a v))] ;; We assume e and a are always given.
       (go-pair
         (->>
-          {:select [:*] ;; e :a :v :tx] ;; TODO: generalize columns.
+          {:select [:e :a :v :tx [1 :added]] ;; TODO: generalize columns.
            :from [:datoms]
            :where (cons :and (map #(vector := %1 %2) [:e :a :v :tx] (take-while (comp not nil?) [e a v tx])))} ;; Must drop nils.
           (sql/format)
@@ -113,7 +113,9 @@
           v (ds/->SQLite schema a v)]
       (go-pair
         (->>
-          {:select [:*] :from [:datoms] :where [:and [:= :a a] [:= :v v]]}
+          {:select [:e :a :v :tx [1 :added]] ;; TODO: generalize columns.
+           :from [:datoms]
+           :where [:and [:= :a a] [:= :v v] [:= :index_avet 1]]}
           (sql/format)
 
           (s/all-rows (:sqlite-connection db))
@@ -134,8 +136,12 @@
             ;; Update materialized datom view.
             (if (.-added datom)
               (<? (exec
-                    ;; TODO: use schema to insert correct indexing flags.
-                    ["INSERT INTO datoms VALUES (?, ?, ?, ?, 0, 0)" e a v tx]))
+                    ["INSERT INTO datoms VALUES (?, ?, ?, ?, ?, ?, ?, ?)" e a v tx
+                     (ds/indexing? (.-schema db) a) ;; index_avet
+                     (ds/ref? (.-schema db) a) ;; index_vaet
+                     (ds/unique-value? (.-schema db) a) ;; unique_value
+                     (ds/unique-identity? (.-schema db) a) ;; unique_identity
+                     ]))
               (<? (exec
                     ;; TODO: verify this is correct.
                     ["DELETE FROM datoms WHERE (e = ? AND a = ? AND v = ?)" e a v]))))))
@@ -202,9 +208,22 @@
 (defn id-literal? [x]
   (and (instance? TempId x)))
 
-(defn temp-literal? [x]
-  (and (id-literal? x)
-       (= :db.part/temp (:part x))))
+(defrecord TxReport [db-before db-after entities tx-data tempids])
+
+(defn- report? [x]
+  (and (instance? TxReport x)))
+
+(defonce -eid (atom (- 0x200 1)))
+
+;; TODO: better here.
+(defn- next-eid [db]
+  (swap! -eid inc))
+
+(defn- allocate-eid
+  [report id-literal eid]
+  {:pre [(report? report) (id-literal? id-literal) (and (integer? eid) (not (neg? eid)))]}
+
+  (assoc-in report [:tempids id-literal] eid))
 
 ;; (def data-readers {'db/id id-literal})
 
@@ -235,7 +254,7 @@
    (go-pair
      (when-not (= sqlite-schema/current-version (<? (sqlite-schema/<ensure-current-version sqlite-connection)))
        (raise "Could not ensure current SQLite schema version."))
-     (let [idents (into (<? (<idents sqlite-connection)) {:db/txInstant 100 :db/ident 101 :x 102 :y 103 :name 104 :aka 105 :test/kw 106})] ;; TODO: pre-populate idents and SQLite tables?
+     (let [idents (into (<? (<idents sqlite-connection)) {:db/txInstant 100 :db/ident 101 :x 102 :y 103 :name 104 :aka 105 :test/kw 106 :age 107 :email 108 :spouse 109})] ;; TODO: pre-populate idents and SQLite tables?
        (map->DB
          {:sqlite-connection sqlite-connection
           :idents            idents
@@ -245,11 +264,6 @@
 (defn connection-with-db [db]
   (map->Connection {:current-db (atom db)}))
 
-(defrecord TxReport [db-before db-after entities tx-data tempids])
-
-(defn- report? [x]
-  (and (instance? TxReport x)))
-
 ;; ;; TODO: persist max-tx and max-eid in SQLite.
 
 (defn maybe-datom->entity [entity]
@@ -265,21 +279,28 @@
     true
     entity))
 
-(defn maybe-explode [schema entity] ;; TODO db?  schema?
-  (cond
-    (map? entity)
-    ;; TODO: reverse refs, lists, nested maps
-    (let [eid (or (:db/id entity)
-                  (id-literal :db.part/temp))] ;; Must upsert if no ID given. TODO: check if this fails in Datomic/DS.
-      (for [[a v] (dissoc entity :db/id)]
-        [:db/add eid a v]))
+(defn explode-entities [schema report]
+  (let [initial-es     (:entities report)
+        initial-report (assoc report :entities [])]
+    (loop [report initial-report
+           es     initial-es]
+      (let [[entity & entities] es]
+        (cond
+          (nil? entity)
+          report
 
-    ;; (raise "Map entities are not yet supported, got " entity
-    ;;        {:error :transact/syntax
-    ;;         :op    entity })
+          (map? entity)
+          ;; TODO: reverse refs, lists, nested maps
+          (if-let [eid (:db/id entity)]
+            (let [exploded (for [[a v] (dissoc entity :db/id)]
+                             [:db/add eid a v])]
+              (recur report (concat exploded entities)))
+            (raise "Map entity missing :db/id, got " entity
+                   {:error :transact/entity-missing-db-id
+                    :op    entity }))
 
-    true
-    [entity]))
+          true
+          (recur (util/conj-in report [:entities] entity) entities))))))
 
 (defn maybe-ident->entid [db [op & entity :as orig]]
   ;; TODO: use something faster than `into` here.
@@ -302,32 +323,30 @@
 (defn preprocess [db report]
   {:pre [(db? db) (report? report)]}
 
-  (let [initial-es (conj (or (:entities report) []) (tx-entity db))]
+  (let [initial-es (or (:entities report) [])]
     (when-not (sequential? initial-es)
       (raise "Bad transaction data " initial-es ", expected sequential collection"
              {:error :transact/syntax, :tx-data initial-es}))
 
-    (->>
-      (->
-        (comp
-          ;; Track the provenance of each assertion for error reporting.
-          (map #(with-meta % {:source %}))
+    ;; TODO: find an approach that generates less garbage.
+    (->
+      report
 
-          ;; Normalize Datoms into :db/add or :db/retract vectors.
-          (map maybe-datom->entity)
+      (update :entities conj (tx-entity db))
 
-          ;; Explode map shorthand, such as {:db/id e :attr value :_reverse ref},
-          ;; to a list of vectors, like
-          ;; [[:db/add e :attr value] [:db/add ref :reverse e]].
-          (mapcat (partial maybe-explode (schema db)))
+      ;; Normalize Datoms into :db/add or :db/retract vectors.
+      (update :entities (partial map maybe-datom->entity))
 
-          ;; Replace idents with entids where possible.
-          (map (partial maybe-ident->entid db))
+      ;; Explode map shorthand, such as {:db/id e :attr value :_reverse ref},
+      ;; to a list of vectors, like
+      ;; [[:db/add e :attr value] [:db/add ref :reverse e]].
+      (->> (explode-entities (schema db)))
 
-          ;; Add tx if not given.
-          (map (partial maybe-add-current-tx (current-tx db))))
-        (transduce conj [] initial-es))
-      (assoc-in report [:entities]))))
+      ;; Replace idents with entids where possible.
+      (update :entities (partial map (partial maybe-ident->entid db)))
+
+      ;; Add tx if not given.
+      (update :entities (partial map (partial maybe-add-current-tx (current-tx db)))))))
 
 (defn- lookup-ref? [x]
   (and (sequential? x)
@@ -376,16 +395,6 @@
                             field)))))
       (assoc-in report [:entities])))) ;; TODO: meta.
 
-(defonce -eid (atom (- 0x200 1)))
-
-;; TODO: better here.
-(defn- next-eid [db]
-  (swap! -eid inc))
-
-(defn- allocate-eid
-  [report id-literal eid]
-  (assoc-in report [:tempids id-literal] eid))
-
 (declare <resolve-id-literals)
 
 (defn <retry-with-tempid [db report es tempid upserted-eid]
@@ -437,17 +446,10 @@
           initial-entities (sort-by keyfn (:entities report))]
       (loop [report initial-report
              es initial-entities]
-        (if (report? initial-report)
-          (update report :tempids #(into {} (filter (comp not temp-literal? first) %)))
-          (raise "fail" {:initial-report report}))
-
         (let [[[op e a v tx :as entity] & entities] es]
           (cond
             (nil? entity)
-            ;; We can add :db.part/temp id-literals; remove them.
-            (if (report? report)
-              (update report :tempids #(into {} (filter (comp not temp-literal? first) %)))
-              (raise "fail" {:report report}))
+            report
 
             (and (not= op :db/add)
                  (not (empty? (filter id-literal? [e a v tx]))))
@@ -457,9 +459,9 @@
 
             ;; Upsert!
             (and (id-literal? e)
-                 (ds/unique-identity? (schema db) a) ;; TODO: schema.
+                 (ds/unique-identity? (schema db) a)
                  (not-any? id-literal? [a v tx]))
-            (let [upserted-eid (:e (first (<? (<avet db [a v]))))  ;; TODO: define this interface.
+            (let [upserted-eid (:e (first (<? (<avet db [a v]))))
                   allocated-eid (get-in report [:tempids e])]
               (if (and upserted-eid allocated-eid (not= upserted-eid allocated-eid))
                 (<? (<retry-with-tempid db initial-report initial-entities e upserted-eid)) ;; TODO: not initial report, just the sorted entities here.
@@ -493,21 +495,73 @@
 (defn- transact-report [report datom]
   (update-in report [:tx-data] conj datom))
 
-(defn- ensure-schema-constraints
-  "Verify that all entities obey the schema constraints."
+(defn- <ensure-schema-constraints
+  "Throw unless all entities in :entities obey the schema constraints."
 
   [db report]
   {:pre [(db? db) (report? report)]}
 
-  ;; TODO: :db/unique :db.unique/value.
   ;; TODO: consider accumulating errors to show more meaningful error reports.
   ;; TODO: constrain entities; constrain attributes.
 
-  (doseq [[op e a v tx] (:entities report)]
-    (ds/ensure-valid-value (schema db) a v))
-  report)
+  (go-pair
+    (doseq [[op e a v tx] (:entities report)]
+      (ds/ensure-valid-value (schema db) a v))
+    report))
 
-(defn <postprocess [db report]
+(defn- <ensure-unique-constraints
+  "Throw unless all datoms in :tx-data obey the uniqueness constraints."
+
+  [db report]
+  {:pre [(db? db) (report? report)]}
+
+  ;; TODO: consider accumulating errors to show more meaningful error reports.
+  ;; TODO: constrain entities; constrain attributes.
+
+  (go-pair
+    ;; TODO: comment on applying datoms that violate uniqueness.
+    (let [unique-datoms (transient {})] ;; map (nil, a, v)|(e, a, nil)|(e, a, v) -> datom.
+      (doseq [[e a v tx added :as datom] (:tx-data report)]
+
+        (when added
+          ;; Check for violated :db/unique constraint between datom and existing store.
+          (when (ds/unique? (schema db) a)
+            (when-let [found (first (<? (<avet db [a v])))]
+              (raise "Cannot add " datom " because of unique constraint: " found
+                     {:error :transact/unique
+                      :attribute a ;; TODO: map attribute back to ident.
+                      :entity datom})))
+
+          ;; Check for violated :db/unique constraint between datoms.
+          (when (ds/unique? (schema db) a)
+            (let [key [nil a v]]
+              (when-let [other (get unique-datoms key)]
+                (raise "Cannot add " datom " and " other " because together they violate unique constraint"
+                       {:error :transact/unique
+                        :attribute a ;; TODO: map attribute back to ident.
+                        :entity datom}))
+              (assoc! unique-datoms key datom)))
+
+          ;; Check for violated :db/cardinality :db.cardinality/one constraint between datoms.
+          (when-not (ds/multival? (schema db) a)
+            (let [key [e a nil]]
+              (when-let [other (get unique-datoms key)]
+                (raise "Cannot add " datom " and " other " because together they violate cardinality constraint"
+                       {:error :transact/unique
+                        :entity datom}))
+              (assoc! unique-datoms key datom)))
+
+          ;; Check for duplicated datoms.  Datomic doesn't allow overlapping writes, and we don't
+          ;; want to guarantee order, so we don't either.
+          (let [key [e a v]]
+            (when-let [other (get unique-datoms key)]
+              (raise "Cannot add duplicate " datom
+                     {:error :transact/unique
+                      :entity datom}))
+            (assoc! unique-datoms key datom)))))
+    report))
+
+(defn <entities->tx-data [db report]
   {:pre [(db? db) (report? report)]}
   (go-pair
     (let [initial-report report]
@@ -546,19 +600,24 @@
   {:pre [(db? db) (report? report)]}
 
   (go-pair
-    (->> report
-         (preprocess db)
+    (->>
+      report
+      (preprocess db)
 
-         (<resolve-lookup-refs db)
-         (<?)
+      (<resolve-lookup-refs db)
+      (<?)
 
-         (<resolve-id-literals db)
-         (<?)
+      (<resolve-id-literals db)
+      (<?)
 
-         (ensure-schema-constraints db)
+      (<ensure-schema-constraints db)
+      (<?)
 
-         (<postprocess db)
-         (<?))))
+      (<entities->tx-data db)
+      (<?)
+
+      (<ensure-unique-constraints db)
+      (<?))))
 
 ;; Normalize as [op int|id-literal int|id-literal value|id-literal tx|id-literal]. ;; TODO: mention lookup-refs.
 
@@ -573,9 +632,8 @@
   "Transactions may add idents, install new partitions, and install new schema attributes.
   Handle :db/ident assertions here."
   [db tx-data]
-  {:pre [(db? db)
-         ;; (report? report)
-         ]}
+  {:pre [(db? db)]}
+
   ;; TODO: use q to filter the report!
   (let [original-db db
         original-ident-assertions (filter (partial is-ident? db) tx-data)]
diff --git a/src/datomish/schema.cljc b/src/datomish/schema.cljc
index c2ef2e02..2af56c0c 100644
--- a/src/datomish/schema.cljc
+++ b/src/datomish/schema.cljc
@@ -32,6 +32,10 @@
           :cljs [^boolean indexing?]) [schema attr]
   (is-attr? schema attr :db/index))
 
+(defn #?@(:clj  [^Boolean unique?]
+          :cljs [^boolean unique?]) [schema attr]
+  (is-attr? schema attr :db/unique))
+
 (defn #?@(:clj  [^Boolean unique-identity?]
           :cljs [^boolean unique-identity?]) [schema attr]
   (is-attr? schema attr :db.unique/identity))
diff --git a/src/datomish/sqlite_schema.cljc b/src/datomish/sqlite_schema.cljc
index ef63975b..f107b211 100644
--- a/src/datomish/sqlite_schema.cljc
+++ b/src/datomish/sqlite_schema.cljc
@@ -18,11 +18,15 @@
 (def current-version 1)
 
 (def v1-statements
-  ["CREATE TABLE datoms (e INTEGER NOT NULL, a SMALLINT NOT NULL, v BLOB NOT NULL, tx INTEGER NOT NULL, index_avet TINYINT NOT NULL DEFAULT 0, index_vaet TINYINT NOT NULL DEFAULT 0)"
+  ["CREATE TABLE datoms (e INTEGER NOT NULL, a SMALLINT NOT NULL, v BLOB NOT NULL, tx INTEGER NOT NULL,
+                         index_avet TINYINT NOT NULL DEFAULT 0, index_vaet TINYINT NOT NULL DEFAULT 0,
+                         unique_value TINYINT NOT NULL DEFAULT 0, unique_identity TINYINT NOT NULL DEFAULT 0)"
    "CREATE INDEX eavt ON datoms (e, a)" ;; No v -- that's an opt-in index.
    "CREATE INDEX aevt ON datoms (a, e)" ;; No v -- that's an opt-in index.
-   "CREATE INDEX avet ON datoms (a, v, e) WHERE index_avet = 1" ;; Opt-in index: only if a has :db/index true.
-   "CREATE INDEX vaet ON datoms (v, a, e) WHERE index_vaet = 1" ;; Opt-in index: only if a has :db/valueType :db.type/ref
+   "CREATE UNIQUE INDEX avet ON datoms (a, v, e) WHERE index_avet = 1" ;; Opt-in index: only if a has :db/index true.
+   "CREATE UNIQUE INDEX vaet ON datoms (v, a, e) WHERE index_vaet = 1" ;; Opt-in index: only if a has :db/valueType :db.type/ref
+   "CREATE UNIQUE INDEX unique_value ON datoms (v) WHERE unique_value = 1" ;; TODO.
+   "CREATE UNIQUE INDEX unique_identity ON datoms (a, v) WHERE unique_identity = 1" ;; TODO.
    "CREATE TABLE transactions (e INTEGER NOT NULL, a SMALLINT NOT NULL, v BLOB NOT NULL, tx INTEGER NOT NULL, added TINYINT NOT NULL DEFAULT 1)"
    "CREATE INDEX tx ON transactions (tx)"
    "CREATE TABLE attributes (name TEXT NOT NULL PRIMARY KEY, a INTEGER UNIQUE NOT NULL)"])
diff --git a/test/datomish/db_test.cljc b/test/datomish/db_test.cljc
index 8c4bf97e..7ad7c1f2 100644
--- a/test/datomish/db_test.cljc
+++ b/test/datomish/db_test.cljc
@@ -45,6 +45,15 @@
         (filter #(not (= :db/txInstant (second %))))
         (set)))))
 
+(defn- <shallow-entity [db eid]
+  (let [entids (zipmap (vals (dm/idents db)) (keys (dm/idents db)))]
+    (go-pair
+      (->>
+        (s/all-rows (:sqlite-connection db) ["SELECT a, v FROM datoms WHERE e = ?" eid])
+        (<?)
+        (mapv #(vector (entids (:a %)) (:v %)))
+        (reduce conj {})))))
+
 (defn- <transactions [db]
   (let [entids (zipmap (vals (dm/idents db)) (keys (dm/idents db)))]
     (go-pair
@@ -57,14 +66,20 @@
   (get-in report [:db-after :current-tx]))
 
 (def test-schema
-  {:x    {:db/unique    :db.unique/identity
-          :db/valueType :db.type/integer}
-   :y    {:db/cardinality :db.cardinality/many
-          :db/valueType   :db.type/integer}
-   :name {:db/unique    :db.unique/identity
-          :db/valueType :db.type/string}
-   :aka  {:db/cardinality :db.cardinality/many
-          :db/valueType   :db.type/string}})
+  {:x     {:db/unique    :db.unique/identity
+           :db/valueType :db.type/integer}
+   :y     {:db/cardinality :db.cardinality/many
+           :db/valueType   :db.type/integer}
+   :name  {:db/unique    :db.unique/identity
+           :db/valueType :db.type/string}
+   :aka   {:db/cardinality :db.cardinality/many
+           :db/valueType   :db.type/string}
+   :age   {:db/valueType :db.type/integer}
+   :email {:db/unique    :db.unique/identity
+           :db/valueType :db.type/string}
+   :spouse {:db/unique    :db.unique/value
+            :db/valueType :db.type/string}
+   })
 
 (deftest-async test-add-one
   (with-tempfile [t (tempfile)]
@@ -159,6 +174,28 @@
         (finally
           (<? (dm/close-db db)))))))
 
+(deftest-async test-unique
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c test-schema))
+          conn (dm/connection-with-db db)
+          now  -1]
+      (try
+        (testing "Multiple :db/unique values in tx-data violate unique constraint, no tempid"
+          (is (thrown-with-msg?
+                ExceptionInfo #"unique constraint"
+                (<? (dm/<transact! conn [[:db/add 1 :x 0]
+                                         [:db/add 2 :x 0]] now)))))
+
+        (testing "Multiple :db/unique values in tx-data violate unique constraint, tempid"
+          (is (thrown-with-msg?
+                ExceptionInfo #"unique constraint"
+                (<? (dm/<transact! conn [[:db/add (dm/id-literal :db.part/user -1) :spouse "Dana"]
+                                         [:db/add (dm/id-literal :db.part/user -2) :spouse "Dana"]] now)))))
+
+        (finally
+          (<? (dm/close-db db)))))))
+
 (deftest-async test-add-ident
   (with-tempfile [t (tempfile)]
     (let [c    (<? (s/<sqlite-connection t))
@@ -219,3 +256,148 @@
 
         (finally
           (<? (dm/close-db db)))))))
+
+(deftest-async test-vector-upsert
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c test-schema))
+          conn (dm/connection-with-db db)
+          now  0xdeadbeef]
+      (try
+        ;; Not having DB-as-value really hurts us here.
+        (let [<with-base-and (fn [entities]
+                               (go-pair
+                                 (<? (s/execute! (:sqlite-connection (dm/db conn)) ["DELETE FROM datoms"]))
+                                 (<? (s/execute! (:sqlite-connection (dm/db conn)) ["DELETE FROM transactions"]))
+                                 ;; TODO: don't rely on explicit IDs.
+                                 (<? (dm/<transact! conn [{:db/id 1 :name "Ivan" :email "@1"}
+                                                          {:db/id 2 :name "Petr" :email "@2"}] now))
+                                 (<? (dm/<transact! conn entities now))))
+              tempids (fn [tx] (into {} (map (juxt (comp :idx first) second) (:tempids tx))))]
+
+          (testing "upsert with tempid"
+            (let [tx (<? (<with-base-and [[:db/add (dm/id-literal :db.part/user -1) :name "Ivan"]
+                                          [:db/add (dm/id-literal :db.part/user -1) :age 12]]))]
+              (is (= (<? (<shallow-entity (dm/db conn) 1))
+                     {:name "Ivan" :age 12 :email "@1"}))
+              (is (= (tempids tx)
+                     {-1 1}))))
+
+          (testing "upsert with tempid, order does not matter"
+            (let [tx (<? (<with-base-and [[:db/add (dm/id-literal :db.part/user -1) :age 12]
+                                          [:db/add (dm/id-literal :db.part/user -1) :name "Ivan"]]))]
+              (is (= (<? (<shallow-entity (dm/db conn) 1))
+                     {:name "Ivan" :age 12 :email "@1"}))
+              (is (= (tempids tx)
+                     {-1 1}))))
+
+          (testing "Conflicting upserts fail"
+            (is (thrown-with-msg? Throwable #"Conflicting upsert: #datomish.db.TempId\{:part :db.part/user, :idx -\d+\} resolves both to \d+ and \d+"
+                                  (<? (dm/<with db [[:db/add (dm/id-literal :db.part/user -1) :name "Ivan"]
+                                                    [:db/add (dm/id-literal :db.part/user -1) :age 35]
+                                                    [:db/add (dm/id-literal :db.part/user -1) :name "Petr"]
+                                                    [:db/add (dm/id-literal :db.part/user -1) :age 36]]))))))
+        (finally
+          (<? (dm/close-db db)))))))
+
+(deftest-async test-map-upsert
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c test-schema))
+          conn (dm/connection-with-db db)
+          now  0xdeadbeef]
+      (try
+        ;; Not having DB-as-value really hurts us here.
+        (let [<with-base-and (fn [entities]
+                               (go-pair
+                                 (<? (s/execute! (:sqlite-connection (dm/db conn)) ["DELETE FROM datoms"]))
+                                 (<? (s/execute! (:sqlite-connection (dm/db conn)) ["DELETE FROM transactions"]))
+                                 ;; TODO: don't rely on explicit IDs.
+                                 (<? (dm/<transact! conn [{:db/id 1 :name "Ivan" :email "@1"}
+                                                          {:db/id 2 :name "Petr" :email "@2"}] now))
+                                 (<? (dm/<transact! conn entities now))))
+              tempids (fn [tx] (into {} (map (juxt (comp :idx first) second) (:tempids tx))))]
+
+          (testing "upsert with tempid"
+            (let [tx (<? (<with-base-and [{:db/id (dm/id-literal :db.part/user -1) :name "Ivan" :age 35}]))]
+              (is (= (<? (<shallow-entity (dm/db conn) 1))
+                     {:name "Ivan" :email "@1" :age 35}))
+              (is (= (tempids tx)
+                     {-1 1}))))
+
+          (testing "upsert by 2 attrs with tempid"
+            (let [tx (<? (<with-base-and [{:db/id (dm/id-literal :db.part/user -1) :name "Ivan" :email "@1" :age 35}]))]
+              (is (= (<? (<shallow-entity (dm/db conn) 1))
+                     {:name "Ivan" :email "@1" :age 35}))
+              (is (= (tempids tx)
+                     {-1 1}))))
+
+          (testing "upsert to two entities, resolve to same tempid, fails due to overlapping writes"
+            (is (thrown-with-msg? Throwable #"cardinality constraint"
+                                  (<? (<with-base-and [{:db/id (dm/id-literal :db.part/user -1) :name "Ivan" :age 35}
+                                                       {:db/id (dm/id-literal :db.part/user -1) :name "Ivan" :age 36}])))))
+
+          (testing "upsert to two entities, two tempids, fails due to overlapping writes"
+            (is (thrown-with-msg? Throwable #"cardinality constraint"
+                                  (<? (<with-base-and [{:db/id (dm/id-literal :db.part/user -1) :name "Ivan" :age 35}
+                                                       {:db/id (dm/id-literal :db.part/user -2) :name "Ivan" :age 36}])))))
+
+          (testing "upsert with existing id"
+            (let [tx (<? (<with-base-and [{:db/id 1 :name "Ivan" :age 35}]))]
+              (is (= (<? (<shallow-entity (dm/db conn) 1))
+                     {:name "Ivan" :email "@1" :age 35}))
+              (is (= (tempids tx)
+                     {}))))
+
+          (testing "upsert by 2 attrs with existing id"
+            (let [tx (<? (<with-base-and [{:db/id 1 :name "Ivan" :email "@1" :age 35}]))]
+              (is (= (<? (<shallow-entity (dm/db conn) 1))
+                     {:name "Ivan" :email "@1" :age 35}))
+              (is (= (tempids tx)
+                     {})))))
+
+        (finally
+          (<? (dm/close-db db)))))))
+
+(deftest-async test-map-upsert-conflicts
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c test-schema))
+          conn (dm/connection-with-db db)
+          now  0xdeadbeef]
+      (try
+        ;; Not having DB-as-value really hurts us here.
+        (let [<with-base-and (fn [entities]
+                               (go-pair
+                                 (<? (s/execute! (:sqlite-connection (dm/db conn)) ["DELETE FROM datoms"]))
+                                 (<? (s/execute! (:sqlite-connection (dm/db conn)) ["DELETE FROM transactions"]))
+                                 ;; TODO: don't rely on explicit IDs.
+                                 (<? (dm/<transact! conn [{:db/id 1 :name "Ivan" :email "@1"}
+                                                          {:db/id 2 :name "Petr" :email "@2"}] now))
+                                 (<? (dm/<transact! conn entities now))))
+              tempids (fn [tx] (into {} (map (juxt (comp :idx first) second) (:tempids tx))))]
+
+          ;; TODO: improve error message to refer to upsert inputs.
+          (testing "upsert conficts with existing id"
+            (is (thrown-with-msg? Throwable #"unique constraint"
+                                  (<? (<with-base-and [{:db/id 2 :name "Ivan" :age 36}])))))
+
+          ;; TODO: improve error message to refer to upsert inputs.
+          (testing "upsert conficts with non-existing id"
+            (is (thrown-with-msg? Throwable #"unique constraint"
+                                  (<? (<with-base-and [{:db/id 3 :name "Ivan" :age 36}])))))
+
+          (testing "upsert by non-existing value resolves as update"
+            (let [tx (<? (<with-base-and [{:db/id (dm/id-literal :db.part/user -1) :name "Ivan" :email "@3" :age 35}]))]
+              (is (= (<? (<shallow-entity (dm/db conn) 1))
+                     {:name "Ivan" :email "@3" :age 35}))
+              (is (= (tempids tx)
+                     {-1 1}))))
+
+          ;; TODO: improve error message to refer to upsert inputs.
+          (testing "upsert by 2 conflicting fields"
+            (is (thrown-with-msg? Throwable #"Conflicting upsert: #datomish.db.TempId\{:part :db.part/user, :idx -\d+\} resolves both to \d+ and \d+"
+                                  (<? (<with-base-and [{:db/id (dm/id-literal :db.part/user -1) :name "Ivan" :email "@2" :age 35}]))))))
+
+        (finally
+          (<? (dm/close-db db)))))))