feat(events): change notifications on DatasetWrapper (on/off)

package.json

@@ -34,7 +34,9 @@
     "Jesse Wright (https://github.com/jeswr)"
   ],
   "license": "MIT",
-  "dependencies": {},
+  "dependencies": {
+    "@jeswr/eventful-dataset": "github:jeswr/eventful-dataset#feat/eventful-dataset"
+  },
   "devDependencies": {
     "@rdfjs/types": "^2",
     "@types/node": "^26",

src/DatasetWrapper.ts

@@ -1,14 +1,26 @@
+import type { EventfulDatasetCore } from "@jeswr/eventful-dataset"
 import type { DataFactory, DatasetCore, Quad, Quad_Graph, Term } from "@rdfjs/types"
 import type { ITermWrapperConstructor } from "./type/ITermWrapperConstructor.js"
+import type { IDatasetChangeListener } from "./type/IDatasetChangeListener.js"
 import type { NamedGraphDataset } from "./NamedGraphDataset.js"
 import type { INamedGraphDatasetConstructor } from "./type/INamedGraphDatasetConstructor.js"
 
+import { ensureEventfulDatasetCore } from "./ensure.js"
 import { RDF } from "./vocabulary/RDF.js"
 
 export class DatasetWrapper implements DatasetCore {
     //#region DatasetCore
 
-    public constructor(private readonly dataset: DatasetCore, protected readonly factory: DataFactory) {
+    private readonly dataset: EventfulDatasetCore<Quad>
+
+    /**
+     * Creates a new instance of {@link DatasetWrapper}.
+     *
+     * @param dataset - The dataset being wrapped. Mutations performed through this wrapper write through to it and notify listeners attached with {@link on}.
+     * @param factory - A collection of methods for creating terms.
+     */
+    public constructor(dataset: DatasetCore, protected readonly factory: DataFactory) {
+        this.dataset = ensureEventfulDatasetCore(dataset)
     }
 
     public get size(): number {
@@ -39,6 +51,104 @@ export class DatasetWrapper implements DatasetCore {
 
     //#endregion
 
+    //#region Events
+
+    private readonly listeners = new Map<IDatasetChangeListener, { added(quad: Quad): void, deleted(quad: Quad): void }>()
+
+    /**
+     * Subscribes `listener` to change notifications for the underlying dataset.
+     *
+     * The listener is invoked synchronously with the type of the change (`"add"` or `"delete"`) and the affected quad whenever the contents of the dataset effectively change, regardless of how the mutation was performed: direct calls to {@link add} or {@link delete}, assigning a mapped property on a {@link TermWrapper}, or mutating a {@link Set} returned by a {@link SetFrom} mapping.
+     *
+     * @remarks
+     * - Only effective changes are notified: adding a quad that the dataset already contains, or deleting one that it does not, does not invoke the listener.
+     * - Property assignments do not deduplicate: mutators remove existing quads before adding the new one, so assigning a value emits a `"delete"` notification for every quad previously matching the property (if any), followed by an `"add"` notification for the new quad - even when the assigned value equals the current one. Clearing an optional property (assigning `undefined`) emits only `"delete"` notifications.
+     * - Subscribing a listener that is already subscribed has no effect. Listeners are notified in subscription order.
+     * - Only mutations performed through this wrapper (or through another wrapper sharing the same underlying eventful dataset, such as views created by {@link named}) are observed. Mutating the wrapped dataset directly does not notify listeners.
+     *
+     * @param listener - The callback to invoke with every change to the contents of the dataset.
+     *
+     * @example Observing property assignments
+     * Assume the following RDF data:
+     * ```turtle
+     * BASE <http://example.com/>
+     *
+     * <someSubject> <someProperty> "some value" .
+     * ```
+     *
+     * Given the mapping
+     * ```ts
+     * class SomeClass extends TermWrapper {
+     *   set someProperty(value: string) {
+     *     RequiredAs.object(this, "http://example.com/someProperty", value, LiteralFrom.string)
+     *   }
+     * }
+     * ```
+     *
+     * mutations can be observed as follows:
+     * ```ts
+     * const wrapper = new DatasetWrapper(dataset, DataFactory) // which has the RDF above loaded
+     * wrapper.on((event, quad) => console.log(`${event} ${quad.object.value}`))
+     *
+     * const instance = new SomeClass("http://example.com/someSubject", wrapper, DataFactory)
+     * instance.someProperty = "some other value"
+     * // logs `delete some value` followed by `add some other value`
+     * ```
+     *
+     * @example Observing direct mutations
+     * ```ts
+     * const wrapper = new DatasetWrapper(dataset, DataFactory)
+     * wrapper.on((event, quad) => console.log(event))
+     *
+     * wrapper.add(someQuad)    // logs `add` (unless the dataset already contained the quad)
+     * wrapper.delete(someQuad) // logs `delete`
+     * ```
+     *
+     * @see
+     * - {@link off} for detaching the listener.
+     * - {@link IDatasetChangeListener} for the listener signature.
+     * - [RDF/JS: Dataset specification](https://rdf.js.org/dataset-spec/)
+     */
+    public on(listener: IDatasetChangeListener): void {
+        if (this.listeners.has(listener)) {
+            return
+        }
+
+        const handlers = {
+            added: (quad: Quad) => listener("add", quad),
+            deleted: (quad: Quad) => listener("delete", quad),
+        }
+
+        this.listeners.set(listener, handlers)
+        this.dataset.on("add", handlers.added)
+        this.dataset.on("delete", handlers.deleted)
+    }
+
+    /**
+     * Unsubscribes `listener` from change notifications for the underlying dataset.
+     *
+     * @remarks
+     * The argument must be the same function reference that was passed to {@link on}. Detaching a listener that is not subscribed has no effect.
+     *
+     * @param listener - The callback to detach.
+     *
+     * @see
+     * - {@link on} for attaching a listener.
+     */
+    public off(listener: IDatasetChangeListener): void {
+        const handlers = this.listeners.get(listener)
+
+        if (handlers === undefined) {
+            return
+        }
+
+        this.listeners.delete(listener)
+        this.dataset.off("add", handlers.added)
+        this.dataset.off("delete", handlers.deleted)
+    }
+
+    //#endregion
+
     //#region Utilities
 
     protected subjectsOf<T>(predicate: string, termWrapper: ITermWrapperConstructor<T>): Iterable<T> {

src/ensure.ts

@@ -1,4 +1,5 @@
-import type { DefaultGraph, Literal, Quad, Term } from "@rdfjs/types"
+import type { DatasetCore, DefaultGraph, Literal, Quad, Term } from "@rdfjs/types"
+import { EventfulDatasetCore } from "@jeswr/eventful-dataset"
 import { TermTypeError } from "./errors/TermTypeError.js"
 import { LiteralDatatypeError } from "./errors/LiteralDatatypeError.js"
 import type { IRdfJsTerm } from "./type/IRdfJsTerm.js"
@@ -58,3 +59,30 @@ export function ensureDefaultGraph(quad: Quad): asserts quad is Quad & { graph:
     throw new NamedGraphError(quad)
 }
 
+/**
+ * Returns `dataset` if it is already an {@link EventfulDatasetCore}, otherwise wraps it in one so that mutations can be observed.
+ *
+ * The wrapped dataset remains the single place of storage: additions and deletions performed through the returned {@link EventfulDatasetCore} write through to `dataset`, so external references to `dataset` observe all changes made through the eventful layer.
+ */
+export function ensureEventfulDatasetCore(dataset: DatasetCore): EventfulDatasetCore<Quad> {
+    if (dataset instanceof EventfulDatasetCore) {
+        return dataset
+    }
+
+    // An EventfulDatasetCore delegates storage to a dataset created by the factory it is constructed with.
+    // This factory returns the dataset being wrapped on the first invocation (the storage of the eventful layer, so that mutations write through to the original dataset) and creates fresh, independent datasets on subsequent invocations (the snapshots returned by EventfulDatasetCore.match).
+    let storage: DatasetCore | undefined = dataset
+
+    return new EventfulDatasetCore<Quad>(undefined, {
+        dataset(quads?: Quad[]): DatasetCore {
+            if (storage !== undefined) {
+                const wrapped = storage
+                storage = undefined
+                return wrapped
+            }
+
+            return new EventfulDatasetCore<Quad>(quads)
+        },
+    })
+}
+

src/mod.ts

@@ -3,6 +3,7 @@ export type * from "./type/ITermWrapperConstructor.js"
 export type * from "./type/INamedGraphDatasetConstructor.js"
 export type * from "./type/ITermFromValueMapping.js"
 export type * from "./type/ILangString.js"
+export type * from "./type/IDatasetChangeListener.js"
 
 export * from "./mapping/TermAs.js"
 export * from "./mapping/LiteralAs.js"

src/type/IDatasetChangeListener.ts

@@ -0,0 +1,51 @@
+import type { Quad } from "@rdfjs/types"
+
+/**
+ * The type of change to the contents of a dataset reported to an {@link IDatasetChangeListener}: `"add"` when a quad was added to the dataset, `"delete"` when a quad was removed from it.
+ *
+ * @see
+ * - {@link DatasetWrapper.on}
+ * - {@link DatasetWrapper.off}
+ */
+export type ChangeEvent = "add" | "delete"
+
+/**
+ * Represents the function signature of callbacks that observe changes to the contents of a {@link DatasetWrapper}.
+ *
+ * Used by this library where lambdas are accepted for reacting to quads being added to or removed from the underlying data.
+ *
+ * @remarks
+ * - Listeners are invoked synchronously, once per quad that is effectively added or removed, regardless of whether the change was performed directly on the {@link DatasetWrapper} or indirectly through a mapped property.
+ * - No restrictions are imposed on what listeners do with the notifications, except for the understanding that mutating the dataset from within a listener triggers further notifications.
+ *
+ * @param event - The type of the change: `"add"` or `"delete"`.
+ * @param quad - The quad that was added to or removed from the dataset.
+ *
+ * @example Listener as TypeScript function
+ * ```ts
+ * function listener(event: ChangeEvent, quad: Quad): void {
+ *   console.log(`${event} ${quad.object.value}`)
+ * }
+ * ```
+ *
+ * @example Listener as JavaScript function
+ * ```js
+ * function listener(event, quad) {
+ *   console.log(`${event} ${quad.object.value}`)
+ * }
+ * ```
+ *
+ * @example Listener as TypeScript typed constant
+ * Authoring listeners as typed TypeScript constants helps by compile-time checking that a lambda adheres to the interface.
+ * ```ts
+ * const listener: IDatasetChangeListener = (event, quad) => console.log(`${event} ${quad.object.value}`)
+ * ```
+ *
+ * @see
+ * - {@link DatasetWrapper.on}
+ * - {@link DatasetWrapper.off}
+ * - {@link Quad}
+ */
+export interface IDatasetChangeListener {
+    (event: ChangeEvent, quad: Quad): void
+}