Advanced

// Note: Client API requires Datomic Cloud or Peer Server
// For Datomic Free, use peer library as shown in earlier examples

// Connect via client API (Datomic Cloud example)
import datomic.client.api.Client;
import datomic.client.api.Datomic;

Client client = Datomic.clientCloud(
    Util.map("server-type", "cloud",
             "region", "us-east-1",
             "system", "my-system",
             "endpoint", "https://...")
);
// => Creates client connection to Datomic Cloud

Connection conn = client.connect(Util.map("db-name", "tutorial"));
// => Connect to specific database

// Query via client API (same datalog)
Collection results = client.q(
    Util.map("query", "[:find ?name :where [?e :person/name ?name]]",
             "args", Util.list(client.db(conn)))
);
// => Client API uses same query language as peer
// => Queries execute remotely on peer server or Cloud

// Transaction via client API
Map txResult = client.transact(
    conn,
    Util.map("tx-data", Util.list(
        Util.map(":person/name", "Remote User",
                 ":person/email", "remote@example.com")
    ))
);
// => :tx-data key wraps transaction data
// => Returns transaction result map

;; Note: Client API requires Datomic Cloud or Peer Server
;; For Datomic Free, use peer library as shown in earlier examples

;; Connect via client API (Datomic Cloud example)
(require '[datomic.client.api :as d-client])

(def client (d-client/client {:server-type :cloud
                               :region "us-east-1"
                               :system "my-system"
                               :endpoint "https://..."}))
;; => Creates client connection to Datomic Cloud

(def conn (d-client/connect client {:db-name "tutorial"}))
;; => Connect to specific database

;; Query via client API (same datalog)
(d-client/q '[:find ?name
              :where [?e :person/name ?name]]
            (d-client/db conn))
;; => Client API uses same query language as peer
;; => Queries execute remotely on peer server or Cloud

;; Transaction via client API
(d-client/transact conn
  {:tx-data [{:person/name "Remote User"
              :person/email "remote@example.com"}]})
;; => :tx-data key wraps transaction data
;; => Returns transaction result map

// Note: Excision requires Datomic Pro or Cloud
// Datomic Free doesn't support excision

// Mark entity for excision (Datomic Pro/Cloud example)
// WARNING: Excision is irreversible - data permanently deleted

// Query entity to excise
Object entityToExcise = Peer.q(
    "[:find ?e . " +
    " :where [?e :person/email \"excise-me@example.com\"]]",
    db
);

// Submit excision request (Datomic Cloud API with client)
// client.transact(
//     conn,
//     Util.map("tx-data", Util.list(
//         Util.list(":db/excise", entityToExcise)
//     ))
// );
// => Schedules excision of entity
// => Background process removes all datoms for this entity from history
// => Cannot be undone - use with extreme caution

// Alternative: Soft delete for Datomic Free
conn.transact(
    Util.list(
        Util.map(":db/ident", ":person/deleted",
                 ":db/valueType", ":db.type/boolean",
                 ":db/cardinality", ":db.cardinality/one")
    )
).get();

conn.transact(
    Util.list(
        Util.map(":db/id", Util.list(":person/email", "alice@example.com"),
                 ":person/deleted", true)
    )
).get();
// => Mark as deleted instead of excising
// => Filter deleted entities in queries

;; Note: Excision requires Datomic Pro or Cloud
;; Datomic Free doesn't support excision

;; Mark entity for excision (Datomic Pro/Cloud example)
;; WARNING: Excision is irreversible - data permanently deleted

;; Query entity to excise
(def entity-to-excise
  (d/q '[:find ?e .
         :where [?e :person/email "excise-me@example.com"]]
       db))

;; Submit excision request (Datomic Cloud API)
(d-client/transact conn
  {:tx-data [[:db/excise entity-to-excise]]})
;; => Schedules excision of entity
;; => Background process removes all datoms for this entity from history
;; => Cannot be undone - use with extreme caution

;; Alternative: Soft delete for Datomic Free
@(d/transact conn
   [{:db/ident       :person/deleted
     :db/valueType   :db.type/boolean
     :db/cardinality :db.cardinality/one}])

@(d/transact conn
   [{:db/id [:person/email "alice@example.com"]
     :person/deleted true}])
;; => Mark as deleted instead of excising
;; => Filter deleted entities in queries

// Statistics by tenant
Collection results = Peer.q(
    "[:find ?tenant (count ?e) (avg ?age) (min ?age) (max ?age) " +
    " :where [?e :person/tenant ?tenant] " +
    "        [?e :person/age ?age]]",
    db
);
// => [["acme" 15 32.5 22 55]
//     ["globex" 12 29.3 21 48]]
// => Returns tenant, count, average age, min age, max age

