graphy/cli.js

#!/usr/bin/env node


/* eslint-disable global-require */
/* eslint-disable no-console */

const gobble = (s_text, s_indent='') => {
	let m_pad = /^(\s+)/.exec(s_text.replace(/^([ \t]*\n)/, ''));
	if(m_pad) {
		return s_indent+s_text.replace(new RegExp(`\\n${m_pad[1]}`, 'g'), '\n'+s_indent.trim()).trim();
	}
	else {
		return s_indent+s_text.trim();
	}
};

const fs =require('fs');
const path = require('path');
const yargs = require('yargs');
const mk_yargs = require('yargs/yargs');
const graphy = require('./api.js');
const factory = require('@graphy/core.data.factory');
const stream = require('@graphy/core.iso.stream');
const dataset_tree = require('@graphy/memory.dataset.fast');

const parse_filter = require('./quad-expression.js').parse;
const expression_handler = require('./expression-handler.js');

const F_ADAPT_STREAM = function(ds_out) {
	let ds_dst = ds_out;

	// non-object mode
	if(!ds_dst._writableState.objectMode) {
		// transform to JSON
		ds_out = stream.quads_to_json();
	}
	// yes object mode and graphy writable
	else if(ds_out.isGraphyWritable) {
		// transform to writable data events
		ds_out = stream.quads_to_writable();
	}
	// forward as-is to super
	else {
		return this.constructor.prototype.pipe.call(this, ds_dst);
	}

	// forward output to super
	this.constructor.prototype.pipe.call(this, ds_out);

	// pipe output to destination
	return ds_out.pipe(ds_dst);
};

const bypass = a_inputs => a_inputs.map((ds_input) => {
	// intercept pipe
	ds_input.pipe = F_ADAPT_STREAM;

	return ds_input;
});

const map_streams = (a_inputs, f_map) => a_inputs.map((ds_input) => {
	let ds_output = f_map(ds_input);

	// intercept pipe
	ds_output.pipe = F_ADAPT_STREAM;

	// pipe input to step and return step
	return ds_input.pipe(ds_output);
});

const warp_term = (z_term, h_prefixes) => {
	// c1 string
	if('string' === typeof z_term) {
		return factory.c1(z_term, h_prefixes);
	}
	// normalize term
	else {
		return factory.fromTerm(z_term);
	}
};

const interpret_item = (z_item, h_prefixes, ds_transform, fke_transform) => {
	// array
	if(Array.isArray(z_item)) {
		// zero-length, skip
		if(!z_item.length) return fke_transform();

		// first object is also array
		if(Array.isArray(z_item[0])) {
			let nl_subs = z_item.length;

			let c_resolves = 0;

			for(let z_sub of z_item) {
				interpret_item(z_sub, h_prefixes, ds_transform, () => {  // eslint-disable-line no-loop-func
					if(++c_resolves === nl_subs) {
						fke_transform();
					}
				});
			}

			// do not consume transform synchronously
			return;
		}
		// triple/quad
		else if(3 === z_item.length || 4 === z_item.length) {
			let a_terms = z_item.map(z => warp_term(z, h_prefixes));

			ds_transform.push(factory.quad(...a_terms));
		}
	}
	// string (trig)
	else if('string' === typeof z_item) {
		graphy.content.trig.read({
			input: {
				string: z_item,

				error(e_read) {
					warn(`The 'transform' command threw an Error while trying to read the returned TriG string: '${z_item}'\n\nThe reader reported: ${e_read.stack}`);

					// done
					fke_transform();
				},
			},

			data(g_quad_read) {
				ds_transform.push(g_quad_read);
			},

			eof() {
				// done
				fke_transform();
			},
		});

		// do not consume transform synchronously
		return;
	}
	// quad
	else if(z_item.subject && z_item.predicate && z_item.object) {
		ds_transform.push(factory.fromQuad(z_item));
	}
	// iterable
	else if(z_item[Symbol.iterator]) {
		for(let g_quad_it of z_item) {
			ds_transform.push(g_quad_it);
		}
	}
	// other
	else {
		exit(`The callback function supplied to the 'transform' command returned an invalid quad value: '${z_item}'`);
	}

	// done
	fke_transform();
};

const dataset_N1QQ = async(g_argv, a_inputs, fe_command, s_operation) => {
	let b_canonicalize = !g_argv.strict;

	// create trees
	let a_trees = a_inputs.map(() => dataset_tree());

	// initial tree
	let k_tree_out = a_trees[0];

	// pairwise readiness
	for(let i_input=0, nl_inputs=a_inputs.length; i_input<nl_inputs; i_input++) {
		let k_tree_b = a_trees[i_input];

		// pipe input stream to tree b
		a_inputs[i_input].pipe(k_tree_b);

		// wait for input stream to finish writing to b
		await k_tree_b.until('finish');

		// canonicalize
		if(b_canonicalize) {
			k_tree_b = a_trees[i_input] = k_tree_b.canonicalize();

			// update out ref
			if(!i_input) k_tree_out = k_tree_b;
		}

		// non-first input
		if(i_input) {
			// perform pairwise operation
			k_tree_out = k_tree_out[s_operation](k_tree_b);
		}
	}

	// return readable tree
	return [k_tree_out];
};

const dataset_21QQ = (g_argv, a_inputs, fe_command, s_operation) => new Promise((fk_resolve) => {
	let b_canonicalize = !g_argv.strict;

	let operate = () => [k_tree_a[s_operation](k_tree_b)];

	// wait for a
	let k_tree_a = dataset_tree();
	let b_finished_a = false;
	k_tree_a.on('finish', () => {
		// canonicalize
		if(b_canonicalize) k_tree_a = k_tree_a.canonicalize();

		// a is finished now
		b_finished_a = true;

		// b is already finished
		if(b_finished_b) fk_resolve(operate());
	});

	// wait for b
	let k_tree_b = dataset_tree();
	let b_finished_b = false;
	k_tree_b.on('finish', () => {
		// canonicalize
		if(b_canonicalize) k_tree_b = k_tree_b.canonicalize();

		// b is finished now
		b_finished_b = true;

		// a is already finished
		if(b_finished_a) fk_resolve(operate());
	});

	// ref both input streams
	let [ds_input_a, ds_input_b] = a_inputs;

	// pipe each to its tree
	ds_input_a.pipe(k_tree_a);
	ds_input_b.pipe(k_tree_b);
});

const dataset_21QR = (g_argv, a_inputs, fe_command, s_operation) => new Promise((fk_resolve) => {
	let b_canonicalize = !g_argv.strict;

	let operate = () => [new AnswerSource(k_tree_a[s_operation](k_tree_b))];

	// wait for a
	let k_tree_a = dataset_tree();
	let b_finished_a = false;
	k_tree_a.on('finish', () => {
		// canonicalize
		if(b_canonicalize) k_tree_a = k_tree_a.canonicalize();

		// a is finished now
		b_finished_a = true;

		// b is already finished
		if(b_finished_b) fk_resolve(operate());
	});

	// wait for b
	let k_tree_b = dataset_tree();
	let b_finished_b = false;
	k_tree_b.on('finish', () => {
		// canonicalize
		if(b_canonicalize) k_tree_b = k_tree_b.canonicalize();

		// b is finished now
		b_finished_b = true;

		// a is already finished
		if(b_finished_a) fk_resolve(operate());
	});

	// ref both input streams
	let [ds_input_a, ds_input_b] = a_inputs;

	// pipe each to its tree
	ds_input_a.pipe(k_tree_a);
	ds_input_b.pipe(k_tree_b);
});

