WIP

src/runner/columns.js

@@ -5,16 +5,22 @@ import { sparqlEscapeUri } from '../db.js';
 // before the leaf is identical, and the leaf is read the same way.
 // rep:self columns (empty path) are filled by assemble.js.
 
-const SUBJECT_CHUNK_SIZE = Number(process.env.SUBJECT_CHUNK_SIZE || 250);
+const SUBJECT_CHUNK_SIZE = Math.max(1, Number(process.env.SUBJECT_CHUNK_SIZE) || 100);
 
-export async function fetchColumns(sessionQuery, subjects, spec, values = new Map(), onChunk = null) {
+export async function fetchColumns(sessionQuery, subjects, spec, values = new Map(), onChunk = null, onProgress = null) {
   const groups = groupColumns(spec.columns);
-
+  // The subject list is complete and the chunk size is fixed, so the number
+  // of batch queries is known before the first one flies. onProgress fires
+  // after every answered batch with { done, total }.
+  const total = groups.length * Math.ceil(subjects.length / SUBJECT_CHUNK_SIZE);
+  let done = 0;
   for (const group of groups) {
     for (let i = 0; i < subjects.length; i += SUBJECT_CHUNK_SIZE) {
       const chunk = subjects.slice(i, i + SUBJECT_CHUNK_SIZE);
       const result = await sessionQuery(groupQuery(group, chunk));
       collect(result, group, values);
+      done += 1;
+      onProgress?.({ done, total });
     }
     if (onChunk) await onChunk(group);
   }

src/runner/csv.js

@@ -1,4 +1,4 @@
-import { mkdirSync, writeFileSync } from 'fs';
+import { mkdirSync, writeFileSync, readFileSync } from 'fs';
 
 // Rows to a CSV file. CSV_SEPARATOR is ';' by default. A cell is quoted
 // only when it holds the separator, a quote or a newline; a quote inside a
@@ -23,4 +23,18 @@ export function writeCsv(name, csv) {
   const file = `${SHARE_DIR}/${name}`;
   writeFileSync(file, csv, 'utf8');
   return file;
+}
+
+// The spec of the run, next to its CSV (same base name, .ttl), so a report
+// can be debugged afterwards. The store keeps only the pointer.
+export function writeSpec(csvName, spec) {
+  mkdirSync(SHARE_DIR, { recursive: true });
+  const name = csvName.replace(/\.csv$/, '') + '.ttl';
+  writeFileSync(`${SHARE_DIR}/${name}`, spec, 'utf8');
+  return `share://${name}`;
+}
+
+export function readSpec(pointer) {
+  if (!pointer.startsWith('share://')) throw new Error(`not a share:// pointer: ${pointer}`);
+  return readFileSync(`${SHARE_DIR}/${pointer.slice('share://'.length)}`, 'utf8');
 }

src/runner/report.js

@@ -44,7 +44,7 @@ export async function registerReport(sessionUpdateFn, title, fileUri, extra = {}
 
   const extraTriples = [];
   if (extra.question) extraTriples.push(`${sparqlEscapeUri(reportUri)} ${sparqlEscapeUri('http://mu.semte.ch/vocabularies/ext/question')} ${sparqlEscapeString(extra.question)} .`);
-  if (extra.spec) extraTriples.push(`${sparqlEscapeUri(reportUri)} ${sparqlEscapeUri('http://mu.semte.ch/vocabularies/ext/spec')} ${sparqlEscapeString(extra.spec)} .`);
+  if (extra.specFile) extraTriples.push(`${sparqlEscapeUri(reportUri)} ${sparqlEscapeUri('http://mu.semte.ch/vocabularies/ext/specFile')} ${sparqlEscapeString(extra.specFile)} .`);
   if (extra.creator) extraTriples.push(`${sparqlEscapeUri(reportUri)} ${sparqlEscapeUri('http://purl.org/dc/terms/creator')} ${sparqlEscapeUri(extra.creator)} .`);
 
   await (sessionUpdateFn || sessionUpdate)(`

src/runner/run.js

@@ -1,28 +1,34 @@
 import { seed } from './seed.js';
 import { fetchColumns } from './columns.js';
 import { assemble } from './assemble.js';
-import { rowsToCsv, writeCsv } from './csv.js';
+import { rowsToCsv, writeCsv, writeSpec } from './csv.js';
 import { registerFile, registerReport } from './report.js';
 import { touch } from './job.js';
 
 // Ties the steps together: seed → columns → assemble → CSV → register.
 // Returns { reportUri, fileUri, rowCount }. The task is touched once per
-// column group so the dashboard shows movement.
+// column group so the dashboard shows movement. onProgress fires on every
+// answered query with what the phase knows — { subjects } per seed page,
+// { done, total } per column batch — so a waiter can tell a live run from
+// a stalled one.
 
-export async function run(query, update, parsed, profile, title, taskUri, extra = {}) {
-  const subjects = await seed(query, parsed);
+export async function run(query, update, parsed, profile, title, taskUri, extra = {}, onProgress = null) {
+  const subjects = await seed(query, parsed,
+    onProgress ? (n) => onProgress({ subjects: n }) : null);
 
   const values = new Map();
   await fetchColumns(query, subjects, parsed, values,
-    taskUri ? () => touch(update, taskUri) : null);
+    taskUri ? () => touch(update, taskUri) : null, onProgress);
 
   const rows = assemble(subjects, values, parsed);
   const fileName = `${slug(title)}.csv`;
   const csv = rowsToCsv(rows);
   const filePath = writeCsv(fileName, csv);
+  const specFile = extra.spec && writeSpec(fileName, extra.spec);
 
   const fileUri = await registerFile(update, fileName, filePath);
-  const reportUri = await registerReport(update, title, fileUri, extra);
+  const reportUri = await registerReport(update, title, fileUri,
+    specFile ? { ...extra, specFile } : extra);
 
   return { reportUri, fileUri, rowCount: rows.length - 1, filePath };
 }

src/runner/seed.js

@@ -1,10 +1,12 @@
 import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime, sparqlEscapeDate } from '../db.js';
 
 // Subject selection. One paged query per page; stops at ROW_LIMIT.
+// onPage(n) fires after every answered page with the subject count so far,
+// so a caller waiting on this run can tell a live one from a stalled one.
 const ROW_LIMIT = Number(process.env.ROW_LIMIT || 200000);
 const PAGE_SIZE = Number(process.env.SEED_PAGE_SIZE || 5000);
 
-export async function seed(sessionQuery, spec) {
+export async function seed(sessionQuery, spec, onPage = null) {
   const subjects = [];
   let offset = 0;
   for (;;) {
@@ -12,6 +14,7 @@ export async function seed(sessionQuery, spec) {
     const result = await sessionQuery(query);
     const rows = result.results.bindings.map(b => b.s.value);
     subjects.push(...rows);
+    onPage?.(subjects.length);
     if (subjects.length > ROW_LIMIT) {
       throw new Error(`the question matches more than ${ROW_LIMIT} subjects. Add a filter, for example a date range or a bestuurseenheid.`);
     }

src/runner/spec.js

@@ -90,8 +90,13 @@ function parsePath(store, pathQuad) {
   while (cur && cur.value !== RDF + 'nil') {
     const item = store.getQuads(cur, RDF + 'first', null)[0]?.object;
     if (!item) throw new Error('sh:path list is broken');
-    if (item.termType !== 'NamedNode') throw new Error('sh:path list items must be URIs');
-    hops.push({ predicate: item.value, inverse: false });
+    if (item.termType === 'BlankNode') {
+      const inv = store.getQuads(item, SH + 'inversePath', null)[0]?.object;
+      if (!inv) throw new Error('sh:path list items must be URIs or [ sh:inversePath <predicate> ]');
+      hops.push({ predicate: inv.value, inverse: true });
+    } else {
+      hops.push({ predicate: item.value, inverse: false });
+    }
     cur = store.getQuads(cur, RDF + 'rest', null)[0]?.object;
   }
   return hops;