// Age distribution (histogram bins)
Collection results2 = Peer.q(
    "[:find ?age-bin (count ?e) " +
    " :where [?e :person/age ?age] " +
    "        [(quot ?age 10) ?age-bin]]",
    db
);
// => [[2 450] [3 380] [4 178]]
// => Groups by decade: 20s, 30s, 40s

// Custom aggregate: percentile (implement as Java function)
public static class Percentile implements clojure.lang.IFn {
    private double p;
    public Percentile(double p) { this.p = p; }
    public Object invoke(Object vals) {
        List<Number> numbers = new ArrayList<>((Collection<Number>) vals);
        Collections.sort(numbers, Comparator.comparingDouble(Number::doubleValue));
        int idx = (int) (p * (numbers.size() - 1));
        return numbers.get(idx);
    }
}

// Use in query (requires registering function)
// Collection results3 = Peer.q(
//     "[:find (my.ns.Percentile/invoke 0.95 ?age) " +
//     " :where [?e :person/age ?age]]",
//     db
// );
// => [[52]]
// => 95th percentile age

;; Statistics by tenant
(d/q '[:find ?tenant (count ?e) (avg ?age) (min ?age) (max ?age)
       :where [?e :person/tenant ?tenant]
              [?e :person/age ?age]]
     db)
;; => [["acme" 15 32.5 22 55]
;;     ["globex" 12 29.3 21 48]]
;; => Returns tenant, count, average age, min age, max age

;; Age distribution (histogram bins)
(d/q '[:find ?age-bin (count ?e)
       :where [?e :person/age ?age]
              [(quot ?age 10) ?age-bin]]
     db)
;; => [[2 450] [3 380] [4 178]]
;; => Groups by decade: 20s, 30s, 40s

;; Custom aggregate: percentile
(defn percentile [p vals]
  (let [sorted (sort vals)
        idx (int (* p (dec (count vals))))]
    (nth sorted idx)))

(d/q '[:find (my-ns/percentile 0.95 ?age)
       :where [?e :person/age ?age]]
     db)
;; => [[52]]
;; => 95th percentile age

// Query returns Collection (not lazy in Java, but can be streamed)
Collection allPeople = Peer.q(
    "[:find ?e ?name " +
    " :where [?e :person/name ?name]]",
    db
);
// => Collection of results

// Process in chunks
int chunkSize = 100;
List<Object> chunk = new ArrayList<>();
Iterator it = allPeople.iterator();
while (it.hasNext()) {
    chunk.add(it.next());
    if (chunk.size() >= chunkSize) {
        System.out.println("Processing " + chunk.size() + " people");
        // Process chunk
        for (Object item : chunk) {
            // Do something with each person
        }
        chunk.clear();
    }
}
// Process remaining
if (!chunk.isEmpty()) {
    System.out.println("Processing " + chunk.size() + " people");
}
// => Processes 100 entities at a time
// => Keeps memory usage bounded

// Transform with streams (Java 8+)
allPeople.stream()
    .map(tuple -> ((List) tuple).get(0))  // Get entity ID
    .map(e -> db.entity(e))                // Convert to entity
    .filter(entity -> ((Integer) entity.get(":person/age")) > 30)
    .limit(10)
    .collect(Collectors.toList());
// => Stream transformation pipeline
// => Only processes first 10 matching results

;; Query returns lazy sequence
(def all-people
  (d/q '[:find ?e ?name
         :where [?e :person/name ?name]]
       db))
;; => Lazy sequence - not fully realized in memory

;; Process in chunks
(doseq [chunk (partition-all 100 all-people)]
  (println "Processing" (count chunk) "people")
  ;; Process chunk
  (doseq [[e name] chunk]
    ;; Do something with each person
    nil))
;; => Processes 100 entities at a time
;; => Keeps memory usage constant