class AnswerSource extends require('stream').Readable {
	constructor(w_datum) {
		super({
			objectMode: true,
		});

		this._w_datum = w_datum;
	}

	// intercept pipe
	pipe(ds_dst) {
		// string out
		if(!ds_dst._writableState.objectMode) {
			// change read mode; push as JSON
			this._read = () => {
				this.push(JSON.stringify(this._w_datum)+'\n', 'utf8');
				this.push(null);
			};
		}

		// forward to super
		return super.pipe(ds_dst);
	}

	// push object
	_read() {
		this.push(this._w_datum);
		this.push(null);
	}
}

function ignore_pipe_warnings(ds_src) {
	let f_pipe = ds_src.pipe;

	// override pipe
	ds_src.pipe = function pipe(ds_dst) {
		// pipe without warning because of worker threads
		let n_maxl_src = this._maxListeners || 10;
		let n_maxl_dst = ds_dst._maxListeners || 10;

		// set infinite max listeners
		this.setMaxListeners(Infinity);
		ds_dst.setMaxListeners(Infinity);

		// call pipe
		let w_ret = f_pipe.call(this, ds_dst);

		// reset max listeners
		this._maxListeners = n_maxl_src;
		ds_dst._maxListeners = n_maxl_dst;

		// return value
		return w_ret;
	};

	// return self
	return ds_src;
}

let b_show_stack_trace = false;
const warn = (s_message) => {
	let s_msg = (new Error(s_message)).stack;

	if(!b_show_stack_trace) {
		s_msg = s_msg.replace(/\n\s+at [^\n]*\n/, '\n')
			.replace(/^Error:/, 'Warning:');
	}

	console.warn(s_msg);
};

const exit = (s_exit) => {
	console.error(s_exit);
	process.exit(1);
};


const S_TRANSFORM_TYPE_NNSQ = 'Transform Type:  N-to-N (map);  (...Strings) --> [...Quads]';
const S_TRANSFORM_TYPE_NNQS = 'Transform Type:  N-to-N (map);  (...Quads) --> [...Strings]';
const S_TRANSFORM_TYPE_NNQQ = 'Transform Type:  N-to-N (map);  (...Quads) --> [...Quads]';
const S_TRANSFORM_TYPE_NNQRB = 'Transform Type:  N-to-N (map);  (...Quads) --> [...ResultValues<Boolean>]';
const S_TRANSFORM_TYPE_NNQRN = 'Transform Type:  N-to-N (map);  (...Quads) --> [...ResultValues<Number>]';
const S_TRANSFORM_TYPE_21QQ = 'Transform Type:  2-to-1 (join);  (Quads, Quads) --> [Quads]';
const S_TRANSFORM_TYPE_N1QQ = 'Transform Type:  N-to-1 (reduce);  (...Quads) --> [Quads]';
const S_TRANSFORM_TYPE_N1AA = 'Transform Type:  N-to-1 (reduce);  (...Any) --> [Any]';
const S_TRANSFORM_TYPE_21QRB = 'Transform Type:  2-to-1 (join);  (Quads, Quads) --> [ResultValues<Boolean>]';
const S_TRANSFORM_TYPE_21QRN = 'Transform Type:  2-to-1 (join);  (Quads, Quads) --> [ResultValues<Number>]';

const S_CATEGORY_IO = 'Input/Output Commands:';
const S_CATEGORY_DATA = 'Quad Manipulation Commands:';
const S_CATEGORY_STREAM = 'Stream Control Commands:';
const S_CATEGORY_SET = 'Dataset Commands:';
const S_CATEGORY_STATS = 'Statistics Commands:';

const S_WARN_JAVASCRIPT = `WARNING: the '-j' / '--js' / '--javascript' option evals the given code. DO NOT allow user input into this option as it will grant them arbitrary code execution with whatever privileges the process is running under.`;

const G_OPTIONS_DATASET = {
	strict: {
		type: 'boolean',
		describe: 'if true, forgoes canonicalization before the set operation',
	},
};

const content_type_flags = f_verb => ({
	// nt: {
	// 	type: 'boolean',
	// 	alias: ['ntriples', 'n-triples'],
	// 	conflicts: ['ttl', 'nq', 'trig'],
	// 	describe: f_verb('N-Triples'),
	// },

	// nq: {
	// 	type: 'boolean',
	// 	alias: ['nquads', 'n-quads'],
	// 	conflicts: ['nt', 'ttl', 'trig'],
	// 	describe: f_verb('N-Quads'),
	// },

	// ttl: {
	// 	type: 'boolean',
	// 	alias: ['turtle'],
	// 	conflicts: ['nt', 'nq', 'trig'],
	// 	describe: f_verb(' Turtle'),
	// },

	// trig: {
	// 	type: 'boolean',
	// 	conflicts: ['nt', 'nq', 'ttl'],
	// 	describe: f_verb('TriG'),
// },
});

