From baec3815b0d0cab4b93c1d4556617498e4a22f91 Mon Sep 17 00:00:00 2001
From: Nick Alexander <nalexander@mozilla.com>
Date: Wed, 27 Jul 2016 14:29:16 -0700
Subject: [PATCH] Implement transactions.

---
 src/datomish/datom.cljc    | 136 ++++++++++
 src/datomish/db.cljc       | 507 ++++++++++++++++++++++++++++++++++++-
 src/datomish/schema.cljc   |  96 +++++++
 test/datomish/db_test.cljc | 149 +++++++++++
 test/datomish/test.cljs    |   2 +
 5 files changed, 876 insertions(+), 14 deletions(-)
 create mode 100644 src/datomish/datom.cljc
 create mode 100644 src/datomish/schema.cljc
 create mode 100644 test/datomish/db_test.cljc

diff --git a/src/datomish/datom.cljc b/src/datomish/datom.cljc
new file mode 100644
index 00000000..78b61b9f
--- /dev/null
+++ b/src/datomish/datom.cljc
@@ -0,0 +1,136 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+;; Purloined from DataScript.
+
+(ns datomish.datom)
+
+(declare hash-datom equiv-datom seq-datom val-at-datom nth-datom assoc-datom)
+
+(deftype Datom [e a v tx added]
+  #?@(:cljs
+      [IHash
+       (-hash [d] (or (.-__hash d)
+                      (set! (.-__hash d) (hash-datom d))))
+       IEquiv
+       (-equiv [d o] (and (instance? Datom o) (equiv-datom d o)))
+
+       ISeqable
+       (-seq [d] (seq-datom d))
+
+       ILookup
+       (-lookup [d k] (val-at-datom d k nil))
+       (-lookup [d k nf] (val-at-datom d k nf))
+
+       IIndexed
+       (-nth [this i] (nth-datom this i))
+       (-nth [this i not-found] (nth-datom this i not-found))
+
+       IAssociative
+       (-assoc [d k v] (assoc-datom d k v))
+
+       IPrintWithWriter
+       (-pr-writer [d writer opts]
+                   (pr-sequential-writer writer pr-writer
+                                         "#datascript/Datom [" " " "]"
+                                         opts [(.-e d) (.-a d) (.-v d) (.-tx d) (.-added d)]))
+       ]
+      :clj
+      [Object
+       (hashCode [d] (hash-datom d))
+
+       clojure.lang.IHashEq
+       (hasheq [d] (hash-datom d))
+
+       clojure.lang.Seqable
+       (seq [d] (seq-datom d))
+
+       clojure.lang.IPersistentCollection
+       (equiv [d o] (and (instance? Datom o) (equiv-datom d o)))
+       (empty [d] (throw (UnsupportedOperationException. "empty is not supported on Datom")))
+       (count [d] 5)
+       (cons [d [k v]] (assoc-datom d k v))
+
+       clojure.lang.Indexed
+       (nth [this i]           (nth-datom this i))
+       (nth [this i not-found] (nth-datom this i not-found))
+
+       clojure.lang.ILookup
+       (valAt [d k] (val-at-datom d k nil))
+       (valAt [d k nf] (val-at-datom d k nf))
+
+       clojure.lang.Associative
+       (entryAt [d k] (some->> (val-at-datom d k nil) (clojure.lang.MapEntry k)))
+       (containsKey [e k] (#{:e :a :v :tx :added} k))
+       (assoc [d k v] (assoc-datom d k v))
+       ]))
+
+(defn ^Datom datom
+  ([e a v tx]       (Datom. e a v tx true))
+  ([e a v tx added] (Datom. e a v tx added)))
+
+(defn datom? [x] (instance? Datom x))
+
+(defn- hash-datom [^Datom d]
+  (-> (hash (.-e d))
+      (hash-combine (hash (.-a d)))
+      (hash-combine (hash (.-v d)))))
+
+(defn- equiv-datom [^Datom d ^Datom o]
+  (and (= (.-e d) (.-e o))
+       (= (.-a d) (.-a o))
+       (= (.-v d) (.-v o))))
+
+(defn- seq-datom [^Datom d]
+  (list (.-e d) (.-a d) (.-v d) (.-tx d) (.-added d)))
+
+;; keep it fast by duplicating for both keyword and string cases
+;; instead of using sets or some other matching func
+(defn- val-at-datom [^Datom d k not-found]
+  (case k
+    :e     (.-e d)        "e"     (.-e d)
+    :a     (.-a d)        "a"     (.-a d)
+    :v     (.-v d)        "v"     (.-v d)
+    :tx    (.-tx d)       "tx"    (.-tx d)
+    :added (.-added d)    "added" (.-added d)
+    not-found))
+
+(defn- nth-datom
+  ([^Datom d ^long i]
+   (case i
+     0 (.-e d)
+     1 (.-a d)
+     2 (.-v d)
+     3 (.-tx d)
+     4 (.-added d)
+     #?(:clj  (throw (IndexOutOfBoundsException.))
+        :cljs (throw (js/Error. (str "Datom/-nth: Index out of bounds: " i))))))
+  ([^Datom d ^long i not-found]
+   (case i
+     0 (.-e d)
+     1 (.-a d)
+     2 (.-v d)
+     3 (.-tx d)
+     4 (.-added d)
+     not-found)))
+
+(defn- ^Datom assoc-datom [^Datom d k v]
+  (case k
+    :e     (Datom. v       (.-a d) (.-v d) (.-tx d) (.-added d))
+    :a     (Datom. (.-e d) v       (.-v d) (.-tx d) (.-added d))
+    :v     (Datom. (.-e d) (.-a d) v       (.-tx d) (.-added d))
+    :tx    (Datom. (.-e d) (.-a d) (.-v d) v        (.-added d))
+    :added (Datom. (.-e d) (.-a d) (.-v d) (.-tx d) v)
+    (throw (IllegalArgumentException. (str "invalid key for #datascript/Datom: " k)))))
+
+;; printing and reading
+
+(defn ^Datom datom-from-reader [vec]
+  (apply datom vec))
+
+#?(:clj
+   (defmethod print-method Datom [^Datom d, ^java.io.Writer w]
+     (.write w (str "#datascript/Datom "))
+     (binding [*out* w]
+       (pr [(.-e d) (.-a d) (.-v d) (.-tx d) (.-added d)]))))
diff --git a/src/datomish/db.cljc b/src/datomish/db.cljc
index cd5c0531..f2a61ca4 100644
--- a/src/datomish/db.cljc
+++ b/src/datomish/db.cljc
@@ -12,31 +12,308 @@
    [datomish.query.projection :as projection]
    [datomish.query.source :as source]
    [datomish.query :as query]
