Refactor/nameservice protocol

src/clj/fluree/db/api.cljc

@@ -18,7 +18,7 @@
  [fluree.db.ledger :as ledger]
  [fluree.db.util.log :as log]
  [fluree.db.query.range :as query-range]
- [fluree.db.nameservice.core :as nameservice]
+ [fluree.db.nameservice :as nameservice]
  [fluree.db.connection :refer [notify-ledger]]
  [fluree.db.json-ld.credential :as cred]
  [fluree.db.reasoner :as reasoner]

src/clj/fluree/db/api/transact.cljc

@@ -5,7 +5,7 @@
  [fluree.db.query.fql.parse :as q-parse]
  [fluree.db.transact :as tx]
  [fluree.db.ledger.json-ld :as jld-ledger]
- [fluree.db.nameservice.core :as nameservice]
+ [fluree.db.nameservice :as nameservice]
  [fluree.db.util.async :refer [<? go-try]]
  [fluree.db.util.core :as util :refer [catch* try*]]
  [fluree.db.util.context :as ctx-util]

src/clj/fluree/db/conn/file.cljc

@@ -12,7 +12,7 @@
  [fluree.db.serde.json :refer [json-serde]]
  [fluree.db.util.bytes :as bytes]
  [fluree.db.util.json :as json]
- [fluree.db.nameservice.filesystem :as ns-filesystem]
+ [fluree.db.nameservice.storage-backed :as storage-ns]
  [fluree.db.storage :as storage]
  [fluree.db.storage.file :as file-storage])
  #?(:clj (:import (java.io Writer))))
@@ -97,11 +97,6 @@
  (binding [*out* w]
  (pr (connection/printer-map conn)))))
 
-(defn default-file-nameservice
- "Returns file nameservice or will throw if storage-path generates an exception."
- [path]
- (ns-filesystem/initialize path))
-
 (defn default-lru-cache
  [cache-max-mb]
  (let [cache-size (conn-cache/memory->cache-size cache-max-mb)]
@@ -116,12 +111,13 @@
  (go
  (let [conn-id (str (random-uuid))
  state (connection/blank-state)
- nameservices* (util/sequential
- (or nameservices (default-file-nameservice storage-path)))
  lru-cache-atom* (or lru-cache-atom
  (default-lru-cache cache-max-mb))
  store* (or store
- (file-storage/open storage-path))]
+ (file-storage/open storage-path))
+ nameservices* (-> nameservices
+ (or (storage-ns/start "fluree:file://" store* true))
+ util/sequential)]
  ;; TODO - need to set up monitor loops for async chans
  (map->FileConnection {:id conn-id
  :store store*

src/clj/fluree/db/conn/ipfs.cljc

@@ -10,7 +10,8 @@
  [clojure.core.async :as async :refer [chan]]
  [fluree.db.serde.json :refer [json-serde]]
  [fluree.db.nameservice.ipns :as ns-ipns]
- [fluree.db.nameservice.filesystem :as ns-filesystem]
+ [fluree.db.nameservice.storage-backed :as storage-ns]
+ [fluree.db.storage.file :as file-storage]
  [fluree.db.conn.cache :as conn-cache]
  [fluree.db.storage :as storage])
  #?(:clj (:import (java.io Writer))))
@@ -95,8 +96,8 @@
 
 (defn default-file-nameservice
  [{:keys [path base-address sync?]}]
- (ns-filesystem/initialize path {:base-address base-address
-  :sync? sync?}))
+ (let [ns-store (file-storage/open path)]
+ (storage-ns/start base-address ns-store sync?)))
 
 (defn connect
  "Creates a new IPFS connection.

src/clj/fluree/db/conn/memory.cljc

@@ -2,13 +2,12 @@
  (:require [clojure.core.async :as async :refer [go]]
  [fluree.db.flake.index.storage :as index-storage]
  [fluree.db.flake.index :as index]
- [fluree.db.nameservice.memory :as ns-memory]
+ [fluree.db.nameservice.storage-backed :as storage-ns]
  [fluree.db.util.core :as util]
  [fluree.db.util.log :as log :include-macros true]
  [fluree.db.connection :as connection]
  [fluree.db.util.async :refer [<? go-try]]
  [fluree.db.conn.cache :as conn-cache]
- [fluree.json-ld :as json-ld]
  [fluree.db.storage :as storage]
  [fluree.db.storage.memory :as memory-storage]
  #?(:cljs [fluree.db.platform :as platform]))
@@ -87,11 +86,6 @@
  (binding [*out* w]
  (pr (connection/printer-map conn)))))
 
-(defn default-memory-nameservice
- "Returns memory nameservice"
- [store]
- (ns-memory/initialize store))
-
 (defn connect
  "Creates a new memory connection."
  [{:keys [parallelism lru-cache-atom cache-max-mb defaults nameservices]}]
@@ -101,16 +95,7 @@
  mem-store (memory-storage/create)
  nameservices* (util/sequential
  (or nameservices
- ;; TODO: We should not reach inside the storage
- ;; implementation to reuse the contents atom
- ;; because that breaks the abstraction and
- ;; implicitly couples components that should be
- ;; independent. We should change the memory
- ;; nameservice to either use a storage
- ;; implementation explicitly, or to use an atom
- ;; independent of the data held in commit and
- ;; index storage.
- (default-memory-nameservice (:contents mem-store))))
+ (storage-ns/start "fluree:memory://" mem-store true)))
  cache-size (conn-cache/memory->cache-size cache-max-mb)
  lru-cache-atom (or lru-cache-atom (atom (conn-cache/create-lru-cache
  cache-size)))]

