Custom SPARQL engines

import { QueryEngine } from "@comunica/query-sparql-rdfjs-lite";
import type * as rdfjs from "@rdfjs/types";
import type {
  SparqlAskResults,
  SparqlBinding,
  SparqlConstructResults,
  SparqlEngineInterface,
  SparqlRequest,
  SparqlResponse,
  SparqlSelectResults,
  SparqlValue,
} from "@worlds/sdk/sparql-engine";

/** The minimal Comunica result shape this adapter consumes. */
interface ComunicaQueryResult {
  resultType: string;
  execute(): Promise<unknown>;
  metadata?(): Promise<{ variables: rdfjs.Variable[] }>;
}

/** A single row from a Comunica bindings stream. */
interface ComunicaBinding {
  get(variable: string): rdfjs.Term | undefined;
}

/** Default SPARQL query timeout in milliseconds (matches the Wazoo engine). */
const DEFAULT_TIMEOUT_MS = 30_000;

export class ComunicaSparqlEngine implements SparqlEngineInterface {
  readonly #engine = new QueryEngine();

  constructor(private readonly store: rdfjs.Store) {}

  async execute(request: SparqlRequest): Promise<SparqlResponse> {
    // One controller composes the caller's signal with the timeout, so
    // whichever fires first cancels the whole request.
    const controller = new AbortController();
    const timer = setTimeout(
      () => controller.abort(new Error("SPARQL query timed out")),
      request.timeoutMs ?? DEFAULT_TIMEOUT_MS,
    );
    const onCallerAbort = () =>
      controller.abort(
        request.signal?.reason instanceof Error
          ? request.signal?.reason
          : new Error("SPARQL query aborted"),
      );
    if (request.signal?.aborted) onCallerAbort();
    else
      request.signal?.addEventListener("abort", onCallerAbort, { once: true });

    try {
      return await Promise.race([
        this.#run(request, controller.signal),
        abortOn(controller.signal),
      ]);
    } finally {
      clearTimeout(timer);
      request.signal?.removeEventListener("abort", onCallerAbort);
    }
  }

  async #run(
    request: SparqlRequest,
    signal: AbortSignal,
  ): Promise<SparqlResponse> {
    const result = (await this.#engine.query(request.query, {
      sources: [this.store],
      baseIRI: request.baseIri,
    })) as ComunicaQueryResult;

    switch (result.resultType) {
      case "bindings":
        return { kind: "select", data: await this.#bindings(result) };
      case "boolean":
        return { kind: "ask", data: await this.#boolean(result) };
      case "quads":
        return { kind: "construct", data: await this.#quads(result) };
      case "void":
        await result.execute();
        return { kind: "void" };
      default:
        throw new Error(
          `Unsupported Comunica result type: ${result.resultType}`,
        );
    }
  }

  async #bindings(result: ComunicaQueryResult): Promise<SparqlSelectResults> {
    if (!result.metadata) {
      throw new Error("SPARQL bindings result is missing metadata.");
    }
    const metadata = await result.metadata();
    const vars = metadata.variables.map((variable) => variable.value);
    const stream = (await result.execute()) as AsyncIterable<ComunicaBinding>;

    const bindings: SparqlBinding[] = [];
    for await (const binding of stream) {
      const row: SparqlBinding = {};
      for (const variable of vars) {
        const term = binding.get(variable);
        if (term) row[variable] = termToSparqlValue(term);
      }
      bindings.push(row);
    }
    return { head: { vars }, results: { bindings } };
  }

  async #boolean(result: ComunicaQueryResult): Promise<SparqlAskResults> {
    const value = await result.execute();
    if (typeof value !== "boolean") {
      throw new Error("Comunica returned a non-boolean ASK result.");
    }
    return { head: {}, boolean: value };
  }

  async #quads(result: ComunicaQueryResult): Promise<SparqlConstructResults> {
    const quads: rdfjs.Quad[] = [];
    for await (const quad of (await result.execute()) as AsyncIterable<rdfjs.Quad>) {
      quads.push(quad);
    }
    return { quads };
  }
}

/** Maps an RDF/JS term to the SPARQL results JSON value shape. */
function termToSparqlValue(term: rdfjs.Term): SparqlValue {
  switch (term.termType) {
    case "NamedNode":
      return { type: "uri", value: term.value };
    case "BlankNode":
      return { type: "bnode", value: term.value };
    case "Literal": {
      const value: SparqlValue = { type: "literal", value: term.value };
      if (term.language) value["xml:lang"] = term.language;
      if (term.datatype.value !== "http://www.w3.org/2001/XMLSchema#string") {
        value.datatype = term.datatype.value;
      }
      return value;
    }
    case "Quad":
      return {
        type: "triple",
        value: {
          subject: termToSparqlValue(term.subject),
          predicate: termToSparqlValue(term.predicate),
          object: termToSparqlValue(term.object),
        },
      };
    default:
      throw new Error(`Unsupported RDF term type: ${term.termType}`);
  }
}

/** Rejects with the controller's reason when the signal fires. */
function abortOn(signal: AbortSignal): Promise<never> {
  return new Promise((_, reject) => {
    if (signal.aborted) {
      reject(
        signal.reason instanceof Error
          ? signal.reason
          : new Error("SPARQL query aborted"),
      );
      return;
    }
    signal.addEventListener(
      "abort",
      () =>
        reject(
          signal.reason instanceof Error
            ? signal.reason
            : new Error("SPARQL query aborted"),
        ),
      { once: true },
    );
  });
}

import { Sdk } from "@worlds/sdk";
import { RdfjsQuadStore, RdfjsSearchIndex } from "@worlds/sdk/rdfjs";
import { Store } from "n3";
import { ComunicaSparqlEngine } from "./comunica-sparql-engine.ts";

const store = new Store();
const client = new Sdk({
  quadStore: new RdfjsQuadStore({ store }),
  searchIndex: new RdfjsSearchIndex(store),
  sparqlEngine: new ComunicaSparqlEngine(store),
});