+   [honeysql.core :as sql]
+   [datomish.datom :as dd :refer [datom datom? #?@(:cljs [Datom])]]
+   [datomish.util :as util #?(:cljs :refer-macros :clj :refer) [raise raise-str cond-let]]
+   [datomish.schema :as ds]
    [datomish.sqlite :as s]
    [datomish.sqlite-schema :as sqlite-schema]
-   [datomish.util :as util #?(:cljs :refer-macros :clj :refer) [raise raise-str]]
    #?@(:clj [[datomish.pair-chan :refer [go-pair <?]]
              [clojure.core.async :as a :refer [chan go <! >!]]])
    #?@(:cljs [[datomish.pair-chan]
-              [cljs.core.async :as a :refer [chan <! >!]]])))
+              [cljs.core.async :as a :refer [chan <! >!]]]))
+  #?(:clj
+     (:import
+      [datomish.datom Datom])))
+
+#?(:clj
+   ;; From https://stuartsierra.com/2015/05/27/clojure-uncaught-exceptions
+   ;; Assuming require [clojure.tools.logging :as log]
+   (Thread/setDefaultUncaughtExceptionHandler
+     (reify Thread$UncaughtExceptionHandler
+       (uncaughtException [_ thread ex]
+         (println ex "Uncaught exception on" (.getName thread))))))
 
 (defprotocol IDB
   (query-context
     [db])
-  (close
-    [db]
-    "Close this database. Returns a pair channel of [nil error]."))
 