// commands
let h_commands = {  // eslint-disable-next-line quote-props
/*
------ Input/Output --------
*/

	read: {
		type: S_TRANSFORM_TYPE_NNSQ,
		category: S_CATEGORY_IO,
		overview: 'Deserialize RDF content',
		description: [
			'Read RDF content, i.e., deserialize it, from 1 or more inputs using a single thread',
		],
		options: {
			c: {
				type: 'string',
				alias: ['content-type'],
				default: 'trig',
				describe: 'either an RDF Content-Type or format selector',
				group: 'Content Selector Options:',
			},
			r: {
				type: 'boolean',
				alias: ['relax'],
				default: undefined,  // eslint-disable-line no-undefined
				describe: 'relax validation of tokens within the RDF document',
			},
			b: {
				type: 'string',
				alias: ['base', 'base-uri', 'base-iri'],
				describe: 'set a base URI on the document',
			},

			...content_type_flags(s => `read from ${s}`),
		},
		examples: [
			`read -c nt`,
			`read -c n-triples`,
			`read -c 'application/n-triples'`,
			`read -c ttl`,
			`read -c turtle`,
			`read -c 'text/turtle'`,
		],

		command(g_argv, a_inputs, fe_command) {
			// select reader
			let f_reader = graphy.content(g_argv['content-type']).read;

			let gc_read = {
				relax: g_argv.relax || false,
			};

			// 'base-uri' => 'baseUri'
			if(g_argv['base-uri']) {
				gc_read.baseUri = g_argv['base-uri'];
			}

			return map_streams(a_inputs, () => f_reader({
				...gc_read,

				error: e => fe_command(e),
			}));
		},
	},

	scan: {
		type: S_TRANSFORM_TYPE_NNSQ,
		category: S_CATEGORY_IO,
		overview: 'Deserialize RDF content using multiple threads',
		description: [
			'Scan RDF content, i.e., deserialize it, from 1 or more inputs using multiple threads',
		],
		options: {
			c: {
				type: 'string',
				alias: ['content-type'],
				default: 'nt',
				describe: 'either an RDF Content-Type or format selector',
				group: 'Content Selector Options:',
			},
			r: {
				type: 'boolean',
				alias: ['relax'],
				default: undefined,  // eslint-disable-line no-undefined
				describe: 'relax validation of tokens within the RDF document',
				group: 'Read Options:',
			},
			// b: {
			// 	type: 'string',
			// 	alias: ['base', 'base-uri', 'base-iri'],
			// 	describe: 'set a base URI on the document',
			// },
			threads: {
				type: 'number',
				describe: 'manually set the total number of threads to use (including the main thread)',
				group: 'Scan Options:',
			},

			...content_type_flags(s => `scan from ${s}`),
		},
		examples: [
			`scan -c nt`,
			`scan -c n-triples`,
			`scan -c 'application/n-triples'`,
			`scan -c nq`,
			`scan -c nquads`,
			`scan -c 'application/n-quads'`,
		],

		command(g_argv, a_inputs, fe_command, a_rest) {
			// select scanner
			let f_scanner = graphy.content(g_argv['content-type']).scan;

			let gc_scan = {
				relax: g_argv.relax || false,
				threads: g_argv.threads || 0,
			};

			// next command
			let a_series = a_rest.shift();
			let si_next = a_series? a_series[0]: null;

			// command
			switch(si_next) {
				// count task
				case 'count': {
					return a_inputs.map(ds_input => new Promise((fk_resolve) => {
						ds_input.on('error', fe_command);

						f_scanner(ds_input, {
							...gc_scan,

							preset: 'count',

							error: fe_command,

							report(c_quads) {
								fk_resolve(ignore_pipe_warnings(new AnswerSource(c_quads)));
							},
						});
					}));
				}

				// scribe / ndjson
				case 'scribe':
				case null: {
					return a_inputs.map((ds_input) => {
						ds_input.on('error', fe_command);

						let ds_output = ignore_pipe_warnings(new stream.PassThrough());

						f_scanner(ds_input, {
							...gc_scan,

							preset: si_next || 'ndjson',

							error: fe_command,

							update(z_scribe, i_worker) {
								ds_output.write(i_worker? Buffer.from(z_scribe).toString(): z_scribe);
							},

							report() {
								ds_output.end();
							},
						});

						return ds_output;
					});
				}

				// other
				default: {
					fe_command(`The 'scan' command only supports a limited set of commands to follow it; '${si_next}' is not supported`);
				}
			}
		},
	},

	scribe: {
		type: S_TRANSFORM_TYPE_NNQS,
		category: S_CATEGORY_IO,
		overview: 'Serialize RDF content fast',
		description: [
			'Scribe RDF content, i.e., serialize it, fast (and possibly ugly) using the given content-type.',
		],
		options: {
			c: {
				type: 'string',
				alias: ['content-type'],
				default: 'trig',
				describe: 'either an RDF Content-Type or format selector',
				group: 'Content Selector Options:',
			},

			...content_type_flags(s => `scribe to ${s}`),
		},
		examples: [
			`scribe -c nt`,
			`scribe -c n-triples`,
			`scribe -c 'application/n-triples'`,
			`scribe -c ttl`,
			`scribe -c turtle`,
			`scribe -c 'text/turtle'`,
			`scribe -c xml`,
		],

		command(g_argv, a_inputs, fe_command) {
			// select scriber
			let f_scriber = graphy.content(g_argv['content-type']).scribe;

			// map input(s) to writer(s)
			return a_inputs.map(ds_input => ds_input.pipe(f_scriber({
				error: e => fe_command(e),
			})));
		},
	},

	write: {
		type: S_TRANSFORM_TYPE_NNQS,
		category: S_CATEGORY_IO,
		overview: 'Serialize RDF content in style (pretty-printing)',
		description: [
			'Write RDF content, i.e., serialize it, in style (pretty-print) using the given content-type.',
		],
		options() {
			let s_group_style = 'Style Options:';
			let s_group_list = 'List Structure Options:';

			return {
				c: {
					type: 'string',
					alias: ['content-type'],
					describe: `either an RDF Content-Type or format selector (defaults to 'trig')`,
					group: 'Content Selector Options:',
				},
				i: {
					type: 'string',
					alias: ['indent'],
					// default: '\\t',  // eslint-disable-line no-undefined
					describe: `sets the whitespace string to use for indentation. Writers use '\\t' by default`,
					group: s_group_style,
				},
				g: {
					// type: 'string',
					alias: ['graph-keyword'],
					describe: `sets the style to use when serializing the optional 'GRAPH' keyword in TriG. Writers omit this keyword by default.
						Passing 'true' or empty with this flag on is shorthand for the all-caps 'GRAPH' keyword`.replace(/\n\s*/g, ' '),
					group: s_group_style,
				},
				s: {
					type: 'boolean',
					alias: ['simplify-default-graph'],
					describe: 'if enabled, omits serializing the surrounding optional graph block for the default graph in TriG.',
					group: s_group_style,
				},
				f: {
					type: 'string',
					alias: ['first'],
					describe: `c1 string: sets the predicate to use for the 'first' relation when serializing list structures`,
					group: s_group_list,
				},
				r: {
					type: 'string',
					alias: ['rest'],
					describe: `c1 string: sets the predicate to use for the 'rest' relation when serializing list structures`,
					group: s_group_list,
				},
				n: {
					type: 'string',
					alias: ['nil'],
					describe: `c1 string: sets the predicate to use for the 'nil' relation when serializing list structures`,
					group: s_group_list,
				},

				...content_type_flags(s => `write to ${s}`),
			};
		},
		examples: [
			`write -c nt`,
			`write -c n-triples`,
			`write -c 'application/n-triples'`,
			`write -c ttl`,
			`write -c turtle`,
			`write -c 'text/turtle'`,
		],

		command(g_argv, a_inputs, fe_command) {
			// default write config
			let gc_write = {};

			// extend style options
			let g_style = gc_write.style || {};

			// content-type selector
			let s_selector = g_argv['content-type'];

			// no selector specified
			if(!s_selector) {
				// default to trig
				s_selector = 'trig';

				// set simplify default graph so that it might also turtle-compatible
				g_style.simplify_default_graph = true;
			}

			// select writer
			let f_writer = graphy.content(s_selector).write;

			// style options
			{
				// indent
				if(g_argv.indent) g_style.indent = g_argv.indent;

				// graph keyword
				if(g_argv['graph-keyword']) {
					let z_graph_keyword = g_argv['graph-keyword'];
					if('boolean' === typeof z_graph_keyword) {
						g_style.graph_keyword = z_graph_keyword;
					}
					else if(/^true$/i.test(z_graph_keyword)) {
						g_style.graph_keyword = true;
					}
					else if(/^false$/i.test(z_graph_keyword)) {
						g_style.graph_keyword = false;
					}
					else if(/^graph$/i.test(z_graph_keyword)) {
						g_style.graph_keyword = z_graph_keyword;
					}
					else {
						return fe_command(`The 'write' command reported an invalid value given to the 'graph-keyword' option: '${z_graph_keyword}'`);
					}
				}

				// simplify default graph
				if(g_argv['simplify-default-graph']) g_style.simplify_default_graph = g_argv['simplify-default-graph'];
			}

			// extend list options
			let g_lists = gc_write.lists || {};
			{
				// first
				if(g_argv.first) g_lists.first = g_argv.first;

				// rest
				if(g_argv.rest) g_lists.rest = g_argv.rest;

				// nil
				if(g_argv.nil) g_lists.nil = g_argv.nil;
			}

			// map input(s) to writer(s)
			return a_inputs.map(ds_input => ds_input.pipe(f_writer({
				...gc_write,

				style: g_style,

				lists: g_lists,

				error: e => fe_command(e),
			})));
		},
	},

/*
------ Quad-Level --------
*/

	skip: {
		type: S_TRANSFORM_TYPE_NNQQ,
		category: S_CATEGORY_DATA,
		overview: 'Skip over some amount of quads in the stream(s)',
		description: [
			'Skip over some amount of data (quads by default) for each input stream before piping the remainder as usual.',
		],
		syntax: '[size=1]',
		positionals: {
			size: {
				type: 'number',
				describe: 'the number of things to skip',
			},
		},
		options: {
			q: {
				type: 'boolean',
				alias: ['quads', 't', 'triples'],
				describe: 'skip the given number of quads',
				conflicts: ['s'],
			},

			s: {
				type: 'boolean',
				alias: ['subjects'],
				describe: 'skip quads until the given number of distinct subjects have been encountered',
				conflicts: ['q'],
			},

			// m: {
			// 	type: 'number',
			// 	alias: ['multiply'],
			// },

			// d: {
			// 	type: 'number',
			// 	alias: ['divisions'],
			// 	describe: 'rather than counting numbers, use equal divisions of the given size',
			// 	example: gobble(`
			// 		Skip the first third of data:  skip -d 3
			// 	`),
			// },

			// r: {
			// 	type: 'number',
			// 	alias: ['ratio'],
			// 	describe: 'rather than counting numbers, use equal divisions of the given size',
			// 	example: gobble(`
			// 		Skip the first twothird of data:  skip -r '2/3'
			// 	`),
			// },
		},

		command(g_argv, a_inputs, fe_command) {
			// size argument
			let [
				n_skip=1,
			] = g_argv._;

			// count subjects
			if(g_argv.subjects) {
				return map_streams(a_inputs, () => {
					let c_subjects = 0;
					let kt_prev = null;

					return new stream.Transform.QuadsToOther({
						error: e => fe_command(e),

						transform(g_quad, s_encoding, fke_transform) {
							// reached length
							if(!g_quad.subject.equals(kt_prev) && ++c_subjects > n_skip) {
								// start pushing
								this.push(g_quad);
							}

							// save subject
							kt_prev = g_quad.subject;

							// done
							fke_transform();
						},
					});
				});
			}
			// count quads
			else {
				return map_streams(a_inputs, () => {
					let c_quads = 0;

					return new stream.Transform.QuadsToOther({
						error: e => fe_command(e),

						transform(g_quad, s_encoding, fke_transform) {
							// reached length
							if(++c_quads > n_skip) {
								// start pushing
								this.push(g_quad);
							}

							// done
							fke_transform();
						},
					});
				});
			}
		},
	},

	head: {
		type: S_TRANSFORM_TYPE_NNQQ,
		category: S_CATEGORY_DATA,
		overview: 'Limit number of quads from top of stream(s)',
		description: [
			'Limit the number of quads that pass through by counting from the top of the stream.',
		],
		syntax: '[size=1]',
		positionals: {
			size: {
				type: 'number',
				describe: 'the number of things to emit',
			},
		},
		options: {
			q: {
				type: 'boolean',
				alias: ['quads', 't', 'triples'],
				describe: 'emit only the given number of quads from the top of a stream',
				conflicts: ['s'],
			},

			s: {
				type: 'boolean',
				alias: ['subjects'],
				describe: 'emit quads until the given number of distinct subjects have been encountered from the top of a stream',
				conflicts: ['q'],
			},
		},

		command(g_argv, a_inputs, fe_command) {
			// size argument
			let [
				n_head=1,
			] = g_argv._;

			// count subjects
			if(g_argv.subjects) {
				return map_streams(a_inputs, (ds_input) => {
					let c_subjects = 0;
					let kt_prev = null;

					return new stream.Transform.QuadsToOther({
						error: e => fe_command(e),

						transform(g_quad, s_encoding, fke_transform) {
							// under limit
							if(g_quad.subject.equals(kt_prev) || ++c_subjects <= n_head) {
								this.push(g_quad);
							}
							// hit limit
							else {
								// push eof
								this.push(null);

								// destroy source
								ds_input.destroy();
							}

							// save subject
							kt_prev = g_quad.subject;

							// done
							fke_transform();
						},
					});
				});
			}
			// count quads
			else {
				return map_streams(a_inputs, (ds_input) => {
					let c_quads = 0;

					return new stream.Transform.QuadsToOther({
						error: e => fe_command(e),

						transform(g_quad, s_encoding, fke_transform) {
							// under limit
							if(++c_quads <= n_head) {
								this.push(g_quad);
							}
							// hit limit
							else {
								// push eof
								this.push(null);

								// destroy source
								ds_input.destroy();
							}

							// done
							fke_transform();
						},
					});
				});
			}
		},
	},

	tail: {
		type: S_TRANSFORM_TYPE_NNQQ,
		category: S_CATEGORY_DATA,
		overview: 'Limit number of quads from bottom of stream(s)',
		description: [
			'Limit the number of quads that pass through by counting from the bottom of the stream.',
			'WARNING: quads must be buffered in memory until the end of the stream is reached. Specifying a large number of quads or subjects might therefore incur lots of memory.',
		],
		syntax: '[size=1]',
		positionals: {
			size: {
				type: 'number',
				describe: 'the number of things to emit',
			},
		},
		options: {
			q: {
				type: 'boolean',
				alias: ['quads', 't', 'triples'],
				describe: 'emit only the given number of quads from the bottom of a stream',
				conflicts: ['s'],
			},

			s: {
				type: 'boolean',
				alias: ['subjects'],
				describe: 'emit quads contained by the given number of distinct subjects from the bottom of a stream',
				conflicts: ['q'],
			},
		},

		command(g_argv, a_inputs, fe_command) {
			// size argument
			let [
				n_tail=1,
			] = g_argv._;

			// count subjects
			if(g_argv.subjects) {
				return map_streams(a_inputs, () => {
					let c_subjects = 0;
					let kt_prev = null;
					let a_batch = null;
					let a_fifo = [];

					return new stream.Transform.QuadsToOther({
						error: e => fe_command(e),

						transform(g_quad, s_encoding, fke_transform) {
							// different subject
							if(!g_quad.subject.equals(kt_prev)) {
								// reset batch
								a_batch = [];

								// push batch to fifo
								a_fifo.push(a_batch);

								// hit limit
								if(++c_subjects > n_tail) {
									a_fifo.shift();
								}
							}

							// save subject
							kt_prev = g_quad.subject;

							// add quad to batch
							a_batch.push(g_quad);

							// done
							fke_transform();
						},

						flush(fk_flush) {
							// push queue
							for(let a_quads of a_fifo) {
								for(let g_quad of a_quads) {
									this.push(g_quad);
								}
							}

							// free to GC
							a_fifo.length = 0;
							a_batch.length = 0;

							// done
							fk_flush();
						},
					});
				});
			}
			// count quads
			else {
				return map_streams(a_inputs, () => {
					let c_quads = 0;
					let a_fifo = [];

					return new stream.Transform.QuadsToOther({
						error: e => fe_command(e),

						transform(g_quad, s_encoding, fke_transform) {
							// under limit
							if(++c_quads <= n_tail) {
								a_fifo.push(g_quad);
							}
							// hit limit
							else {
								// shift off bottom
								a_fifo.shift();

								// push to top
								a_fifo.push(g_quad);
							}

							// done
							fke_transform();
						},

						flush(fk_flush) {
							// push queue
							for(let g_quad of a_fifo) {
								this.push(g_quad);
							}

							// free to GC
							a_fifo.length = 0;

							// done
							fk_flush();
						},
					});
				});
			}
		},
	},

	filter: {
		type: S_TRANSFORM_TYPE_NNQQ,
		category: S_CATEGORY_DATA,
		overview: 'Filter quads in the stream(s) via expression',
		description: [
			'Filter quads by using either a Quad Filter Expression or a bit of JavaScript.',
			'For documentation on the Quad Filter Expression syntax, see:  https://graphy.link/quad-filter-expressions',
			S_WARN_JAVASCRIPT,
		],
		options: {
			x: {
				type: 'string',
				alias: ['expression'],
				describe: 'filter quads using the given quad filter expression',
				conflicts: ['j'],
			},

			j: {
				type: 'string',
				alias: ['js', 'javascript'],
				describe: 'filter quads using the given JavaScript expression which will be evaluated as a callback function passed the quad and current prefix map as arguments',
				conflicts: ['x'],
			},

			v: {
				type: 'boolean',
				alias: ['verbose'],
				describe: 'prints the compiled quad filter expression to stderr',
			},
		},

		examples: [
			[
				`Filter quads equivalent to the triple pattern: '?s rdf:type dbo:Plant'`,
				`filter -x '; a; dbo:Plant'`,
			],
			[
				`Filter quads equivalent to the SPARQL fragment: 'dbr:Banana ?p ?o. filter(!isLiteral(?o))'`,
				`filter -x 'dbr:Banana;; !{literal}'`,
			],
			[
				`Filter quads equivalent to the SPARQL fragment: '?s ?p ?o. filter(?o > 10e3)]`,
				`filter --js 'g => g.object.number > 10e3'`,
			],
			[
				`Filter quads equivalent to the SPARQL fragment: '?s ?p ?o. filter(strStarts(str(?s), str(?o)))'`,
				`filter --js 'g => g.object.value.startsWith(g.subject.value)'`,
			],
			// [
			// 	`Filter quads equivalent to the SPARQL fragment: '?s ?p ?o. filter(strStarts(str(?s), str(?o)))'`,
			// 	`filter --js '(g, h) => g.subject.concise(h).startsWith("db")'`,
			// ],
		],

		command(g_argv, a_inputs, fe_command) {
			// quad filter expression
			if(g_argv.expression) {
				let g_parse = parse_filter(g_argv.expression);

				let sj_eval = expression_handler.prepare(g_parse);

				if(g_argv.verbose) {
					console.warn(`The compiled quad filter expression from 'transform' command: () => {\n${sj_eval.replace(/^|\n/g, '\n\t')}\n}\n`);
				}

				let f_filter = new Function('factory', 'stream', sj_eval);  // eslint-disable-line no-new-func

				return map_streams(a_inputs, () => f_filter(factory, stream));
			}
			// javascript expression
			else if(g_argv.javascript) {
				let f_build = new Function('factory', /* syntax: js */ `return (${g_argv.javascript}) || null;`);  // eslint-disable-line no-new-func

				let f_filter = f_build(factory);

				// filter exists
				if(f_filter) {
					// invalid type
					if('function' !== typeof f_filter) {
						exit(`The 'filter' command expects -j/--javascript expression to evaluate to a function, instead found '${typeof f_filter}'`);
					}

					return map_streams(a_inputs, () => {
						let h_prefixes = {};

						return new stream.Transform.QuadsToOther({
							error: e => fe_command(e),

							prefix(si_prefix, p_iri) {
								h_prefixes[si_prefix] = p_iri;
							},

							transform(g_quad, s_encoding, fke_transform) {
								if(f_filter(g_quad, h_prefixes)) {
									return fke_transform(null, g_quad);
								}

								fke_transform();
							},
						});
					});
				}
			}

			// neither used (bypass filter)
			warn(`The 'filter' command was not used and is being ignored.`);
			return bypass(a_inputs);
		},
	},

	// transform
	transform: {
		type: S_TRANSFORM_TYPE_NNQQ,
		category: S_CATEGORY_DATA,
		overview: 'Apply a custom transform function to each quad in the stream(s)',
		description: [
			'$1',
			S_WARN_JAVASCRIPT,
		],
		options: {
			j: {
				type: 'string',
				alias: ['js', 'javascript'],
				describe: 'transform quads using the given JavaScript expression which will be evaluated as a callback function passed the quad and current prefix map as arguments',
				demandOption: true,
				example: [
					`transform -j 'g => [g.object, g.predicate, g.subject]'`,
					`transform -j 'g => ({
						[factory.blankNode()]: {
							a: 'rdf:Statement',
							'rdf:subject': g.subject,
							'rdf:predicate': g.predicate,
							'rdf:object': g.object,
						},
					})'`,
				].join('\n'),
			},
		},

		command(g_argv, a_inputs, fe_command) {
			// javascript expression
			if(g_argv.javascript) {
				let f_build = new Function('factory', 'c3', 'c4', /* syntax: js */ `return (${g_argv.javascript}) || null;`);  // eslint-disable-line no-new-func

				let f_transform = f_build(factory, factory.c3, factory.c4);

				// transform exists
				if(f_transform) {
					// invalid type
					if('function' !== typeof f_transform) {
						exit(`The 'filter' command expects -j/--javascript expression to evaluate to a function, instead found '${typeof f_filter}'`);
					}

					return map_streams(a_inputs, () => {
						let h_prefixes = {};

						return new stream.Transform.QuadsToOther({
							error: e => fe_command(e),

							prefix(si_prefix, p_iri) {
								h_prefixes[si_prefix] = p_iri;
							},

							transform(g_quad, s_encoding, fke_transform) {
								// alias quad property access
								g_quad.s = g_quad.subject;
								g_quad.p = g_quad.predicate;
								g_quad.o = g_quad.object;
								g_quad.g = g_quad.graph;

								// try to apply transform callback
								let z_item;
								try {
									z_item = f_transform(g_quad, h_prefixes);
								}
								catch(e_transform) {
									warn(`The 'transform' command threw an Error while applying the given callback function:\n${e_transform.stack}`);
									return fke_transform();
								}

								// item was returned
								if(z_item) {
									return interpret_item(z_item, h_prefixes, this, fke_transform);
								}

								// done
								fke_transform();
							},
						});
					});
				}
			}

			// nothing used (bypass filter)
			warn(`The 'transform' command was not used and is being ignored.`);
			return bypass(a_inputs);
		},
	},

/*
------ Stream Control --------
*/

	concat: {
		type: S_TRANSFORM_TYPE_N1AA,
		category: S_CATEGORY_STREAM,
		overview: 'Join stream data in order via concatentation',
		description: [
			'Concatenate quads from all input streams in order.',
		],
		options: {},

		command(g_argv, a_inputs, fe_command) {
			let nl_inputs = a_inputs.length;

			// single input, bypass passthrough
			if(1 === nl_inputs) return a_inputs;

			// input index
			let i_input = 0;

			// single output stream
			let ds_out = new stream.PassThrough();

			// stream consumer
			let f_next = () => {
				// done consuming inputs; end output stream
				if(i_input >= nl_inputs) return ds_out.end();

				// next input
				let ds_input = a_inputs[i_input++];

				// once it ends; consume next input
				ds_input.on('end', f_next);

				// catch stream errors
				ds_input.on('error', fe_command);

				// pipe to passthrough
				ds_input.pipe(ds_out, {end:false});
			};

			// start concatenating
			f_next();

			// return stream
			return [ds_out];

			// return [new stream.Readable({
			// 	objectMode: true,

			// 	read() {
			// 		// while there are inputs
			// 		for(; i_input<nl_inputs; i_input++) {
			// 			// ref input
// 			let ds_input = a_inputs[i_input];

			// 			// read chunk from input and push to output
			// 			let w_chunk;
			// 			while((w_chunk = ds_input.read()) && this.push(w_chunk)) {
			// 				; // eslint-disable-line no-empty
// 			}

			// 			// input not fully consumed; try again next read
			// 			if(!w_chunk.readableEnded) break;
			// 		}
			// 	},
		// })];
		},
	},

	merge: {
		type: S_TRANSFORM_TYPE_N1AA,
		category: S_CATEGORY_STREAM,
		overview: `Join stream data on a 'first come, first serve' basis`,
		description: [
			'Merge quads from all input streams without order.',
		],
		options: {},

		command(g_argv, a_inputs, fe_command) {
			let nl_inputs = a_inputs.length;

			// single input, bypass passthrough
			if(1 === nl_inputs) return a_inputs;

			// input index
			let i_input = 0;

			// single output stream
			let ds_out = new stream.PassThrough();

			// stream consumer
			let f_next = () => {
				// done consuming inputs; end output stream
				if(i_input >= nl_inputs) return ds_out.end();

				// next input
				let ds_input = a_inputs[i_input++];

				// once it ends; consume next input
				ds_input.on('end', f_next);

				// catch stream errors
				ds_input.on('error', fe_command);

				// pipe to passthrough
				ds_input.pipe(ds_out, {end:false});
			};

			// start concatenating
			f_next();

			// return stream
			return [ds_out];
		},
	},

/*
------ Dataset --------
*/

	tree: {
		type: S_TRANSFORM_TYPE_NNQQ,
		category: S_CATEGORY_SET,
		overview: 'Put all quads into a tree data structure to remove duplicates',
		description: [
			'$1',
		],
		options: {},

		command(g_argv, a_inputs, fe_command) {
			return a_inputs.map(ds_input => ds_input.pipe(dataset_tree()));
		},
	},

	canonical: {
		alias: 'canonicalize',
		type: S_TRANSFORM_TYPE_NNQQ,
		category: S_CATEGORY_SET,
		overview: 'Canonicalize a set of quads using RDF Dataset Normalization Algorithm (URDNA2015)',
		description: [
			'$1',
		],
		options: {},

		command(g_argv, a_inputs, fe_command) {
			return a_inputs.map(ds_input => ds_input.pipe(dataset_tree({canonicalize:true})));
		},
	},

	union: {
		type: S_TRANSFORM_TYPE_N1QQ,
		category: S_CATEGORY_SET,
		overview: 'Compute the set union of 1 or more inputs',
		description: [
			'$1',
		],
		options: {
			...G_OPTIONS_DATASET,
		},

		command(g_argv, a_inputs, fe_command) {
			return dataset_N1QQ(g_argv, a_inputs, fe_command, 'union');
		},
	},

	intersect: {
		alias: 'intersection',
		type: S_TRANSFORM_TYPE_N1QQ,
		category: S_CATEGORY_SET,
		overview: 'Compute the set intersection of 1 or more inputs',
		description: [
			'$1',
		],
		options: {
			...G_OPTIONS_DATASET,
		},

		command(g_argv, a_inputs, fe_command) {
			return dataset_N1QQ(g_argv, a_inputs, fe_command, 'intersection');
		},
	},

	diff: {
		alias: 'difference',
		type: S_TRANSFORM_TYPE_21QQ,
		category: S_CATEGORY_SET,
		overview: 'Compute the set difference between 2 inputs',
		description: [
			'$1',
		],
		options: {
			...G_OPTIONS_DATASET,
		},

		command(g_argv, a_inputs, fe_command) {
			return dataset_21QQ(g_argv, a_inputs, fe_command, 'difference');
		},
	},

	minus: {
		alias: ['subtract', 'subtraction'],
		type: S_TRANSFORM_TYPE_21QQ,
		category: S_CATEGORY_SET,
		overview: 'Subtract the second input from the first: A - B',
		description: [
			'$1',
		],
		options: {
			...G_OPTIONS_DATASET,
		},

		command(g_argv, a_inputs, fe_command) {
			return dataset_21QQ(g_argv, a_inputs, fe_command, 'minus');
		},
	},

	equals: {
		alias: 'equal',
		type: S_TRANSFORM_TYPE_21QRB,
		category: S_CATEGORY_SET,
		overview: 'Test if 2 inputs are equivalent',
		description: [
			'$1',
		],
		options: {
			...G_OPTIONS_DATASET,
		},

		command(g_argv, a_inputs, fe_command) {
			return dataset_21QR(g_argv, a_inputs, fe_command, 'equals');
		},
	},

	disjoint: {
		type: S_TRANSFORM_TYPE_21QRB,
		category: S_CATEGORY_SET,
		overview: 'Test if 2 inputs are completely disjoint from one another',
		description: [
			'$1',
		],
		options: {
			...G_OPTIONS_DATASET,
		},

		command(g_argv, a_inputs, fe_command) {
			return dataset_21QR(g_argv, a_inputs, fe_command, 'disjoint');
		},
	},

	contains: {
		alias: 'contain',
		type: S_TRANSFORM_TYPE_21QRB,
		category: S_CATEGORY_SET,
		overview: 'Test if the first input completely contains the second',
		description: [
			'$1',
		],
		options: {
			...G_OPTIONS_DATASET,
		},

		command(g_argv, a_inputs, fe_command) {
			return dataset_21QR(g_argv, a_inputs, fe_command, 'contains');
		},
	},


/*
------ Statistics --------
*/

	count: {
		type: S_TRANSFORM_TYPE_NNQRN,
		category: S_CATEGORY_STATS,
		overview: 'Count the number of events',
		description: [
			'Count the number of events in each steam',
		],

		command(g_argv, a_inputs, fe_command) {
			return a_inputs.map(ds_input => new Promise((fk_resolve) => {
				let c_items = 0;

				ds_input.on('data', () => {
					c_items += 1;
				});

				ds_input.on('error', fe_command);

				ds_input.on('end', () => {
					fk_resolve(new AnswerSource(c_items));
				});
			}));
		},
	},

	distinct: {
		type: S_TRANSFORM_TYPE_NNQRN,
		category: S_CATEGORY_STATS,
		overview: 'Count the number of distinct things',
		description: [
			'Count the number of distinct things, such as quads, triples, subjects, etc.',
		],
		options() {
			let h_options = {
				q: {
					alias: ['quads'],
					type: 'boolean',
					describe: 'count the number of distinct quads',
				},
				t: {
					alias: ['triples'],
					type: 'boolean',
					describe: 'count the number of distinct triples by ignoring the graph component',
				},
				s: {
					alias: ['subjects'],
					type: 'boolean',
					describe: 'count the number of distinct subjects',
				},
				p: {
					alias: ['predicates'],
					type: 'boolean',
					describe: 'count the number of distinct predicates',
				},
				o: {
					alias: ['objects'],
					type: 'boolean',
					describe: 'count the number of distinct objects',
				},
				g: {
					alias: ['graphs'],
					type: 'boolean',
					describe: 'count the number of distinct graphs',
				},
				// l: {
				// 	alias: ['literals'],
				// 	type: 'boolean',
				// 	describe: 'count the number of distinct literals',
			// },
			};

			let a_others = Object.keys(h_options);
			for(let [si_option, g_option] of Object.entries(h_options)) {
				let as_conflicts = new Set(a_others);
				as_conflicts.delete(si_option);
				g_option.conflicts = [...as_conflicts];
			}

			return h_options;
		},

		command(g_argv, a_inputs, fe_command) {
			// quad component
			let s_component = null;
			{
				if(g_argv.subjects) s_component = 'subject';
				if(g_argv.predicates) s_component = 'predicate';
				if(g_argv.objects) s_component = 'object';
				if(g_argv.graphs) s_component = 'graph';
			}

			// distinct number of a certain component
			if(s_component) {
				return a_inputs.map(ds_input => new Promise((fk_resolve) => {
					// term set
					let as_terms = new Set();

					// simply count
					ds_input.on('data', (g_quad) => {
						// concise term
						let sc1_term = g_quad[s_component].concise();

						// add to set
						as_terms.add(sc1_term);
					});

					// error handling
					ds_input.on('error', fe_command);

					// once it ends
					ds_input.on('end', () => {
						fk_resolve(new AnswerSource(as_terms.size));
					});
				}));
			}
			// distinct number of triples
			else if(g_argv.triples) {
				return a_inputs.map(async(ds_input) => {
					// remove graph component
					let ds_explode = new stream.Transform.QuadsToOther({
						transform(g_quad, s_encoding, fke_transform) {
							// push quad
							this.push(factory.quad(g_quad.subject, g_quad.predicate, g_quad.object));

							// done
							fke_transform();
						},
					});

					// create dataset
					let k_dataset = dataset_tree();

					// create pipeline
					ds_input.pipe(ds_explode)
						.pipe(k_dataset);

					// await dataset finish
					await k_dataset.until('finish');

					// return size
					return new AnswerSource(k_dataset.size);
				});
			}
			// distinct number of quads
			else {
				return a_inputs.map(async(ds_input) => {
					let k_dataset = dataset_tree();

					ds_input.pipe(k_dataset);

					await k_dataset.until('finish');

					return new AnswerSource(k_dataset.size);
				});
			}
		},
	},

	// boilerplate: {
	// 	type: S_TRANSFORM_TYPE_N1QQ,
	// 	overview: '',
	// 	description: [
	// 		'Some decsription',
	// 	],
// 	options: {},

	// 	command(g_argv, a_inputs, fe_command) {
	// 		return map_streams(a_inputs, () => new Transform({
	// 			error: e => fe_command(e),
	// 		}));
	// 	},
	// },

};

