storeForest.js

import { Readable } from 'stream';
import fs from 'fs';
import crypto from 'crypto';
import {v4 as uuidv4} from 'uuid';
import { rebaseTerm, escapeLiteral } from './turtleEncoding.js';
import { generateParameterLabel, mergePreambles } from './queryHandling.js';
import SparqlGraphConnection from './SparqlGraphConnection.js';
import httpCall from './httpCall.js';
import { buildStoreGraphUrl } from './queryEndpoint.js';
import { stringify } from 'csv-stringify';
import RdfFileStreamConnection from './RdfFileStreamConnection.js';
const JSON_DATATYPE_URI = 'https://www.iana.org/assignments/media-types/application/json';

function md5(str) {
    return crypto.createHash('md5').update(str).digest("hex")
}

function escapeLsqId(lsqId) {
    return lsqId.replaceAll('-', '_')
}

const LSQ_QUERIES_PREFIX = 'http://lsq.aksw.org/lsqQuery-';

export default class ParametricQueriesStorage {
    queryCount = 0;

    constructor(options) {
        this.options = options;
        const {resourcesNs, outputGraphStoreURL, outputDirPath, outputGraphnameId, outputGraphname} = this.options;
        this.preamble = `
        PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
        PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
        PREFIX lsqv: <http://lsq.aksw.org/vocab#>
        PREFIX prov: <http://www.w3.org/ns/prov#>
        PREFIX prv: <http://purl.org/net/provenance/ns#>
        PREFIX prvTypes: <http://purl.org/net/provenance/types#>
        PREFIX wfprov: <http://purl.org/wf4ever/wfprov#>
        PREFIX sh: <http://www.w3.org/ns/shacl#>
        PREFIX schema: <https://schema.org/>
        PREFIX spaclus: <http://sparql-clustering/vocab/>
        PREFIX dcterms: <http://purl.org/dc/terms/>
        PREFIX void: <http://rdfs.org/ns/void#>

        PREFIX lsqQueries: <http://lsq.aksw.org/lsqQuery->
        PREFIX actions: <${resourcesNs}actions/>
        PREFIX templates: <${resourcesNs}templates/>
        PREFIX executions: <${resourcesNs}executions/>
        PREFIX queryGenerations: <${resourcesNs}queryGenerations/>
        PREFIX params: <${resourcesNs}params/>
        PREFIX bindings: <${resourcesNs}bindings/>
        PREFIX paramPaths: <${resourcesNs}paramPaths/>
        `;
        this.metadataPreamble = `
        PREFIX schema: <https://schema.org/>
        PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
        PREFIX sd: <http://www.w3.org/ns/sparql-service-description#>
        PREFIX spaclus: <http://sparql-clustering/vocab/>
        
        PREFIX actions: <${resourcesNs}actions/>
        PREFIX graphs: <${resourcesNs}graphs/>
        PREFIX services: <${resourcesNs}services/>
        PREFIX datasets: <${resourcesNs}datasets/>
        `;

        if (outputGraphStoreURL) {
            const outputUrl = buildStoreGraphUrl(outputGraphStoreURL, outputGraphname);
            this.outputGraphConnection = new SparqlGraphConnection(outputUrl, {
                preamble: this.preamble
            });
        } else if (outputDirPath) {
            const outputFilename = outputGraphnameId || 'output';
            const extension =
                (options.format === 'application/n-quads' ? 'nq' : 'nt') +
                (options.compress ? '.gz' : '');
            this.outputGraphConnection = new RdfFileStreamConnection (
                outputDirPath + outputFilename + '.' + extension, {
                    ...options,
                    preamble: this.preamble,
                    outputGraphname: outputGraphname
                }
            );
        }

    }