-(defrecord DB [sqlite-connection]
+  (close-db
+    [db]
+    "Close this database. Returns a pair channel of [nil error].")
+
+  (schema
+    [db]
+    "Return the schema of this database.")
+
+  (idents
+    [db]
+    "Return the known idents of this database, as a map from keyword idents to entids.")
+
+  (current-tx
+    [db]
+    "TODO: document this interface.")
+
+  (<eavt
+    [db pattern]
+    "Search for datoms using the EAVT index.")
+
+  (<avet
+    [db pattern]
+    "Search for datoms using the AVET index.")
+
+  (<apply-datoms
+    [db datoms]
+    "Apply datoms to the store.")
+
+  (<advance-tx
+    [db]
+    "TODO: document this interface."))
+
+;; TODO: handle _?
+(defn search->sql-clause [pattern]
+  (merge
+    {:select [:*] ;; e :a :v :tx] ;; TODO: generalize columns.
+     :from [:datoms]}
+    (if-not (empty? pattern)
+      {:where (cons :and (map #(vector := %1 (if (keyword? %2) (str %2) %2)) [:e :a :v :tx] pattern))} ;; TODO: use schema to process v.
+      {})))
+
+(defrecord DB [sqlite-connection schema idents current-tx]
+  ;; idents is map {ident -> entid} of known idents.  See http://docs.datomic.com/identity.html#idents.
   IDB
   (query-context [db] (context/->Context (source/datoms-source db) nil nil))
-  (close [db] (s/close (.-sqlite-connection db))))
 
-(defn <with-sqlite-connection [sqlite-connection]
+  (schema [db] (.-schema db))
+
+  (idents [db] (.-idents db))
+
+  (current-tx
+    [db]
+    (inc (:current-tx db)))
+
+  ;; TODO: use q for searching?  Have q use this for searching for a single pattern?
+  (<eavt [db pattern]
+    (go-pair
+      ;; TODO: find a better expression of this pattern.
+      (let [rows (<? (->>
+                       (search->sql-clause pattern)
+                       (sql/format)
+                       (s/all-rows (:sqlite-connection db))))]
+        (mapv #(Datom. (:e %) (:a %) (:v %) (:tx %) true) rows)))) ;; TODO: map values according to schema.
+
+  (<avet [db pattern]
+    (go-pair
+      ;; TODO: find a better expression of this pattern.
+      (let [[a v] pattern
+            rows (<? (->>
+                       {:select [:*] :from [:datoms] :where [:and [:= :a a] [:= :v v]]}
+                       (sql/format)
+                       (s/all-rows (:sqlite-connection db))))]
+        (mapv #(Datom. (:e %) (:a %) (:v %) (:tx %) true) rows)))) ;; TODO: map values according to schema.
+
+  (<apply-datoms [db datoms]
+    (go-pair
+      (let [exec (partial s/execute! (:sqlite-connection db))]
+        ;; TODO: batch insert, batch delete.
+        (doseq [datom datoms]
+          (let [[e a v tx added] datom]
+            ;; Append to transaction log.
+            (<? (exec
+                  ["INSERT INTO transactions VALUES (?, ?, ?, ?, ?)" e a v tx added]))
+            ;; 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]))
+              (<? (exec
+                    ;; TODO: verify this is correct.
+                    ["DELETE FROM datoms WHERE (e = ? AND a = ? AND v = ?)" e a v]))))))
+      db))
+
+  (<advance-tx [db]
+    (go-pair
+      (let [exec (partial s/execute! (:sqlite-connection db))]
+        ;; (let [ret (<? (exec
+        ;;                 ;; TODO: be more clever about UPDATE OR ...?
+        ;;                 ["UPDATE metadata SET current_tx = ? WHERE current_tx = ?" (inc (:current-tx db)) (:current-tx db)]))]
+
+        ;; TODO: handle exclusion across transactions here.
+        (update db :current-tx inc))))
+  ;;  )
+
+  (close-db [db] (s/close (.-sqlite-connection db))))
+
+(defn db? [x]
+  (and (satisfies? IDB x)))
+
+(defprotocol IConnection
+  (close
+    [conn]
+    "Close this connection. Returns a pair channel of [nil error].")
+
+  (db
+    [conn]
+    "Get the current DB associated with this connection.")
+
+  (history
+    [conn]
+    "Get the full transaction history DB associated with this connection."))
+
+(defrecord Connection [current-db]
+  IConnection
+  (close [conn] (close-db @(:current-db conn)))
+
+  (db [conn] @(:current-db conn))
+
+  (history [conn]
+    (raise "Datomic's history is not yet supported." {})))
+
+(defn conn? [x]
+  (and (satisfies? IConnection x)))
+
+;; ----------------------------------------------------------------------------
+;; define data-readers to be made available to EDN readers. in CLJS
+;; they're magically available. in CLJ, data_readers.clj may or may
+;; not work, but you can always simply do
+;;
+;;  (clojure.edn/read-string {:readers datomish/data-readers} "...")
+;;
+
+(defonce -id-literal-idx (atom -1000000))
+
+(defrecord TempId [part idx])
+
+(defn id-literal
+  ([part]
+   (if (sequential? part)
+     (apply id-literal part)
+     (->TempId part (swap! -id-literal-idx dec))))
+  ([part idx]
+   (->TempId part idx)))
+
+(defn id-literal? [x]
+  (and (instance? TempId x)))
+
+(defn temp-literal? [x]
+  (and (id-literal? x)
+       (= :db.part/temp (:part x))))
+
+;; (def data-readers {'db/id id-literal})
+
+;; #?(:cljs
+;;    (doseq [[tag cb] data-readers] (cljs.reader/register-tag-parser! tag cb)))
+
+;; TODO: implement support for DB parts?
+(def tx0 0x2000000)
+
+(defn <idents [sqlite-connection]
   (go-pair
-    (when-not (= sqlite-schema/current-version (<? (sqlite-schema/<ensure-current-version sqlite-connection)))
-      (raise-str "Could not ensure current SQLite schema version."))
-    (->DB sqlite-connection)))
+    (let [rows (<? (->>
+                     {:select [:e :v] :from [:datoms] :where [:= :a ":db/ident"]} ;; TODO: don't stringify?
+                     (sql/format)
+                     (s/all-rows sqlite-connection)))]
+      (into {} (map #(-> {(keyword (:v %)) (:e %)})) rows))))
+
+(defn <db-with-sqlite-connection
+  ([sqlite-connection]
+   (<db-with-sqlite-connection sqlite-connection {}))
+
+  ([sqlite-connection schema]
+   (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})] ;; TODO: pre-populate idents and SQLite tables?
+       (map->DB
+         {:sqlite-connection sqlite-connection
+          :idents            idents
+          :schema            (ds/schema (into {} (map (fn [[k v]] [(k idents) v]) schema)))
+          :current-tx        tx0})))))
+
+(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]
+  (cond
+    (datom? entity)
+    (->
+      (let [[e a v tx added] entity]
+        (if added
+          [:db/add [e a v tx]]
+          [:db/retract [e a v tx]]))
+      (with-meta (get (meta entity) :source)))
+
+    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]))
+
+    ;; (raise "Map entities are not yet supported, got " entity
+    ;;        {:error :transact/syntax
+    ;;         :op    entity })
+
+    true
+    [entity]))
+
+(defn maybe-ident->entid [db [op & entity :as orig]]
+  ;; TODO: use something faster than `into` here.
+  (->
+    (into [op] (for [field entity]
+                 (get (idents db) field field))) ;; TODO: schema, not db.
+    ;; (with-meta (get (meta orig) :source {:source orig}))
+    ))
+
+(defrecord Transaction [db tempids entities])
+
+(defn- tx-entity [db]
+  (let [tx (current-tx db)]
+    [:db/add tx :db/txInstant 0xdeadbeef tx])) ;; TODO: now.
+
+(defn maybe-add-current-tx [current-tx entity]
+  (let [[op e a v tx] entity]
+    [op e a v (or tx current-tx)]))
+
+(defn preprocess [db report]
+  (let [initial-es (conj (or (:entities report) []) (tx-entity db))]
+    (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 %}))
+
+          ;; Normalize Datoms into :db/add or :db/retract vectors.
+          (map maybe-datom->entity)
+
+          ;; 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)))
+
+          ;; Replace idents with entids where possible.
+          (map (partial maybe-ident->entid db))
+
+          ;; Add tx if not given.
+          (map (partial maybe-add-current-tx (current-tx db))))
+        (transduce conj [] initial-es))
+      (assoc-in report [:entities]))))
 
 (defn <?run
   "Execute the provided query on the provided DB.
@@ -45,8 +322,8 @@
   [db find args]
   (let [parsed (query/parse find)
         context (-> db
-                  query-context
-                  (query/find-into-context parsed))
+                    query-context
+                    (query/find-into-context parsed))
         row-pair-transducer (projection/row-pair-transducer context)
         sql (query/context->sql-string context args)
         chan (chan 50 row-pair-transducer)]
@@ -67,4 +344,206 @@
   [db find args]
   (a/reduce (partial reduce-error-pair conj) [[] nil]
             (<?run db find args)))