// command aliases
let h_aliases = {};

let n_width_column = Object.keys(h_commands)
	.reduce((n, s) => Math.max(n, s.length), 0);

// group command by category
let h_categories = {};
for(let [si_command, g_command] of Object.entries(h_commands)) {
	let s_category = g_command.category;

	let g_category = (h_categories[s_category] = h_categories[s_category] || {
		commands: [],
		overview: [],
	});

	g_category.commands.push(g_command);
	let s_aliases = '';
	if(g_command.alias) {
		let z_aliases = g_command.alias;
		if(Array.isArray(z_aliases)) {
			s_aliases = ` [aliases: ${z_aliases.join(', ')}]`;

			// add aliases
			for(let s_alias of z_aliases) {
				h_aliases[s_alias] = si_command;
			}

			g_command.aliases = z_aliases;
		}
		else {
			s_aliases = ` [alias: ${z_aliases}]`;

			// add alias
			h_aliases[z_aliases] = si_command;

			g_command.aliases = [z_aliases];
		}
	}
	else {
		g_command.aliases = [];
	}

	g_category.overview.push(`  ${si_command.padEnd(n_width_column, ' ')}  ${g_command.overview}${s_aliases}`);
}

// terminal width
let n_width_terminal = Math.max(80, yargs.terminalWidth()-10);

