Implement schema management proposal. Fixes #95. r=grisha

src/common/datomish/api.cljc

@@ -35,6 +35,7 @@
 (def <close transact/close)
 
 (def id-literal db/id-literal)
+(def id-literal? db/id-literal?)
 
 (def lookup-ref db/lookup-ref)
 

src/common/datomish/pair_chan.cljc

@@ -71,6 +71,13 @@
      (consume-pair (cljs.core.async/<! ~pc-chan))
      (consume-pair (clojure.core.async/<! ~pc-chan))))
 
+(defmacro <??
+  "Takes from the channel if it's non-nil."
+  [pc-chan]
+  `(let [c# ~pc-chan]
+     (when c#
+       (datomish.pair-chan/<? c#))))
+
 (defn consume-pair
   "When passed a [value nil] pair, returns value. When passed a [nil error] pair,
   throws error. See also `<?`."

src/common/datomish/schema.cljc

@@ -178,7 +178,7 @@
       (raise "Unknown attribute " attr ", expected one of " (sorted-set (keys schema))
              {:error :schema/valueType, :attribute attr}))))
 
-(defn- validate-schema [schema]
+(defn validate-schema [schema]
   (doseq [[a kv] schema]
     (when-not (:db/valueType kv)
       (throw (ex-info (str "Bad attribute specification for " a ": should have {:db/valueType ...}")

src/common/datomish/schema_management.cljc

@@ -0,0 +1,348 @@
+;; 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.schema-management
+  #?(:cljs
+     (:require-macros
+      [datomish.pair-chan :refer [go-pair <?? <?]]))
+  (:require
+   [clojure.data :refer [diff]]
+   #?@(:clj [[datomish.pair-chan :refer [go-pair <?? <?]]])
+   [clojure.set]
+   [datomish.api :as d]
+   [datomish.schema]        ; For validation.
+   [datomish.util :as util
+    #?(:cljs :refer-macros :clj :refer) [raise raise-str cond-let]]))
+
+;; For testing.
+(def log (fn [& args]) #_ println)
+
+;; This code implements the concept described in
+;; https://github.com/mozilla/datomish/wiki/Proposal:-application-schema-coordination-and-versioning
+;;
+;; This is a high-level API: it's built on top of the schema alteration
+;; primitives and plain ol' storage layer that Datomish exposes.
+;;
+;; Schema fragments are described by name and version number.
+;; The existing schema in the store is diffed against incoming fragments.
+;;
+;; Validation:
+;; - No attribute should be mentioned in a different fragment in the
+;;   store and the input. Attributes cannot move between fragments.
+;; - Fragments with the same name and version should be congruent,
+;;   with the only acceptable changes being to add new attributes.
+;; - Fragments with an increased version number can make changes to
+;;   attributes:
+;;   - Adding new attributes.
+;;   - Rename existing attributes.
+;;   - Altering cardinality, uniqueness, or indexing properties.
+;;
+;; The inputs to the diffing process are:
+;; - The set of schema fragments.
+;; - The existing database (and implicitly its active schema).
+;; - A collection of attribute renames.
+;; - A set of app- and fragment-scoped pre/post functions that will
+;;   be run before and after schema changes are applied.
+;;
+;; The output of the diffing process, if validation succeeds, is a
+;; set of operations to perform on the knowledge base. If there are
+;; no version changes, no pre/post functions will be included.
+;;
+;; Potential outcomes:
+;; - An attribute is mentioned in two incoming fragments: error.
+;; - An attribute is in fragment A in the store and fragment B in input: error.
+;; - An attribute changed between the store and input, but the version number is the same: error.
+;; - An attribute is present in input, but not the store, and the version number is the same: add the attribute.
+;; - An attribute is present in the store, but not input, and the version number is the same: do nothing.
+;; - A fragment's version number is higher in the store than in the input:
+;;   - If the input is a subset of the fragment in the store, then do nothing.
+;;   - If the input differs, then error.
+;; - A fragment's version number is higher in the input than in the store:
+;;   - Run app 'pre' and 'post'.
+;;   - Run this fragment's 'pre' and 'post'.
+;;   - Alter the store to match. If altering fails due to a consistency error, roll back and error out.
+;;
+;; The core data format here, which we call a "managed schema fragment" is:
+;;
+;; {:name :org.mozilla.foo
+;;  :version 4
+;;  :attributes {:foo/bar {:db/valueType ...}}}
+;;
+;; This can be trivially expanded from the 'simple schema' format used by
+;; JS callers:
+;;
+;; {"name": "org.mozilla.foo",
+;;  "version": 4,
+;;  "attributes": [
+;;    {"name": "foo/bar",
+;;     "type": ...}]}
+;;
+;; and it can be trivially collapsed into the format understood by the
+;; transactor, which we call "schema datoms":
+;;
+;; [{:db/ident :org.mozilla.foo
+;;   :db.schema/version 4}
+;;  {:db/name :foo/bar
+;;   :db.schema/_attribute :org.mozilla.foo
+;;   :db/valueType ...}]
+
+(defn- attribute->datoms [schema-fragment-id [attribute-name attribute-pairs]]
+  (let [attribute-id (d/id-literal :db.part/user)]
+    [(assoc
+       attribute-pairs
+       :db.install/_attribute :db.part/db
+        ;; Point back to the fragment.
+       :db.schema/_attribute schema-fragment-id
+       :db/id attribute-id
+       :db/ident attribute-name)]))
+
+(defn managed-schema-fragment->datoms [{:keys [name version attributes]}]
+  (let [fragment-id (d/id-literal :db.part/db)]
+    (conj
+      (mapcat (partial attribute->datoms fragment-id) attributes)
+      {:db/id fragment-id
+       :db/ident name
+       :db.schema/version version})))
+
+(defn <collect-schema-fragment-versions
+  "Return a map, like {:org.mozilla.foo 5, :org.mozilla.core 2}."
+  [db]
+  (let [ident (partial d/ident db)]
+    (go-pair
+      (into {}
+        (map
+          (fn [[s v]] [(ident s) v])
+          (<?
+            (d/<q db '[:find ?s ?v
+                       :in $
+                       :where [?s :db.schema/version ?v]])))))))
+
+(defn <collect-schema-fragment-attributes
+  "Return a map, like {:foo/name :org.mozilla.foo}.
+   Attributes that are not linked to a fragment will not be returned."
+  [db]
+  (let [ident (partial d/ident db)]
+    (go-pair
+      (into {}
+        (map
+          (fn [[a f]] [(ident a) (ident f)])
+          (<?
+            (d/<q db '[:find ?a ?f
+                       :in $
+                       :where [?f :db.schema/attribute ?a]])))))))
+
+(defn db->symbolic-schema [db]
+  (:symbolic-schema db))
+
+(defn changed-attribute->datoms [schema-fragment-id attribute-name existing new-values]
+  (let [differences (first (diff new-values existing))]
+    (when-not (empty? differences)
+      [(merge
+         {:db/id (d/id-literal :db.part/user)
+          :db/ident attribute-name
+          ;; Point back to the fragment.
+          :db.schema/_attribute schema-fragment-id
+          :db.alter/_attribute :db.part/db}
+         differences)])))
+
+(defn changed-schema-fragment->datoms [schema-fragment-id existing-schema name attributes version]
+  (conj
+    (mapcat (fn [[attribute-name new-values]]
+              (let [existing (get existing-schema attribute-name)]
+                (if existing
+                  (changed-attribute->datoms
+                    schema-fragment-id
+                    attribute-name
+                    existing
+                    new-values)
+                  (attribute->datoms
+                    schema-fragment-id [attribute-name new-values]))))
+            attributes)
+    {:db.schema/version version
+     :db/ident name
+     :db/id (d/id-literal :db.part/db)}))
+
+(defn- prepare-schema-application-for-fragments
+  "Given a non-empty collection of fragments known to be new or outdated,
+   yield a migration sequence containing the necessary pre/post ops and
+   transact bodies."
+  [db
+   symbolic-schema
+   schema-fragment-versions
+   schema-fragment-attributes
+   {:keys [fragments pre post fragment-pre fragment-post] :as args}]
+
+  (when-let
+    [body
+     (mapcat
+       (fn [{:keys [name version attributes] :as fragment}]
+         (if-let [existing-version (get schema-fragment-versions name)]
+           ;; It's a change.
+           ;; Spit out any pre/post for this fragment, with a
+           ;; transact of the datoms to effect the change and
+           ;; bump the schema fragment version in the middle.
+           (concat
+             (when-let [fragment-pre-for-this
+                        (get-in fragment-pre [name existing-version])]
+               [[:call fragment-pre-for-this]])
+             [[:transact
+               (changed-schema-fragment->datoms (d/entid db name)
+                                                symbolic-schema
+                                                name
+                                                attributes
+                                                version)]]
+             (when-let [fragment-post-for-this
+                        (get-in fragment-post [name existing-version])]
+               [[:call fragment-post-for-this]]))
+
+           ;; It's new! Just do it.
+           ;; There can't be any fragment pre/post, 'cos there's no previous
+           ;; version to come from.
+           [[:transact (managed-schema-fragment->datoms fragment)]]))
+       fragments)]
+
+    (concat
+      (when pre [[:call pre]])
+      body
+      (when post [[:call post]]))))
+
+(defn- <prepare-schema-application*
+  [db {:keys [fragments pre post fragment-pre fragment-post] :as args}]
+  {:pre [(map? (first fragments))]}
+  (go-pair
+    (let [symbolic-schema (db->symbolic-schema db)
+          schema-fragment-versions (<? (<collect-schema-fragment-versions db))
+          schema-fragment-attributes (<? (<collect-schema-fragment-attributes db))]
+
+      ;; Filter out any incoming fragments that are already present with
+      ;; the correct version. Err if any of the fragments are outdated,
+      ;; or contain attributes that are already present elsewhere.
+      (let [to-apply
+            (filter
+              (fn [{:keys [name version attributes]}]
+                {:pre [(not (nil? name))
+                       (integer? version)
+                       (not (empty? attributes))]}
+
+                ;; Make sure that every attribute in this fragment is either
+                ;; already associate with its ident, or not associated with
+                ;; anything. We do this before we even check the version.
+                (doseq [attribute-name (keys attributes)]
+                  (when-let [existing-fragment (get schema-fragment-attributes attribute-name)]
+                    (when-not (= existing-fragment name)
+                      (raise "Attribute " attribute-name
+                             " already belongs to schema fragment "
+                             existing-fragment ", not " name "."
+                             {:error :schema/different-fragment
+                              :existing existing-fragment
+                              :fragment name}))))
+
+                ;; Now we know that every attribute is either this fragment's or
+                ;; not assigned to one.
+                ;; Check our fragment version against the store.
+                (let [existing-version (get schema-fragment-versions name)]
+                  (log "Schema" name "at existing version" existing-version)
+                  (or (nil? existing-version)
+                      (> version existing-version)
+                      (when (< version existing-version)
+                        (raise "Existing version of " name " is " existing-version
+                               ", which is later than requested " version "."
+                               {:error :schema/outdated-version
+                                :name name
+                                :version version
+                                :existing existing-version})))))
+              fragments)]
+
+        (if (empty? to-apply)
+          (do
+            (log "No fragments left to apply.")
+            nil)
+          (prepare-schema-application-for-fragments
+            db
+            symbolic-schema
+            schema-fragment-versions
+            schema-fragment-attributes
+            (assoc args :fragments to-apply)))))))
+
+(defn <prepare-schema-application
+  "Take a database and a sequence of managed schema fragments,
+   along with migration tools, and return a migration operation."
+  [db {:keys [fragments pre post fragment-pre fragment-post] :as args}]
+  (when-not (contains? args :fragments)
+    (raise-str "Missing :fragments argument to <prepare-schema-application."))
+  (if (empty? fragments)
+    (go-pair nil)
+    (do
+      ;; Validate each fragment.
+      (doseq [fragment fragments]
+        (datomish.schema/validate-schema (:attributes fragment)))
+      (let [repeated-attributes (util/repeated-keys (map :attributes fragments))]
+        (when-not (empty? repeated-attributes)
+          (raise "Attributes appear in more than one fragment: " repeated-attributes
+                 {:error :schema/repeated-attributes
+                  :repeated repeated-attributes}))
+
+        ;; At this point we know we have schema fragments to apply,
+        ;; and that they don't overlap. They might still cross fragment
+        ;; boundaries when compared to the store, and they might still
+        ;; be inconsistent, but we can proceed to the next step.
+        (<prepare-schema-application* db args)))))
+
+(defn- <schema-fragment-versions-match?
+  "Quickly return true if every provided fragment matches the
+   version in the store."
+  [db fragments]
+  (go-pair
+    (let [schema-fragment-versions (<? (<collect-schema-fragment-versions db))]
+      (every?
+        (fn [{:keys [name version]}]
+          (= (get schema-fragment-versions name)
+             version))
+        fragments))))
+
+(defn <apply-schema-alteration
+  "Take a database and a sequence of managed schema fragments,
+  along with migration tools, and transact a migration operation.
+  Throws and rolls back if any step of the operation fails. Returns
+  nil if no work was done, or the last db-report otherwise."
+  [conn args]
+  (go-pair
+    (if (or (empty? (:fragments args))
+            (<? (<schema-fragment-versions-match?
+                  (d/db conn)
+                  (:fragments args))))
+      (log "No schema work to do.")
+
+      ;; Do the real fragment op computation inside the transaction.
+      ;; This avoids a check-then-write race.
+      (<?
+        (d/<transact!
+          conn
+          (fn [db do-transact]
+            (go-pair
+              (let [last-report (atom {:db-after db})
+                    ops (<? (<prepare-schema-application db args))]
+                (doseq [[op op-arg] ops]
+                  (case op
+                        :transact
+                        (reset! last-report
+                                (<? (do-transact
+                                      (:db-after @last-report)
+                                      op-arg)))
+
+                        :call
+                        ;; We use <?? so that callers don't accidentally
+                        ;; break us if they pass a function that returns
+                        ;; nil rather than a *channel* that returns nil.
+                        (when-let [new-report
+                                   (<?? (op-arg (:db-after @last-report)
+                                                do-transact))]
+                          (when-not (:db-after new-report)
+                            (raise "Function didn't return a valid report."
+                                   {:error :schema/invalid-report
+                                    :function op-arg
+                                    :returned new-report}))
+
+                          (reset! last-report new-report))))
+                @last-report))))))))

src/common/datomish/transact.cljc

@@ -791,14 +791,14 @@
       (-> report
           (assoc-in [:db-after] db-after)))))
 
-(defn- <with [db tx-data]
+(defn <transact-tx-data-in-transaction! [db tx-data]
   (let [fail-touch-attr  (fn [old new] (raise "Altering schema attributes is not yet supported, got " new " altering existing schema attribute " old
                                               {:error :schema/alter-schema :old old :new new}))]
     (<with-internal db tx-data fail-touch-attr)))
 
-(defn <db-with [db tx-data]
+(defn <db-transact-tx-data! [db tx-data]
   (go-pair
-    (:db-after (<? (<with db tx-data)))))
+    (:db-after (<? (<transact-tx-data-in-transaction! db tx-data)))))
 
 (defn <transact!
   "Submits a transaction to the database for writing.
@@ -806,11 +806,15 @@
   Returns a pair-chan resolving to `[result error]`."
   ([conn tx-data]
    (<transact! conn tx-data (a/chan 1) true))
-  ([conn tx-data result close?]
+  ([conn tx-data-or-fn result close?]
    {:pre [(conn? conn)]}
    ;; Any race to put! is a real race between callers of <transact!.  We can't just park on put!,
    ;; because the parked putter that is woken is non-deterministic.
-   (let [closed? (not (a/put! (:transact-chan conn) [:sentinel-transact tx-data result close?]))]
+   (let [op (if (fn? tx-data-or-fn)
+              :sentinel-fn
+              :sentinel-transact)
+         closed? (not (a/put! (:transact-chan conn)
+                              [op tx-data-or-fn result close?]))]
      (go-pair
        ;; We want to return a pair-chan, no matter what kind of channel result is.
        (if closed?
@@ -823,30 +827,50 @@
       (>! token-chan (gensym "transactor-token"))
       (loop []
         (when-let [token (<! token-chan)]
-          (when-let [[sentinel tx-data result close?] (<! (:transact-chan conn))]
+          (when-let [[sentinel tx-data-or-fn result close?] (<! (:transact-chan conn))]
-            (let [pair
+            (let
-                  (<! (go-pair ;; Catch exceptions, return the pair.
+              [pair
-                        (case sentinel
+               (<!
-                          :sentinel-close
+                 (go-pair ;; Catch exceptions, return the pair.
-                          ;; Time to close the underlying DB.
+                   (case sentinel
-                          (<? (db/close-db @(:current-db conn)))
+                     :sentinel-close
+                     ;; Time to close the underlying DB.
+                     (<? (db/close-db @(:current-db conn)))
 
-                          ;; Default: process the transaction.
+                     ;; Default: process the transaction.
-                          (do
+                     (do
-                            (when @(:closed? conn)
+                       (when @(:closed? conn)
-                              ;; Drain enqueued transactions.
+                         ;; Drain enqueued transactions.
-                              (raise "Connection is closed" {:error :transact/connection-closed}))
+                         (raise "Connection is closed" {:error :transact/connection-closed}))
-                            (let [db (db conn)
+
-                                  report (<? (db/in-transaction!
+                       (let [db (db conn)
-                                               db
+                             in-transaction-fn
-                                               #(-> (<with db tx-data))))]
+                             (case sentinel
-                              (when report
+                               :sentinel-fn
-                                ;; <with returns non-nil or throws, but we still check report just in
+                               ;; This is a function that we'd like to run
-                                ;; case.  Here, in-transaction! function completed and returned non-nil,
+                               ;; within a database transaction. See
-                                ;; so the transaction has committed.
+                               ;; db/in-transaction! for details.
-                                (reset! (:current-db conn) (:db-after report))
+                               ;; The function is invoked with two arguments:
-                                (>! (:listener-source conn) report))
+                               ;; the db and a function that takes (db,
-                              report)))))]
+                               ;; tx-data) and transacts it to return a
+                               ;; TxReport.
+                               ;; The function must return a TxReport.
+                               ;; The function must not itself call
+                               ;; `in-transaction!` or `<transact!`.
+                               (partial tx-data-or-fn db <transact-tx-data-in-transaction!)
+
+                               :sentinel-transact
+                               ;; This is data. Apply it with `<transact-tx-data-in-transaction!`.
+                               (partial <transact-tx-data-in-transaction! db tx-data-or-fn))
+
+                               report (<? (db/in-transaction! db in-transaction-fn))]
+                         (when report
+                           ;; <r-t-t-d! returns non-nil or throws, but we still check report just in
+                           ;; case.  Here, in-transaction! function completed and returned non-nil,
+                           ;; so the transaction has committed.
+                           (reset! (:current-db conn) (:db-after report))
+                           (>! (:listener-source conn) report))
+                         report)))))]
               ;; Even when report is nil (transaction not committed), pair is non-nil.
               (>! result pair))
             (>! token-chan token)

src/common/datomish/transact/bootstrap.cljc

@@ -38,6 +38,15 @@
                           :db/cardinality :db.cardinality/one}
    :db.alter/attribute   {:db/valueType   :db.type/ref
                           :db/cardinality :db.cardinality/many}
+
+   :db.schema/version    {:db/valueType   :db.type/long
+                          :db/cardinality :db.cardinality/one}
+
+   ;; unique-value because an attribute can only belong to a single
+   ;; schema fragment.
+   :db.schema/attribute  {:db/valueType   :db.type/ref
+                          :db/unique      :db.unique/value
+                          :db/cardinality :db.cardinality/many}
    })
 
 (def idents
@@ -76,6 +85,8 @@
    :db.unique/value      33
    :db.unique/identity   34
    :db/doc               35
+   :db.schema/version    36    ; Fragment -> version.
+   :db.schema/attribute  37    ; Fragment -> attribute.
    })
 
 (def parts

src/common/datomish/util.cljc

@@ -174,3 +174,22 @@
         (let [[k v] (f x)]
           (assoc! ret k (conj (get ret k []) v))))
       (transient {}) coll)))
+
+(defn repeated-keys
+  "Takes a seq of maps.
+   Returns the set of keys that appear in more than one map."
+  [maps]
+  (if (not (seq (rest maps)))
+    #{}
+    ;; This is a perfect use case for transients, except that
+    ;; you can't use them for intersection due to CLJ-700.
+    ;; http://dev.clojure.org/jira/browse/CLJ-700
+    (loop [overlapping #{}
+           seen #{}
+           key-sets (map (comp set keys) maps)]
+      (if-let [ks (first key-sets)]
+        (let [overlap (clojure.set/intersection seen ks)]
+          (recur (clojure.set/union overlapping overlap)
+                 (clojure.set/union seen ks)
+                 (rest key-sets)))
+        overlapping))))

test/datomish/schema_management_test.cljc

@@ -0,0 +1,456 @@
+;; 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.schema-management-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.schema-management :as sm]
+   [datomish.api :as d]
+   #?@(:clj [[datomish.jdbc-sqlite]
+             [datomish.pair-chan :refer [go-pair <?]]
+             [tempfile.core :refer [tempfile with-tempfile]]
+             [datomish.test-macros :refer [deftest-async deftest-db]]
+             [clojure.test :as t :refer [is are deftest testing]]
+             [clojure.core.async :refer [go <! >!]]])
+   #?@(:cljs [[datomish.js-sqlite]
+              [datomish.pair-chan]
+              [datomish.test-macros :refer-macros [deftest-async deftest-db]]
+              [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])))
+
+#?(:cljs
+   (def Throwable js/Error))
+
+(def id-schema (d/id-literal :db.part/db))
+(def id-foobar (d/id-literal :db.part/user))
+
+(def trivial-schema-managed-fragment-v1
+  {:name :com.example.foo
+   :version 1
+   :attributes
+   {:foo/bar
+    {:db/cardinality :db.cardinality/one
+     :db/valueType :db.type/string}}})
+
+(def additional-schema-managed-fragment-v7
+  {:name :com.example.bar
+   :version 7
+   :attributes
+   {:bar/noo
+    {:db/cardinality :db.cardinality/one
+     :db/unique :db.unique/value
+     :db/valueType :db.type/long}}})
+
+(def additional-schema-managed-fragment-v8
+  {:name :com.example.bar
+   :version 8
+   :attributes
+   {:bar/choo
+    {:db/cardinality :db.cardinality/many
+     :db/fulltext true
+     :db/valueType :db.type/string}
+    :bar/noo
+    {:db/cardinality :db.cardinality/one
+     :db/unique :db.unique/value
+     :db/valueType :db.type/long}}})
+
+(def trivial-schema-managed-fragment-v2
+  {:name :com.example.foo
+   :version 2
+   :attributes
+   {:foo/bar
+    {:db/cardinality :db.cardinality/many
+     :db/valueType :db.type/string}}})
+
+(def trivial-schema-v1
+  (sm/managed-schema-fragment->datoms trivial-schema-managed-fragment-v1))
+
+;; TODO
+(deftest test-managed-schema-fragment->datoms)
+
+(defn <initialize-with-schema [conn schema]
+  (go-pair
+    (let [tx (<? (d/<transact! conn schema))]
+      (let [idents (map :db/ident schema)
+            db (d/db conn)]
+        (into {}
+          (map (fn [ident]
+                 [ident (d/entid db ident)])
+               idents))))))
+
+(deftest-db test-schema-management-downgrades conn
+  (is (empty? (<? (sm/<collect-schema-fragment-versions (d/db conn)))))
+  (testing "Downgrades cause errors."
+    (let [fragment {:name :com.a
+                    :version 4
+                    :attributes
+                    {:foo/bar
+                     {:db/cardinality :db.cardinality/one
+                      :db/valueType :db.type/string}}}]
+      (<? (<initialize-with-schema
+            conn
+            (sm/managed-schema-fragment->datoms fragment)))
+
+      (let [current (<? (sm/<collect-schema-fragment-versions (d/db conn)))]
+        (is (= {:com.a 4} current)))
+
+      (is (thrown-with-msg?
+            Throwable
+            #"Existing version of :com.a is 4, which is later than requested 3"
+            (<?
+              (sm/<prepare-schema-application
+                (d/db conn)
+                {:fragments
+                 [(assoc fragment :version 3)]})))))))
+
+(deftest-db test-schema-management-conflicting-ownership conn
+  (is (empty? (<? (sm/<collect-schema-fragment-versions (d/db conn)))))
+  (testing "Conflicting managed fragments cause errors."
+    (is (thrown-with-msg?
+          Throwable #"Attributes appear in more than one"
+          (<? (sm/<prepare-schema-application
+                (d/db conn)
+                {:fragments
+                 [{:name :com.a
+                   :version 1
+                   :attributes
+                   {:foo/bar
+                    {:db/cardinality :db.cardinality/one
+                     :db/valueType :db.type/string}}}
+                  {:name :com.b
+                   :version 1
+                   :attributes
+                   {:foo/bar
+                    {:db/cardinality :db.cardinality/one
+                     :db/valueType :db.type/string}}}]}))))))
+
+(defn- without-tempids
+  "Return the map without any k-v pairs for which v is a TempId."
+  [m]
+  (into {} (keep (fn [[k v]] (when-not (d/id-literal? v) [k v])) m)))
+
+(defn- op-seq-without-tempids [ops]
+  (map (fn [[op val]]
+         (if (= op :transact)
+           [op (map without-tempids val)]
+           [op val]))
+       ops))
+
+;; This is much more convenient than trying to make tempids match
+;; in our tests.
+(defn- =-op-seq-without-tempids
+  "Compare two sequences of operations (e.g., [[:call foo]])
+   without comparing TempIds within :transact operations."
+  [expected actual]
+  (= (op-seq-without-tempids expected)
+     (op-seq-without-tempids actual)))
+
+;;;
+;;; In this test we use simple strings to denote pre/post functions.
+;;; That makes everything a little easier to test.
+;;;
+;;; Note carefully that we manually transact some schema fragments,
+;;; then we use the schema management functions to get a _plan_. That's
+;;; not the same as executing the plan.
+;;;
+(deftest-db test-schema-management conn
+  (is (empty? (<? (sm/<collect-schema-fragment-versions (d/db conn)))))
+
+  ;; Start off with the trivial schema at v1.
+  (let [attrs (<? (<initialize-with-schema conn trivial-schema-v1))
+        db (d/db conn)]
+
+    (testing "We have a version, and it's 1. There are no other schemas."
+      (is (= {:com.example.foo 1}
+             (<? (sm/<collect-schema-fragment-versions db)))))
+
+    (testing "This schema fragment has one attribute."
+      (is (= {:foo/bar :com.example.foo}
+             (<? (sm/<collect-schema-fragment-attributes db)))))
+
+    (testing "The symbolic schema contains our attribute."
+      (is (= :db.type/string
+             (get-in (sm/db->symbolic-schema db)
+                     [:foo/bar :db/valueType]))))
+
+    (testing "An empty fragment yields no work."
+      (is (nil? (<? (sm/<prepare-schema-application
+                      db
+                      {:pre "X"
+                       :post "Y"
+                       :fragments []})))))
+
+    (testing "The same fragment, expressed as a managed schema fragment, yields no work."
+      (is (nil? (<? (sm/<prepare-schema-application
+                      db
+                      {:fragments [trivial-schema-managed-fragment-v1]})))))
+
+    (testing "If we try to add a second fragment, we run all-pre, new fragment, all-post."
+      (is (=-op-seq-without-tempids
+            [[:call "X"]
+             [:transact (sm/managed-schema-fragment->datoms
+                          additional-schema-managed-fragment-v7)]
+             [:call "Y"]]
+            (<? (sm/<prepare-schema-application
+                  db
+                  {:pre "X"
+                   :post "Y"
+                   :fragment-pre {}
+                   :fragment-post {}
+                   :fragments [additional-schema-managed-fragment-v7]})))))
+
+    (testing "If we upgrade one of the fragments, we run all-pre, the appropriate fragment pre, changed fragment, fragment post, all-post."
+      (is (=-op-seq-without-tempids
+            [[:call "X"]
+             [:call "A"]
+             [:transact
+              [{:db/ident :com.example.foo
+                :db.schema/version 2}
+               {:db/ident :foo/bar
+                :db.alter/_attribute :db.part/db
+                :db.schema/_attribute (d/entid db :com.example.foo)
+                :db/cardinality :db.cardinality/many}]]
+             [:call "B"]
+             [:call "Y"]]
+            (<? (sm/<prepare-schema-application
+                  db
+                  {:pre "X"
+                   :post "Y"
+                   :fragment-pre {:com.example.foo {1 "A"}}
+                   :fragment-post {:com.example.foo {1 "B"}}
+                   :fragments [trivial-schema-managed-fragment-v2]})))))
+
+    (testing "If we upgrade both, we run all-pre, each fragment pre, fragment, fragment post, all post."
+      (let [db (:db-after
+                  (<?
+                    (d/<transact! conn (sm/managed-schema-fragment->datoms
+                                         additional-schema-managed-fragment-v7))))
+
+            next-up
+            {:pre "XX"
+             :post "YY"
+             :fragment-pre {:com.example.foo {1 "AA"}}
+             :fragment-post {:com.example.foo {1 "BB"}
+                             :com.example.bar {7 "CC"}}
+             :fragments [trivial-schema-managed-fragment-v2
+                         additional-schema-managed-fragment-v8
+                         ]}
+
+            counter (atom 0)
+            pre (atom nil)
+            post (atom nil)
+            fragment-pre-foo (atom nil)
+            fragment-post-foo (atom nil)
+            fragment-post-bar (atom nil)
+
+            next-up-but-with-functions
+            {:pre (fn [db _]
+                    (reset! pre (swap! counter inc))
+                    nil)
+             :post (fn [db _]
+                     (reset! post (swap! counter inc))
+                     nil)
+             :fragment-pre {:com.example.foo
+                            {1 (fn [db _]
+                                 (reset! fragment-pre-foo (swap! counter inc))
+                                 nil)}}
+             :fragment-post {:com.example.foo
+                             {1 (fn [db _]
+                                  (reset! fragment-post-foo (swap! counter inc))
+                                  nil)}
+                             :com.example.bar
+                             {7 (fn [db _]
+                                  (reset! fragment-post-bar (swap! counter inc))
+                                  nil)}}
+             :fragments [trivial-schema-managed-fragment-v2
+                         additional-schema-managed-fragment-v8]}]
+
+        (testing "Make sure our fragment was added correctly."
+          (let [bar (d/entid db :com.example.bar)]
+            (is (integer? bar))
+            (is (= :com.example.bar (d/ident db bar))))
+
+          (is (= {:com.example.foo 1
+                  :com.example.bar 7}
+                 (<? (sm/<collect-schema-fragment-versions db)))))
+
+        ;; Now we'll see the addition of the new attribute, the
+        ;; bump of both versions, and the change to foo.
+        (is (=-op-seq-without-tempids
+              [[:call "XX"]
+               [:call "AA"]
+               [:transact
+                [{:db/ident :com.example.foo
+                  :db.schema/version 2}
+                 {:db/ident :foo/bar
+                  :db.alter/_attribute :db.part/db
+                  :db.schema/_attribute (d/entid db :com.example.foo)
+                  :db/cardinality :db.cardinality/many}]]
+               [:call "BB"]
+               [:transact
+                [{:db/ident :com.example.bar
+                  :db.schema/version 8}
+                 {:db/ident :bar/choo
+                  :db.install/_attribute :db.part/db
+                  :db.schema/_attribute (d/entid db :com.example.bar)
+                  :db/fulltext true
+                  :db/valueType :db.type/string
+                  :db/cardinality :db.cardinality/many}]]
+               [:call "CC"]
+               [:call "YY"]]
+              (<? (sm/<prepare-schema-application
+                    db
+                    next-up))))
+
+        ;; No change? Nothing to do.
+        (is (nil? (<? (sm/<apply-schema-alteration conn {:fragments []}))))
+
+        ;; Now let's try the write!
+        (let [report (<? (sm/<apply-schema-alteration conn next-up-but-with-functions))
+              db (:db-after report)]
+          (is (not (nil? report)))
+
+          (testing "Things happened in the right order."
+            (is (= 1 @pre))
+            (is (= 2 @fragment-pre-foo))
+            (is (= 3 @fragment-post-foo))
+            (is (= 4 @fragment-post-bar))
+            (is (= 5 @post)))
+
+          (testing "Our changes were applied."
+            (is (= {:com.example.foo 2
+                    :com.example.bar 8}
+                   (<? (sm/<collect-schema-fragment-versions db))))
+            (is (= {:foo/bar :com.example.foo
+                    :bar/choo :com.example.bar
+                    :bar/noo :com.example.bar}
+                   (<? (sm/<collect-schema-fragment-attributes db))))))))))
+
+(deftest-db test-functions-can-do-work conn
+  (let
+    [;; Use an atom to keep this long test fairly flat.
+     db (atom (d/db conn))
+
+     v1-fragment
+     {:name :com.example.bar
+      :version 1
+      :attributes
+      {:bar/thoo
+       {:db/valueType :db.type/long
+        :db/cardinality :db.cardinality/many}
+       :bar/choo
+       {:db/cardinality :db.cardinality/one
+        :db/fulltext true
+        :db/valueType :db.type/string}}}
+
+     ;; This is what we'll change the schema into.
+     v2-fragment
+     (assoc
+       (assoc-in v1-fragment [:attributes :bar/thoo :db/cardinality] :db.cardinality/one)
+       :version 2)
+
+     ;; This is the migration we're going to test.
+     ;; We're going to remove all existing :bar/thoo, and turn it into
+     ;; cardinality one. We're going to make :bar/choo
+     ;; unique, and we're going to use :bar/thoo to count the number of
+     ;; chars in an entity's :bar/choo.
+     ;; Note that the contents of the migration work:
+     ;; - Look just like regular queries and transacts, except for
+     ;;   working with the internal API
+     ;; - Don't have to transact the schema change at all.
+     migration
+     {:fragments [v2-fragment]
+      :pre (fn [db do-transact]
+             ;; Retract all existing uses of :bar/thoo.
+             (go-pair
+               (<?
+                 (do-transact
+                   db
+                   (map (fn [e]
+                          [:db.fn/retractAttribute e :bar/thoo])
+                        (<? (d/<q db '[:find [?e ...]
+                                       :in $
+                                       :where [?e :bar/thoo _]])))))))
+      :post (fn [db do-transact]
+              ;; Transact some new use for :bar/thoo.
+              (go-pair
+                (<?
+                  (do-transact
+                    db
+                    (map (fn [[e c]]
+                           [:db/add e :bar/thoo (count c)])
+                         (<? (d/<q db '[:find ?e ?c
+                                        :in $
+                                        :where [?e :bar/choo ?c]])))))))}]
+
+    ;; We start empty.
+    (is (empty? (<? (sm/<collect-schema-fragment-versions @db))))
+
+    ;; Apply the v1 schema.
+    (reset!
+      db
+      (:db-after
+         (<? (sm/<apply-schema-alteration
+               conn
+               {:fragments [v1-fragment]}))))
+
+    ;; Make sure it's applied.
+    (testing "We can begin."
+      (is (= {:com.example.bar 1}
+             (<? (sm/<collect-schema-fragment-versions @db))))
+      (is (= {:bar/choo :com.example.bar
+              :bar/thoo :com.example.bar}
+             (<? (sm/<collect-schema-fragment-attributes @db)))))
+
+    ;; Add some data.
+    (let [x (d/id-literal :db.part/user)
+          y (d/id-literal :db.part/user)]
+      (reset!
+        db
+        (:db-after
+           (<?
+             (d/<transact! conn [{:db/id x
+                                  :bar/thoo 99}
+                                 {:db/id y
+                                  :bar/thoo 88}
+                                 {:db/id x
+                                  :bar/choo "hello, world."}])))))
+
+    ;; Alter the schema to v2, running our migration.
+    (reset! db (:db-after (<? (sm/<apply-schema-alteration conn migration))))
+
+    ;; It worked!
+    (is (not (nil? @db)))
+
+    ;; It persisted!
+    (is (= @db @(:current-db conn)))
+
+    (testing "The schema change applied."
+      (is (= :db.cardinality/one
+             (get-in (sm/db->symbolic-schema @db) [:bar/thoo :db/cardinality]))))
+
+    (is (= (count "hello, world.")   ; 13.
+           (<? (d/<q @db
+                     '[:find ?thoo .
+                       :in $
+                       :where
+                       [?x :bar/choo "hello, world."]
+                       [?x :bar/thoo ?thoo]]))))
+
+    ;; Now that we changed its cardinality, we can transact another :bar/thoo
+    ;; for the same entity, and we will get only the new value.
+    (let [x (<? (d/<q @db '[:find ?x . :in $ :where [?x :bar/choo "hello, world."]]))]
+      (is (= [13] (<? (d/<q @db [:find ['?thoo '...] :in '$ :where [x :bar/thoo '?thoo]]))))
+
+      (<? (d/<transact! conn [[:db/add x :bar/thoo 1492]]))
+      (is (= [1492] (<? (d/<q @db [:find ['?thoo '...] :in '$ :where [x :bar/thoo '?thoo]])))))))

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.schema-changes-test
+   datomish.schema-management-test
    datomish.places.import-test
    datomish.promise-sqlite-test
    datomish.db-test
@@ -19,6 +20,7 @@
 
 (doo-tests
   'datomish.schema-changes-test
+  'datomish.schema-management-test
   'datomish.places.import-test
   'datomish.promise-sqlite-test
   'datomish.db-test

test/datomish/util_test.cljc

@@ -51,3 +51,18 @@
   (are [m xs] (= m (util/group-by-kv identity xs))
     {:a [1 2] :b [3]}
     [[:a 1] [:a 2] [:b 3]]))
+
+(deftest test-repeated-keys
+  (let [abc {:a 1 :b 2 :c 3}
+        def {:d 1 :e 2 :f 3}
+        bcd {:b 1 :c 2 :d 3}
+        efg {:e 1 :f 2 :g 3}
+        empty {}]
+    (is (= #{} (util/repeated-keys [])))
+    (is (= #{} (util/repeated-keys [empty])))
+    (is (= #{} (util/repeated-keys [empty empty])))
+    (is (= #{} (util/repeated-keys [abc empty empty])))
+    (is (= #{} (util/repeated-keys [abc def empty])))
+    (is (= #{:b :c} (util/repeated-keys [bcd abc])))
+    (is (= #{:b :c :d} (util/repeated-keys [abc def bcd])))
+    (is (= #{:b :c :d :e :f :g} (util/repeated-keys [abc efg def efg bcd])))))