feat(talos): merge representations from selected env directories

.changeset/pink-eggs-deny.md

@@ -0,0 +1,5 @@
+---
+"@hydrofoil/talos": minor
+---
+
+Resources across multiple source directories are now merged in memory and save as a whole to the store

packages/talos/lib/bootstrap.ts

@@ -1,90 +1,47 @@
-import path from 'path'
-import { NamedNode, DefaultGraph } from 'rdf-js'
-import walk from '@fcostarodrigo/walk'
+import { NamedNode } from 'rdf-js'
 import StreamClient, { StreamClientOptions } from 'sparql-http-client'
-import $rdf from 'rdf-ext'
+import type DatasetExt from 'rdf-ext/lib/Dataset'
 import clownface from 'clownface'
 import { ResourcePerGraphStore } from '@hydrofoil/knossos/lib/store'
 import { hydra } from '@tpluscode/rdf-ns-builders'
-import TermMap from '@rdfjs/term-map'
-import type DatasetExt from 'rdf-ext/lib/Dataset'
-import { getPatchedStream } from './fileStream'
+import { isNamedNode } from 'is-graph-pointer'
 import { log } from './log'
-import { optionsFromPrefixes } from './prefixHandler'
+import { talosNs } from './ns'
 
-type Options = StreamClientOptions & {
-  api: string
+type Bootstrap = StreamClientOptions & {
+  dataset: DatasetExt
   apiUri: NamedNode
-  cwd: string
 }
 
-interface ResourceOptions {
-  existingResource: 'merge' | 'overwrite' | 'skip'
-}
-
-export async function bootstrap({ api, apiUri, cwd, ...options }: Options): Promise<void> {
+export async function bootstrap({ dataset, apiUri, ...options }: Bootstrap): Promise<void> {
   const store = new ResourcePerGraphStore(new StreamClient(options))
 
-  for await (const file of walk(cwd)) {
-    const relative = path.relative(cwd, file)
-    const resourcePath = path.relative(cwd, file)
-      .replace(/\.[^.]+$/, '')
-      .replace(/\/?index$/, '')
-
-    const url = resourcePath === ''
-      ? encodeURI(api)
-      : encodeURI(`${api}/${resourcePath}`)
-
-    const parserStream = getPatchedStream(file, cwd, api, url)
-    if (!parserStream) {
+  const graph = clownface({ dataset, graph: talosNs.resources })
+  const resources = graph.has(talosNs.action)
+  for (const resource of resources.toArray().filter(isNamedNode)) {
+    const pointer = clownface({ dataset })
+      .namedNode(resource)
+      .addOut(hydra.apiDocumentation, apiUri)
+
+    const action = resource.out(talosNs.action).value
+    const exists = await store.exists(pointer.term)
+    if (exists && action === 'skip') {
+      log(`Skipping resource ${resource}`)
       continue
     }
 
-    const resourceOptions: ResourceOptions = {
-      existingResource: 'overwrite',
-    }
-    parserStream.on('prefix', optionsFromPrefixes(resourceOptions))
-    const graphs = new TermMap<NamedNode | DefaultGraph, DatasetExt>()
-    try {
-      for await (const { subject, predicate, object, graph } of parserStream) {
-        let dataset = graphs.get(graph)
-        if (!dataset) {
-          dataset = $rdf.dataset()
-          graphs.set(graph, dataset)
-        }
-
-        dataset.add($rdf.quad(subject, predicate, object))
-      }
-    } catch (e: any) {
-      log('Failed to parse %s: %s', relative, e.message)
-      continue
-    }
-
-    for (const [graph, dataset] of graphs) {
-      const resource = graph.termType === 'DefaultGraph' ? url : graph.value
-      const pointer = clownface({ dataset })
-        .namedNode(resource)
-        .addOut(hydra.apiDocumentation, apiUri)
-
-      const exists = await store.exists(pointer.term)
-      if (exists && resourceOptions.existingResource === 'skip') {
-        log(`Skipping resource ${resource}`)
-        continue
-      }
-
-      if (exists) {
-        if (resourceOptions.existingResource === 'overwrite') {
-          log(`Replacing resource ${resource}`)
-        } else {
-          log(`Merging existing resource ${resource}`)
-          const current = await store.load(pointer.term)
-          pointer.dataset.addAll(current.dataset)
-        }
+    if (exists) {
+      if (action === 'overwrite') {
+        log(`Replacing resource ${resource}`)
       } else {
-        log(`Creating resource ${resource}`)
+        log(`Merging existing resource ${resource}`)
+        const current = await store.load(pointer.term)
+        pointer.dataset.addAll(current.dataset)
       }
-
-      await store.save(pointer)
+    } else {
+      log(`Creating resource ${resource}`)
     }
+
+    await store.save(pointer)
   }
 }

packages/talos/lib/command/put.ts

@@ -1,9 +1,7 @@
-import path from 'path'
-import * as fs from 'fs'
 import $rdf from 'rdf-ext'
 import { bootstrap } from '../bootstrap'
 import { deleteApi } from '../deleteApi'
-import { log } from '../log'
+import { fromDirectories } from '../resources'
 import type { Command } from '.'
 
 export interface Put extends Command {
@@ -13,28 +11,17 @@ export interface Put extends Command {
 
 export async function put(directories: string[], { token, api, endpoint, updateEndpoint, user, password, apiPath = '/api' }: Put) {
   const apiUri = $rdf.namedNode(`${api}${apiPath}`)
-  for (const dir of directories) {
-    const cwd = path.resolve(process.cwd(), dir)
 
-    if (!fs.existsSync(cwd)) {
-      log('Skipping path %s which does not exist', dir)
-      continue
-    }
-    if (!fs.statSync(cwd).isDirectory()) {
-      log('Skipping path %s which is not a directory', dir)
-      continue
-    }
+  const dataset = await fromDirectories(directories, api)
 
-    await bootstrap({
-      api,
-      apiUri,
-      cwd,
-      endpointUrl: endpoint,
-      updateUrl: updateEndpoint || endpoint,
-      user,
-      password,
-    })
-  }
+  await bootstrap({
+    dataset,
+    apiUri,
+    endpointUrl: endpoint,
+    updateUrl: updateEndpoint || endpoint,
+    user,
+    password,
+  })
 
   await deleteApi({ apiUri, token })
 }

packages/talos/lib/log.ts

@@ -1,4 +1,6 @@
-import debug from 'debug'
+import Debugger from 'debug'
 
-export const log = debug('talos')
+export const log = Debugger('talos')
 log.enabled = true
+
+export const debug = log.extend('debug')

packages/talos/lib/ns.ts

@@ -0,0 +1,3 @@
+import namespace from '@rdfjs/namespace'
+
+export const talosNs = namespace<'resources' | 'action'>('urn:talos:')

packages/talos/lib/resources.ts

@@ -0,0 +1,88 @@
+import path from 'path'
+import fs from 'fs'
+import { NamedNode } from 'rdf-js'
+import walk from '@fcostarodrigo/walk'
+import $rdf from 'rdf-ext'
+import type DatasetExt from 'rdf-ext/lib/Dataset'
+import TermSet from '@rdfjs/term-set'
+import clownface from 'clownface'
+import { log, debug } from './log'
+import { getPatchedStream } from './fileStream'
+import { optionsFromPrefixes } from './prefixHandler'
+import { talosNs } from './ns'
+
+interface ResourceOptions {
+  existingResource: 'merge' | 'overwrite' | 'skip'
+}
+
+export async function fromDirectories(directories: string[], api: string): Promise<DatasetExt> {
+  const validDirs = directories.filter(isValidDir)
+  return validDirs.reduce(toGraphs(api), Promise.resolve($rdf.dataset()))
+}
+
+function toGraphs(api: string) {
+  return async function (previous: Promise<DatasetExt>, dir: string): Promise<DatasetExt> {
+    const dataset = await previous
+
+    debug('Processing dir %s', dir)
+
+    for await (const file of walk(dir)) {
+      const relative = path.relative(dir, file)
+      const resourcePath = path.relative(dir, file)
+        .replace(/\.[^.]+$/, '')
+        .replace(/\/?index$/, '')
+
+      const url = resourcePath === ''
+        ? encodeURI(api)
+        : encodeURI(`${api}/${resourcePath}`)
+
+      const parserStream = getPatchedStream(file, dir, api, url)
+      if (!parserStream) {
+        continue
+      }
+
+      debug('Parsing %s', relative)
+      const parsedResourceOptions: Partial<ResourceOptions> = { }
+      parserStream.on('prefix', optionsFromPrefixes(parsedResourceOptions))
+
+      const resources = new TermSet<NamedNode>()
+      const resourceOptions = clownface({ dataset, graph: talosNs.resources })
+      try {
+        for await (const { subject, predicate, object, ...quad } of parserStream) {
+          const graph: NamedNode = quad.graph.equals($rdf.defaultGraph()) ? $rdf.namedNode(url) : quad.graph
+
+          if (!resources.has(graph)) {
+            debug('Parsed resource %s', graph.value)
+          }
+          resources.add(graph)
+          dataset.add($rdf.quad(subject, predicate, object, graph))
+        }
+
+        resources.forEach(id => {
+          const action = parsedResourceOptions.existingResource || 'default'
+          resourceOptions
+            .node(id)
+            .deleteOut(talosNs.action, $rdf.literal('default'))
+            .addOut(talosNs.action, action)
+        })
+      } catch (e: any) {
+        log('Failed to parse %s: %s', relative, e.message)
+      }
+    }
+
+    return dataset
+  }
+}
+
+function isValidDir(dir: string) {
+  if (!fs.existsSync(dir)) {
+    log('Skipping path %s which does not exist', dir)
+    return false
+  }
+  if (!fs.statSync(dir).isDirectory()) {
+    log('Skipping path %s which is not a directory', dir)
+    return false
+  }
+
+  return true
+}

packages/talos/package.json

@@ -23,7 +23,9 @@
     "@hydrofoil/knossos": "^0.9.7",
     "@hydrofoil/vocabularies": "^0.3",
     "@rdfjs/formats-common": "^2.2.0",
+    "@rdfjs/namespace": "^1.1.0",
     "@rdfjs/term-map": "^1",
+    "@rdfjs/term-set": "^1",
     "@tpluscode/rdf-ns-builders": "^2.0.0",
     "@tpluscode/rdf-string": "^0.2.26",
     "@tpluscode/sparql-builder": "^0.3.23",
@@ -32,21 +34,22 @@
     "commander": "^9.4.0",
     "debug": "^4.3.4",
     "is-absolute-url": "^3",
+    "is-graph-pointer": "^1.2.2",
     "mime-types": "^2.1.35",
     "node-fetch": "^2.6.7",
     "rdf-ext": "^1.3.5",
     "replacestream": "^4.0.3",
     "sparql-http-client": "^2.4.1"
   },
   "devDependencies": {
-    "@rdfjs/namespace": "^1.1.0",
     "@types/mime-types": "^2.1.1",
     "@types/replacestream": "^4.0.1",
     "@wikibus/vocabularies": "^0.2.2",
     "chai": "^4.3.6",
     "get-stream": "^5",
     "into-stream": "^6.0.0",
     "mocha": "^10.0.0",
+    "rdf-dataset-ext": "^1.0.1",
     "rdf-ext": "^1.3.5",
     "sinon": "^14"
   }

packages/talos/test/lib/__snapshots__/resources.test.ts.snap

@@ -0,0 +1,6 @@
+exports["@hydrofoil/talos/lib/resources fromDirectories merges resources from multiple graph documents"] = "<https://example.com/#part> <http://schema.org/maxValue> \"100\"^^<http://www.w3.org/2001/XMLSchema#integer> <https://example.com> .\n<https://example.com/#part> <http://schema.org/minValue> \"10\"^^<http://www.w3.org/2001/XMLSchema#integer> <https://example.com> .\n<https://example.com> <http://schema.org/hasPart> <https://example.com/#part> <https://example.com> .\n<https://example.com> <http://schema.org/name> \"Bar environment\" <https://example.com> .\n<https://example.com> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://schema.org/Thing> <https://example.com> .\n";
+
+exports["@hydrofoil/talos/lib/resources fromDirectories merges resources from dataset and graph documents"] = "<https://example.com/trig/users/john-doe> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://xmlns.com/foaf/0.1/Person> <https://example.com/trig/users/john-doe> .\n<https://example.com/trig/users/john-doe> <http://www.w3.org/ns/earl#test> \"trig\" <https://example.com/trig/users/john-doe> .\n<https://example.com/trig/users/john-doe> <http://xmlns.com/foaf/0.1/name> \"John Doe\" <https://example.com/trig/users/john-doe> .\n";
+
+exports["@hydrofoil/talos/lib/resources fromDirectories merges resources from multiple dataset documents"] = "<https://example.com/trig/users/jane-doe> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://xmlns.com/foaf/0.1/Person> <https://example.com/trig/users/jane-doe> .\n<https://example.com/trig/users/jane-doe> <http://www.w3.org/ns/earl#test> \"trig\" <https://example.com/trig/users/jane-doe> .\n<https://example.com/trig/users/jane-doe> <http://xmlns.com/foaf/0.1/name> \"Jane Doe\" <https://example.com/trig/users/jane-doe> .\n";
+