feat: Add support for SPARQL updates on ResourceStores

src/storage/PatchingStore.ts

@@ -0,0 +1,46 @@
+import { Conditions } from './Conditions';
+import { Patch } from '../ldp/http/Patch';
+import { PatchHandler } from './patch/PatchHandler';
+import { Representation } from '../ldp/representation/Representation';
+import { RepresentationPreferences } from '../ldp/representation/RepresentationPreferences';
+import { ResourceIdentifier } from '../ldp/representation/ResourceIdentifier';
+import { ResourceStore } from './ResourceStore';
+
+/**
+ * {@link ResourceStore} using decorator pattern for the `modifyResource` function.
+ * If the original store supports the {@link Patch}, behaviour will be identical,
+ * otherwise one of the {@link PatchHandler}s supporting the given Patch will be called instead.
+ */
+export class PatchingStore implements ResourceStore {
+  private readonly source: ResourceStore;
+  private readonly patcher: PatchHandler;
+
+  public constructor(source: ResourceStore, patcher: PatchHandler) {
+    this.source = source;
+    this.patcher = patcher;
+  }
+
+  public async addResource(container: ResourceIdentifier, representation: Representation, conditions?: Conditions): Promise<ResourceIdentifier> {
+    return this.source.addResource(container, representation, conditions);
+  }
+
+  public async deleteResource(identifier: ResourceIdentifier, conditions?: Conditions): Promise<void> {
+    return this.source.deleteResource(identifier, conditions);
+  }
+
+  public async getRepresentation(identifier: ResourceIdentifier, preferences: RepresentationPreferences, conditions?: Conditions): Promise<Representation> {
+    return this.source.getRepresentation(identifier, preferences, conditions);
+  }
+
+  public async setRepresentation(identifier: ResourceIdentifier, representation: Representation, conditions?: Conditions): Promise<void> {
+    return this.source.setRepresentation(identifier, representation, conditions);
+  }
+
+  public async modifyResource(identifier: ResourceIdentifier, patch: Patch, conditions?: Conditions): Promise<void> {
+    try {
+      return await this.source.modifyResource(identifier, patch, conditions);
+    } catch (error) {
+      return this.patcher.handleSafe({ identifier, patch });
+    }
+  }
+}

src/storage/patch/PatchHandler.ts