// args
let a_argv = process.argv.slice(2);
let n_args = a_argv.length;

// no arguments
if(!a_argv.length) {
	exit('no arguments given');
}

// inputs
let a_inputs = [];

// pipeline
let a_pipeline = [];
{
	let a_series = [];

	for(let i_argv=0; i_argv<n_args; i_argv++) {
		let s_arg = a_argv[i_argv];

		// after first arg
		if(i_argv) {
			// internal pipe
			if('--pipe' === s_arg) {
				a_pipeline.push(a_series);
				if(i_argv === n_args) {
					exit(`was expecting pipe destination after --pipe: ${a_argv}`);
				}
				a_series = [];
				continue;
			}
			// shorthand internal pipe
			else if('/' === s_arg) {
				a_pipeline.push(a_series);
				if(i_argv === n_args) {
					exit(`was expecting pipe destination after internal pipe character '/': ${a_argv}`);
				}
				a_series = [];
				continue;
			}
			// inputs follow
			else if('--inputs' === s_arg) {
				// convert to readable streams
				a_inputs.push(...a_argv.slice(i_argv+1).map(p => fs.createReadStream(p)));
				break;
			}
		}
		// first arg, main option
		else if('-h' === s_arg || '--help' === s_arg || 'help' === s_arg) {
			// command overview
			let s_overview = '';
			for(let [s_category, g_category] of Object.entries(h_categories)) {
				s_overview += `${s_category}\n${g_category.overview.join('\n')}\n\n`;
			}

			// eslint-disable-next-line no-console
			console.log(`\nUsage:  graphy [OPTIONS] COMMAND [ / COMMAND]* [--inputs FILES...]\n\n`
				+`Tip: Use the internal pipe operator ' / ' to string together a series of commands.\n\n`
				+s_overview
				+gobble(`
					Graphy Commands:
					  examples        Alias for '--examples'
					  help            Alias for '--help'
					  version         Alias for '--version'

					Options:
					  -e, --examples  Print some examples and exit
					  -h, --help      Print this help message and exit
					  -v, --version   Print the version info and exit

					More Options:
					  --show-stack-trace  Show the stack trace when printing error messages
				`)+'\n\n'
				+`\nRun 'graphy COMMAND --help' for more information on a command.\n`
				+`\nRun 'graphy --examples' to see some examples.\n`,
			);

			process.exit(0);
		}
		// version
		else if('-v' === s_arg || '--version' === s_arg || 'version' === s_arg) {
			// eslint-disable-next-line no-console
			console.log(require(path.join(__dirname, './package.json')).version);

			process.exit(0);
		}
		// examples
		else if('-e' === s_arg || '--examples' === s_arg || 'examples' === s_arg) {
			console.log(gobble(`
				Examples:
				  1) Count the number of distinct triples in a Turtle file:

				     graphy read -c ttl / distinct --triples   < input.ttl

				  2) Count the distinct number of subjects that are of type dbo:Place in an N-Quads file:

				     graphy read -c nq / filter -x '; a; dbo:Place' / distinct --subjects   < input.nq

				  3) Compute the difference between two RDF datasets 'a.ttl' and 'b.ttl':

				     graphy read / diff / write  --inputs a.ttl b.ttl   > diff.trig

				  4) Compute the canonicalized union of a bunch of RDF datasets in the 'data/' directory:

				     graphy read / union / write   --inputs data/*.{nt,nq,ttl,trig}   > output.trig

				  5) Extract the first 2 million quads of a Turtle file:

				     graphy read -c ttl / head 2e6 / write -c ttl  < in.ttl  > view-2M.ttl

				  6) Find all owl:sameAs triples where the object is a node and different from
				     the subject, then swap the subject and object:
				  
				     graphy read / filter -x '!$object; owl:sameAs; {node}' / transform -j  \\
				         't => [t.o, t.p, t.s]' / write -c ttl   < input.ttl   > output.ttl

			`)+'\n');

			process.exit(0);
		}
		// print stack trace in error messages
		else if('--show-stack-trace' === s_arg) {
			b_show_stack_trace = true;

			// consume arg
			continue;
		}

		a_series.push(s_arg);
	}

	// empty series
	if(a_series.length) {
		a_pipeline.push(a_series);
	}
}

