Add membership filter link queue wrapper

packages/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter/README.md

@@ -0,0 +1,37 @@
+# Comunica Wrapper Membership RDF Resolve Hypermedia Links Queue Actor
+
+[![npm version](https://badge.fury.io/js/%40comunica%2Factor-rdf-resolve-hypermedia-links-queue-wrapper-membership.svg)](https://www.npmjs.com/package/@comunica/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership)
+
+An [RDF Resolve Hypermedia Links Queue](https://github.com/comunica/comunica/tree/master/packages/bus-rdf-resolve-hypermedia-links-queue) actor
+that wraps over another link queue provided by the bus,
+and imposes a limit on the maximum number of links that can be pushed into it.
+
+This module is part of the [Comunica framework](https://github.com/comunica/comunica),
+and should only be used by [developers that want to build their own query engine](https://comunica.dev/docs/modify/).
+
+[Click here if you just want to query with Comunica](https://comunica.dev/docs/query/).
+
+## Install
+
+```bash
+$ yarn add @comunica/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership
+```
+
+## Configure
+
+After installing, this package can be added to your engine's configuration as follows:
+```json
+{
+  "@context": [
+    ...
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter/^0.0.0/components/context.jsonld"
+  ],
+  "actors": [
+    ...
+    {
+      "@id": "urn:comunica:default:rdf-resolve-hypermedia-links/queue#wrapper-membership-filter",
+      "@type": "ActorRdfResolveHypermediaLinksQueueWrapperMembershipFilter"
+    }
+  ]
+}
+```

packages/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter/lib/ActorRdfResolveHypermediaLinksQueueWrapperMembershipFilter.ts

@@ -0,0 +1,65 @@
+import type {
+  MediatorRdfResolveHypermediaLinksQueue,
+  IActionRdfResolveHypermediaLinksQueue,
+  IActorRdfResolveHypermediaLinksQueueOutput,
+} from '@comunica/bus-rdf-resolve-hypermedia-links-queue';
+import { ActorRdfResolveHypermediaLinksQueue } from '@comunica/bus-rdf-resolve-hypermedia-links-queue';
+import type { IActorArgs, IActorTest } from '@comunica/core';
+import { ActionContextKey } from '@comunica/core';
+import { LinkQueueMembershipFilter } from './LinkQueueMembershipFilter';
+
+/**
+   * A comunica Wrapper Limit Count RDF Resolve Hypermedia Links Queue Actor.
+   */
+export class ActorRdfResolveHypermediaLinksQueueWrapperMembershipFilter extends ActorRdfResolveHypermediaLinksQueue {
+  private readonly mediatorRdfResolveHypermediaLinksQueue: MediatorRdfResolveHypermediaLinksQueue;
+  private readonly ignorePatterns?: string[];
+  private readonly members: string[];
+
+  public constructor(args: IActorRdfResolveHypermediaLinksQueueWrapperMembershipFilterArgs) {
+    super(args);
+    this.mediatorRdfResolveHypermediaLinksQueue = args.mediatorRdfResolveHypermediaLinksQueue;
+    this.ignorePatterns = args.ignorePatterns;
+    this.members = args.members;
+  }
+
+  public async test(action: IActionRdfResolveHypermediaLinksQueue): Promise<IActorTest> {
+    if (action.context.get(KEY_CONTEXT_WRAPPED)) {
+      throw new Error('Unable to wrap one link queues multiple times');
+    }
+    return true;
+  }
+
+  public async run(action: IActionRdfResolveHypermediaLinksQueue): Promise<IActorRdfResolveHypermediaLinksQueueOutput> {
+    const context = action.context.set(KEY_CONTEXT_WRAPPED, true);
+    const { linkQueue } = await this.mediatorRdfResolveHypermediaLinksQueue.mediate({ ...action, context });
+    return {
+      linkQueue: new LinkQueueMembershipFilter(
+        linkQueue,
+        context,
+        this.members,
+        this.ignorePatterns,
+      ),
+    };
+  }
+}
+
+export interface IActorRdfResolveHypermediaLinksQueueWrapperMembershipFilterArgs
+  extends IActorArgs<IActionRdfResolveHypermediaLinksQueue, IActorTest, IActorRdfResolveHypermediaLinksQueueOutput> {
+  /**
+   * The hypermedia links queue mediator
+   */
+  mediatorRdfResolveHypermediaLinksQueue: MediatorRdfResolveHypermediaLinksQueue;
+  /**
+   * The RegExp patterns that should be used to ignore filtering when the incoming or outgoing URI matches one of them
+   */
+  ignorePatterns?: string[];
+  /**
+   * Limit the targets of filters to consider, from quad members { s, p, o, g, spog }
+   */
+  members: string[];
+}
+
+export const KEY_CONTEXT_WRAPPED = new ActionContextKey<boolean>(
+  '@comunica/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter:wrapped',
+);

packages/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter/lib/LinkQueueMembershipFilter.ts

@@ -0,0 +1,117 @@
+import type { IMembershipFilter, IMembershipFilterStorage } from '@comunica/bus-rdf-parse-membership-filter';
+import { KeyMembershipFilterStorage } from '@comunica/bus-rdf-parse-membership-filter';
+import type { ILinkQueue, ILink } from '@comunica/bus-rdf-resolve-hypermedia-links-queue';
+import { LinkQueueWrapper } from '@comunica/bus-rdf-resolve-hypermedia-links-queue';
+import { KeysInitQuery } from '@comunica/context-entries';
+import type { IActionContext } from '@comunica/types';
+import type * as RDF from '@rdfjs/types';
+import type { Algebra } from 'sparqlalgebrajs';
+import { Util } from 'sparqlalgebrajs';
+
+/**
+ * A link queue that filters out links based on known membership filters.
+ */
+export class LinkQueueMembershipFilter extends LinkQueueWrapper {
+  private readonly queryTerms: Map<string, RDF.Term[]>;
+  private readonly filterStorage: IMembershipFilterStorage;
+  private readonly ignorePatterns: RegExp[] | undefined;
+  private readonly members: string[];
+
+  public constructor(
+    linkQueue: ILinkQueue,
+    context: IActionContext,
+    members: string[],
+    ignorePatterns?: string[],
+  ) {
+    super(linkQueue);
+    this.ignorePatterns = ignorePatterns?.map(pattern => new RegExp(pattern, 'u'));
+    this.filterStorage = context.getSafe<IMembershipFilterStorage>(KeyMembershipFilterStorage);
+    this.queryTerms = this.extractQueryTermsFromContext(context);
+    this.members = members;
+  }
+
+  private extractQueryTermsFromContext(context: IActionContext): Map<string, RDF.Term[]> {
+    const operation: Algebra.Operation = context.getSafe(KeysInitQuery.query);
+    const patterns: Algebra.Pattern[] = [];
+
+    Util.recurseOperation(operation, {
+      pattern(pattern) {
+        patterns.push(pattern);
+        return true;
+      },
+    });
+
+    const subjectTerms: Set<RDF.Term> = new Set();
+    const predicateTerms: Set<RDF.Term> = new Set();
+    const objectTerms: Set<RDF.Term> = new Set();
+    const graphTerms: Set<RDF.Term> = new Set();
+    const allTerms: Set<RDF.Term> = new Set();
+
+    for (const pattern of patterns) {
+      if (pattern.subject.termType === 'NamedNode') {
+        subjectTerms.add(pattern.subject);
+        allTerms.add(pattern.subject);
+      }
+      if (pattern.predicate.termType === 'NamedNode') {
+        predicateTerms.add(pattern.predicate);
+        allTerms.add(pattern.predicate);
+      }
+      if (pattern.object.termType === 'NamedNode') {
+        objectTerms.add(pattern.object);
+        allTerms.add(pattern.object);
+      }
+      if (pattern.graph.termType === 'NamedNode') {
+        graphTerms.add(pattern.graph);
+        allTerms.add(pattern.graph);
+      }
+    }
+
+    return new Map<string, RDF.Term[]>([
+      [ 's', [ ...subjectTerms.values() ]],
+      [ 'p', [ ...predicateTerms.values() ]],
+      [ 'o', [ ...objectTerms.values() ]],
+      [ 'g', [ ...graphTerms.values() ]],
+      [ 'spog', [ ...allTerms.values() ]],
+    ]);
+  }
+
+  /**
+   * Determine whether a link should be accepted by the queue, using membership filters.
+   * @param link The link entering or leaving the queue
+   * @returns Whether the link should be accepted
+   */
+  private acceptable(link: ILink): boolean {
+    if (this.members.length > 0 && !this.ignorePatterns?.some(pattern => pattern.test(link.url))) {
+      const filtersMatchingLink: IMembershipFilter[] = this.filterStorage.findForMembers(
+        link.url,
+        this.members,
+      );
+      if (filtersMatchingLink.length > 0) {
+        let accepted = false;
+        for (const [ members, terms ] of this.queryTerms) {
+          if (
+            this.members.includes(members) &&
+            terms.some(term => filtersMatchingLink.some(filter => filter.test(term)))
+          ) {
+            accepted = true;
+            break;
+          }
+        }
+        return accepted;
+      }
+    }
+    return true;
+  }
+
+  public pop(): ILink | undefined {
+    let link: ILink | undefined = super.pop();
+    while (link && !this.acceptable(link)) {
+      link = super.pop();
+    }
+    return link;
+  }
+
+  public push(link: ILink, parent: ILink): boolean {
+    return this.acceptable(link) ? super.push(link, parent) : false;
+  }
+}

packages/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter/lib/index.ts

@@ -0,0 +1,2 @@
+export * from './ActorRdfResolveHypermediaLinksQueueWrapperMembershipFilter';
+export * from './LinkQueueMembershipFilter';

packages/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter/package.json

@@ -0,0 +1,19 @@
+{
+  "name": "@comunica/actor-rdf-resolve-hypermedia-links-queue-wrapper-membership-filter",
+  "version": "0.0.0",
+  "description": "Comunica link queue wrapper actor for membership filters",
+  "main": "lib/index.js",
+  "types": "lib/index",
+  "repository": "https://github.com/surilindur/comunica-components",
+  "author": "surilindur",
+  "license": "MIT",
+  "private": true,
+  "lsd:module": true,
+  "dependencies": {
+    "@comunica/bus-rdf-parse-membership-filter": "^0.0.0",
+    "@comunica/bus-rdf-resolve-hypermedia-links-queue": "^2.0.0",
+    "@comunica/context-entries": "^2.0.0",
+    "@comunica/core": "^2.0.0",
+    "sparqlalgebrajs": "^4.0.0"
+  }
+}