Add rdf:type information to Member and fix ReadableStream types (#34)

* Add `rdf:type` information to Member and fix ReadableStream types * Default to tree:Relation for a tree relation --------- Co-authored-by: ajuvercr <[email protected]>
rdf-connect · Jun 3, 2024 · 5bb915f · 5bb915f
1 parent 2039a84
commit 5bb915f
Show file tree

Hide file tree

Showing 5 changed files with 109 additions and 73 deletions.
diff --git a/lib/client.ts b/lib/client.ts
@@ -10,7 +10,6 @@ import { Quad_Object, Term } from "@rdfjs/types";
 import {
   enhanced_fetch,
   extractMainNodeShape,
-  FetchConfig,
   getObjects,
   ModulatorFactory,
   Notifier,
@@ -23,6 +22,7 @@ import { OrderedStrategy, StrategyEvents, UnorderedStrategy } from "./strategy";
 import debug from "debug";
 import type { Writer } from "@ajuvercr/js-runner";
 
+import { ReadableStream } from "stream/web";
 export { intoConfig } from "./config";
 export { retry_fetch, extractMainNodeShape } from "./utils";
 export type { Member, Page, Relation } from "./page";
@@ -359,7 +359,7 @@ export class Client {
 
   stream(strategy?: {
     highWaterMark?: number;
-    size?: (chunk: Member) => number;
+    size?: (chunk: Member | undefined) => number;
   }): ReadableStream<Member> {
     const emitted = longPromise();
     const config: UnderlyingDefaultSource = {

diff --git a/lib/memberManager.ts b/lib/memberManager.ts
@@ -5,7 +5,11 @@ import { CBDShapeExtractor } from "extract-cbd-shape";
 import { RDF, TREE } from "@treecg/types";
 import { LDESInfo } from "./client";
 import debug from "debug";
-import { getObjects, Notifier } from "./utils";
+import {
+  getObjects,
+  memberFromQuads,
+  Notifier,
+} from "./utils";
 import { RdfStore } from "rdf-stores";
 
 const log = debug("manager");
@@ -131,41 +135,18 @@ export class Manager {
     member: Term,
     data: RdfStore,
   ): Promise<Member | undefined> {
-    const quads: Quad[] = await this.extractMemberQuads(member, data);
+    if (this.state.has(member.value)) return;
 
-    if (this.state.has(member.value)) {
-      return;
-    }
+    const quads: Quad[] = await this.extractMemberQuads(member, data);
 
     if (quads.length > 0) {
       this.state.add(member.value);
-
-      // Get timestamp
-      let timestamp: Date | string | undefined;
-      if (this.timestampPath) {
-        const ts = quads.find(
-          (x) =>
-            x.subject.equals(member) && x.predicate.equals(this.timestampPath),
-        )?.object.value;
-        if (ts) {
-          try {
-            timestamp = new Date(ts);
-          } catch (ex: any) {
-            timestamp = ts;
-          }
-        }
-      }
-
-      let isVersionOf: string | undefined;
-      if (this.isVersionOfPath) {
-        isVersionOf = quads.find(
-          (x) =>
-            x.subject.equals(member) &&
-            x.predicate.equals(this.isVersionOfPath),
-        )?.object.value;
-      }
-
-      return { id: member, quads, timestamp, isVersionOf };
+      return memberFromQuads(
+        member,
+        quads,
+        this.timestampPath,
+        this.isVersionOfPath
+      );
     }
   }
 

diff --git a/lib/page.ts b/lib/page.ts
@@ -3,13 +3,14 @@ import { RDF, TREE } from "@treecg/types";
 import { CBDShapeExtractor } from "extract-cbd-shape";
 import { State } from "./state";
 import { RdfStore } from "rdf-stores";
-import { getObjects } from "./utils";
+import { getObjects, memberFromQuads } from "./utils";
 
 export interface Member {
   id: Term;
   quads: Quad[];
   timestamp?: string | Date;
   isVersionOf?: string;
+  type?: Term;
 }
 
 export interface Relation {
@@ -36,27 +37,10 @@ export function extractMembers(
   isVersionOfPath?: Term,
 ): Promise<void>[] {
   const members = getObjects(store, stream, TREE.terms.member, null);
-
-  const extractMember = async (member: Term) => {
-    state.add(member.value);
+  async function extractMember(member: Term) {
     const quads = await extractor.extract(store, member, shapeId);
-    // Get timestamp
-    let timestamp: string | undefined;
-    if (timestampPath) {
-      timestamp = quads.find(
-        (x) => x.subject.equals(member) && x.predicate.equals(timestampPath),
-      )?.object.value;
-    }
-
-    let isVersionOf: string | undefined;
-    if (isVersionOfPath) {
-      isVersionOf = quads.find(
-        (x) => x.subject.equals(member) && x.predicate.equals(isVersionOfPath),
-      )?.object.value;
-    }
-    // Get isVersionof
-    cb({ quads, id: member, isVersionOf, timestamp });
-  };
+    cb(memberFromQuads(member, quads, timestampPath, isVersionOfPath));
+  }
 
   const out: Promise<void>[] = [];
   for (let member of members) {
@@ -88,7 +72,7 @@ export function extractRelations(
 
   for (let relationId of relationIds) {
     const node = getObjects(store, relationId, TREE.terms.node, null)[0];
-    const ty = getObjects(store, relationId, RDF.terms.type, null);
+    const ty = getObjects(store, relationId, RDF.terms.type, null)[0] || TREE.Relation;
     const path = getObjects(store, relationId, TREE.terms.path, null)[0];
     const value = getObjects(store, relationId, TREE.terms.value, null);
 
@@ -97,36 +81,52 @@ export function extractRelations(
       const assessableRelations = [];
 
       if (after) {
-        assessableRelations.push(...[TREE.LessThanRelation, TREE.LessThanOrEqualToRelation]);
+        assessableRelations.push(
+          ...[TREE.LessThanRelation, TREE.LessThanOrEqualToRelation]
+        );
         if (before) {
-          assessableRelations.push(...[TREE.GreaterThanRelation, TREE.GreaterThanOrEqualToRelation]);
+          assessableRelations.push(
+            ...[TREE.GreaterThanRelation, TREE.GreaterThanOrEqualToRelation]
+          );
           // This filter applies for all cardinal relations
-          if (assessableRelations.includes(ty[0].value)) {
-            if (ty[0].value === TREE.LessThanRelation && after >= new Date(value[0].value)) {
+          if (assessableRelations.includes(ty.value)) {
+            if (
+              ty.value === TREE.LessThanRelation &&
+              after >= new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
                 allowedNodes.delete(node.value);
               }
               continue;
             }
-            if (ty[0].value === TREE.LessThanOrEqualToRelation && after > new Date(value[0].value)) {
+            if (
+              ty.value === TREE.LessThanOrEqualToRelation &&
+              after > new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
                 allowedNodes.delete(node.value);
               }
               continue;
             }
-            if (ty[0].value === TREE.GreaterThanRelation && before <= new Date(value[0].value)) {
+            if (
+              ty.value === TREE.GreaterThanRelation &&
+              before <= new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
                 allowedNodes.delete(node.value);
               }
               continue;
             }
-            if (ty[0].value === TREE.GreaterThanOrEqualToRelation && before < new Date(value[0].value)) {
+            if (
+              ty.value === TREE.GreaterThanOrEqualToRelation &&
+              before < new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
@@ -137,16 +137,22 @@ export function extractRelations(
           }
         } else {
           // This filter only applies for tree:LessThanRelation and tree:LessThanOrEqualToRelation
-          if (assessableRelations.includes(ty[0].value)) {
-            if (ty[0].value === TREE.LessThanRelation && after >= new Date(value[0].value)) {
+          if (assessableRelations.includes(ty.value)) {
+            if (
+              ty.value === TREE.LessThanRelation &&
+              after >= new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
                 allowedNodes.delete(node.value);
               }
               continue;
             }
-            if (ty[0].value === TREE.LessThanOrEqualToRelation && after > new Date(value[0].value)) {
+            if (
+              ty.value === TREE.LessThanOrEqualToRelation &&
+              after > new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
@@ -158,18 +164,26 @@ export function extractRelations(
         }
       } else {
         if (before) {
-          assessableRelations.push(...[TREE.GreaterThanRelation, TREE.GreaterThanOrEqualToRelation]);
+          assessableRelations.push(
+            ...[TREE.GreaterThanRelation, TREE.GreaterThanOrEqualToRelation]
+          );
           // This filter only applies for tree:GreaterThanRelation and tree:GreaterThanOrEqualToRelation
-          if (assessableRelations.includes(ty[0].value)) {
-            if (ty[0].value === TREE.GreaterThanRelation && before <= new Date(value[0].value)) {
+          if (assessableRelations.includes(ty.value)) {
+            if (
+              ty.value === TREE.GreaterThanRelation &&
+              before <= new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
                 allowedNodes.delete(node.value);
               }
               continue;
             }
-            if (ty[0].value === TREE.GreaterThanOrEqualToRelation && before < new Date(value[0].value)) {
+            if (
+              ty.value === TREE.GreaterThanOrEqualToRelation &&
+              before < new Date(value[0].value)
+            ) {
               filteredNodes.add(node.value);
               if (allowedNodes.has(node.value)) {
                 // In case a permissive relation had allowed this node before
@@ -178,15 +192,17 @@ export function extractRelations(
               continue;
             }
           }
-        } else { /* No filters, everything is allowed */ }
+        } else {
+          /* No filters, everything is allowed */
+        }
       }
     }
 
     if (!filteredNodes.has(node.value)) {
       allowedNodes.set(node.value, {
         source,
         node: node.value,
-        type: ty[0],
+        type: ty,
         path,
         value,
       });

diff --git a/lib/utils.ts b/lib/utils.ts
@@ -1,9 +1,10 @@
-import { NamedNode, Stream, Term } from "@rdfjs/types";
+import { NamedNode, Quad, Stream, Term } from "@rdfjs/types";
 import { BaseQuad } from "n3";
 import { StateFactory, StateT } from "./state";
 import { RdfStore } from "rdf-stores";
 import { RDF, SHACL } from "@treecg/types";
 import debug from "debug";
+import { Member } from "./page";
 
 export type Notifier<Events, S> = {
   [K in keyof Events]: (event: Events[K], state: S) => void;
@@ -449,3 +450,40 @@ export function retry_fetch(
 
   return retry;
 }
+
+export function memberFromQuads(
+  member: Term,
+  quads: Quad[],
+  timestampPath: Term | undefined,
+  isVersionOfPath: Term | undefined,
+): Member {
+  // Get timestamp
+  let timestamp: string | Date | undefined;
+  if (timestampPath) {
+    const ts = quads.find(
+      (x) => x.subject.equals(member) && x.predicate.equals(timestampPath)
+    )?.object.value;
+    if (ts) {
+      try {
+        timestamp = new Date(ts);
+      } catch (ex: any) {
+        timestamp = ts;
+      }
+    }
+  }
+
+  // Get isVersionof
+  let isVersionOf: string | undefined;
+  if (isVersionOfPath) {
+    isVersionOf = quads.find(
+      (x) => x.subject.equals(member) && x.predicate.equals(isVersionOfPath)
+    )?.object.value;
+  }
+
+  // Get type
+  let type: Term | undefined;
+  type = quads.find(
+    (x) => x.subject.equals(member) && x.predicate.value === RDF.type
+  )?.object;
+  return { quads, id: member, isVersionOf, timestamp, type };
+}
diff --git a/tests/helper.ts b/tests/helper.ts
@@ -2,6 +2,7 @@ import { Quad } from "@rdfjs/types";
 import { Parser, Writer } from "n3";
 import { Member } from "../lib/page";
 import { TREE } from "@treecg/types";
+import { ReadableStream } from "stream/web";
 
 export type FragmentId = number;