// empty command list
if(!a_pipeline.length) {
	exit('no commands given');
}

(async() => {
	// failure handler
	let fe_command = (z_error) => {
		let e_command = 'string' === typeof z_error? new Error(z_error): z_error;
		debugger;
		exit(e_command.stack);
	};

	// starting inputs default to stdin if no explicit inputs given
	let a_prev = a_inputs.length? a_inputs: [process.stdin];

	// // each series in pipeline
	// for(let a_series of a_pipeline) {

	// each series in pipeline
	while(a_pipeline.length) {
		let a_series = a_pipeline.shift();

		// start with command string
		let s_command = a_series[0];

		// command not found
		if(!(s_command in h_commands)) {
			// command alias
			if(s_command in h_aliases) {
				s_command = h_aliases[s_command];
			}
			// no such command
			else {
				exit(`no such command '${s_command}'`);
			}
		}

		try {
			// ref command
			let g_command = h_commands[s_command];

			let g_options = 'function' === typeof g_command.options
				? g_command.options()
				: (g_command.options || {});

			let a_decsribes = g_command.description;
			let s_describe = '';
			if(a_decsribes.length) {
				s_describe = '\nDescription:'+a_decsribes
					.map(s => '  '+s.replace(/^\$1$/, g_command.overview+'.'))
					.join('\n\n');
			}

			let s_usage = [s_command, ...g_command.aliases]
				.reduce((s_out, s, i) => `${s_out}\n${i? 'Or': 'Usage'}:  $0${S_TRANSFORM_TYPE_NNSQ === g_command.type? '': ' [...]'} ${s}${g_command.syntax? ' '+g_command.syntax: ''} [OPTIONS] [ / COMMAND]*`, '');

			let s_positionals = '\n'+g_command.type+'\n';
			if(g_command.positionals) {
				let n_width_positionals = Object.entries(g_command.positionals)
					.reduce((n, [s]) => Math.max(n, s.length), 10);

				s_positionals += '\nArguments:';
				for(let [si_pos, g_pos] of Object.entries(g_command.positionals)) {
					s_positionals += `\n  ${si_pos.padEnd(n_width_positionals, ' ')}  [${g_pos.type}] ${g_pos.describe}`;
				}
			}

			let s_examples = '';
			if(g_command.examples && g_command.examples.length) {
				s_examples = `Examples:\n`;

				let a_egs = g_command.examples;
				for(let i_eg=0, nl_egs=a_egs.length; i_eg<nl_egs; i_eg++) {
					let z_eg = a_egs[i_eg];

					s_examples += `  ${i_eg+1}) `;
					if('string' === typeof z_eg) {
						s_examples += z_eg+'\n';
					}
					else {
						s_examples += z_eg[0]+'\n     '+z_eg[1]+'\n';
					}
				}
			}

			// build yargs
			let g_argv = mk_yargs()
				.strict()
				.usage(s_usage+'\n'+s_describe+'\n'+s_positionals)
				.options(g_options)
				.help()
				.epilog(s_examples)
				.version(false)
				.wrap(n_width_terminal)
				.parse(a_series.slice(1));

			// no inputs
			if(!a_prev.length) {
				return fe_command(`The '${s_command}' command requires at least 1 input stream but 0 were piped in.`);
			}

			// check input cardinality
			switch(g_command.type) {
				case S_TRANSFORM_TYPE_21QRB:
				case S_TRANSFORM_TYPE_21QRN:
				case S_TRANSFORM_TYPE_21QQ: {
					if(2 !== a_prev.length) {
						let nl_inputs = a_inputs.length;
						return fe_command(`The '${s_command}' command expects exactly 2 input streams but ${1 === nl_inputs? 'only 1 was': nl_inputs+' were'} piped in.`);
					}

					break;
				}

				default: {
					break;
				}
			}

			// eval command with its args
			let a_curr = await g_command.command(g_argv, a_prev, fe_command, a_pipeline);

			// advance inputs
			a_prev = await Promise.all(a_curr);
		}
		catch(e_command) {
			if(b_show_stack_trace) {
				exit(e_command.stack);
			}
			else {
				exit(e_command.message);
			}
		}
	}

	// expect single output
	if(1 !== a_prev.length) {
		exit(`expected a single output stream but last command produces ${a_prev.length} streams`);
	}

	// pipe output to stdout
	a_prev[0].pipe(process.stdout);
})();