-    
+
+(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]
+  (if (contains? (:tempids report) tempid)
+    (go-pair
+      (raise "Conflicting upsert: " tempid " resolves"
+             " both to " upserted-eid " and " (get (:tempids report) tempid)
+             { :error :transact/upsert }))
+    ;; try to re-run from the beginning
+    ;; but remembering that `old-eid` will resolve to `upserted-eid`
+    (<resolve-id-literals db
+                          (->
+                            report
+                            (assoc-in [:tempids tempid] upserted-eid)
+                            (assoc-in [:entities] es)))))
+
+(defn- transact-entity [report entity]
+  (update-in report [:entities] conj entity))
+
+(defn <resolve-id-literals
+  "Upsert uniquely identified literals when possible and allocate new entids for all other id literals.
+
+  It's worth noting that some amount of trial and error is probably
+  necessary here, since [[-1 :ref -2] [-2 :ref -1]] is a valid input.
+  It's my belief that no graph algorithm can correctly order the
+  id-literals in quasi-linear time, since that algorithm will need to
+  accept all permutations of the id-literals.  Therefore, we simplify
+  by accepting that we may process the input multiple times, and we
+  regain some efficiency by sorting so that upserts happen earlier and
+  we are most likely to find a successful entid allocation without
+  multiple trials.
+
+  Concretely, we sort [-1 a v] < [-1 a -2] < [e a -1] < [e a v].  This
+  means simple upserts will be processed early, followed by entities
+  with multiple id-literals that we hope will reduce to simple upserts
+  based on the earlier upserts.  After that, we handle what should be
+  simple allocations."
+
+  [db report]
+  (go-pair
+    (let [keyfn (fn [[op e a v tx]]
+                  (if (and (id-literal? e)
+                           (not-any? id-literal? [a v tx]))
+                    (- 5)
+                    (- (count (filter id-literal? [e a v tx])))))
+          initial-report (dissoc report :entities) ;; TODO.
+          initial-entities (sort-by keyfn (:entities report))]
+      (loop [report initial-report
+             es initial-entities]
+        (let [[[op e a v tx :as entity] & entities] es]
+          (cond
+            (nil? entity)
+            ;; We can add :db.part/temp id-literals; remove them.
+            (update report :tempids #(into {} (filter (comp not temp-literal? first) %)))
+
+            (and (not= op :db/add)
+                 (not (empty? (filter id-literal? [e a v tx]))))
+            (raise "id-literals are resolved for :db/add only"
+                   {:error :transact/syntax
+                    :op    entity })
+
+            ;; Upsert!
+            (and (id-literal? e)
+                 (ds/unique-identity? (schema db) a) ;; TODO: schema.
+                 (not-any? id-literal? [a v tx]))
+            (let [upserted-eid (:e (first (<? (<avet db [a v]))))  ;; TODO: define this interface.
+                  allocated-eid (get-in report [:tempids e])]
+              (if (and upserted-eid allocated-eid (not= upserted-eid allocated-eid))
+                (do
+                  (<? (<retry-with-tempid db initial-report initial-entities e upserted-eid)) ;; TODO: not initial report, just the sorted entities here.
+                  )
+                (let [eid (or upserted-eid allocated-eid (next-eid db))]
+                  (recur (allocate-eid report e eid) (cons [op eid a v tx] entities)))))
+
+            ;; Start allocating and retrying.  We try with e last, so as to eventually upsert it.
+            (id-literal? tx)
+            ;; TODO: enforce tx part only?
+            (let [eid (or (get-in report [:tempids tx]) (next-eid db))]
+              (recur (allocate-eid report tx eid) (cons [op e a v eid] entities)))
+
+            (id-literal? v)
+            ;; We can't fail with unbound literals here, since we could have multiple
+            (let [eid (or (get-in report [:tempids v]) (next-eid db))]
+              (recur (allocate-eid report v eid) (cons [op e a eid tx] entities)))
+
+            (id-literal? a)
+            ;; TODO: should we even allow id-literal attributes?  Datomic fails in some cases here.
+            (let [eid (or (get-in report [:tempids a]) (next-eid db))]
+              (recur (allocate-eid report a eid) (cons [op e eid v tx] entities)))
+
+            (id-literal? e)
+            (let [eid (or (get-in report [:tempids e]) (next-eid db))]
+              (recur (allocate-eid report e eid) (cons [op eid a v tx] entities)))
+
+            true
+            (recur (transact-entity report entity) entities)
+            ))))))
+
+(defn- transact-report [report datom]
+  (update-in report [:tx-data] conj datom))
+
+(defn <postprocess [db report]
+  (go-pair
+    (let [initial-report report]
+      (loop [report initial-report
+             es (:entities initial-report)]
+        (let [[[op e a v tx :as entity] & entities] es]
+          (cond
+            (nil? entity)
+            report
+
+            (= op :db/add)
+            (if (ds/multival? (schema db) a)
+              (if (empty? (<? (<eavt db [e a v])))
+                (recur (transact-report report (datom e a v tx true)) entities)
+                (recur report entities))
+              (if-let [^Datom old-datom (first (<? (<eavt db [e a])))]
+                (if  (= (.-v old-datom) v)
+                  (recur report entities)
+                  (recur (-> report
+                             (transact-report (datom e a (.-v old-datom) tx false))
+                             (transact-report (datom e a v tx true)))
+                         entities))
+                (recur (transact-report report (datom e a v tx true)) entities)))
+
+            (= op :db/retract)
+            (if (first (<? (<eavt db [e a v])))
+              (recur (transact-report report (datom e a v tx false)) entities)
+              (recur report entities))
+
+            true
+            (raise "Unknown operation at " entity ", expected :db/add, :db/retract"
+                   {:error :transact/syntax, :operation op, :tx-data entity})))))))
+
+(defn <transact-tx-data
+  [db now initial-report]
+  {:pre [(db? db)]}
+
+  (go-pair
+    (->> initial-report
+         (preprocess db)
+
+         (<resolve-id-literals db)
+         (<?)
+
+         (<postprocess db)
+         (<?))))
+
+;; Normalize as [op int|id-literal int|id-literal value|id-literal tx|id-literal]. ;; TODO: mention lookup-refs.
+
+;; Replace lookup-refs with entids where possible.
+
+;; Upsert or allocate id-literals.
+
+(defn <with [db tx-data]
+  (go-pair
+    (let [report (<? (<transact-tx-data db 0xdeadbeef ;; TODO
+                                        (map->TxReport
+                                          {:db-before  db
+                                           :db-after   db
+                                           ;; :current-tx current-tx
+                                           :entities   tx-data
+                                           :tx-data    []
+                                           :tempids    {}})))
+          db-after (->
+                     db
+
+                     (<apply-datoms (:tx-data report))
+                     (<?)
+
+                     (<advance-tx)
+                     (<?))]
+
+      (-> report
+          (assoc-in [:db-after] db-after)))))
+
+(defn <db-with [db tx-data]
+  (go-pair
+    (:db-after (<? (<with db tx-data)))))
+
+(defn <transact!
+  ([conn tx-data]
+   (<transact! conn tx-data 0xdeadbeef)) ;; TODO: timestamp!
+  ([conn tx-data now]
+   {:pre [(conn? conn)]}
+   (let [db (db conn)] ;; TODO: be careful with swapping atoms.
+     (s/in-transaction!
+       (:sqlite-connection db)
+       #(go-pair
+          (let [report (<? (<with db tx-data))] ;; TODO: timestamp!
+            (reset! (:current-db conn) (:db-after report))
+            report))))))
diff --git a/src/datomish/schema.cljc b/src/datomish/schema.cljc
new file mode 100644
index 00000000..fd3e614c
--- /dev/null
+++ b/src/datomish/schema.cljc
@@ -0,0 +1,96 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+;; Purloined from DataScript.
+
+(ns datomish.schema)
+
+(defprotocol ISchema
+  (attrs-by
+    [schema property]
+    "TODO: document this, think more about making this part of the schema."))
+
+(defn- #?@(:clj  [^Boolean is-attr?]
+           :cljs [^boolean is-attr?]) [schema attr property]
+  (contains? (attrs-by schema property) attr))
+
+(defn #?@(:clj  [^Boolean multival?]
+          :cljs [^boolean multival?]) [schema attr]
+  (is-attr? schema attr :db.cardinality/many))
+
+(defn #?@(:clj  [^Boolean ref?]
+          :cljs [^boolean ref?]) [schema attr]
+  (is-attr? schema attr :db.type/ref))
+
+(defn #?@(:clj  [^Boolean component?]
+          :cljs [^boolean component?]) [schema attr]
+  (is-attr? schema attr :db/isComponent))
+
+(defn #?@(:clj  [^Boolean indexing?]
+          :cljs [^boolean indexing?]) [schema attr]
+  (is-attr? schema attr :db/index))
+
+(defn #?@(:clj  [^Boolean unique-identity?]
+          :cljs [^boolean unique-identity?]) [schema attr]
+  (is-attr? schema attr :db.unique/identity))
+
+(defn #?@(:clj  [^Boolean unique-value?]
+          :cljs [^boolean unique-value?]) [schema attr]
+  (is-attr? schema attr :db.unique/value))
+
+(defrecord Schema [schema rschema]
+  ISchema
+  (attrs-by [schema property]
+    ((.-rschema schema) property)))
+
+(defn- attr->properties [k v]
+  (cond
+    (= [k v] [:db/isComponent true]) [:db/isComponent]
+    (= v :db.type/ref)               [:db.type/ref :db/index]
+    (= v :db.cardinality/many)       [:db.cardinality/many]
+    (= v :db.unique/identity)        [:db/unique :db.unique/identity :db/index]
+    (= v :db.unique/value)           [:db/unique :db.unique/value    :db/index]
+    (= [k v] [:db/index true])       [:db/index]))
+
+(defn- multimap [e m]
+  (reduce
+    (fn [acc [k v]]
+      (update-in acc [k] (fnil conj e) v))
+    {} m))
+
+(defn- rschema [schema]
+  (->>
+    (for [[a kv] schema
+          [k v]  kv
+          prop   (attr->properties k v)]
+      [prop a])
+    (multimap #{})))
+
+(defn- validate-schema-key [a k v expected]
+  (when-not (or (nil? v)
+                (contains? expected v))
+    (throw (ex-info (str "Bad attribute specification for " (pr-str {a {k v}}) ", expected one of " expected)
+                    {:error :schema/validation
+                     :attribute a
+                     :key k
+                     :value v}))))
+
+(defn- validate-schema [schema]
+  (doseq [[a kv] schema]
+    (let [comp? (:db/isComponent kv false)]
+      (validate-schema-key a :db/isComponent (:db/isComponent kv) #{true false})
+      (when (and comp? (not= (:db/valueType kv) :db.type/ref))
+        (throw (ex-info (str "Bad attribute specification for " a ": {:db/isComponent true} should also have {:db/valueType :db.type/ref}")
+                        {:error     :schema/validation
+                         :attribute a
+                         :key       :db/isComponent}))))
+    (validate-schema-key a :db/unique (:db/unique kv) #{:db.unique/value :db.unique/identity})
+    (validate-schema-key a :db/valueType (:db/valueType kv) #{:db.type/ref})
+    (validate-schema-key a :db/cardinality (:db/cardinality kv) #{:db.cardinality/one :db.cardinality/many}))
+  schema)
+
+(defn schema [schema]
+  {:pre [(or (nil? schema) (map? schema))]}
+  (map->Schema {:schema  (validate-schema schema)
+                :rschema (rschema schema)}))
diff --git a/test/datomish/db_test.cljc b/test/datomish/db_test.cljc
new file mode 100644
index 00000000..4255bd14
--- /dev/null
+++ b/test/datomish/db_test.cljc
@@ -0,0 +1,149 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+(ns datomish.db-test
+  #?(:cljs
+     (:require-macros
+      [datomish.pair-chan :refer [go-pair <?]]
+      [datomish.node-tempfile-macros :refer [with-tempfile]]
+      [cljs.core.async.macros :as a :refer [go]]))
+  (:require
+   [datomish.util :as util #?(:cljs :refer-macros :clj :refer) [raise cond-let]]
+   [datomish.sqlite :as s]
+   [datomish.datom]
+
+   [datascript.core :as d]
+   [datascript.db :as db]
+
+   [datomish.db :as dm]
+   #?@(:clj [[datomish.pair-chan :refer [go-pair <?]]
+             [tempfile.core :refer [tempfile with-tempfile]]
+             [datomish.test-macros :refer [deftest-async]]
+             [clojure.test :as t :refer [is are deftest testing]]
+             [clojure.core.async :refer [go <! >!]]])
+   #?@(:cljs [[datomish.pair-chan]
+              [datomish.test-macros :refer-macros [deftest-async]]
+              [datomish.node-tempfile :refer [tempfile]]
+              [cljs.test :as t :refer-macros [is are deftest testing async]]
+              [cljs.core.async :as a :refer [<! >!]]]))
+  #?(:clj
+     (:import [clojure.lang ExceptionInfo]))
+  #?(:clj
+     (:import [datascript.db DB])))
+
+(defn- <datoms [db]
+  (let [entids (zipmap (vals (dm/idents db)) (keys (dm/idents db)))]
+    (go-pair
+      (->>
+        (s/all-rows (:sqlite-connection db) ["SELECT e, a, v, tx FROM datoms"])
+        (<?)
+        (mapv #(vector (:e %) (entids (:a %)) (:v %)))
+        (filter #(not (= :db/txInstant (second %))))
+        (set)))))
+
+(defn- <transactions [db]
+  (let [entids (zipmap (vals (dm/idents db)) (keys (dm/idents db)))]
+    (go-pair
+      (->>
+        (s/all-rows (:sqlite-connection db) ["SELECT e, a, v, tx, added FROM transactions ORDER BY tx ASC, e, a, v, added"])
+        (<?)
+        (mapv #(vector (:e %) (entids (:a %)) (:v %) (:tx %) (:added %)))))))
+
+(defn tx [report]
+  (get-in report [:db-after :current-tx]))
+
+(deftest-async test-add-one
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c))
+          conn (dm/connection-with-db db)
+          now  0xdeadbeef]
+      (try
+        (let [;; TODO: drop now, allow to set :db/txInstant.
+              report (<? (dm/<transact! conn [[:db/add 0 :x "valuex"]] now))
+              tx     (tx report)]
+          (is (= (<? (<datoms (dm/db conn)))
+                 #{[0 :x "valuex"]}))
+          (is (= (<? (<transactions (dm/db conn)))
+                 [[0 :x "valuex" tx 1] ;; TODO: true, not 1.
+                  [tx :db/txInstant now tx 1]])))
+        (finally
+          (<? (dm/close-db db)))))))
+
+(deftest-async test-add-two
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c
+                                                  {:x {:db/unique :db.unique/identity} ;; TODO: :name and :aka.
+                                                   :y {:db/cardinality :db.cardinality/many}}))
+          conn (dm/connection-with-db db)
+          now  0xdeadbeef]
+      (try
+        (let [tx1 (tx (<? (dm/<transact! conn [[:db/add 1 :x "Ivan"]] now)))
+              tx2 (tx (<? (dm/<transact! conn [[:db/add 1 :x "Petr"]] now)))
+              tx3 (tx (<? (dm/<transact! conn [[:db/add 1 :y "Tupen"]] now)))
+              tx4 (tx (<? (dm/<transact! conn [[:db/add 1 :y "Devil"]] now)))]
+          (is (= (<? (<datoms (dm/db conn)))
+                 #{[1 :x "Petr"]
+                   [1 :y "Tupen"]
+                   [1 :y "Devil"]}))
+
+          (is (= (<? (<transactions (dm/db conn)))
+                 [[1 :x "Ivan" tx1 1] ;; TODO: true, not 1.
+                  [tx1 :db/txInstant now tx1 1]
+                  [1 :x "Ivan" tx2 0]
+                  [1 :x "Petr" tx2 1]
+                  [tx2 :db/txInstant now tx2 1]
+                  [1 :y "Tupen" tx3 1]
+                  [tx3 :db/txInstant now tx3 1]
+                  [1 :y "Devil" tx4 1]
+                  [tx4 :db/txInstant now tx4 1]])))
+        (finally
+          (<? (dm/close-db db)))))))
+
+;; TODO: fail multiple :add and :retract of the same datom in the same transaction.
+(deftest-async test-retract
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c))
+          conn (dm/connection-with-db db)
+          now  0xdeadbeef]
+      (try
+        (let [txa (tx (<? (dm/<transact! conn [[:db/add     0 :x "valuex"]] now)))
+              txb (tx (<? (dm/<transact! conn [[:db/retract 0 :x "valuex"]] now)))]
+          (is (= (<? (<datoms db))
+                 #{}))
+          (is (= (<? (<transactions db))
+                 [[0 :x "valuex" txa 1] ;; TODO: true, not 1.
+                  [txa :db/txInstant now txa 1]
+                  [0 :x "valuex" txb 0]
+                  [txb :db/txInstant now txb 1]])))
+        (finally
+          (<? (dm/close-db db)))))))
+
+(deftest-async test-id-literal-1
+  (with-tempfile [t (tempfile)]
+    (let [c    (<? (s/<sqlite-connection t))
+          db   (<? (dm/<db-with-sqlite-connection c))
+          conn (dm/connection-with-db db)
+          now  -1]
+      (try
+        (let [report (<? (dm/<transact! conn [[:db/add (dm/id-literal :db.part/user -1) :x 0]
+                                              [:db/add (dm/id-literal :db.part/user -1) :y 1]
+                                              [:db/add (dm/id-literal :db.part/user -2) :y 2]
+                                              [:db/add (dm/id-literal :db.part/user -2) :y 3]] now))]
+          (is (= (keys (:tempids report)) ;; TODO: include values.
+                 [(dm/id-literal :db.part/user -1)
+                  (dm/id-literal :db.part/user -2)]))
+
+          (let [eid1 (get-in report [:tempids (dm/id-literal :db.part/user -1)])
+                eid2 (get-in report [:tempids (dm/id-literal :db.part/user -2)])]
+            (is (= (<? (<datoms db))
+                   #{[eid1 :x 0]
+                     [eid1 :y 1]
+                     [eid2 :y 2]
+                     [eid2 :y 3]}))))
+
+        (finally
+          (<? (dm/close-db db)))))))
diff --git a/test/datomish/test.cljs b/test/datomish/test.cljs
index 97eab507..b7327d03 100644
--- a/test/datomish/test.cljs
+++ b/test/datomish/test.cljs
@@ -3,6 +3,7 @@
    [doo.runner :refer-macros [doo-tests doo-all-tests]]
    [cljs.test :as t :refer-macros [is are deftest testing]]
    datomish.promise-sqlite-test
+   datomish.db-test
    datomish.sqlite-user-version-test
    datomish.test.util
    datomish.test.transforms
@@ -11,6 +12,7 @@
 
 (doo-tests
   'datomish.promise-sqlite-test
+  'datomish.db-test
   'datomish.sqlite-user-version-test
   'datomish.test.util
   'datomish.test.transforms