@@ -0,0 +1,5 @@
+import { AsyncHandler } from '../../util/AsyncHandler';
+import { Patch } from '../../ldp/http/Patch';
+import { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
+
+export abstract class PatchHandler extends AsyncHandler<{identifier: ResourceIdentifier; patch: Patch}> {}

src/storage/patch/SimpleSparqlUpdatePatchHandler.ts

@@ -0,0 +1,82 @@
+import { Algebra } from 'sparqlalgebrajs';
+import { BaseQuad } from 'rdf-js';
+import { defaultGraph } from '@rdfjs/data-model';
+import { PatchHandler } from './PatchHandler';
+import { Readable } from 'stream';
+import { Representation } from '../../ldp/representation/Representation';
+import { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
+import { ResourceLocker } from '../ResourceLocker';
+import { ResourceStore } from '../ResourceStore';
+import { someTerms } from 'rdf-terms';
+import { SparqlUpdatePatch } from '../../ldp/http/SparqlUpdatePatch';
+import { Store } from 'n3';
+import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
+
+/**
+ * PatchHandler that supports specific types of SPARQL updates.
+ * Currently all DELETE/INSERT types are supported that have empty where bodies and no variables.
+ */
+export class SimpleSparqlUpdatePatchHandler extends PatchHandler {
+  private readonly source: ResourceStore;
+  private readonly locker: ResourceLocker;
+
+  public constructor(source: ResourceStore, locker: ResourceLocker) {
+    super();
+    this.source = source;
+    this.locker = locker;
+  }
+
+  public async canHandle(input: {identifier: ResourceIdentifier; patch: SparqlUpdatePatch}): Promise<void> {
+    if (input.patch.dataType !== 'algebra' || !input.patch.algebra) {
+      throw new UnsupportedHttpError('Only SPARQL update patch requests are supported.');
+    }
+  }
+
+  public async handle(input: {identifier: ResourceIdentifier; patch: SparqlUpdatePatch}): Promise<void> {
+    const op = input.patch.algebra;
+    if (!this.isDeleteInsert(op)) {
+      throw new UnsupportedHttpError('Only DELETE/INSERT SPARQL update operations are supported.');
+    }
+
+    const def = defaultGraph();
+    const deletes = op.delete || [];
+    const inserts = op.insert || [];
+
+    if (!deletes.every((pattern): boolean => pattern.graph.equals(def))) {
+      throw new UnsupportedHttpError('GRAPH statements are not supported.');
+    }
+    if (!inserts.every((pattern): boolean => pattern.graph.equals(def))) {
+      throw new UnsupportedHttpError('GRAPH statements are not supported.');
+    }
+    if (op.where || deletes.some((pattern): boolean => someTerms(pattern, (term): boolean => term.termType === 'Variable'))) {
+      throw new UnsupportedHttpError('WHERE statements are not supported.');
+    }
+
+    const lock = await this.locker.acquire(input.identifier);
+    const quads = await this.source.getRepresentation(input.identifier, { type: [{ value: 'internal/quads', weight: 1 }]});
+    const store = new Store<BaseQuad>();
+    const importEmitter = store.import(quads.data);
+    await new Promise((resolve, reject): void => {
+      importEmitter.on('end', resolve);
+      importEmitter.on('error', reject);
+    });
+    store.removeQuads(deletes);
+    store.addQuads(inserts);
+    const representation: Representation = {
+      data: store.match() as Readable,
+      dataType: 'quad',
+      metadata: {
+        raw: [],
+        profiles: [],
+        contentType: 'internal/quads',
+      },
+    };
+    await this.source.setRepresentation(input.identifier, representation);
+
+    await lock.release();
+  }
+
+  private isDeleteInsert(op: Algebra.Operation): op is Algebra.DeleteInsert {
+    return op.type === Algebra.types.DELETE_INSERT;
+  }
+}

test/unit/storage/PatchingStore.test.ts

@@ -0,0 +1,67 @@
+import { PatchHandler } from '../../../src/storage/patch/PatchHandler';
+import { PatchingStore } from '../../../src/storage/PatchingStore';
+import { ResourceStore } from '../../../src/storage/ResourceStore';
+
+describe('A PatchingStore', (): void => {
+  let store: PatchingStore;
+  let source: ResourceStore;
+  let patcher: PatchHandler;
+  let handleSafeFn: jest.Mock<Promise<void>, []>;
+
+  beforeEach(async(): Promise<void> => {
+    source = {
+      getRepresentation: jest.fn(async(): Promise<any> => 'get'),
+      addResource: jest.fn(async(): Promise<any> => 'add'),
+      setRepresentation: jest.fn(async(): Promise<any> => 'set'),
+      deleteResource: jest.fn(async(): Promise<any> => 'delete'),
+      modifyResource: jest.fn(async(): Promise<any> => 'modify'),
+    };
+
+    handleSafeFn = jest.fn(async(): Promise<any> => 'patcher');
+    patcher = { handleSafe: handleSafeFn } as unknown as PatchHandler;
+
+    store = new PatchingStore(source, patcher);
+  });
+
+  it('calls getRepresentation directly from the source.', async(): Promise<void> => {
+    await expect(store.getRepresentation({ path: 'getPath' }, null)).resolves.toBe('get');
+    expect(source.getRepresentation).toHaveBeenCalledTimes(1);
+    expect(source.getRepresentation).toHaveBeenLastCalledWith({ path: 'getPath' }, null, undefined);
+  });
+
+  it('calls addResource directly from the source.', async(): Promise<void> => {
+    await expect(store.addResource({ path: 'addPath' }, null)).resolves.toBe('add');
+    expect(source.addResource).toHaveBeenCalledTimes(1);
+    expect(source.addResource).toHaveBeenLastCalledWith({ path: 'addPath' }, null, undefined);
+  });
+
+  it('calls setRepresentation directly from the source.', async(): Promise<void> => {
+    await expect(store.setRepresentation({ path: 'setPath' }, null)).resolves.toBe('set');
+    expect(source.setRepresentation).toHaveBeenCalledTimes(1);
+    expect(source.setRepresentation).toHaveBeenLastCalledWith({ path: 'setPath' }, null, undefined);
+  });
+
+  it('calls deleteResource directly from the source.', async(): Promise<void> => {
+    await expect(store.deleteResource({ path: 'deletePath' }, null)).resolves.toBe('delete');
+    expect(source.deleteResource).toHaveBeenCalledTimes(1);
+    expect(source.deleteResource).toHaveBeenLastCalledWith({ path: 'deletePath' }, null);
+  });
+
+  it('calls modifyResource directly from the source if available.', async(): Promise<void> => {
+    await expect(store.modifyResource({ path: 'modifyPath' }, null)).resolves.toBe('modify');
+    expect(source.modifyResource).toHaveBeenCalledTimes(1);
+    expect(source.modifyResource).toHaveBeenLastCalledWith({ path: 'modifyPath' }, null, undefined);
+  });
+
+  it('calls its patcher if modifyResource failed.', async(): Promise<void> => {
+    source.modifyResource = jest.fn(async(): Promise<any> => {
+      throw new Error('dummy');
+    });
+    await expect(store.modifyResource({ path: 'modifyPath' }, null)).resolves.toBe('patcher');
+    expect(source.modifyResource).toHaveBeenCalledTimes(1);
+    expect(source.modifyResource).toHaveBeenLastCalledWith({ path: 'modifyPath' }, null, undefined);
+    await expect((source.modifyResource as jest.Mock).mock.results[0].value).rejects.toThrow('dummy');
+    expect(handleSafeFn).toHaveBeenCalledTimes(1);
+    expect(handleSafeFn).toHaveBeenLastCalledWith({ identifier: { path: 'modifyPath' }, patch: null });
+  });
+});

test/unit/storage/patch/SimpleSparqlUpdatePatchHandler.test.ts

@@ -0,0 +1,187 @@
+import arrayifyStream from 'arrayify-stream';
+import { Lock } from '../../../../src/storage/Lock';
+import { Quad } from 'rdf-js';
+import { ResourceLocker } from '../../../../src/storage/ResourceLocker';
+import { ResourceStore } from '../../../../src/storage/ResourceStore';
+import { SimpleSparqlUpdatePatchHandler } from '../../../../src/storage/patch/SimpleSparqlUpdatePatchHandler';
+import { SparqlUpdatePatch } from '../../../../src/ldp/http/SparqlUpdatePatch';
+import streamifyArray from 'streamify-array';
+import { translate } from 'sparqlalgebrajs';
+import { UnsupportedHttpError } from '../../../../src/util/errors/UnsupportedHttpError';
+import { namedNode, quad } from '@rdfjs/data-model';
+
+describe('A SimpleSparqlUpdatePatchHandler', (): void => {
+  let handler: SimpleSparqlUpdatePatchHandler;
+  let locker: ResourceLocker;
+  let lock: Lock;
+  let release: () => Promise<void>;
+  let source: ResourceStore;
+  let startQuads: Quad[];
+  let order: string[];
+
+  beforeEach(async(): Promise<void> => {
+    order = [];
+
+    startQuads = [ quad(
+      namedNode('http://test.com/startS1'),
+      namedNode('http://test.com/startP1'),
+      namedNode('http://test.com/startO1'),
+    ), quad(
+      namedNode('http://test.com/startS2'),
+      namedNode('http://test.com/startP2'),
+      namedNode('http://test.com/startO2'),
+    ) ];
+
+    source = {
+      getRepresentation: jest.fn(async(): Promise<any> => {
+        order.push('getRepresentation');
+        return {
+          dataType: 'quads',
+          data: streamifyArray([ ...startQuads ]),
+          metadata: null,
+        };
+      }),
+      addResource: null,
+      setRepresentation: jest.fn(async(): Promise<any> => {
+        order.push('setRepresentation');
+      }),
+      deleteResource: null,
+      modifyResource: jest.fn(async(): Promise<any> => {
+        throw new Error('noModify');
+      }),
+    };
+
+    release = jest.fn(async(): Promise<any> => order.push('release'));
+    locker = {
+      acquire: jest.fn(async(): Promise<any> => {
+        order.push('acquire');
+        lock = { release };
+        return lock;
+      }),
+    };
+
+    handler = new SimpleSparqlUpdatePatchHandler(source, locker);
+  });
+
+  const basicChecks = async(quads: Quad[]): Promise<void> => {
+    expect(source.getRepresentation).toHaveBeenCalledTimes(1);
+    expect(source.getRepresentation).toHaveBeenLastCalledWith({ path: 'path' }, { type: [{ value: 'internal/quads', weight: 1 }]});
+    expect(source.setRepresentation).toHaveBeenCalledTimes(1);
+    expect(order).toEqual([ 'acquire', 'getRepresentation', 'setRepresentation', 'release' ]);
+    const setParams = (source.setRepresentation as jest.Mock).mock.calls[0];
+    expect(setParams[0]).toEqual({ path: 'path' });
+    expect(setParams[1]).toEqual(expect.objectContaining({
+      dataType: 'quad',
+      metadata: { raw: [], profiles: [], contentType: 'internal/quads' },
+    }));
+    await expect(arrayifyStream(setParams[1].data)).resolves.toBeRdfIsomorphic(quads);
+  };
+
+  it('only accepts SPARQL updates.', async(): Promise<void> => {
+    const input = { identifier: { path: 'path' }, patch: { dataType: 'algebra', algebra: {}} as SparqlUpdatePatch };
+    await expect(handler.canHandle(input)).resolves.toBeUndefined();
+    input.patch.dataType = 'notAlgebra';
+    await expect(handler.canHandle(input)).rejects.toThrow(UnsupportedHttpError);
+  });
+
+  it('handles INSERT DATA updates.', async(): Promise<void> => {
+    await handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'INSERT DATA { <http://test.com/s1> <http://test.com/p1> <http://test.com/o1>. ' +
+        '<http://test.com/s2> <http://test.com/p2> <http://test.com/o2> }',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await basicChecks(startQuads.concat(
+      [ quad(namedNode('http://test.com/s1'), namedNode('http://test.com/p1'), namedNode('http://test.com/o1')),
+        quad(namedNode('http://test.com/s2'), namedNode('http://test.com/p2'), namedNode('http://test.com/o2')) ],
+    ));
+  });
+
+  it('handles DELETE DATA updates.', async(): Promise<void> => {
+    await handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'DELETE DATA { <http://test.com/startS1> <http://test.com/startP1> <http://test.com/startO1> }',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await basicChecks(
+      [ quad(namedNode('http://test.com/startS2'), namedNode('http://test.com/startP2'), namedNode('http://test.com/startO2')) ],
+    );
+  });
+
+  it('handles DELETE WHERE updates with no variables.', async(): Promise<void> => {
+    await handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'DELETE WHERE { <http://test.com/startS1> <http://test.com/startP1> <http://test.com/startO1> }',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await basicChecks(
+      [ quad(namedNode('http://test.com/startS2'), namedNode('http://test.com/startP2'), namedNode('http://test.com/startO2')) ],
+    );
+  });
+
+  it('handles DELETE/INSERT updates with empty WHERE.', async(): Promise<void> => {
+    await handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'DELETE { <http://test.com/startS1> <http://test.com/startP1> <http://test.com/startO1> }\n' +
+        'INSERT { <http://test.com/s1> <http://test.com/p1> <http://test.com/o1>. }\n' +
+        'WHERE {}',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await basicChecks(
+      [ quad(namedNode('http://test.com/startS2'), namedNode('http://test.com/startP2'), namedNode('http://test.com/startO2')),
+        quad(namedNode('http://test.com/s1'), namedNode('http://test.com/p1'), namedNode('http://test.com/o1')) ],
+    );
+  });
+
+  it('rejects GRAPH inserts.', async(): Promise<void> => {
+    const handle = handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'INSERT DATA { GRAPH <http://test.com/graph> { <http://test.com/startS1> <http://test.com/startP1> <http://test.com/startO1> } }',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await expect(handle).rejects.toThrow('GRAPH statements are not supported.');
+    expect(order).toEqual([]);
+  });
+
+  it('rejects GRAPH deletes.', async(): Promise<void> => {
+    const handle = handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'DELETE DATA { GRAPH <http://test.com/graph> { <http://test.com/startS1> <http://test.com/startP1> <http://test.com/startO1> } }',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await expect(handle).rejects.toThrow('GRAPH statements are not supported.');
+    expect(order).toEqual([]);
+  });
+
+  it('rejects DELETE/INSERT updates with a non-empty WHERE.', async(): Promise<void> => {
+    const handle = handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'DELETE { <http://test.com/startS1> <http://test.com/startP1> <http://test.com/startO1> }\n' +
+        'INSERT { <http://test.com/s1> <http://test.com/p1> <http://test.com/o1>. }\n' +
+        'WHERE { ?s ?p ?o }',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await expect(handle).rejects.toThrow('WHERE statements are not supported.');
+    expect(order).toEqual([]);
+  });
+
+  it('rejects DELETE WHERE updates with variables.', async(): Promise<void> => {
+    const handle = handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'DELETE WHERE { ?v <http://test.com/startP1> <http://test.com/startO1> }',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await expect(handle).rejects.toThrow('WHERE statements are not supported.');
+    expect(order).toEqual([]);
+  });
+
+  it('rejects non-DELETE/INSERT updates.', async(): Promise<void> => {
+    const handle = handler.handle({ identifier: { path: 'path' },
+      patch: { algebra: translate(
+        'MOVE DEFAULT TO GRAPH <http://test.com/newGraph>',
+        { quads: true },
+      ) } as SparqlUpdatePatch });
+    await expect(handle).rejects.toThrow('Only DELETE/INSERT SPARQL update operations are supported.');
+    expect(order).toEqual([]);
+  });
+});