srap/src/kernel.js

"use strict";

const Promise = require("bluebird");
const defaultValue = require("default-value");
const chalk = require("chalk");
const util = require("util");
const syncpipe = require("syncpipe");

const rateLimit = require("@promistream/rate-limit");
const simpleSink = require("@promistream/simple-sink");
const pipe = require("@promistream/pipe");
const parallelize = require("@promistream/parallelize");

const logStatus = require("./log-status");
const { validateOptions } = require("@validatem/core");
const isValidConfiguration = require("./validators/is-valid-configuration");
const createPrometheus = require("./prometheus");
const generateTaskGraph = require("./generate-task-graph");
const unreachable = require("@joepie91/unreachable")("srap");

// FIXME: *Require* a taskInterval to be set, even if explicitly null, to prevent accidentally forgetting it

module.exports = async function createKernel(_configuration) {
	let configuration = validateOptions(arguments, isValidConfiguration);

	let state = {
		... createPrometheus(),
		tasks: generateTaskGraph({
			tags: configuration.tags,
			tasks: configuration.tasks
		})
	};

	let { metrics, tasks } = state;

	const createBackend = require("./database-backends")(state);

	let attachToGlobalRateLimit = (configuration.taskInterval != null)
		? rateLimit.clonable(configuration.taskInterval)
		: undefined;

	let backend = await createBackend({
		backend: configuration.backend,
		options: configuration.database
	});

	Object.assign(state, { backend: backend });

	const createTaskKernel = require("./task-kernel")(state);

	function checkLockedTasks() {
		return Promise.try(() => {
			return backend.topLevel.countLockedTasks();
		}).then((lockedCount) => {
			if (lockedCount > 0) {
				console.log(`${chalk.bold.red("WARNING:")} There are ${lockedCount} tasks currently locked, and they will not be run! This may be caused by a process crash in the past. See the documentation for more details on how to solve this issue.`);
			}
		});
	}

	let databasePreparePromise;
	async function prepareDatabase() {
		if (databasePreparePromise == null) {
			databasePreparePromise = Promise.all([
				checkLockedTasks(),
				backend.topLevel.insertSeeds(configuration.seed)
			]);
		}

		return databasePreparePromise;
	}

	return {
		run: async function runKernel() {
			console.log(`Starting ${tasks.size} tasks...`);

			await prepareDatabase();
			
			return Promise.map(tasks.values(), (task) => {
				return pipe([
					createTaskKernel(task),
					simpleSink(({ status, item, error }) => {
						if (status === "completed") {
							metrics.successfulItems.inc(1);
							metrics.successfulItems.labels({ task: task }).inc(1);
						} else if (status === "failed") {
							metrics.failedItems.inc(1);
							metrics.failedItems.labels({ task: task }).inc(1);
						} else {
							unreachable(`Unrecognized status '${status}'`);
						}
					})
				]).read();
			});
		},
		simulate: async function simulate({ itemID, task }) {
			await prepareDatabase();
			let simulatedBackend = backend.simulate();
			
			return simulateTask(itemID, task);
		},
		execute: async function simulate({ itemID, task }) {
			await prepareDatabase();
			return executeTask(itemID, task);
		},
		shutdown: function () {
			// TODO: Properly lock all public methods after shutdown is called, and wait for any running tasks to have completed
			return backend.shutdown();
		},
		getMetrics: function () {
			return Promise.try(() => {
				return state.prometheusRegistry.metrics();
			}).then((metrics) => {
				return {
					contentType: state.prometheusRegistry.contentType,
					metrics: metrics
				};
			});
		}
	};















	function runTaskStreams() {

		

		

		return Promise.map(Object.entries(tasks), ([ task, tags ]) => {
			let taskConfiguration = configuration.tasks[task];

			if (taskConfiguration != null) {
				let taskStream = createTaskStream({
					task: task,
					tags: tags,
					taskVersion: defaultValue(taskConfiguration.version, "0"),
					taskInterval: taskConfiguration.taskInterval,
					parallelTasks: taskConfiguration.parallelTasks,
					ttl: taskConfiguration.ttl,
					run: taskConfiguration.run,
					globalRateLimiter: (attachToGlobalRateLimit != null)
						? attachToGlobalRateLimit()
						: null,
					globalParallelize: (configuration.parallelTasks != null)
						? parallelize(configuration.parallelTasks)
						: null,
					taskDependencies: dependencyMap[task],
					taskDependents: dependentMap[task]
				});

				return pipe([
					taskStream,
					simpleSink((completedItem) => {
						metrics.successfulItems.inc(1);
						metrics.successfulItems.labels({ task: task }).inc(1);
						logStatus(task, chalk.bold.green, "completed", completedItem.id);
					})
				]).read();
			} else {
				throw new Error(`Task '${task}' is defined to run for tags [${tags}], but no such task is defined`);
			}
		}).catch((error) => {
			console.dir(error, { depth: null, colors: true });
			throw error;
		});
	}
	
	function executeTask(id, task) {
		let taskConfiguration = configuration.tasks[task];

		return knex.transaction((tx) => {
			return Promise.try(() => {
				return queries.getItem(knex, id);
			}).then((item) => {
				let queue = createDatabaseQueue({
					tx,
					item,
					task,
					taskVersion: defaultValue(taskConfiguration.version, "0"),
					taskDependents: dependentMap[task],
					taskDependencies: dependencyMap[task]
				});

				return Promise.try(() => {
					return taskConfiguration.run({
						id: item.id,
						data: item.data,
						getItem: function (id) {
							return queries.getItem(knex, id);
						},
						... queue.api
					});
				}).then(() => {
					return queue.execute();
				});	
			});
		}, { doNotRejectOnRollback: false });
	}
	
	function simulateTask(id, task) {
		let taskConfiguration = configuration.tasks[task];

		let methods = [ "createItem", "renameItem", "mergeItem", "deleteItem", "createAlias", "deleteAlias", "updateData", "updateMetadata", "expire", "expireDependents" ];

		let simulatedMethods = syncpipe(methods, [
			(_) => _.map((method) => [ method, function() {
				console.log(`${chalk.bold.yellow.bgBlack(`${method} (simulated):`)} ${util.inspect(arguments, { colors: true, depth: null })}`);
			}]),
			(_) => Object.fromEntries(_)
		]);
		
		return Promise.try(() => {
			return queries.getItem(knex, id);
		}).then((item) => {
			return taskConfiguration.run({
				id: item.id,
				data: item.data,
				getItem: function (id) {
					return queries.getItem(knex, id);
				},
				... simulatedMethods
			});
		});
	}

	return {
		run: function runKernel() {
			return Promise.try(() => {
				return insertSeeds();
			}).then(() => {
				return checkLockedTasks();
			}).then(() => {
				return runTaskStreams();
			});
		},
		simulate: function simulate({ itemID, task }) {
			return Promise.try(() => {
				return insertSeeds();
			}).then(() => {
				return checkLockedTasks();
			}).then(() => {
				return simulateTask(itemID, task);
			});
		},
		execute: function simulate({ itemID, task }) {
			return Promise.try(() => {
				return insertSeeds();
			}).then(() => {
				return checkLockedTasks();
			}).then(() => {
				return executeTask(itemID, task);
			});
		},
		shutdown: function () {
			// TODO: Properly lock all public methods after shutdown is called, and wait for any running tasks to have completed
			knex.destroy();
		},
		getMetrics: function () {
			return Promise.try(() => {
				return prometheusRegistry.metrics();
			}).then((metrics) => {
				return {
					contentType: prometheusRegistry.contentType,
					metrics: metrics
				};
			});
		}
	};
};