    async recordProcessStart() {
        const {
            inputEndpointURL, inputGraphnames = null,
            metadataUpdateURL, metadataGraphname = null,
            overwriteOutputGraph = true
        } = this.options

        const actionId = uuidv4()
        const inputGraphStoreId = encodeURIComponent(inputEndpointURL)

        const metadataSetup = this.metadataPreamble + `
        INSERT DATA {
            ${metadataGraphname ? `GRAPH <${metadataGraphname}> {` : ''}
                actions:${actionId} a schema:Action, spaclus:SparqlClustering;
                    schema:startTime '${new Date().toISOString()}'^^xsd:dateTime;
                    schema:actionStatus schema:ActiveActionStatus;
                    spaclus:parameters ${escapeLiteral(JSON.stringify(this.options))}^^<${JSON_DATATYPE_URI}>.
            
                services:${inputGraphStoreId} a sd:Service;
                    sd:endpoint <${inputEndpointURL}>;
                    sd:defaultDataset datasets:${inputGraphStoreId}.

                ${this.options.cluster ?
                    `actions:${actionId} schema:object <${this.options.cluster}>` :
                    (inputGraphnames || [null]).map(inputGraphname => {
                        const inputGraphId = encodeURIComponent(inputEndpointURL) + (inputGraphname ? '_' + encodeURIComponent(inputGraphname) : '')
                        return `
                        actions:${actionId} schema:object graphs:${inputGraphId}.
                        graphs:${inputGraphId} a ${inputGraphname ? `sd:NamedGraph; sd:name <${inputGraphname}>` : 'sd:Graph'}.
                        datasets:${inputGraphStoreId} a sd:Dataset;
                            ${inputGraphname ? 'sd:namedGraph' : 'sd:defaultGraph'} graphs:${inputGraphId}.`;
                    }).join('')}
            ${metadataGraphname ? '}' : ''}
        };`

        await httpCall(metadataUpdateURL, {
            method: 'POST',
            headers: {'Content-Type': 'application/sparql-update'},
            body: metadataSetup
        });

        if (this.outputGraphConnection) {
            if (overwriteOutputGraph) {
                await this.outputGraphConnection.delete();
                if (this.options.dataset) {
                    await this.outputGraphConnection.post(`
                    <${this.options.dataset}> a void:Dataset.
                    ${this.options.datasetId ?
                        `<${this.options.dataset}> dcterms:identifier ${escapeLiteral(this.options.datasetId)}.` :
                        ''}
                    `);
                }
            }

            if (this.options.cluster) {
                await this.outputGraphConnection.post(`
                <${this.options.cluster}> a spaclus:QueryCluster.
                ${this.options.dataset ? `<${this.options.cluster}> dcterms:isPartOf <${this.options.dataset}>.` : ''}
                ${'clusterId' in this.options ? `<${this.options.cluster}> dcterms:identifier ${this.options.clusterId}.` : ''}
                `);
            }
        }

        return actionId
    }

    async storeForest(forest, parentQueryId, actionId) {

        if (this.outputGraphConnection) {
            for (const query of forest) {
                const {text, instances, specializations, preamble} = query;

                const queryId = ++this.queryCount;

                await this.outputGraphConnection.post((
                    parentQueryId ?
                        `templates:${queryId} prov:specializationOf  templates:${parentQueryId}.` :
                        // `templates:${queryId} prov:wasGeneratedBy actions:${actionId}.`) + `
                        // `actions:${actionId} schema:result templates:${queryId}.
                        this.options.cluster ?
                            `templates:${queryId} prov:specializationOf <${this.options.cluster}>.` :
                            this.options.dataset ?
                                `templates:${queryId} dcterms:isPartOf <${this.options.dataset}>.` :
                                `actions:${actionId} schema:result templates:${queryId}.`) + `
                    templates:${queryId}
                        a prvTypes:QueryTemplate, sh:Parametrizable;
                        dcterms:identifier ${queryId};
                        lsqv:text ${escapeLiteral(text)}.
                `);

                for (const paramIndex of instances[0].bindings.map(({}, paramIndex) => paramIndex)) {
                    const paramName = generateParameterLabel(paramIndex).slice(1);
                    await this.outputGraphConnection.post(`
                        templates:${queryId} sh:parameter params:${queryId}_${paramIndex}.

                        params:${queryId}_${paramIndex}
                            a sh:Parameter;
                            sh:path paramPaths:${paramName};
                            sh:description "${paramName}".
                    `);
                }

                for (const {id: lsqIdUnesc, bindings} of instances) {
                    const lsqId = escapeLsqId(lsqIdUnesc)
                    await this.outputGraphConnection.post(`
                        queryGenerations:${queryId}_${lsqId}
                            a  prvTypes:DataCreation;
                            prv:usedGuideline templates:${queryId}.

                        <${LSQ_QUERIES_PREFIX}${lsqIdUnesc}>
                            a prvTypes:SPARQLQuery;
                            prv:createdBy queryGenerations:${queryId}_${lsqId}.
                    `);

                    for (const [paramIndex, bindingValue] of bindings.entries()) {
                        let rdfValue;
                        try {
                            rdfValue = rebaseTerm(bindingValue, mergePreambles(this.options.defaultPreamble, preamble));
                        } catch(exception) {}
                        await this.outputGraphConnection.post(`
                            queryGenerations:${queryId}_${lsqId} prv:usedData bindings:${lsqId}_${paramIndex}.
                            bindings:${lsqId}_${paramIndex}
                                a wfprov:Artifact;
                                wfprov:describedByParameter params:${queryId}_${paramIndex};
                                ${rdfValue ? `rdf:value ${rdfValue};` : ''}
                                lsqv:text ${escapeLiteral(bindingValue)}. 
                        `); // wfprov:usedInput
                    }
                }

                await this.storeForest(specializations, queryId);
            }
        }

    }

    async linkSingleQueries(queryLsqIds, actionId) {

        if (this.outputGraphConnection) {
            for (const lsqId of queryLsqIds) {
                var queryTurtle = `
                actions:${actionId} schema:result <${LSQ_QUERIES_PREFIX}${lsqId}>.
                ${this.options.cluster ?
                    `<${this.options.cluster}> rdfs:member <${LSQ_QUERIES_PREFIX}${lsqId}>.` :
                    this.options.dataset ?
                        `<${this.options.dataset}> rdfs:member <${LSQ_QUERIES_PREFIX}${lsqId}> .` :
                        `actions:${actionId} schema:result <${LSQ_QUERIES_PREFIX}${lsqId}>.`}
                <${LSQ_QUERIES_PREFIX}${lsqId}> a prvTypes:SPARQLQuery.
                `;
                this.outputGraphConnection.post(queryTurtle);
            }
        }

    }

    storeStats(groupingResult) {
        const {
            queryForest, nonClusterizedQueryIds,
            totalQueries, totalExecutions,
            timeOfFirstExecution, timeOfLastExecution
        } =
            'queryForest' in groupingResult ?
                groupingResult :
                {queryForest: groupingResult};
        const {statsFilePath} = this.options;
        const queryStatsStream = new Readable({objectMode: true});
        queryStatsStream.pipe(stringify({header: true})).pipe(fs.createWriteStream(statsFilePath));
        if (totalQueries) {
            queryStatsStream.push({
                queryId: 0,
                text: '',
                numOfInstances: totalQueries,
                numOfExecutions: totalExecutions,
                timeOfFirstExecution, timeOfLastExecution,
                sumOfLogOfNumOfExecutions: 0,
                numOfSpecializations: nonClusterizedQueryIds ? nonClusterizedQueryIds.length : 0,
                numOfOriginalRdfTerms: 0,
                numOfParameters: 0
            })
        }
        queryForest.map(({
            text, numOfInstances, numOfExecutions,
            timeOfFirstExecution, timeOfLastExecution,
            sumOfLogOfNumOfExecutions,
            specializations,
            instances
        }, queryIndex) => ({
            queryId: queryIndex + 1, text,
            numOfInstances, numOfExecutions,
            timeOfFirstExecution, timeOfLastExecution,
            sumOfLogOfNumOfExecutions,
            numOfSpecializations: specializations.length,
            numOfOriginalRdfTerms: instances[0].numOfOriginalRdfTerms,
            numOfParameters: instances[0].bindings.length
        })).sort((a, b) => b.numOfExecutions - a.numOfExecutions).forEach(queryStats => {
            queryStatsStream.push(queryStats);
        });
        queryStatsStream.push(null);
    }

    async recordProcessCompletion(actionId) {
        const {
            outputGraphStoreURL, outputGraphname = null,
            metadataUpdateURL, metadataGraphname = null
        } = this.options
    
        if (this.outputGraphConnection) {
            await this.outputGraphConnection.sync();
        }

        const outputGraphId = encodeURIComponent(outputGraphStoreURL) + (outputGraphname ? '_' + encodeURIComponent(outputGraphname) : '')
        const outputGraphStoreId = encodeURIComponent(outputGraphStoreURL)

        const metadataUpdate = this.metadataPreamble + `
        DELETE DATA {
            ${metadataGraphname ? `GRAPH <${metadataGraphname}> {` : ''}
                actions:${actionId} schema:actionStatus schema:ActiveActionStatus
            ${metadataGraphname ? '}' : ''}
        };
        
        INSERT DATA {
            ${metadataGraphname ? `GRAPH <${metadataGraphname}> {` : ''}
                actions:${actionId} schema:actionStatus schema:CompletedActionStatus;
                    schema:endTime '${new Date().toISOString()}'^^xsd:dateTime;
                    schema:result graphs:${outputGraphId}.

                graphs:${outputGraphId} a ${outputGraphname ? `sd:NamedGraph; sd:name <${outputGraphname}>` : 'sd:Graph'}.

                services:${outputGraphStoreId} a sd:Service;
                    sd:endpoint <${outputGraphStoreURL}>;
                    sd:defaultDataset datasets:${outputGraphStoreId}.
        
                datasets:${outputGraphStoreId} a sd:Dataset;
                    ${outputGraphname ? 'sd:namedGraph' : 'sd:defaultGraph'} graphs:${outputGraphId}.
            ${metadataGraphname ? '}' : ''}
        };`

        await httpCall(metadataUpdateURL, {
            method: 'POST',
            headers: {'Content-Type': 'application/sparql-update'},
            body: metadataUpdate
        });
    }
    
    async recordProcessFailure(actionId, error) {
        const {
            metadataUpdateURL, metadataGraphname = null
        } = this.options
    
        if (this.outputGraphConnection) {
            try {
                await this.outputGraphConnection.sync();
            } catch(e) {}
        }

        const metadataUpdate = this.metadataPreamble + `
        DELETE DATA {
            ${metadataGraphname ? `GRAPH <${metadataGraphname}> {` : ''}
                actions:${actionId} schema:actionStatus schema:ActiveActionStatus
            ${metadataGraphname ? '}' : ''}
        };
        
        INSERT DATA {
            ${metadataGraphname ? `GRAPH <${metadataGraphname}> {` : ''}
                actions:${actionId} schema:actionStatus schema:FailedActionStatus;
                    schema:endTime '${new Date().toISOString()}'^^xsd:dateTime;
                    schema:error ${escapeLiteral('' + error)}.
            ${metadataGraphname ? '}' : ''}
        };`

        await httpCall(metadataUpdateURL, {
            method: 'POST',
            headers: {'Content-Type': 'application/sparql-update'},
            body: metadataUpdate
        });
    }
    
}