Merge pull request #60 from rdf-connect/chore/update-dependencies

Chore/update dependencies

Author: julianrojas87

bun.lock

@@ -1,6 +1,6 @@
 {
     "name": "@rdfc/sparql-ingest-processor-ts",
-    "version": "2.0.2",
+    "version": "2.0.3",
     "description": "SPARQL Update function to be within RDF-Connect pipelines",
     "author": "Julián Rojas",
     "contributors": [
@@ -10,7 +10,7 @@
     "scripts": {
         "build": "tspc && tsc-alias",
         "prepublishOnly": "npm run build",
-        "test": "vitest run --coverage --coverage.include lib",
+        "test": "vitest run --coverage --coverage.include src",
         "prepare": "husky"
     },
     "files": [
@@ -28,29 +28,29 @@
     },
     "dependencies": {
         "@treecg/types": "^0.4.6",
-        "n3": "^1.26.0",
+        "n3": "^2.0.1",
         "rdf-data-factory": "^2.0.2",
         "rdf-stores": "^2.1.1",
         "undici": "^7.20.0",
         "winston": "^3.19.0"
     },
     "devDependencies": {
-        "@comunica/query-sparql": "^4.5.0",
+        "@comunica/query-sparql": "^5.1.3",
         "@rdfc/js-runner": "^3.0.2",
         "@rdfjs/types": "^2.0.1",
         "@types/n3": "^1.26.1",
-        "@types/node": "^24.10.10",
+        "@types/node": "^25.2.1",
         "@typescript-eslint/eslint-plugin": "^8.54.0",
         "@typescript-eslint/parser": "^8.54.0",
-        "@vitest/coverage-v8": "^3.2.4",
-        "eslint": "^9.34.0",
+        "@vitest/coverage-v8": "^4.0.18",
+        "eslint": "^9.39.2",
         "eslint-config-prettier": "^10.1.8",
         "fastify": "^5.7.4",
         "husky": "^9.1.7",
         "ts-patch": "^3.3.0",
         "tsc-alias": "^1.8.16",
         "typescript": "^5.9.3",
-        "vite-tsconfig-paths": "^5.1.4",
-        "vitest": "^3.2.4"
+        "vite-tsconfig-paths": "^6.0.5",
+        "vitest": "^4.0.18"
     }
 }

package-lock.json

@@ -34,6 +34,16 @@ rdfc:SPARQLIngest rdfc:jsImplementationOf rdfc:Processor;
 [ ] a sh:NodeShape;
   sh:targetClass rdfc:IngestConfig;
   sh:property [
+    sh:path rdfc:operationMode;
+    sh:datatype xsd:string;
+    sh:name "operationMode";
+    sh:maxCount 1;
+  ], [
+    sh:path rdfc:memberBatchSize;
+    sh:datatype xsd:integer;
+    sh:name "memberBatchSize";
+    sh:maxCount 1;
+  ], [
     sh:path rdfc:memberIsGraph;
     sh:datatype xsd:boolean;
     sh:name "memberIsGraph";

package.json

@@ -1,15 +1,16 @@
-import { extendLogger, Processor, Reader, Writer } from "@rdfc/js-runner";
+import { Processor, extendLogger } from "@rdfc/js-runner";
 import { SDS } from "@treecg/types";
 import { DataFactory } from "rdf-data-factory";
 import { RdfStore } from "rdf-stores";
-import { Parser } from "n3";
+import { Parser, Writer as N3Writer } from "n3";
 import { writeFile } from "fs/promises";
 import { CREATE, DELETE, UPDATE } from "./SPARQLQueries";
 import { doSPARQLRequest, getObjects, sanitizeQuads } from "./Utils";
-
-import type { Quad_Subject, Term } from "@rdfjs/types";
 import { Logger } from "winston";
 
+import type { Quad, Quad_Subject, Term } from "@rdfjs/types";
+import type { Reader, Writer } from "@rdfc/js-runner";
+
 const df = new DataFactory();
 
 // TODO: This should be obtained from an SDS metadata stream
@@ -33,7 +34,14 @@ export type PerformanceConfig = {
    failureIsFatal?: boolean;
 };
 
+export enum OperationMode {
+   REPLICATION = "Replication",
+   SYNC = "Sync"
+}
+
 export type IngestConfig = {
+   operationMode?: OperationMode;
+   memberBatchSize?: number;
    memberIsGraph?: boolean;
    memberShapes?: string[]; // TODO: This should be obtained from an SDS metadata stream
    changeSemantics?: ChangeSemantics;
@@ -45,7 +53,7 @@ export type IngestConfig = {
    measurePerformance?: PerformanceConfig;
 };
 
-export type TransactionMember = {
+type TransactionMember = {
    memberId: string,
    transactionId: string,
    store: RdfStore
@@ -59,17 +67,32 @@ type SPARQLIngestArgs = {
 
 export class SPARQLIngest extends Processor<SPARQLIngestArgs> {
    protected transactionMembers: TransactionMember[] = [];
+   protected memberBatch: Quad[] = [];
    protected requestsPerformance: number[] = [];
+   protected batchCount = 0;
 
    protected createTransactionQueriesLogger: Logger;
    protected doSPARQLRequestLogger: Logger;
 
    async init(this: SPARQLIngestArgs & this): Promise<void> {
       this.createTransactionQueriesLogger = extendLogger(this.logger, "createTransactionQueries");
       this.doSPARQLRequestLogger = extendLogger(this.logger, "doSPARQLRequest");
+
+      if (!this.config.operationMode) {
+         this.config.operationMode = OperationMode.SYNC;
+      }
+
+      if (!this.config.memberBatchSize) {
+         this.config.memberBatchSize = 100;
+      }
+
+      if (this.config.accessToken === "") {
+         this.config.accessToken = undefined;
+      }
    }
 
    async transform(this: SPARQLIngestArgs & this): Promise<void> {
+
       for await (const rawQuads of this.memberStream.strings()) {
          this.logger.debug(`Raw member data received: \n${rawQuads}`);
          const quads = new Parser().parse(rawQuads);
@@ -177,7 +200,11 @@ export class SPARQLIngest extends Processor<SPARQLIngestArgs> {
                   this.transactionMembers = [];
                } else {
                   // Determine if we have a named graph (either explicitly configured or as the member itself)
-                  const ng = this.getNamedGraphIfAny(memberIRI, this.config.memberIsGraph, this.config.targetNamedGraph);
+                  const ng = this.getNamedGraphIfAny(
+                     memberIRI,
+                     this.config.memberIsGraph,
+                     this.config.targetNamedGraph
+                  );
                   // Get the type of change
                   // TODO: use rdf-lens to support complex paths
                   const ctv = store.getQuads(
@@ -212,19 +239,37 @@ export class SPARQLIngest extends Processor<SPARQLIngestArgs> {
                   this.logger.info(`Preparing 'DELETE {} WHERE {} + INSERT DATA {}' SPARQL query for transaction member ${memberIRI.value}`);
                   query = UPDATE(store, this.config.forVirtuoso, this.config.targetNamedGraph);
                } else {
-                  // Determine if we have a named graph (either explicitly configure or as the member itself)
-                  const ng = this.getNamedGraphIfAny(memberIRI, this.config.memberIsGraph, this.config.targetNamedGraph);
-                  // No change semantics are provided so we do a DELETE/INSERT query by default
-                  this.logger.info(`Preparing 'DELETE {} WHERE {} + INSERT DATA {}' SPARQL query for member ${memberIRI.value}`);
-                  query = UPDATE(store, this.config.forVirtuoso, ng);
+                  // Check operation mode
+                  if (this.config.operationMode === OperationMode.REPLICATION) {
+                     this.memberBatch.push(...store.getQuads(null, null, null, null));
+                     this.batchCount++;
+                     if (this.batchCount < this.config.memberBatchSize!) {
+                        continue;
+                     }
+                  } else {
+                     // Determine if we have a named graph (either explicitly configure or as the member itself)
+                     const ng = this.getNamedGraphIfAny(memberIRI, this.config.memberIsGraph, this.config.targetNamedGraph);
+                     // No change semantics are provided so we do a DELETE/INSERT query by default
+                     this.logger.info(`Preparing 'DELETE {} WHERE {} + INSERT DATA {}' SPARQL query for member ${memberIRI.value}`);
+                     query = UPDATE(store, this.config.forVirtuoso, ng);
+                  }
                }
             }
          } else {
             // Non-SDS data
 
-            // TODO: Handle change semantics(?) and transactions for non-SDS data
-            this.logger.info(`Preparing 'DELETE {} WHERE {} + INSERT DATA {}' SPARQL query for received triples (${store.size})`);
-            query = UPDATE(store, this.config.forVirtuoso, this.config.targetNamedGraph);
+            // Check operation mode
+            if (this.config.operationMode === OperationMode.REPLICATION) {
+               this.memberBatch.push(...store.getQuads(null, null, null, null));
+               this.batchCount++;
+               if (this.batchCount < this.config.memberBatchSize!) {
+                  continue;
+               }
+            } else {
+               // TODO: Handle change semantics(?) and transactions for non-SDS data
+               this.logger.info(`Preparing 'DELETE {} WHERE {} + INSERT DATA {}' SPARQL query for received triples (${store.size})`);
+               query = UPDATE(store, this.config.forVirtuoso, this.config.targetNamedGraph);
+            }
          }
 
          // Execute the update query
@@ -255,7 +300,56 @@ export class SPARQLIngest extends Processor<SPARQLIngestArgs> {
                await this.sparqlWriter.string(query.join("\n"));
             }
          } else {
-            this.logger.warn(`No query generated for member ${memberIRI.value}`);
+            if (this.config.operationMode === OperationMode.REPLICATION) {
+               try {
+                  // Execute the ingestion of the collected member batch via the SPARQL Graph Store protocol
+                  const t0 = Date.now();
+                  await doSPARQLRequest(this.memberBatch, this.config, this.doSPARQLRequestLogger);
+                  const reqTime = Date.now() - t0;
+                  if (this.config.measurePerformance) {
+                     this.requestsPerformance.push(reqTime);
+                  }
+                  this.logger.info(`Executed query on remote SPARQL server ${this.config.graphStoreUrl} (took ${reqTime} ms)`);
+                  this.batchCount = 0;
+                  this.memberBatch = [];
+               } catch (error) {
+                  if (!this.config.measurePerformance || this.config.measurePerformance.failureIsFatal) {
+                     this.logger.error(`Error executing query on remote SPARQL server ${this.config.graphStoreUrl}: ${error}`);
+                     throw error;
+                  } else {
+                     if (this.config.measurePerformance) {
+                        this.requestsPerformance.push(-1); // -1 indicates a failure
+                     }
+                  }
+               }
+            } else {
+               this.logger.warn(`No query generated for member ${memberIRI.value}`);
+            }
+         }
+      }
+
+      // Flush remaining member batch if any
+      if (this.config.operationMode === OperationMode.REPLICATION && this.memberBatch.length > 0) {
+         try {
+            // Execute the ingestion of the collected member batch via the SPARQL Graph Store protocol
+            const t0 = Date.now();
+            await doSPARQLRequest(this.memberBatch, this.config, this.doSPARQLRequestLogger);
+            const reqTime = Date.now() - t0;
+            if (this.config.measurePerformance) {
+               this.requestsPerformance.push(reqTime);
+            }
+            this.logger.info(`Executed query on remote SPARQL server ${this.config.graphStoreUrl} (took ${reqTime} ms)`);
+            this.batchCount = 0;
+            this.memberBatch = [];
+         } catch (error) {
+            if (!this.config.measurePerformance || this.config.measurePerformance.failureIsFatal) {
+               this.logger.error(`Error executing query on remote SPARQL server ${this.config.graphStoreUrl}: ${error}`);
+               throw error;
+            } else {
+               if (this.config.measurePerformance) {
+                  this.requestsPerformance.push(-1); // -1 indicates a failure
+               }
+            }
          }
       }
 

processors.ttl

@@ -2,10 +2,11 @@ import { XSD } from "@treecg/types";
 import { DataFactory } from "rdf-data-factory";
 import { RdfStore } from "rdf-stores";
 import { Agent } from "undici";
+import { Writer as N3Writer } from "n3";
 
-import type { Term, Quad_Subject, Quad_Object } from "@rdfjs/types";
+import type { Term, Quad_Subject, Quad_Object, Quad } from "@rdfjs/types";
 import type { IngestConfig } from "./SPARQLIngest";
-import {Logger} from "winston";
+import { Logger } from "winston";
 
 const df = new DataFactory();
 
@@ -109,23 +110,55 @@ export function sanitizeQuads(store: RdfStore): void {
     }
 }
 
-export async function doSPARQLRequest(query: string[], config: IngestConfig, logger: Logger): Promise<void> {
+export async function doSPARQLRequest(
+    query: string[] | Quad[],
+    config: IngestConfig,
+    logger: Logger
+): Promise<void> {
     try {
+        const timeout = config.measurePerformance?.queryTimeout || 1800; // Default to 30 minutes if not specified
+        // Check if the query is a Quad array (for Replication mode)
+        if (query.length > 0 && typeof query[0] !== 'string') {
+            const quads = query as Quad[];
+            const writer = new N3Writer();
+            const serialized = writer.quadsToString(quads);
+            const url = new URL(config.graphStoreUrl!);
+            if (config.accessToken) {
+                url.searchParams.append("access-token", config.accessToken);
+            }
+
+            logger.debug(`Executing SPARQL Graph Store request (POST) with ${quads.length} quads.`);
+            const res = await fetch(url.toString(), {
+                method: "POST",
+                headers: {
+                    'Content-Type': 'application/n-quads',
+                },
+                body: serialized,
+                dispatcher: new Agent({
+                    headersTimeout: timeout * 1000,
+                    bodyTimeout: timeout * 1000,
+                }),
+            });
+
+            if (!res.ok) {
+                throw new Error(`HTTP request failed with code ${res.status} and message: \n${await res.text()}`);
+            }
+            return;
+        }
+
         let queries: string[] = [];
-        const jointQuery = query.join("\n");
+        const jointQuery = (query as string[]).join("\n");
 
         if (config.forVirtuoso && Buffer.byteLength(jointQuery, 'utf8') > 1e6) {
             // We need to split the query across multiple requests for Virtuoso,
             // when the query is too big (see https://community.openlinksw.com/t/virtuosoexception-sq199/1950).
-            // We set 1MB as the maximum query size empirally, aiming to maximize the query size without hitting the limit.
-            queries = query;
+            // We set 1MB as the maximum query size empirically, aiming to maximize the query size without hitting the limit.
+            queries = query as string[];
         }
         else {
             queries.push(jointQuery);
         }
 
-        const timeout = config.measurePerformance?.queryTimeout || 1800; // Default to 30 minutes if not specified
-
         for (const q of queries) {
             logger.debug(`Executing SPARQL query: \n${q}`);
             const res = await fetch(config.graphStoreUrl!, {