;; Transform lazily
(->> (d/q '[:find ?e :where [?e :person/name]] db)
     (map first)
     (map (partial d/entity db))
     (filter #(> (:person/age % 0) 30))
     (take 10))
;; => Lazy transformation pipeline
;; => Only realizes 10 results, not entire dataset

// Materialize age-to-people index in Map
Map<Integer, List<Object>> ageIndex = new ConcurrentHashMap<>();

void rebuildAgeIndex(Database db) {
    ageIndex.clear();
    Collection results = Peer.q(
        "[:find ?e ?age " +
        " :where [?e :person/age ?age]]",
        db
    );
    for (Object obj : results) {
        List tuple = (List) obj;
        Object entityId = tuple.get(0);
        Integer age = (Integer) tuple.get(1);
        ageIndex.computeIfAbsent(age, k -> new ArrayList<>()).add(entityId);
    }
}

rebuildAgeIndex(db);

// Query custom index (O(1) lookup)
List<Object> people30 = ageIndex.get(30);
// => [entity-id-1, entity-id-2, ...]
// => Instant lookup vs query scan

// Update index incrementally after transactions
void updateAgeIndex(Collection txData) {
    for (Object obj : txData) {
        datomic.Datom datom = (datomic.Datom) obj;
        if (datom.a().equals(db.entid(":person/age"))) {
            Integer age = (Integer) datom.v();
            Object entityId = datom.e();
            if ((Boolean) datom.added()) {
                ageIndex.computeIfAbsent(age, k -> new ArrayList<>()).add(entityId);
            } else {
                List<Object> entities = ageIndex.get(age);
                if (entities != null) {
                    entities.remove(entityId);
                }
            }
        }
    }
}

;; Materialize age-to-people index in atom
(def age-index (atom {}))

(defn rebuild-age-index [db]
  (reset! age-index
    (reduce (fn [acc [e age]]
              (update acc age (fnil conj []) e))
            {}
            (d/q '[:find ?e ?age
                   :where [?e :person/age ?age]]
                 db))))

(rebuild-age-index db)

;; Query custom index (O(1) lookup)
(get @age-index 30)
;; => [entity-id-1 entity-id-2 ...]
;; => Instant lookup vs query scan

;; Update index incrementally after transactions
(defn update-age-index [tx-data]
  (doseq [[e a v tx added] tx-data]
    (when (= a :person/age)
      (if added
        (swap! age-index update v (fnil conj []) e)
        (swap! age-index update v (fn [es] (remove #{e} es)))))))

// Note: Datomic doesn't support distributed transactions across databases
// => Each database has independent transaction log
// Use application-level coordination (e.g., outbox pattern)
// => Saga pattern, two-phase commit simulation, or eventual consistency

// Two databases
String uri1 = "datomic:mem://db1";
// => First database URI (in-memory)
String uri2 = "datomic:mem://db2";
// => Second database URI (separate in-memory database)
Peer.createDatabase(uri1);
// => Creates first database
Peer.createDatabase(uri2);
// => Creates second database (independent of first)
Connection conn1 = Peer.connect(uri1);
// => Connection to first database
Connection conn2 = Peer.connect(uri2);
// => Connection to second database (separate connection object)

// Application-level two-phase commit simulation
Map distributedTransact(Connection conn1, List tx1, Connection conn2, List tx2) {
    // => Function simulates coordinated transaction across two databases
    // => conn1, tx1: first database and its transaction data
    // => conn2, tx2: second database and its transaction data
    try {
        // Phase 1: Prepare (validate transactions)
        conn1.db().with(tx1);
        // => Validates tx1 against conn1 database without committing
        // => with() simulates transaction to check for conflicts/errors
        // => Throws exception if validation fails
        conn2.db().with(tx2);
        // => Validates tx2 against conn2 database without committing
        // => Both validations must succeed before commit phase

        // Phase 2: Commit (both or neither)
        Map result1 = conn1.transact(tx1).get();
        // => Commits tx1 to first database
        // => .get() blocks until transaction completes
        // => Returns transaction result map with :db-after, :tx-data, :tempids
        Map result2 = conn2.transact(tx2).get();
        // => Commits tx2 to second database
        // => WARNING: If this fails, tx1 already committed (no automatic rollback)
        // => Returns transaction result map

        return Util.map(":success", true,
                       // => Indicates both transactions succeeded
                       ":results", Util.list(result1, result2));
                       // => Returns list of both transaction results
    } catch (Exception e) {
        // Rollback if either fails
        // => LIMITATION: No automatic rollback if Phase 2 partially succeeds
        // => Application must handle compensating transactions manually
        return Util.map(":success", false,
                       // => Indicates failure
                       ":error", e.getMessage());
                       // => Returns error message
    }
}

// Use with care - not true distributed transactions
distributedTransact(
    // => Attempts to coordinate transaction across two databases
    conn1, Util.list(Util.map(":person/name", "User in DB1")),
    // => First transaction: add person to db1
    conn2, Util.list(Util.map(":project/name", "Project in DB2"))
    // => Second transaction: add project to db2
);
// => WARNING: If second transaction fails after first succeeds,
// => db1 has "User in DB1" but db2 has no "Project in DB2"
// => Application must implement compensating transaction to undo

;; Note: Datomic doesn't support distributed transactions across databases
;; => Each database has independent transaction log
;; Use application-level coordination (e.g., outbox pattern)
;; => Saga pattern, two-phase commit simulation, or eventual consistency

;; Two databases
(def uri-1 "datomic:mem://db1")
;; => First database URI (in-memory)
(def uri-2 "datomic:mem://db2")
;; => Second database URI (separate in-memory database)
(d/create-database uri-1)
;; => Creates first database
(d/create-database uri-2)
;; => Creates second database (independent of first)
(def conn-1 (d/connect uri-1))
;; => Connection to first database
(def conn-2 (d/connect uri-2))
;; => Connection to second database (separate connection object)

;; Application-level two-phase commit simulation
(defn distributed-transact [conn-1 tx-1 conn-2 tx-2]
  ;; => Function simulates coordinated transaction across two databases
  ;; => conn-1, tx-1: first database and its transaction data
  ;; => conn-2, tx-2: second database and its transaction data
  (try
    ;; Phase 1: Prepare (validate transactions)
    (d/with (d/db conn-1) tx-1)
    ;; => Validates tx-1 against conn-1 database without committing
    ;; => with() simulates transaction to check for conflicts/errors
    ;; => Throws exception if validation fails
    (d/with (d/db conn-2) tx-2)
    ;; => Validates tx-2 against conn-2 database without committing
    ;; => Both validations must succeed before commit phase
    ;; Phase 2: Commit (both or neither)
    (let [result-1 @(d/transact conn-1 tx-1)
          ;; => Commits tx-1 to first database
          ;; => @ blocks until transaction completes
          ;; => Returns transaction result map with :db-after, :tx-data, :tempids
          result-2 @(d/transact conn-2 tx-2)]
          ;; => Commits tx-2 to second database
          ;; => WARNING: If this fails, tx-1 already committed (no automatic rollback)
          ;; => Returns transaction result map
      {:success true
       ;; => Indicates both transactions succeeded
       :results [result-1 result-2]})
       ;; => Returns vector of both transaction results
    (catch Exception e
      ;; Rollback if either fails
      ;; => LIMITATION: No automatic rollback if Phase 2 partially succeeds
      ;; => Application must handle compensating transactions manually
      {:success false
       ;; => Indicates failure
       :error (.getMessage e)})))
       ;; => Returns error message

;; Use with care - not true distributed transactions
(distributed-transact
  ;; => Attempts to coordinate transaction across two databases
  conn-1 [{:person/name "User in DB1"}]
  ;; => First transaction: add person to db1
  conn-2 [{:project/name "Project in DB2"}])
  ;; => Second transaction: add project to db2
;; => WARNING: If second transaction fails after first succeeds,
;; => db1 has "User in DB1" but db2 has no "Project in DB2"
;; => Application must implement compensating transaction to undo

// Note: Memory tuning applies to Datomic Pro/Cloud peer servers
// For Datomic Free embedded peer, use JVM heap settings

// JVM settings for peer (example)
// -Xmx4g -Xms4g              (4GB heap)
// -Ddatomic.objectCacheMax=1g (1GB object cache)
// -Ddatomic.memoryIndexMax=512m (512MB memory index)

// Monitor cache statistics (Datomic Pro)
// Map metrics = (Map) Peer.metrics(conn);
// => Returns metrics including cache hit rates

// Application-level caching of database values
static class DbCache {
    private volatile Database db;
    private volatile long validUntil;

    public synchronized Database getCachedDb(Connection conn, long cacheMs) {
        long now = System.currentTimeMillis();
        if (now > validUntil) {
            db = conn.db();
            validUntil = now + cacheMs;
        }
        return db;
    }
}

DbCache dbCache = new DbCache();

// Use cached db (refreshes every 1 second)
Database cachedDb = dbCache.getCachedDb(conn, 1000);
// => Returns cached database value (saves connection overhead)

;; Note: Memory tuning applies to Datomic Pro/Cloud peer servers
;; For Datomic Free embedded peer, use JVM heap settings

;; JVM settings for peer (example)
;; -Xmx4g -Xms4g              (4GB heap)
;; -Ddatomic.objectCacheMax=1g (1GB object cache)
;; -Ddatomic.memoryIndexMax=512m (512MB memory index)

;; Monitor cache statistics (Datomic Pro)
;; (d/metrics conn)
;; => Returns metrics including cache hit rates

;; Application-level caching of database values
(def db-cache (atom {:db nil :valid-until 0}))

(defn cached-db [conn cache-ms]
  (let [now (System/currentTimeMillis)
        cached @db-cache]
    (if (> now (:valid-until cached))
      (let [new-db (d/db conn)]
        (reset! db-cache {:db new-db
                          :valid-until (+ now cache-ms)})
        new-db)
      (:db cached)))

;; Use cached db (refreshes every 1 second)
(cached-db conn 1000)
;; => Returns cached database value (saves connection overhead)

// Schema versioning attribute
conn.transact(
    // => Add schema version tracking attribute
    Util.list(
        Util.map(":db/ident", ":schema/version",
                 // => Attribute name: :schema/version
                 ":db/valueType", ":db.type/long",
                 // => Stores version number as long integer
                 ":db/cardinality", ":db.cardinality/one")
                 // => Single version number per schema
    )
).get();
// => Schema version attribute now available

// Initial schema (v1)
List schemaV1 = Util.list(
    // => Version 1 schema with single full-name attribute
    Util.map(":db/ident", ":person/full-name",
             // => Original attribute: :person/full-name
             ":db/valueType", ":db.type/string",
             // => Stores full name as single string (e.g., "Alice Johnson")
             ":db/cardinality", ":db.cardinality/one")
             // => Single full name per person
);

List combined = new ArrayList(schemaV1);
// => Copy schemaV1 list to new ArrayList
combined.add(Util.map(":db/ident", ":schema/version", ":schema/version", 1));
// => Add version marker: schema version = 1
// => Uses :db/ident as lookup ref to set :schema/version attribute value
conn.transact(combined).get();
// => Transacts v1 schema + version marker
// => Database now has :person/full-name attribute and schema version 1

// Schema v2: Split full-name into first-name and last-name
List schemaV2 = Util.list(
    // => Version 2 schema with split name attributes
    Util.map(":db/ident", ":person/first-name",
             // => New attribute: :person/first-name
             ":db/valueType", ":db.type/string",
             // => Stores first name as string
             ":db/cardinality", ":db.cardinality/one"),
             // => Single first name per person
    Util.map(":db/ident", ":person/last-name",
             // => New attribute: :person/last-name
             ":db/valueType", ":db.type/string",
             // => Stores last name as string
             ":db/cardinality", ":db.cardinality/one")
             // => Single last name per person
);

// Migration function
void migrateV1ToV2(Connection conn) throws Exception {
    // => Migrates data from v1 schema (full-name) to v2 schema (first-name + last-name)
    // Add new attributes
    conn.transact(schemaV2).get();
    // => Adds :person/first-name and :person/last-name attributes
    // => Additive schema change - doesn't remove :person/full-name

    // Transform data
    Database db = conn.db();
    // => Get current database value
    Collection people = Peer.q(
        // => Query all entities with :person/full-name
        "[:find ?e ?full-name " +
        " :where [?e :person/full-name ?full-name]]",
        db
    );
    // => Returns collection of tuples: [[entity-id full-name], ...]
    // => Example: [[17592186045418, "Alice Johnson"], [17592186045419, "Bob Smith"]]

    List txData = new ArrayList();
    // => Accumulates transaction data for batch update
    for (Object obj : people) {
        // => Iterate each person entity
        List tuple = (List) obj;
        // => Extract tuple [entity-id, full-name]
        Object e = tuple.get(0);
        // => Entity ID (e.g., 17592186045418)
        String fullName = (String) tuple.get(1);
        // => Full name string (e.g., "Alice Johnson")
        String[] parts = fullName.split(" ");
        // => Split on space: ["Alice", "Johnson"]
        String firstName = parts[0];
        // => First part: "Alice"
        String lastName = parts.length > 1 ? parts[parts.length - 1] : parts[0];
        // => Last part if exists, else use first part (handles single names)
        // => Example: "Alice Johnson" → "Johnson", "Madonna" → "Madonna"

        txData.add(Util.map(":db/id", e,
                           // => Target entity ID for update
                           ":person/first-name", firstName,
                           // => Assert first name (e.g., "Alice")
                           ":person/last-name", lastName));
                           // => Assert last name (e.g., "Johnson")
    }
    // => txData now contains updates for all people
    conn.transact(txData).get();
    // => Batch transaction adds :person/first-name and :person/last-name to all entities
    // => Original :person/full-name remains (backward compatibility)

    // Update version
    conn.transact(Util.list(
        Util.map(":db/ident", ":schema/version", ":schema/version", 2)
        // => Update schema version to 2 using :db/ident lookup ref
    )).get();
    // => Schema version now 2
}

// Check current version
Entity schemaVersionEntity = conn.db().entity(":schema/version");
// => Fetch :schema/version entity using :db/ident lookup
Integer currentVersion = (Integer) schemaVersionEntity.get(":schema/version");
// => Get :schema/version attribute value
// => Example: 1 (if v1), 2 (if v2), null (if unversioned)

// Run migration if needed
if (currentVersion < 2) {
    // => Check if migration needed
    migrateV1ToV2(conn);
    // => Execute migration to v2
}
// => Safe to run multiple times (idempotent if version check correct)

;; Schema versioning attribute
@(d/transact conn
   ;; => Add schema version tracking attribute
   [{:db/ident       :schema/version
     ;; => Attribute name: :schema/version
     :db/valueType   :db.type/long
     ;; => Stores version number as long integer
     :db/cardinality :db.cardinality/one}])
     ;; => Single version number per schema
;; => Schema version attribute now available

;; Initial schema (v1)
(def schema-v1
  ;; => Version 1 schema with single full-name attribute
  [{:db/ident       :person/full-name
    ;; => Original attribute: :person/full-name
    :db/valueType   :db.type/string
    ;; => Stores full name as single string (e.g., "Alice Johnson")
    :db/cardinality :db.cardinality/one}])
    ;; => Single full name per person

@(d/transact conn (conj schema-v1 {:db/ident :schema/version :schema/version 1}))
;; => Transacts v1 schema + version marker
;; => conj adds version map to schema-v1 vector
;; => {:db/ident :schema/version :schema/version 1} uses :db/ident as lookup ref
;; => Database now has :person/full-name attribute and schema version 1

;; Schema v2: Split full-name into first-name and last-name
(def schema-v2
  ;; => Version 2 schema with split name attributes
  [{:db/ident       :person/first-name
    ;; => New attribute: :person/first-name
    :db/valueType   :db.type/string
    ;; => Stores first name as string
    :db/cardinality :db.cardinality/one}
    ;; => Single first name per person
   {:db/ident       :person/last-name
    ;; => New attribute: :person/last-name
    :db/valueType   :db.type/string
    ;; => Stores last name as string
    :db/cardinality :db.cardinality/one}])
    ;; => Single last name per person

;; Migration function
(defn migrate-v1-to-v2 [conn]
  ;; => Migrates data from v1 schema (full-name) to v2 schema (first-name + last-name)
  ;; Add new attributes
  @(d/transact conn schema-v2)
  ;; => Adds :person/first-name and :person/last-name attributes
  ;; => Additive schema change - doesn't remove :person/full-name
  ;; Transform data
  (let [db (d/db conn)
        ;; => Get current database value
        people (d/q '[:find ?e ?full-name
                      ;; => Query all entities with :person/full-name
                      :where [?e :person/full-name ?full-name]]
                    db)]
        ;; => Returns set of tuples: #{[entity-id full-name], ...}
        ;; => Example: #{[17592186045418 "Alice Johnson"] [17592186045419 "Bob Smith"]}
    @(d/transact conn
       ;; => Batch transaction to add first-name and last-name to all people
       (for [[e full-name] people
             ;; => Destructure each tuple: e = entity-id, full-name = string
             :let [parts (clojure.string/split full-name #" ")
                   ;; => Split on space: ["Alice" "Johnson"]
                   first-name (first parts)
                   ;; => First part: "Alice"
                   last-name (last parts)]]
                   ;; => Last part: "Johnson" (handles single names: (last ["Madonna"]) → "Madonna")
         {:db/id e
          ;; => Target entity ID for update
          :person/first-name first-name
          ;; => Assert first name (e.g., "Alice")
          :person/last-name last-name})))
          ;; => Assert last name (e.g., "Johnson")
  ;; => All people now have :person/first-name and :person/last-name
  ;; => Original :person/full-name remains (backward compatibility)
  ;; Update version
  @(d/transact conn [{:db/ident :schema/version :schema/version 2}]))
  ;; => Update schema version to 2 using :db/ident lookup ref
  ;; => Schema version now 2

;; Check current version
(def current-version
  ;; => Fetch current schema version
  (:schema/version (d/entity (d/db conn) :schema/version)))
  ;; => (d/entity (d/db conn) :schema/version) fetches :schema/version entity using :db/ident lookup
  ;; => (:schema/version ...) gets :schema/version attribute value
  ;; => Example: 1 (if v1), 2 (if v2), nil (if unversioned)

;; Run migration if needed
(when (< current-version 2)
  ;; => Check if migration needed (current version less than 2)
  (migrate-v1-to-v2 conn))
  ;; => Execute migration to v2
;; => Safe to run multiple times (idempotent if version check correct)

void monitorLargeTransactions(Connection conn, int threshold) {
    datomic.Log log = conn.log();
    Iterable recentTxs = log.txRange(null, null);

    int count = 0;
    for (Object obj : recentTxs) {
        if (count++ >= 100) break;

        Map tx = (Map) obj;
        Collection data = (Collection) tx.get(Keyword.intern("data"));
        int datomCount = data.size();

        if (datomCount > threshold) {
            System.out.println("ALERT: Large transaction detected: " +
                             "tx-id " + tx.get(Keyword.intern("t")) +
                             " datoms " + datomCount);
            // Trigger alert (send to monitoring system)
        }
    }
}

// Run monitoring
monitorLargeTransactions(conn, 1000);
// => Alerts on transactions with >1000 datoms

// Monitor specific attribute changes
void monitorSensitiveAttributes(Connection conn, Set<String> attributes) {
    Database db = conn.db();
    datomic.Log log = conn.log();
    Iterable recentTxs = log.txRange(null, null);

    int count = 0;
    for (Object obj : recentTxs) {
        if (count++ >= 100) break;

        Map tx = (Map) obj;
        Collection data = (Collection) tx.get(Keyword.intern("data"));

        for (Object datomObj : data) {
            datomic.Datom datom = (datomic.Datom) datomObj;
            Object attr = db.ident(datom.a());

            if (attributes.contains(attr.toString())) {
                System.out.println("ALERT: Sensitive attribute changed: " +
                                 "entity " + datom.e() +
                                 " attribute " + attr +
                                 " tx " + tx.get(Keyword.intern("t")));
            }
        }
    }
}

monitorSensitiveAttributes(conn, new HashSet<>(Arrays.asList(":person/email", ":person/password-hash")));

(defn monitor-large-transactions [conn threshold]
  (let [log (d/log conn)
        recent-txs (take 100 (d/tx-range log nil nil))]
    (doseq [tx recent-txs]
      (let [datom-count (count (:data tx))]
        (when (> datom-count threshold)
          (println "ALERT: Large transaction detected:"
                   "tx-id" (:t tx)
                   "datoms" datom-count)
          ;; Trigger alert (send to monitoring system)
          )))))

;; Run monitoring
(monitor-large-transactions conn 1000)
;; => Alerts on transactions with >1000 datoms

;; Monitor specific attribute changes
(defn monitor-sensitive-attributes [conn attributes]
  (let [db (d/db conn)
        log (d/log conn)
        recent-txs (take 100 (d/tx-range log nil nil))]
    (doseq [tx recent-txs]
      (doseq [datom (:data tx)
              :let [attr (d/ident db (:a datom))]
              :when (attributes attr)]
        (println "ALERT: Sensitive attribute changed:"
                 "entity" (:e datom)
                 "attribute" attr
                 "tx" (:t tx))))))

(monitor-sensitive-attributes conn #{:person/email :person/password-hash})

// Note: Backup strategies differ by storage backend
// Datomic Free (dev storage): backup not supported (in-memory or local files)
// Datomic Pro: use backup-db API
// Datomic Cloud: AWS backup snapshots

// Datomic Pro backup example (requires Pro license)
// Peer.backupDb(conn, "s3://my-bucket/backups/tutorial");

// For Datomic Free: export data manually
void exportDatabase(Connection conn, String outputFile) throws IOException {
    Database db = conn.db(); // Get current database value
    Iterable datoms = db.datoms(Peer.EAVT); // Get all datoms in EAVT index order

    try (PrintWriter w = new PrintWriter(new FileWriter(outputFile))) {
        for (Object obj : datoms) {
            datomic.Datom datom = (datomic.Datom) obj; // Cast to Datom
            w.println(datom.toString()); // Write datom as string
        }
    }
}

// Export all data
exportDatabase(conn, "backup-2026-01-29.edn"); // Create backup file

// Point-in-time query (no restore needed - just query)
Database dbYesterday = conn.db().asOf(yesterdayTxId); // Get database as of yesterday
Long count = (Long) Peer.q(
    "[:find (count ?e) . :where [?e :person/name]]", // Count people
    dbYesterday
);
// => Count of people as of yesterday
// => No restore needed - immutability enables time-travel

;; Note: Backup strategies differ by storage backend
;; Datomic Free (dev storage): backup not supported (in-memory or local files)
;; Datomic Pro: use backup-db API
;; Datomic Cloud: AWS backup snapshots

;; Datomic Pro backup example (requires Pro license)
;; (d/backup-db conn "s3://my-bucket/backups/tutorial")

;; For Datomic Free: export data manually
(defn export-database [conn output-file]
  (let [db (d/db conn)
        all-datoms (d/datoms db :eavt)]
    (with-open [w (clojure.java.io/writer output-file)]
      (doseq [datom all-datoms]
        (.write w (pr-str datom))
        (.write w "\n")))))

;; Export all data
(export-database conn "backup-2026-01-29.edn")

;; Point-in-time query (no restore needed - just query)
(def db-yesterday (d/as-of (d/db conn) yesterday-tx-id))
(d/q '[:find (count ?e) . :where [?e :person/name]] db-yesterday)
;; => Count of people as of yesterday
;; => No restore needed - immutability enables time-travel

// Simulate message queue (use real queue in production)
List<Map<String, Object>> messageQueue = new ArrayList<>(); // In-memory queue

void publishChangeEvent(Long entityId, Keyword attribute,
                       Object oldValue, Object newValue) {
    Map<String, Object> event = new HashMap<>(); // Create event map
    event.put("entity", entityId); // Entity ID
    event.put("attribute", attribute); // Attribute changed
    event.put("old", oldValue); // Old value (null if new)
    event.put("new", newValue); // New value
    event.put("timestamp", new java.util.Date()); // Event timestamp
    messageQueue.add(event); // Add to queue
}

// Process transactions and publish changes
void processTransactionForEvents(Map txResult) {
    Database dbBefore = (Database) txResult.get(Keyword.intern("db-before")); // DB before tx
    Database dbAfter = (Database) txResult.get(Keyword.intern("db-after")); // DB after tx
    Collection txData = (Collection) txResult.get(Keyword.intern("tx-data")); // Transaction data

    for (Object obj : txData) {
        datomic.Datom datom = (datomic.Datom) obj; // Cast to Datom
        Long e = datom.e(); // Entity ID
        Object a = datom.a(); // Attribute
        Object v = datom.v(); // Value
        boolean added = datom.added(); // Was added (not retracted)

        Keyword attrIdent = (Keyword) dbAfter.ident(a); // Get attribute ident
        if (added && "person".equals(attrIdent.getNamespace())) { // Filter person namespace
            Object oldValue = dbBefore.entity(e).get(attrIdent); // Get old value
            publishChangeEvent(e, attrIdent, oldValue, v); // Publish event
        }
    }
}

// Transact and publish
Map txResult = conn.transact(
    Util.list(
        Util.map(":person/email", "queue-test@example.com",
                 ":person/name", "Queue Test",
                 ":person/age", 35)
    )
).get(); // Block for result

processTransactionForEvents(txResult); // Process and publish events

// Check message queue
System.out.println(messageQueue);
// => [{entity=123, attribute=:person/name, old=null, new=Queue Test, ...}
//     {entity=123, attribute=:person/age, old=null, new=35, ...}]

;; Simulate message queue (use real queue in production)
(def message-queue (atom []))

(defn publish-change-event [entity-id attribute old-value new-value]
  (swap! message-queue conj
    {:entity entity-id
     :attribute attribute
     :old old-value
     :new new-value
     :timestamp (java.util.Date.)}))

;; Process transactions and publish changes
(defn process-transaction-for-events [tx-result]
  (let [db-before (:db-before tx-result)
        db-after (:db-after tx-result)
        tx-data (:tx-data tx-result)]
    (doseq [[e a v tx added] tx-data]
      (let [attr-ident (d/ident db-after a)]
        (when (and added (= (namespace attr-ident) "person"))
          (let [old-value (get (d/entity db-before e) attr-ident)]
            (publish-change-event e attr-ident old-value v)))))))

;; Transact and publish
(def tx-result
  @(d/transact conn
     [{:person/email "queue-test@example.com"
       :person/name "Queue Test"
       :person/age 35}]))

(process-transaction-for-events tx-result)

;; Check message queue
@message-queue
;; => [{:entity 123 :attribute :person/name :old nil :new "Queue Test" ...}
;;     {:entity 123 :attribute :person/age :old nil :new 35 ...}]

// Thread 1: Long-running read
Database readerDb = conn.db(); // Capture immutable database snapshot
new Thread(() -> {
    try {
        Thread.sleep(5000); // Simulate slow processing
        Long count = (Long) Peer.q(
            "[:find (count ?e) . :where [?e :person/name]]", // Count people
            readerDb // Using captured snapshot
        );
        System.out.println("Reader sees count: " + count); // Print count
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start(); // Start reader thread

// Thread 2: Write during read
Thread.sleep(1000); // Wait 1 second
conn.transact(
    Util.list(
        Util.map(":person/email", "concurrent@example.com",
                 ":person/name", "Concurrent User")
    )
).get(); // Block for result

// Output:
// Reader sees count: 1008
// => Reader's database value is immutable
// => Unaffected by concurrent writes

// New reader sees updated count
Long newCount = (Long) Peer.q(
    "[:find (count ?e) . :where [?e :person/name]]", // Count people
    conn.db() // Get current database value
);
// => 1009
// => New database value includes concurrent write

;; Thread 1: Long-running read
(def reader-db (d/db conn))
(future
  (Thread/sleep 5000) ;; Simulate slow processing
  (println "Reader sees count:"
           (d/q '[:find (count ?e) . :where [?e :person/name]] reader-db)))

;; Thread 2: Write during read
(Thread/sleep 1000)
@(d/transact conn
   [{:person/email "concurrent@example.com"
     :person/name "Concurrent User"}])

;; Output:
;; Reader sees count: 1008
;; => Reader's database value is immutable
;; => Unaffected by concurrent writes

;; New reader sees updated count
(d/q '[:find (count ?e) . :where [?e :person/name]] (d/db conn))
;; => 1009
;; => New database value includes concurrent write