src/clj/fluree/db/conn/s3.clj

@@ -1,7 +1,7 @@
 (ns fluree.db.conn.s3
  (:require [cognitect.aws.client.api :as aws]
  [clojure.string :as str]
- [fluree.db.nameservice.s3 :as ns-s3]
+ [fluree.db.nameservice.storage-backed :as storage-ns]
  [clojure.core.async :as async :refer [go]]
  [fluree.db.conn.cache :as conn-cache]
  [fluree.db.connection :as connection]
@@ -91,28 +91,21 @@
  (binding [*out* w]
  (pr (connection/printer-map conn))))
 
-(defn default-S3-nameservice
- "Returns S3 nameservice or will throw if storage-path generates an exception."
- [s3-client s3-bucket s3-prefix]
- (ns-s3/initialize s3-client s3-bucket s3-prefix))
-
 (defn connect
  "Create a new S3 connection."
  [{:keys [defaults parallelism s3-endpoint s3-bucket s3-prefix lru-cache-atom
  cache-max-mb serializer nameservices]
  :or {serializer (json-serde)} :as _opts}]
  (go
- (let [aws-opts (cond-> {:api :s3}
- s3-endpoint (assoc :endpoint-override s3-endpoint))
- client (aws/client aws-opts)
- conn-id (str (random-uuid))
+ (let [conn-id (str (random-uuid))
  state (connection/blank-state)
- nameservices* (util/sequential
- (or nameservices (default-S3-nameservice client s3-bucket s3-prefix)))
+ s3-store (s3-storage/open s3-bucket s3-prefix s3-endpoint)
+ nameservices* (-> nameservices
+ (or (storage-ns/start "fluree:s3://" s3-store true))
+ util/sequential)
  cache-size (conn-cache/memory->cache-size cache-max-mb)
  lru-cache-atom (or lru-cache-atom
- (atom (conn-cache/create-lru-cache cache-size)))
- s3-store (s3-storage/open s3-bucket s3-prefix s3-endpoint)]
+ (atom (conn-cache/create-lru-cache cache-size)))]
  (map->S3Connection {:id conn-id
  :store s3-store
  :state state

src/clj/fluree/db/json_ld/api.cljc

@@ -19,7 +19,7 @@
  [fluree.db.util.log :as log]
  [fluree.db.query.api :as query-api]
  [fluree.db.query.range :as query-range]
- [fluree.db.nameservice.core :as nameservice]
+ [fluree.db.nameservice :as nameservice]
  [fluree.db.connection :refer [notify-ledger]]
  [fluree.db.json-ld.credential :as cred]
  [fluree.db.reasoner :as reasoner]

src/clj/fluree/db/json_ld/migrate/sid.cljc

@@ -10,7 +10,7 @@
  [fluree.db.json-ld.iri :as iri]
  [fluree.db.json-ld.reify :as reify]
  [fluree.db.ledger.json-ld :as jld-ledger]
- [fluree.db.nameservice.core :as nameservice]
+ [fluree.db.nameservice :as nameservice]
  [fluree.db.query.exec.update :as update]
  [fluree.db.util.async :refer [<? go-try]]
  [fluree.db.util.core :as util :refer [get-first get-first-id get-first-value]]

src/clj/fluree/db/ledger/json_ld.cljc

@@ -13,8 +13,7 @@
  [fluree.db.json-ld.reify :as jld-reify]
  [clojure.string :as str]
  [fluree.db.util.core :as util :refer [get-first get-first-value]]
- [fluree.db.nameservice.proto :as ns-proto]
- [fluree.db.nameservice.core :as nameservice]
+ [fluree.db.nameservice :as nameservice]
  [fluree.db.connection :as connection :refer [register-ledger release-ledger]]
  [fluree.db.json-ld.commit-data :as commit-data]
  [fluree.json-ld :as json-ld]
@@ -380,7 +379,7 @@
  [conn db-alias commit-map]
  (or (get-first-value commit-map const/iri-alias)
  (->> (connection/-nameservices conn)
- (some #(ns-proto/-alias % db-alias)))))
+ (some #(nameservice/-alias % db-alias)))))
 
 ;; TODO - once we have a different delimiter than `/` for branch/t-value this can simplified
 (defn address->alias

src/clj/fluree/db/nameservice/core.cljc → src/clj/fluree/db/nameservice.cljc

@@ -1,13 +1,97 @@
-(ns fluree.db.nameservice.core
- (:refer-clojure :exclude [exists?])
+(ns fluree.db.nameservice
+ (:refer-clojure :exclude [-lookup exists?])
  (:require [clojure.string :as str]
  [fluree.db.connection :as connection]
- [fluree.db.nameservice.proto :as ns-proto]
  [fluree.db.util.async :refer [<? go-try]]
  [fluree.db.util.log :as log]))
 
 #?(:clj (set! *warn-on-reflection* true))
 
+(defprotocol iNameService
+ (-lookup [nameservice ledger-address] [nameservice ledger-alias opts] "Performs lookup operation on ledger alias and returns map of latest commit and other metadata")
+ (-sync? [nameservice] "Indicates if nameservice updates should be performed synchronously, before commit is finalized. Failure will cause commit to fail")
+ (-close [nameservice] "Closes all resources for this nameservice")
+ (-alias [nameservice ledger-address] "Given a ledger address, returns ledger's default alias name else nil, if not avail")
+ (-address [nameservice ledger-alias branch] "Returns full nameservice address/iri which will get published in commit. If 'private', return nil."))
+
+(defprotocol Publisher
+ (-push [nameservice commit-data] "Pushes new commit to nameservice."))
+
+(defprotocol Publication
+ (-subscribe [nameservice ledger-alias callback] "Creates a subscription to nameservice(s) for ledger events. Will call callback with event data as received.")
+ (-unsubscribe [nameservice ledger-alias] "Unsubscribes to nameservice(s) for ledger events"))
+
+(defn full-address
+ [prefix ledger-alias]
+ (str prefix ledger-alias))
+
+(defn ns-record
+ "Generates nameservice metadata map for JSON storage. For now, since we only
+ have a single branch possible, always sets default-branch. Eventually will
+ need to merge changes from different branches into existing metadata map"
+ [ns-address {address "address", alias "alias", branch "branch", :as commit-jsonld}]
+ (let [branch-iri (str ns-address "(" branch ")")]
+ {"@context" "https://ns.flur.ee/ledger/v1"
+ "@id" ns-address
+ "defaultBranch" branch-iri
+ "ledgerAlias" alias
+ "branches" [{"@id" branch-iri
+ "address" address
+ "commit" commit-jsonld}]}))
+
+(defn commit-address-from-record
+ [record branch]
+ (let [branch-iri (if branch
+ (str (get record "@id") "(" branch ")")
+ (get record "defaultBranch"))]
+ (->> (get record "branches")
+ (some #(when (= (get % "@id") branch-iri)
+ (get % "address"))))))
+
+(defn address-path
+ [address]
+ (let [[_ _ path] (str/split address #":")]
+ (subs path 2)))
+
+(defn address->alias
+ [ledger-address]
+ (-> ledger-address
+ address-path
+ (str/split #"/")
+ (->> (drop-last 2) ; branch-name, head
+ (str/join #"/"))))
+
+(defn extract-branch
+ "Splits a given namespace address into its nameservice and branch parts.
+ Returns two-tuple of [nameservice branch].
+ If no branch is found, returns nil as branch value and original ns-address as the nameservice."
+ [ns-address]
+ (if (str/ends-with? ns-address ")")
+ (let [[_ ns branch] (re-matches #"(.*)\((.*)\)" ns-address)]
+ [ns branch])
+ [ns-address nil]))
+
+(defn resolve-address
+ "Resolves a provided namespace address, which might be relative or absolute,
+ into three parts returned as a map:
+ - :alias - ledger alias
+ - :branch - branch (or nil if default)
+ - :address - absolute namespace address (including branch if provided)
+ If 'branch' parameter is provided will always use it as the branch regardless
+ of if a branch is specificed in the ns-address."
+ [base-address ns-address branch]
+ (let [[ns-address* extracted-branch] (extract-branch ns-address)
+ branch* (or branch extracted-branch)
+ absolute? (str/starts-with? ns-address base-address)
+ [ns-address** alias] (if absolute?
+ [ns-address* (subs ns-address* (count base-address))]
+ [(str base-address ns-address*) ns-address*])]
+ {:alias alias
+ :branch branch*
+ :address (if branch*
+ (str ns-address** "(" branch* ")")
+ ns-address*)}))
+
 (defn nameservices
  [conn]
  (connection/-nameservices conn))
@@ -19,7 +103,7 @@
 (defn ns-address
  "Returns async channel"
  [nameservice ledger-alias branch]
- (ns-proto/-address nameservice ledger-alias {:branch branch}))
+ (-address nameservice ledger-alias branch))
 
 (defn addresses
  "Retrieve address for each nameservices based on a relative ledger-alias.
@@ -60,10 +144,10 @@
  (go-try
  (loop [nameservices* nameservices]
  (when-let [ns (first nameservices*)]
- (let [sync? (ns-proto/-sync? ns)]
+ (let [sync? (-sync? ns)]
  (if sync?
- (<? (ns-proto/-push ns json-ld-commit))
- (ns-proto/-push ns json-ld-commit))
+ (<? (-push ns json-ld-commit))
+ (-push ns json-ld-commit))
  (recur (rest nameservices*))))))))
 
 (defn lookup-commit
@@ -74,7 +158,7 @@
  (go-try
  (loop [nameservices* nameservices]
  (when-let [ns (first nameservices*)]
- (let [commit-address (<? (ns-proto/-lookup ns ledger-address))]
+ (let [commit-address (<? (-lookup ns ledger-address))]
  (if commit-address
  commit-address
  (recur (rest nameservices*)))))))))
@@ -108,7 +192,7 @@
  (go-try
  (loop [nameservices* nameservices]
  (if-let [ns (first nameservices*)]
- (let [exists? (<? (ns-proto/-lookup ns ledger-alias))]
+ (let [exists? (<? (-lookup ns ledger-alias))]
  (if exists?
  true
  (recur (rest nameservices*))))
@@ -129,7 +213,7 @@
  (go-try
  (loop [nameservices* nameservices]
  (when-let [ns (first nameservices*)]
- (<? (ns-proto/-subscribe ns ledger-alias callback))
+ (<? (-subscribe ns ledger-alias callback))
  (recur (rest nameservices*)))))))
 
 (defn unsubscribe-ledger
@@ -139,5 +223,5 @@
  (go-try
  (loop [nameservices* nameservices]
  (when-let [ns (first nameservices*)]
- (<? (ns-proto/-unsubscribe ns ledger-alias))
+ (<? (-unsubscribe ns ledger-alias))
  (recur (rest nameservices*)))))))