diff --git a/src/API/API.ts b/src/API/API.ts index 903086b..2d57280 100644 --- a/src/API/API.ts +++ b/src/API/API.ts @@ -1,7 +1,8 @@ import { IBPMNServer, IEngine, IDefinition, IConfiguration, IDataStore, EXECUTION_EVENT - , IExecution, IInstanceData, IItemData, ISecureUser + , IExecution, IInstanceData, IItemData, ISecureUser, + InstanceQuery, ItemQuery, InputData, AssignmentData, MatchingQuery, FindOption } from '../'; /** @@ -85,46 +86,46 @@ export interface IAPIEngine { @param {ISecureUser} user - user object {} */ - - start(modelName, - data: {}, + + start(modelName: string, + data: InputData, user: ISecureUser, options?:IEngineOptions): Promise; /** continue with the execution of a particular item that is in a wait state, typically a user task */ - invoke(query, data: {}, user?: ISecureUser, options?:IEngineOptions): Promise; + invoke(query: ItemQuery, data: InputData, user?: ISecureUser, options?:IEngineOptions): Promise; /** provide assignment data to a user task Also, updates item data */ - assign(query, data, assignment, user?: ISecureUser, options?:IEngineOptions): Promise; + assign(query: ItemQuery, data: InputData, assignment: AssignmentData, user?: ISecureUser, options?:IEngineOptions): Promise; /** throw a message with an id, system will identify receiving item */ - throwMessage(messageId, data, messageMatchingKey, user?: ISecureUser, options?:IEngineOptions): Promise; + throwMessage(messageId: string, data: InputData, messageMatchingKey: MatchingQuery, user?: ISecureUser, options?:IEngineOptions): Promise; /** throw a signal with an id, system will identify receiving item(s) */ - throwSignal(signalId, data, messageMatchingKey, user?: ISecureUser, options?:IEngineOptions): Promise; + throwSignal(signalId: string, data: InputData, messageMatchingKey: MatchingQuery, user?: ISecureUser, options?:IEngineOptions): Promise; /** start a second event node (in a subprocess) for a running instance */ - startEvent(query, elementId, data: {}, user?: ISecureUser, options?:IEngineOptions): Promise; + startEvent(query: InstanceQuery, elementId: string, data: InputData, user?: ISecureUser, options?:IEngineOptions): Promise; /** - * + * * restarting an already completed instance at a particular node * this function requires `dataStore.enableSavePoints` to be true in configuration.ts * this add a savePoint for each item, allowing you to select that item to restore it * - * + * * @param itemQuery - Query to find a single item * @param inputData - * + * */ - restart(itemQuery, data:any,userName, options?) :Promise; + restart(itemQuery: ItemQuery, data: InputData, user: ISecureUser, options?: IEngineOptions) :Promise; /** * upgrade running instances with the latest revised bpmn model @@ -153,20 +154,20 @@ export interface IAPIData { ``` */ - getPendingUserTasks(query, user: ISecureUser): Promise; + getPendingUserTasks(query: InstanceQuery, user: ISecureUser): Promise; /** returns list of `item`s */ - findItems(query, user?: ISecureUser): Promise; - findItem(query, user?: ISecureUser): Promise; + findItems(query: InstanceQuery, user?: ISecureUser): Promise; + findItem(query: InstanceQuery, user?: ISecureUser): Promise; /** returns list of `instance`s */ - findInstances(query, user?: ISecureUser, options?): Promise; + findInstances(query: InstanceQuery, user?: ISecureUser, options?: FindOption): Promise; /** deletes the `instance`s */ - deleteInstances(query, user?: ISecureUser); + deleteInstances(query: InstanceQuery, user?: ISecureUser); } /** common parameters: diff --git a/src/API/AccessManager.ts b/src/API/AccessManager.ts index a9bf341..4d956be 100644 --- a/src/API/AccessManager.ts +++ b/src/API/AccessManager.ts @@ -6,38 +6,35 @@ enum USER_ROLE { DESIGNER = 'DESIGNER' } -var byPass = false; class SecureUser implements ISecureUser { userName; userGroups; tenantId?; modelsOwner?; - constructor(params: IUserInfo) { - Object.assign(this, params); - if (typeof process !=='undefined' && (process.env.REQUIRE_AUTHENTICATION === 'false' || process.env.ENFORCE_SECURITY === 'false')) { - console.log('****Security is disabled as requested in .env****'); - byPass = true; - } - else - console.log('Security is enabled as requested in .env'); + static isSecurityBypassed(): boolean { + return typeof process !== 'undefined' && + (process.env.REQUIRE_AUTHENTICATION === 'false' || process.env.ENFORCE_SECURITY === 'false'); + } + constructor(params: IUserInfo) { + Object.assign(this, params); } static SystemUser() { return new SecureUser({ userName: 'system', userGroups: [USER_ROLE.SYSTEM], tenantId: null, modelsOwner: null }); } isAdmin(): boolean { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; return (this.userGroups.includes(USER_ROLE.ADMIN) || this.userGroups.includes(USER_ROLE.SYSTEM)); } isSystem(): boolean { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; return (this.userGroups.includes(USER_ROLE.SYSTEM)); } inGroup(userGroup) :boolean { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; return (this.userGroups.includes(userGroup)) } @@ -47,7 +44,7 @@ class SecureUser implements ISecureUser { * @returns query */ qualifyInstances(query) { - if (byPass) + if (SecureUser.isSecurityBypassed()) return query; if (this.tenantId) query['tenantId'] = this.tenantId; @@ -74,7 +71,7 @@ class SecureUser implements ISecureUser { * @returns query */ qualifyItems(query) { - if (byPass) + if (SecureUser.isSecurityBypassed()) return query; return this.qualifyInstances(query); @@ -119,7 +116,7 @@ class SecureUser implements ISecureUser { * @returns query */ qualifyModels(query) { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; if (this.modelsOwner) query['owner'] = this.modelsOwner; diff --git a/src/API/SecureUser.ts b/src/API/SecureUser.ts index 96a472f..2dbdf54 100644 --- a/src/API/SecureUser.ts +++ b/src/API/SecureUser.ts @@ -30,41 +30,40 @@ enum USER_ROLE { * Warn the User that instance information has changed since last seen * * */ -var byPass = false; - class SecureUser implements ISecureUser { userName; userGroups; tenantId?; modelsOwner?; - constructor(params: IUserInfo) { - Object.assign(this, params); - - if (typeof process !=='undefined' && (process.env.REQUIRE_AUTHENTICATION === 'false' || process.env.ENFORCE_SECURITY === 'false')) { - console.log('****Security is disabled as requested in .env****'); - byPass = true; - } -// else -// console.log('Security is enabled as requested in .env'); -// console.log('new SecureUser', this); + /** + * Returns true when security enforcement is disabled via environment variables. + * Evaluated on every call so toggling the env var takes effect immediately + * without requiring a process restart or leaving a permanent bypass in place. + */ + static isSecurityBypassed(): boolean { + return typeof process !== 'undefined' && + (process.env.REQUIRE_AUTHENTICATION === 'false' || process.env.ENFORCE_SECURITY === 'false'); + } + constructor(params: IUserInfo) { + Object.assign(this, params); } static SystemUser() { return new SecureUser({ userName: 'system', userGroups: [USER_ROLE.SYSTEM], tenantId: null, modelsOwner: null }); } isAdmin(): boolean { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; return (this.userGroups.includes(USER_ROLE.ADMIN) || this.userGroups.includes(USER_ROLE.SYSTEM)); } isSystem(): boolean { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; return (this.userGroups.includes(USER_ROLE.SYSTEM)); } inGroup(userGroup) :boolean { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; return (this.userGroups.includes(userGroup)) } @@ -74,25 +73,13 @@ class SecureUser implements ISecureUser { * @returns query */ qualifyInstances(query) { - if (byPass) + if (SecureUser.isSecurityBypassed()) return query; if (this.tenantId) query['tenantId'] = this.tenantId; if (!this.isAdmin()) { const grpQuery = []; -/* old - grpQuery.push({ 'items.assignee': null, 'items.candidateUsers': null, 'items.candidateGroups': null }); - - grpQuery.push({ 'items.assignee': this.userName }); - grpQuery.push({ 'items.candidateUsers': this.userName, 'items.assignee': null }); - - this.userGroups.forEach(grp => { - grpQuery.push({ 'items.candidateGroups': grp, 'items.assignee': null }); - }); - - query['$or'] = grpQuery; -*/ grpQuery.push({ 'items.assignee': null, 'items.candidateUsers': null, 'items.candidateGroups': null }); grpQuery.push({ 'items.assignee': this.userName }); @@ -113,7 +100,7 @@ class SecureUser implements ISecureUser { * @returns query */ qualifyItems(query) { - if (byPass) + if (SecureUser.isSecurityBypassed()) return query; return this.qualifyInstances(query); @@ -158,7 +145,7 @@ class SecureUser implements ISecureUser { * @returns query */ qualifyModels(query) { - if (byPass) + if (SecureUser.isSecurityBypassed()) return true; if (this.modelsOwner) query['owner'] = this.modelsOwner; diff --git a/src/common/DefaultConfiguration.ts b/src/common/DefaultConfiguration.ts index 6725d36..ba7092a 100644 --- a/src/common/DefaultConfiguration.ts +++ b/src/common/DefaultConfiguration.ts @@ -52,7 +52,10 @@ export class Configuration implements IConfiguration { this.templatesPath = templatesPath; this.timers = timers; this.database = database; - this.apiKey = apiKey; + this.apiKey = apiKey || process.env.API_KEY; + if (!this.apiKey) { + console.warn('Warning: apiKey is not set. Set the API_KEY environment variable or pass apiKey in configuration.'); + } this.logger = logger; this.definitions = definitions; this.appDelegate = appDelegate; @@ -78,7 +81,7 @@ export const defaultConfiguration = new Configuration( db: 'bpmn' } }, - apiKey: '1234', + apiKey: process.env.API_KEY, logger: function (server) { new Logger(server); }, diff --git a/src/datastore/DataStore.ts b/src/datastore/DataStore.ts index ef28236..cafae12 100644 --- a/src/datastore/DataStore.ts +++ b/src/datastore/DataStore.ts @@ -12,7 +12,13 @@ import { InstanceLocker } from './'; import { QueryTranslator } from './QueryTranslator'; import { Aggregate} from './Aggregate' - +/** + * Persistence layer for BPMN process instances. + * + * Manages saving, loading, querying, and deleting instance state in MongoDB. + * Provides locking via {@link InstanceLocker} to prevent concurrent writes, + * and supports optional save-points for instance restart. + */ class DataStore extends ServerComponent implements IDataStore { dbConfiguration; @@ -63,8 +69,6 @@ class DataStore extends ServerComponent implements IDataStore { } const instanceData = recs[0]; -// this.logger.log(" instance obj found" + instanceData.id); - return { instance: instanceData, items: this.getItemsFromInstances([instanceData]) }; } /* @@ -106,7 +110,6 @@ class DataStore extends ServerComponent implements IDataStore { // save instance to DB static seq = 0; private async saveInstance(instance,options={}) { -// this.logger.log("Saving..."); let saveObject= { version: instance.version,startedAt: instance.startedAt,endedAt: instance.endedAt, status: instance.status, saved: instance.saved, @@ -254,15 +257,12 @@ class DataStore extends ServerComponent implements IDataStore { * @param query */ async findItems(query) : Promise { - // let us rebuild the query form {status: value} to > "tokens.items.status": "wait" const trans = new QueryTranslator('items'); const result = trans.translateCriteria(query); const projection = { id: 1, data: 1, name: 1, version:1, "items": 1 , "tokens":1}; var records = await this.db.find(this.dbConfiguration.db, this.dbConfiguration.Instance_collection, result, projection); - // console.log('...find items for query:', query, " translated to :", JSON.stringify(result), " recs:" , records.length) const items=this.getItemsFromInstances(records, result,trans); - // this.logger.log('...find items for ' + JSON.stringify(query) + " result :" + JSON.stringify(result) + " instances:" + records.length+ " items: "+items.length); return items; } diff --git a/src/datastore/MongoDB.ts b/src/datastore/MongoDB.ts index 1cb3051..570362a 100644 --- a/src/datastore/MongoDB.ts +++ b/src/datastore/MongoDB.ts @@ -80,9 +80,7 @@ class MongoDB { else console.log('error',err); resolve(null); - } else - { - // self.logger.log(" inserted " + result.result); + } else { console.log('index named "'+result+'" was created for collection "'+collName+'"'); resolve(result); } @@ -108,8 +106,6 @@ class MongoDB { if (err) { reject(err); } else { - // self.logger.log(" inserted " + result.result.n); - // console.log(result); resolve(result.result.n); } }); @@ -129,15 +125,16 @@ class MongoDB { return new Promise(function (resolve, reject) { self.profilerStart('>mongo.update:'+collName); - collection.updateOne(query, updateObject, options, + collection.updateOne(query, updateObject, options, function (err, result) { self.profilerEnd(); if (err) { reject(err); } else { - self.logger.log(" updated " + JSON.parse(result).n ); - resolve(JSON.parse(result).n ); + const n = result.modifiedCount ?? result.matchedCount ?? 0; + self.logger.log(" updated " + n); + resolve(n); } }); }); @@ -154,14 +151,15 @@ class MongoDB { return new Promise(function (resolve, reject) { self.profilerStart('>mongo.update:'+collName); - collection.update(query, updateObject, options, + collection.updateMany(query, updateObject, options, function (err, result) { self.profilerEnd(); if (err) { reject(err); } else { - self.logger.log(" updated " + JSON.parse(result).n); - resolve(JSON.parse(result).n); + const n = result.modifiedCount ?? result.matchedCount ?? 0; + self.logger.log(" updated " + n); + resolve(n); } }); }); @@ -189,7 +187,7 @@ class MongoDB { reject(err); } else { - self.logger.log("remove done for " + JSON.parse(result).n + " docs in " + collName); + self.logger.log("remove done for " + result.deletedCount + " docs in " + collName); resolve(result); } @@ -217,7 +215,7 @@ class MongoDB { reject(err); } else { - self.logger.log("remove done for " + id + " >" + JSON.parse(result).n ); + self.logger.log("remove done for " + id + " >" + result.deletedCount); resolve(result); } @@ -226,24 +224,24 @@ class MongoDB { } async connect() { - // Return new promise + // Return new promise const MongoClient = require('mongodb').MongoClient; const client = new MongoClient(this.dbConfig.db_url , { useUnifiedTopology: true }); - return new Promise(function (resolve, reject) { - - // Use connect method to connect to the Server - client.connect(function (err) { - // Do async job - if (err) { - reject(err); - client.close(); - } else { - resolve(client); - } + return new Promise(function (resolve, reject) { + + // Use connect method to connect to the Server + client.connect(function (err) { + // Do async job + if (err) { + reject(err); + client.close(); + } else { + resolve(client); + } + }) }) - }) } } diff --git a/src/elements/Gateway.ts b/src/elements/Gateway.ts index d5e53ef..6193277 100644 --- a/src/elements/Gateway.ts +++ b/src/elements/Gateway.ts @@ -109,13 +109,12 @@ class Gateway extends Node { // fix to bug #233 -- gateway mixes incoming flow for loops // loops need to be split by using keys // Rule: gateway inside a loop only waits of the loop flows - if (token.itemsKey!==null && item.token.itemsKey!==null) - { - if ((item.token.itemsKey+'.'+token.itemsKey).startsWith(token.itemsKey+'.')) - related.push(token); + if (token.itemsKey !== null && item.token.itemsKey !== null) { + if ((item.token.itemsKey + '.' + token.itemsKey).startsWith(token.itemsKey + '.')) + related.push(token); } - else // normal case no keys - related.push(token); + else // normal case no keys + related.push(token); } } @@ -169,11 +168,11 @@ class Gateway extends Node { if (this.type == BPMN_TYPE.ExclusiveGateway) { item.token.log('Gateway(' + item.element.name + '|' + item.element.id + ').start: cancel other pendingTokens.length=' + result.pendingTokens.length); - result.pendingTokens.forEach(async t => { + for (const t of result.pendingTokens) { item.token.log("..cancel ending token #" + t.id); t.currentItem.status = ITEM_STATUS.end; await t.terminate(); - }); + } } else { @@ -188,22 +187,17 @@ class Gateway extends Node { item.token.log('Gateway(' + item.element.name+'|'+item.element.id + ').start: let us converge now waitingTokens.length=' + result.waitingTokens.length); item.token.log('..let us converge now '); - result.waitingTokens.forEach(async t => { + for (const t of result.waitingTokens) { item.token.log('Gateway(' + item.element.name+'|'+item.element.id + ').start: ..converging ending token ' + t.id); item.token.log("..converging ending token #" + t.id); t.currentItem.status = ITEM_STATUS.end; await t.end(true); - //await t.terminate(); - }); + } + - // ------------------------------------------------------------------------------------------------- // Create a new Token at converging gateway // ------------------------------------------------------------------------------------------------- - //item.token.log('Gateway(' + item.element.name+'|'+item.element.id + ').start: Creating a new Token at converging gateway ..... '); - //await Token.startNewToken(TOKEN_TYPE.Primary,item.token.execution, item.token.currentNode, null, item.token, item, null); - //item.token.log('Gateway(' + item.element.name+'|'+item.element.id + ').start: new Token created '); - item.token.log('Gateway(' + item.element.name+'|'+item.element.id + ').start: converged! all waiting tokens ended'); const oldCurrentToken = item.token; @@ -235,15 +229,14 @@ class Gateway extends Node { } } else { // there are still pending tokens need to be cancelled or ended - result.waitingTokens.forEach(async t => { + for (const t of result.waitingTokens) { item.token.log('Gateway(' + item.element.name+'|'+item.element.id + ').start: ..converging ending token ' + t.id); item.token.log("..converging ending token #" + t.id); t.currentItem.status = ITEM_STATUS.end; await t.end(true); - //await t.terminate(); - }); - return NODE_ACTION.continue; - } + } + return NODE_ACTION.continue; + } } } @@ -294,8 +287,7 @@ class EventBasedGateway extends Gateway { this.working = true; const self = this; - endingItem.token.execution.tokens.forEach(async function (token) { - + for (const token of endingItem.token.execution.tokens.values()) { if (token.status == TOKEN_STATUS.wait && token.currentItem.id != endingItem.id) { if (token.originItem && token.originItem.node.id == self.id) { @@ -305,7 +297,7 @@ class EventBasedGateway extends Gateway { await token.terminate(); } } - }); + } this.working = false; } } diff --git a/src/elements/Node.ts b/src/elements/Node.ts index 1c61f30..95967c5 100644 --- a/src/elements/Node.ts +++ b/src/elements/Node.ts @@ -9,6 +9,12 @@ import { ScriptHandler } from '../'; // --------------------------------------------- +/** + * Base class for all BPMN nodes (tasks, events, gateways, subprocesses). + * + * Manages the node lifecycle (enter, start, run, continue, end), + * boundary event attachments, outbound flow evaluation, and script execution. + */ class Node extends Element { name=''; process; @@ -24,11 +30,18 @@ class Node extends Element { candidateGroups; candidateUsers; scripts = new Map(); + /** Returns the id of the owning process. */ get processId() : any { return this.process.id; - } + } + /** + * @param id BPMN element id + * @param process the owning Process + * @param type BPMN type string (e.g. 'bpmn:UserTask') + * @param def the raw BPMN definition object + */ constructor(id, process , type, def) { super(); this.id = id; @@ -43,6 +56,7 @@ class Node extends Element { BehaviourLoader.load(this); } + /** Fires the node_validate event and throws if validation returns an error. */ async validate(item: Item) { let validate = await item.node.doEvent(item, EXECUTION_EVENT.node_validate,item.status); @@ -56,12 +70,20 @@ class Node extends Element { }); } + /** + * Executes scripts registered for the given event, updates item status, + * and emits the event through the execution listener. + * + * @param item the current item + * @param event the execution event type + * @param newStatus if provided, sets the item status before running scripts + * @param eventDetails additional event details passed to listeners + */ async doEvent(item: Item, event: EXECUTION_EVENT, newStatus: ITEM_STATUS=null,eventDetails={}) { item.token.log('Node('+this.name+'|'+this.id+').doEvent: executing script for event:' + event + ' newStatus:'+newStatus); if (newStatus) item.status = newStatus; - ///item.token.log('..>' + event + ' ' + this.id); const scripts = this.scripts.get(event); const rets = []; if (scripts) { @@ -95,14 +117,13 @@ class Node extends Element { */ async setInput(item: Item, input) { item.token.log('Node('+this.name+'|'+this.id+').setInput: input' + JSON.stringify(input)); - // - //item.token.log('--setting input ' + JSON.stringify(input)); const data = await this.getInput(item, input); item.token.appendData(data,item); } + /** Stores raw input and fires the transform_input event, returning the (possibly transformed) input. */ async getInput(item: Item, input) { item.token.log('Node('+this.name+'|'+this.id+').getInput: input' + JSON.stringify(input)); @@ -121,6 +142,7 @@ class Node extends Element { return item.output; } + /** Called when the token first visits this node; records the start time. */ enter(item: Item) { item.token.log('Node('+this.name+'|'+this.id+').enter: item=' + item.id); item.startedAt = new Date(); @@ -130,18 +152,18 @@ class Node extends Element { * does the need require to go into wait * tasks like UserTasks, receiveTask, timer */ + /** Whether this node requires the token to wait (overridden by UserTask, ReceiveTask, etc.). */ get requiresWait() { return false; } - /* - * Can the Node be invoked externally (not from the workflow) - * tasks like UserTasks, receiveTask, or events like receive,signal can be invoked - */ + /** Whether this node can be invoked externally (overridden by UserTask, events, etc.). */ get canBeInvoked() { return false; } + /** Returns the loop characteristics behaviour if defined, or undefined. */ get loopDefinition() { return this.getBehaviour(Behaviour_names.LoopCharacteristics); } - get isCatching(): boolean { return false; } // catching events and tasks + /** Whether this is a catching event or task (overridden by CatchEvent, ReceiveTask). */ + get isCatching(): boolean { return false; } /** * this is the primary exectuion method for a node * @@ -212,10 +234,7 @@ class Node extends Element { // -------- // Save before performing the work await item.token.execution.save(); - //item.token.info(`{"seq":${item.seq},"type":'${this.type}',"id":'${this.id}',"action":'Started'}`); - item.token.log('Node('+this.name+'|'+this.id+').execute: execute run ...token:'+item.token.id); - //item.token.log('..>run ' + this.id); ret = await this.run(item); switch (ret) { @@ -238,14 +257,13 @@ class Node extends Element { return ret2; } - /* - * called by execute or by token.invoke to continue work already started - */ + /** Called after run() to finalize the node and trigger the end phase. */ async continue(item: Item) { item.token.log('Node('+this.name+'|'+this.id+').continue: item=' + item.id); await this.end(item); return; } + /** Starts boundary events and returns wait if the node requires it. Override for custom start logic. */ async start(item: Item): Promise { item.token.log('Node('+this.name+'|'+this.id+').start: item=' + item.id); @@ -256,10 +274,12 @@ class Node extends Element { return NODE_ACTION.continue; } + /** Performs the node's work. Override in subclasses (ScriptTask, ServiceTask, etc.). */ async run(item: Item): Promise { item.token.log('Node('+this.name+'|'+this.id+').run: item=' + item.id); return NODE_ACTION.end; } + /** Cancels sibling branches of an EventBasedGateway when one branch completes. */ async cancelEBG(item) { const ebgItem=item.token.originItem; if (ebgItem && ebgItem.node.type===BPMN_TYPE.EventBasedGateway) @@ -268,6 +288,7 @@ class Node extends Element { await ebg.cancelAllBranched(item); } } + /** Terminates all boundary event tokens attached to this node. */ async cancelBoundaryEvents(item) { // cancel boundary events let i,t; @@ -306,6 +327,7 @@ class Node extends Element { } } } + /** Returns the current items of all boundary event tokens attached to this node. */ getBoundaryEventItems(item) { let i,t; @@ -344,6 +366,13 @@ class Node extends Element { return boundaryItems; } + /** + * Ends the node: fires end event, runs exit behaviours, cancels boundary events, + * processes message flows, and sets the item status and endedAt timestamp. + * + * @param item the item to end + * @param cancel if true, marks as cancelled (no endedAt timestamp) + */ async end(item: Item,cancel:Boolean=false) { if (!item || item.status==ITEM_STATUS.end) return; @@ -358,10 +387,10 @@ class Node extends Element { /** * Rule: boundary events are canceled when owner task status is 'end' * */ - this.behaviours.forEach(async function (b) { await b.end(item); }); + for (const b of this.behaviours.values()) { await b.end(item); } await this.doEvent(item, EXECUTION_EVENT.node_end, ITEM_STATUS.end, {cancel}); item.token.log('Node('+this.name+'|'+this.id+').end: setting item status to end itemId=' + item.id + ' itemStatus=' + item.status + ' cancel: '+cancel+' endedat '+item.endedAt); - this.behaviours.forEach(async function (b) { await b.exit(item); }); + for (const b of this.behaviours.values()) { await b.exit(item); } item.token.log('Node(' + this.name + '|' + this.id + ').end: finished'); let result=await this.cancelBoundaryEvents(item); @@ -398,8 +427,7 @@ class Node extends Element { init(item: Item) { } - /* to be overwritten by XOR gateway */ - + /** Evaluates outbound flows and returns items for flows that should be taken. Overridden by XOR gateway. */ async getOutbounds(item: Item): Promise { item.token.log('Node('+this.name+'|'+this.id+').getOutbounds: itemId='+item.id); const outbounds = []; @@ -416,23 +444,11 @@ class Node extends Element { flowItem.token=null; } - }/* - this.outbounds.forEach(async flow => { - if (flow.type == BPMN_TYPE.MessageFlow) { - - } - else { - let flowItem = new Item(flow, item.token); - if (await flow.run(flowItem) == FLOW_ACTION.take) - outbounds.push(flowItem); - else - flowItem.token=null; - } - }); */ - //item.token.log('..return outbounds' + outbounds.length); + } item.token.log('Node('+this.name+'|'+this.id+').getOutbounds: return outbounds'+outbounds.length); return outbounds; } + /** Creates tokens for all non-compensate boundary events attached to this node. */ async startBoundaryEvents(item,token) { let i; // check for attachments - boundary events: @@ -444,6 +460,7 @@ class Node extends Element { } + /** Returns a description array of this node's configuration (initiator, assignee, scripts). */ describe() { var desc = []; if (this.initiator) diff --git a/src/engine/Execution.ts b/src/engine/Execution.ts index 4d90bdd..fb57cb8 100644 --- a/src/engine/Execution.ts +++ b/src/engine/Execution.ts @@ -9,10 +9,12 @@ import { ServerComponent } from '../server'; import { InstanceObject } from './Model'; /** - * is accessed two ways: - * execute - start process - * signal - invoke a node (userTask, event, etc.) - * */ + * Manages the runtime state of a single BPMN process instance. + * + * Accessed two ways: + * - `execute` — start a new process from a start node + * - `signal` — invoke a waiting node (userTask, event, etc.) + */ // --------------------------------------------- class Execution extends ServerComponent implements IExecution { @@ -32,19 +34,26 @@ class Execution extends ServerComponent implements IExecution { operation; svg:string; + /** Instance unique identifier. */ get id() { return this.instance.id; } + /** Process model name. */ get name() { return this.instance.name; } + /** Current execution status (running, end, etc.). */ get status():EXECUTION_STATUS { return this.instance.status;} + /** Backward-compatible self-reference. */ get execution() { return this;} // backward compatible // TODO: this is not updated action = NODE_ACTION.stop; + /** + * Waits for any background worker to finish and returns this execution. + */ async tillDone() { const res = await this.worker; return this; } - - // end move from ExecutionContext; + + /** Returns the server's event listener for emitting execution events. */ get listener() { return this.server.listener; } @@ -69,12 +78,18 @@ class Execution extends ServerComponent implements IExecution { this.definition = new Definition(name, source, this.server); } + /** Retrieves a node from the process definition by its BPMN id. */ public getNodeById(id) { return this.definition.getNodeById(id); } + /** Retrieves a token by its numeric id from the execution's token map. */ public getToken(id: number): Token { return this.tokens.get(id); } + /** + * Checks whether all tokens have ended; if so, ends the execution. + * EventSubProcess tokens are excluded from the active count. + */ private async checkEnd() { let active = 0; this.tokens.forEach(t => { @@ -88,6 +103,7 @@ class Execution extends ServerComponent implements IExecution { this.end(); } } + /** Marks the execution as ended, notifies the process, and fires end events. */ async end() { this.log(".execution ended."); this.instance.endedAt = new Date(); @@ -119,6 +135,17 @@ class Execution extends ServerComponent implements IExecution { }); } + /** + * Starts execution of the process from the given start node. + * + * Loads the definition, creates the primary token, and runs until + * all tokens reach a wait or end state, then saves. + * + * @param startNodeId start node id (null for default start) + * @param inputData initial process data + * @param options execution options + * @param userName user initiating the execution + */ public async execute(startNodeId = null, inputData = {}, options = {},userName=null) { this.log('^ACTION:execute:'); @@ -131,62 +158,73 @@ class Execution extends ServerComponent implements IExecution { inputData})); this.operation='execute'; this.options=options; - await this.definition.load(); - this.servicesProvider = await this.appDelegate.getServicesProvider(); - - this.instance.status = EXECUTION_STATUS.running; - this.appDelegate.executionStarted(this); - - if (inputData) - this.instance.data = Object.assign({}, inputData); - else - this.instance.data = {}; + try { + await this.definition.load(); + this.servicesProvider = await this.appDelegate.getServicesProvider(); - this.instance.startedAt = new Date(); + this.instance.status = EXECUTION_STATUS.running; + this.appDelegate.executionStarted(this); - let startNode; - if (!startNodeId) - startNode = this.definition.getStartNode(); - else - startNode = this.getNodeById(startNodeId); - if (!startNode) { - this.error("No Start Node"); - return; + if (inputData) + this.instance.data = Object.assign({}, inputData); + else + this.instance.data = {}; - } + this.instance.startedAt = new Date(); - this.process = startNode.process; - //await this.doExecutionEvent(this, EXECUTION_EVENT.process_loaded); + let startNode; + if (!startNodeId) + startNode = this.definition.getStartNode(); + else + startNode = this.getNodeById(startNodeId); + if (!startNode) { + this.error("No Start Node"); + return; + } - await this.doExecutionEvent(this, EXECUTION_EVENT.process_start); + this.process = startNode.process; + await this.doExecutionEvent(this, EXECUTION_EVENT.process_start); - this.log('..starting at :' + startNode.id); - let token = await Token.startNewToken(TOKEN_TYPE.Primary,this, startNode, null, null, null, null,inputData,true); + this.log('..starting at :' + startNode.id); + let token = await Token.startNewToken(TOKEN_TYPE.Primary,this, startNode, null, null, null, null,inputData,true); - // start all event sub processes for the process + // start all event sub processes for the process - const proc = startNode.process; - await proc.start(this, token); + const proc = startNode.process; + await proc.start(this, token); - await token.execute(inputData); + await token.execute(inputData); - await this.checkEnd(); - - await Promise.all(this.promises); - this.log('.execute returned'); - await this.doExecutionEvent(this.process, EXECUTION_EVENT.process_wait); + await this.checkEnd(); - this.report(); - await Promise.all(this.promises); - await this.save(); + await Promise.all(this.promises); + this.log('.execute returned'); + await this.doExecutionEvent(this.process, EXECUTION_EVENT.process_wait); + this.report(); + await Promise.all(this.promises); + await this.save(); + } + catch(exc) { + this.instance.status = EXECUTION_STATUS.error; + this.logger.error('execute failed: ' + exc.message); + await this.save(); + throw exc; + } } /** - * @param executionId - * @param inputData - * + * Assigns or updates a waiting item without completing it. + * + * Merges data and assignment fields (e.g. assignee, candidateGroups) + * into the item, fires the node_assign event, and validates. + * + * @param executionId the item id to assign + * @param inputData data to merge into the item + * @param assignment assignment fields to set on the item + * @param userName user performing the assignment + * @param options execution options */ public async assign(executionId, inputData: any, assignment = {}, userName,options={}) { @@ -215,79 +253,97 @@ class Execution extends ServerComponent implements IExecution { this.log('Execution('+this.name+').assign: finished!'); } /** - * - * invoke scenarios: - * itemId - * elementId - but only one is active - * elementId - for a startEvent in a secondary process - * - * @param executionId - * @param inputData - * + * Phase 1 of invoking a waiting item: sets up execution context, passes + * input data to the token, and completes execution (unless noWait is set). + * + * When `options.noWait` is true, returns immediately after `token.signal()` + * — the caller must follow up with `signalItemContinue()` to finish. + * + * @param itemId the item id to signal + * @param inputData data to pass to the item + * @param userName user performing the invocation + * @param options execution options; `noWait: true` for async two-phase */ public async signalItem(itemId, inputData:any,userName, options={}) :Promise { this.log('Execution('+this.name+').signalItem: executionId=' + itemId + ' data '+JSON.stringify(inputData)); this.operation = 'signal'; this.options=options; this.userName=userName; - let token = null; - this.servicesProvider = await this.appDelegate.getServicesProvider(); + try { + let token = null; - // get process - let firstToken=this.tokens.get(0); - this.process=firstToken.path[0].node.process; - // - this.appDelegate.executionStarted(this); - await this.doExecutionEvent(this.process,EXECUTION_EVENT.process_invoke); + this.servicesProvider = await this.appDelegate.getServicesProvider(); - this.tokens.forEach(t => { - if (t.currentItem && t.currentItem.id == itemId /*&& t.status=='wait' */) { - this.item = t.currentItem; - token = t; - } - }); - this.info(JSON.stringify({type:'execution', label:'Invoke Item', - action:'signalInput', - id:token.currentNodeId, - itemId, - userName, - options, - inputData})); + // get process + let firstToken=this.tokens.get(0); + this.process=firstToken.path[0].node.process; + // + this.appDelegate.executionStarted(this); + await this.doExecutionEvent(this.process,EXECUTION_EVENT.process_invoke); - if (token) { - this.log('Execution('+this.name+').signal: .. launching a token signal'); + this.tokens.forEach(t => { + if (t.currentItem && t.currentItem.id == itemId /*&& t.status=='wait' */) { + this.item = t.currentItem; + token = t; + } + }); + this.info(JSON.stringify({type:'execution', label:'Invoke Item', + action:'signalInput', + id:token.currentNodeId, + itemId, + userName, + options, + inputData})); - let result=await token.signal(inputData,options); + if (token) { + this.log('Execution('+this.name+').signal: .. launching a token signal'); - if (options['noWait']==true) - return this; - - this.log('Execution('+this.name+').signalItem: .. signal token is done'); - } - - this.log('Execution('+this.name+').signalItem: returning .. waiting for promises status:' + this.instance.status + " id: " + itemId); - - await this.checkEnd(); + let result=await token.signal(inputData,options); - await Promise.all(this.promises); + if (options['noWait']==true) + return this; + this.log('Execution('+this.name+').signalItem: .. signal token is done'); + } - await this.doExecutionEvent(this.process,EXECUTION_EVENT.process_invoked); + this.log('Execution('+this.name+').signalItem: returning .. waiting for promises status:' + this.instance.status + " id: " + itemId); - this.log('Execution('+this.name+').signalItem: returned process status:' + this.instance.status + " id: " + itemId); + await this.checkEnd(); - this.report(); + await Promise.all(this.promises); - await Promise.all(this.promises); - await this.save(); - this.log('Execution('+this.name+').signalItem: finished!'); - return this; + await this.doExecutionEvent(this.process,EXECUTION_EVENT.process_invoked); + + this.log('Execution('+this.name+').signalItem: returned process status:' + this.instance.status + " id: " + itemId); + + this.report(); + + await Promise.all(this.promises); + await this.save(); + this.log('Execution('+this.name+').signalItem: finished!'); + + return this; + } + catch(exc) { + this.instance.status = EXECUTION_STATUS.error; + this.logger.error('signalItem failed: ' + exc.message); + await this.save(); + throw exc; + } } - public async signalItem2(itemId) :Promise { - this.log('Execution('+this.name+').signalItem2: executionId=' + itemId); - this.operation = 'signal2'; + /** + * Phase 2 of a noWait invocation: completes execution after signalItem returned early. + * + * Calls `token.signalContinue()` (the deferred continuation), then checks for end, + * resolves promises, fires process_invoked event, and saves state. + * + * @param itemId the item id that was signaled in phase 1 + */ + public async signalItemContinue(itemId) :Promise { + this.log('Execution('+this.name+').signalItemContinue: executionId=' + itemId); + this.operation = 'signalContinue'; let token = null; this.tokens.forEach(t => { if (t.currentItem && t.currentItem.id == itemId /*&& t.status=='wait' */) { @@ -297,11 +353,11 @@ class Execution extends ServerComponent implements IExecution { }); if (token) { - this.log('Execution('+this.name+').signal2: .. launching a token signal'); + this.log('Execution('+this.name+').signalContinue: .. launching a token signal'); - let result=await token.signal2(); + let result=await token.signalContinue(); - this.log('Execution('+this.name+').signalItem2: .. signal token is done'); + this.log('Execution('+this.name+').signalItemContinue: .. signal token is done'); } this.log('Execution('+this.name+').signalItem: returning .. waiting for promises status:' + this.instance.status + " id: " + itemId); @@ -339,24 +395,42 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { this.log('Execution('+this.name+').signal: executionId=' + executionId + ' data '+JSON.stringify(inputData)); @@ -368,32 +442,33 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { - if (t.currentNode && t.currentNode.id == executionId /*&& t.status=='wait' */) - token = t; - }); - } - if (token) { - this.log('Execution('+this.name+').signal: .. launching a token signal'); - let result=await token.signal(inputData,options); - this.log('Execution('+this.name+').signal: .. signal token is done'); - } - - else + if (!token) { + this.tokens.forEach(t => { + if (t.currentNode && t.currentNode.id == executionId /*&& t.status=='wait' */) + token = t; + }); + } + + if (token) { + this.log('Execution('+this.name+').signal: .. launching a token signal'); + let result=await token.signal(inputData,options); + this.log('Execution('+this.name+').signal: .. signal token is done'); + } + else { - // checking for node - let node; - // check for startEvent of a secondary process + // checking for node + let node; + // check for startEvent of a secondary process const startedNodeId = this.tokens.get(0).path[0].elementId; let restart=false; if (options['restart'] && options['restart']==true ) @@ -411,13 +486,10 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { if (i.id==executionId) { @@ -437,18 +509,35 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { this.log('Execution('+this.name+').signalRepeatTimer: executionId=' + executionId + ' data '+JSON.stringify(inputData)); this.operation = 'signalRepeatTime'; @@ -480,9 +569,8 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { @@ -507,16 +596,16 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { items.push(item.save()); }); return items; } - /* - * return the execution State as a Json object - * to be saved for retrieval later and used in restore + /** + * Returns the full execution state as a serializable object. + * Includes instance data, tokens, items, loops, source, and SVG. */ - getState() : IInstanceData { const tokens = []; const loopsMap = new Map(); @@ -541,6 +630,7 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { @@ -650,11 +738,13 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { t.resume();}); } + /** Recursively logs a token and its children for debugging. */ reportToken(token:Token,level) { const branch = token.originItem ? token.originItem.elementId : 'root'; @@ -673,6 +763,7 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise '400fa120-5e9f-411e-94bd-2a23f6695704' + return uuid(); } - + /** Emits an execution-level event to the listener. */ async doExecutionEvent(process,event,eventDetails={}) { - //this.item = null; await this.listener.emit(event, { event, context: this ,eventDetails }); await this.listener.emit('all', { event, context: this ,eventDetails}); } + /** Emits an item-level event and sets the current item reference. */ async doItemEvent(item, event,eventDetails={}) { this.item = item; await this.listener.emit(event, { event, context: this ,eventDetails }); @@ -743,20 +837,23 @@ public async restart(itemId, inputData:any,userName, options={}) :Promise { while (true) { diff --git a/src/engine/ScriptHandler.ts b/src/engine/ScriptHandler.ts index ae66d88..05a6894 100644 --- a/src/engine/ScriptHandler.ts +++ b/src/engine/ScriptHandler.ts @@ -1,8 +1,16 @@ import { Item ,Execution,Token} from "."; import { IScriptHandler } from "../interfaces"; import { spawn } from 'child_process'; +import * as vm from 'vm'; +/** + * Handles JavaScript and Python script execution within BPMN processes. + * + * Provides expression evaluation (with $ prefix), input expression parsing + * (strings, CSV arrays, dates), and full script execution with scope-appropriate + * variable bindings. + */ class ScriptHandler implements IScriptHandler{ @@ -45,7 +53,6 @@ class ScriptHandler implements IScriptHandler{ val = new Date(val); return val; - //console.log('----setAttVal', attr, exp, val); } // old name :scopeEval(scope, script) { /** @@ -58,8 +65,6 @@ class ScriptHandler implements IScriptHandler{ async evaluateExpression(scope: Item|Token, expression) { let script=expression; - let result; - let ret; if (!expression) return; @@ -67,77 +72,63 @@ class ScriptHandler implements IScriptHandler{ script=expression.substring(1); try { - var js = ScriptHandler.getJSvars(scope) + ` - return (${script});`; - - result = await Function(js).bind(scope)(); - - - if (result instanceof Promise) - { - ret = await result; - } - else - ret =result; + const sandbox = ScriptHandler.buildSandbox(scope); + const context = vm.createContext(sandbox); + let result = vm.runInContext(script, context, { + timeout: ScriptHandler.SCRIPT_TIMEOUT_MS, + }); + + if (result instanceof Promise) { + result = await result; + } + return result; } catch (exc) { - console.log('error in script evaluation', js); + console.log('error in script evaluation', script); console.log(exc); throw new Error(exc); } - return ret; } - // used to be called scopeJS + /** + * Executes a full JavaScript (or Python with $py prefix) script in the given scope. + * + * @param scope the Item or Execution to bind as `this` + * @param script the script body to execute + */ async executeScript(scope: Item|Execution, script) { - let result; - let ret; - - /* old - try { - let result; - var js = this.getJSvars(scope) + ` - ${script};`; - - const func=new AsyncFunction(js); - result = await func.bind(scope)(); - } - catch (exc) { - console.log('error in script execution', js); - console.log(exc); - } - - return result; - */ try { if (script.startsWith('$py')) { - result=await this.runPython(scope,script.substring(3)); - console.log('python result:',result) + const result = await this.runPython(scope, script.substring(3)); return result; } - script = script.replace('#','') //remove symbol '#' - //require return - var js = ScriptHandler.getJSvars(scope) + ` - ${script};`; - result = await Function(js).bind(scope)(); - - if (result instanceof Promise) - { - ret = await result; - //console.log(result,ret); - } - else - ret =result; + script = script.replace('#','') //remove symbol '#' + + const sandbox = ScriptHandler.buildSandbox(scope); + const context = vm.createContext(sandbox); + let result = vm.runInContext(script, context, { + timeout: ScriptHandler.SCRIPT_TIMEOUT_MS, + }); + + if (result instanceof Promise) { + result = await result; + } + return result; } catch (exc) { - console.log('error in script execution', js); + console.log('error in script execution', script); console.log(exc); - throw new Error(exc); - } - - return ret; + throw new Error(exc); + } } + /** + * Generates JavaScript variable declarations appropriate for the scope type. + * + * - Token scope: exposes data, instance, input, output, appDelegate, appServices, item + * - Execution scope: exposes appDelegate, instance, appServices + * - Item scope: exposes item, data, instance (via token.execution), appDelegate, appServices + */ static getJSvars(scope) { let isToken = scope.hasOwnProperty('startNodeId'); let isExecution = scope.hasOwnProperty('tokens'); @@ -178,57 +169,127 @@ class ScriptHandler implements IScriptHandler{ } } - - async runPython(item,code: string, input: any={}): Promise { - return new Promise((resolve, reject) => { - - const pythonCmd = process.env.PYTHON_CMD||'python'; - item.data['a']=3; - item.data['b']=4; - const pyCode=` + /** + * Builds a restricted VM context from the scope, exposing only the + * variables that BPMN scripts need — no require, process, or filesystem. + */ + static buildSandbox(scope): Record { + let isToken = scope.hasOwnProperty('startNodeId'); + let isExecution = scope.hasOwnProperty('tokens'); + + const sandbox: Record = { + console: { log: console.log, warn: console.warn, error: console.error }, + JSON, + Date, + Math, + parseInt, + parseFloat, + isNaN, + isFinite, + Promise, + }; + + if (isToken) { + sandbox.data = scope.data; + sandbox.instance = scope.execution?.instance; + sandbox.input = scope.input; + sandbox.output = scope.output; + sandbox.appDelegate = scope.execution?.appDelegate; + sandbox.appServices = scope.execution?.servicesProvider; + sandbox.appUtils = scope.execution?.appDelegate?.appUtils; + sandbox.item = scope; + } else if (isExecution) { + sandbox.appDelegate = scope.appDelegate; + sandbox.instance = scope.instance; + sandbox.appServices = scope.servicesProvider; + sandbox.appUtils = scope.appDelegate?.appUtils; + } else { + sandbox.item = scope; + sandbox.data = scope.data; + sandbox.instance = scope.token?.execution?.instance; + sandbox.input = scope.input; + sandbox.output = scope.output; + sandbox.appDelegate = scope.token?.execution?.appDelegate; + sandbox.appServices = scope.token?.execution?.servicesProvider; + sandbox.appUtils = scope.token?.execution?.appDelegate?.appUtils; + } + + return sandbox; + } + + static readonly SCRIPT_TIMEOUT_MS = parseInt(process.env.SCRIPT_TIMEOUT_MS || '5000', 10); + + /** + * Spawns a Python subprocess to execute the given code. + * Passes item data as JSON to the script and parses JSON output. + * + * @param item the current item (provides data context) + * @param code Python code to execute + * @param input optional input data sent via stdin + */ + static readonly PYTHON_TIMEOUT_MS = parseInt(process.env.PYTHON_TIMEOUT_MS || '10000', 10); + + async runPython(item, code: string, input: any = {}): Promise { + return new Promise((resolve, reject) => { + + const pythonCmd = process.env.PYTHON_CMD || 'python'; + + // Safe bootstrap: data and item metadata are passed via stdin as JSON, + // not interpolated into the Python source code. + const pyCode = ` import sys, json -input = json.loads(sys.stdin.read()) -data =${JSON.stringify(item.data)} -item =${JSON.stringify({id:item.id,name:item.name,elementId:item.elementId})} -result = data["a"] + data["b"] -print(json.dumps({ "sum": result }), flush=True) +_stdin = json.loads(sys.stdin.read()) +data = _stdin.get("data", {}) +item = _stdin.get("item", {}) +input = _stdin.get("input", {}) ${code.trim()}`; - const python = spawn(pythonCmd, ['-c', pyCode]); - - let output = ''; - let error = ''; - - // Handle stdout - python.stdout.on('data', (data) => { - output += data.toString(); - }); - - // Handle stderr - python.stderr.on('data', (data) => { - error += data.toString(); - }); - - // Handle close - python.on('close', (code) => { - if (code !== 0 || error) { - reject(new Error(`Python error: ${error || 'Exit code ' + code}`)); - } else { - try { - const parsed = JSON.parse(output); - resolve(parsed); - } catch (e) { - resolve(output.trim()); // fallback: plain string - } - } - }); - // Send input if provided - if (input !== undefined) { - python.stdin.write(JSON.stringify(input)); - python.stdin.end(); + const python = spawn(pythonCmd, ['-c', pyCode]); + + let output = ''; + let error = ''; + let killed = false; + + // Kill the process if it exceeds the timeout + const timer = setTimeout(() => { + killed = true; + python.kill('SIGKILL'); + }, ScriptHandler.PYTHON_TIMEOUT_MS); + + python.stdout.on('data', (chunk) => { + output += chunk.toString(); + }); + + python.stderr.on('data', (chunk) => { + error += chunk.toString(); + }); + + python.on('close', (exitCode) => { + clearTimeout(timer); + if (killed) { + reject(new Error(`Python script timed out after ${ScriptHandler.PYTHON_TIMEOUT_MS}ms`)); + } else if (exitCode !== 0 || error) { + reject(new Error(`Python error: ${error || 'Exit code ' + exitCode}`)); + } else { + try { + const parsed = JSON.parse(output); + resolve(parsed); + } catch (e) { + resolve(output.trim()); + } + } + }); + + // Pass all data safely via stdin as JSON + const stdinPayload = { + data: item.data || {}, + item: { id: item.id, name: item.name, elementId: item.elementId }, + input: input, + }; + python.stdin.write(JSON.stringify(stdinPayload)); + python.stdin.end(); + }); } - }); -} } export {ScriptHandler} diff --git a/src/engine/Token.ts b/src/engine/Token.ts index e272145..757d301 100644 --- a/src/engine/Token.ts +++ b/src/engine/Token.ts @@ -49,6 +49,12 @@ enum TOKEN_TYPE { EventSubProcess='EventSubProces', BoundaryEvent ='BoundaryEvent' ,AdHoc ='AdHoc' } // --------------------------------------------- +/** + * Represents a pointer moving through the BPMN flow graph. + * + * Each token tracks its current node, the path of items visited, and its + * relationship to parent/child tokens (for subprocesses, gateways, loops). + */ class Token implements IToken { id; type: TOKEN_TYPE; @@ -62,6 +68,7 @@ class Token implements IToken { path: Item[]; // keep track of all nodes and flow taken loop: Loop; _currentNode: Node; + /** The node the token is currently positioned at. */ get currentNode():Node {return this._currentNode} processId; status: TOKEN_STATUS; @@ -70,16 +77,20 @@ class Token implements IToken { messageMatchingKey: {}; itemsKey=null; // for loop items + /** Returns the instance data scoped to this token's dataPath. */ get data():any { return this.execution.getData(this.dataPath); } + /** The most recent item in this token's path. */ get currentItem() : Item { return this.path[this.path.length - 1]; } + /** The first item in this token's path. */ get firstItem(): Item { return this.path[0]; } + /** Returns true if any item in the path visited the given node id. */ hasNode(nodeId): Boolean { let match=false; this.path.forEach(i => { @@ -88,6 +99,7 @@ class Token implements IToken { }); return match; } + /** Returns the second-to-last non-flow item, or null if fewer than two exist. */ get lastItem() : Item { let nodes = this.path.filter(function (value) { return (value.element.type == 'bpmn:SequenceFlow') ? false : true; @@ -98,18 +110,30 @@ class Token implements IToken { else return null; } + /** Returns all tokens whose parentToken is this token. */ get childrenTokens(): Token[] { const list = []; this.execution.tokens.forEach(t => { if (t.parentToken && t.parentToken.id == this.id) list.push(t); }); return list; } + /** Returns the concatenated path from root parent down to this token. */ getFullPath(path=[]) : Item[] { if (this.parentToken) path=this.parentToken.getFullPath(path); this.path.forEach(i => { path.push(i); }); return path; } + /** + * Creates a new token starting at the given node. + * + * @param type token type (Primary, SubProcess, Diverge, etc.) + * @param execution the owning execution + * @param startNode the node where this token begins + * @param dataPath explicit data path (falls back to parent's dataPath) + * @param parentToken parent token for subprocess/diverge tokens + * @param originItem the item that caused this token to be created (e.g. gateway item) + */ constructor(type: TOKEN_TYPE, execution: Execution, startNode: Node, dataPath? ,parentToken?: Token, originItem?: Item) { this.execution = execution; this.type = type; @@ -171,6 +195,7 @@ class Token implements IToken { await token.execute(data); return token; } + /** Serializes this token's state for persistence. */ save() { let parentToken, originItem, loopId; if (this.parentToken) @@ -186,38 +211,46 @@ class Token implements IToken { currentNode: this.currentNode.id , itemsKey: this.itemsKey }; } - static load(execution: Execution , da :any ) : Token { - const startNode = execution.getNodeById(da.startNodeId); - const parentToken = execution.getToken(da.parentToken); - const currentNode = execution.getNodeById(da.currentNode); - - const token = new Token(da.type,execution, startNode, da.dataPath, parentToken, null); - token.id = da.id; - token.startNodeId = da.startNodeId; + /** Restores a token from saved state data. */ + /** + * Reconstructs a Token from its persisted state. + * + * @param execution the parent Execution instance + * @param savedData the serialized token state loaded from the datastore + */ + static load(execution: Execution , savedData :any ) : Token { + const startNode = execution.getNodeById(savedData.startNodeId); + const parentToken = execution.getToken(savedData.parentToken); + const currentNode = execution.getNodeById(savedData.currentNode); + + const token = new Token(savedData.type,execution, startNode, savedData.dataPath, parentToken, null); + token.id = savedData.id; + token.startNodeId = savedData.startNodeId; token._currentNode = currentNode; - token.status = da.status; - token.itemsKey=da.itemsKey; + token.status = savedData.status; + token.itemsKey=savedData.itemsKey; token.path = []; return token; } + /** Placeholder for stopping a token (currently no-op). */ stop() { - + } - /* - * is fired once after the execution is resumed from restrt - * - * fire resume for all existing items to wakeup the timers - * + /** + * Fires after the execution is resumed from a restart. + * Wakes up timers and other waiting behaviours on the current item. */ resume() { this.currentNode.resume(this.currentItem); } + /** Notifies all items in the path that the execution has been restored from persistence. */ restored() { this.path.forEach(item => { item.element.restored(item); }); } + /** Walks up the parent chain to find the nearest SubProcess or AdHoc token, or null. */ getSubProcessToken() : Token { if (this.type==TOKEN_TYPE.SubProcess || this.type==TOKEN_TYPE.AdHoc) return this; @@ -227,6 +260,7 @@ class Token implements IToken { return this.parentToken.getSubProcessToken(); } + /** Returns all direct child tokens of this token (same as childrenTokens getter). */ getChildrenTokens() { const children = []; this.execution.tokens.forEach(token => { @@ -282,25 +316,10 @@ class Token implements IToken { if (input) await this.currentNode.setInput(item,input); -// if (!await this.preExecute()) -// return; // loop logic will take care of it - this.log('Token('+this.id +').execute: executing currentNodeId='+ this.currentNode.id); ret = await this.currentNode.execute(item); -/* - // check for subprocess - if (this.currentNode.type == 'bpmn:SubProcess') { - this.log('..executing a sub process item:' + this.currentNode.id + " " + item.id + " is done"); - const subProcess = this.currentNode as SubProcess; - const proc = subProcess.childProcess; - const startNode = proc.getStartNode(); - - const newToken = await Token.startNewToken(this.execution, startNode, null, this, this.currentNode, null); - - } -*/ if (ret == NODE_ACTION.wait) { this.status = TOKEN_STATUS.wait; @@ -309,7 +328,6 @@ class Token implements IToken { } else if (ret == NODE_ACTION.error) { - // await this.processError(); done by the event with code } else if (ret == NODE_ACTION.abort) { this.execution.terminate(); @@ -329,11 +347,19 @@ class Token implements IToken { return result; } + /** Appends an item to the token's path and updates the current node. */ addItemToPath(item) { this.path.push(item); this.setCurrentNode(item.node); } + /** + * Handles a BPMN error by finding the matching error boundary event or + * event subprocess. If no handler is found, terminates the execution. + * + * @param errorCode the BPMN error code to match + * @param callingEvent the item that raised the error + */ async processError(errorCode,callingEvent) { let errorHandlerToken = this.getScopeCatchEvent('error',errorCode); @@ -352,6 +378,13 @@ class Token implements IToken { } } + /** + * Searches the token hierarchy for a matching catch event (error, escalation, or cancel). + * First checks direct boundary events, then walks up parent tokens. + * + * @param type the event type to find + * @param code the error/escalation code to match (null matches any) + */ getScopeCatchEvent(type:'error'|'escalation'|'cancel',code) { let contextItem: Item = this.currentItem; @@ -360,65 +393,60 @@ class Token implements IToken { let errorHandlerToken = null; let tokens=[]; try { + // two types of error handlers + // 1. eventSubProcess + // 2. boundaryEvents + let bhName=Behaviour_names.ErrorEventDefinition; + let propertyName='errorId'; + let nodeSubType=NODE_SUBTYPE.error; + if (type=='escalation') { + bhName=Behaviour_names.EscalationEventDefinition; + propertyName='escalationId'; + nodeSubType=NODE_SUBTYPE.escalation + } + else if (type=='cancel') { + bhName=Behaviour_names.CancelEventDefinition; + propertyName=null; + nodeSubType=NODE_SUBTYPE.cancel + } - - // two types of error handlers - // 1. eventSubProcess - // 2. boundaryEvents - let bhName=Behaviour_names.ErrorEventDefinition; - let propertyName='errorId'; - let nodeSubType=NODE_SUBTYPE.error; - if (type=='escalation') - { - bhName=Behaviour_names.EscalationEventDefinition; - propertyName='escalationId'; - nodeSubType=NODE_SUBTYPE.escalation - } - else if (type=='cancel') - { - bhName=Behaviour_names.CancelEventDefinition; - propertyName=null; - nodeSubType=NODE_SUBTYPE.cancel - } - - let directEvents=contextItem.node.getBoundaryEventItems(contextItem); + let directEvents=contextItem.node.getBoundaryEventItems(contextItem); - directEvents.forEach(ev=>{ + directEvents.forEach(ev=>{ tokens.push(ev.token); - }); - - let handler=this.checkTokensForError(tokens,bhName,propertyName,code); - - if (handler) - return handler; + }); - // second phase - tokens=[]; - while (contextToken && errorHandlerToken == null) { - contextToken.childrenTokens.forEach(ct => { - if ((ct.type == TOKEN_TYPE.EventSubProcess || ct.type == TOKEN_TYPE.BoundaryEvent) - && ct.currentNode.subType == nodeSubType) { + let handler=this.checkTokensForError(tokens,bhName,propertyName,code); + + if (handler) + return handler; + + // second phase + tokens=[]; + while (contextToken && errorHandlerToken == null) { + contextToken.childrenTokens.forEach(ct => { + if ((ct.type == TOKEN_TYPE.EventSubProcess || ct.type == TOKEN_TYPE.BoundaryEvent) + && ct.currentNode.subType == nodeSubType) { + tokens.push(ct); + } + }); + contextToken = contextToken.parentToken; + } + this.execution.tokens.forEach(ct=>{ + if (ct.type == TOKEN_TYPE.EventSubProcess && ct.currentNode.subType == nodeSubType) tokens.push(ct); - } }); - contextToken = contextToken.parentToken; - } - this.execution.tokens.forEach(ct=>{ - if (ct.type == TOKEN_TYPE.EventSubProcess && ct.currentNode.subType == nodeSubType) - tokens.push(ct); - }); - - return this.checkTokensForError(tokens,bhName,propertyName,code); - - } - catch (exc) { - console.log(exc); - return null; - } + return this.checkTokensForError(tokens,bhName,propertyName,code); + } + catch (exc) { + console.log(exc); + return null; + } } + /** Finds the best matching error handler token: prefers code-specific over catch-all. */ private checkTokensForError(tokens,bhName,propertyName,errorCode) { let handlingCodeEventToken=null; @@ -446,6 +474,7 @@ class Token implements IToken { return handlingAllEventToken; } + /** Handles a BPMN cancel event by finding and signaling the cancel boundary event. */ async processCancel(callingEvent) { let errorHandlerToken = this.getScopeCatchEvent('cancel',null); @@ -454,6 +483,12 @@ class Token implements IToken { } } + /** + * Handles a BPMN escalation by finding the matching escalation handler and signaling it. + * + * @param escalationCode the escalation code to match + * @param callingEvent the item that raised the escalation + */ async processEscalation (escalationCode,callingEvent) { let errorHandlerToken = this.getScopeCatchEvent('escalation',escalationCode); @@ -501,7 +536,6 @@ class Token implements IToken { return; this.log('Token('+this.id +').terminate: terminating ....'); - //await this.currentNode.end(this.currentItem,true); await this.end(true); this.status=TOKEN_STATUS.terminated; @@ -526,10 +560,15 @@ class Token implements IToken { await this.goNext(); } - /* - * is called to invoke an element like userTask, or trigger an envent or signal - * - */ + /** + * Signals a waiting token with input data to resume execution. + * + * Handles restart mode, validates the item, runs the node, and + * advances to the next node (unless noWait is set). + * + * @param data input data for the current item + * @param options execution options (restart, recover, noWait) + */ async signal(data,options={}) { // check if valid node and valid status // find the item @@ -541,7 +580,6 @@ class Token implements IToken { recover=options['recover']; const item = this.currentItem; - //this.log(`..token.signal ${this.currentNode.id} ${this.currentNode.type}`); this.logS('Token('+this.id +').signal: invoking '+this.currentNode.id+' '+this.currentNode.type+' with data='+JSON.stringify(data)); await this.currentNode.setInput(item, data); @@ -576,43 +614,27 @@ class Token implements IToken { this.logE('Token('+this.id +').signal: invoke '+this.currentNode.id+' '+this.currentNode.type+' finished!'); } - /* - * is called to invoke an element like userTask, or trigger an envent or signal - * - */ - async signal2() { - // check if valid node and valid status - // find the item - let self=this; + /** + * Phase 2 of a noWait signal: waits briefly then advances to the next node. + * Called after signalItem returns early in the noWait path. + */ + async signalContinue() { await delay(5,null); - /* - await new Promise(function (resolve) { - setTimeout(async function () { - await self.goNext(); - }, 5); - });*/ - await self.goNext(); + await this.goNext(); return this; } - /* - * is called to mark this token end - - Child Scenarios: - diverge - subprocess/trans/adHoc - loop - + /** + * Ends this token, terminating children and continuing the parent if subprocess. + * + * @param cancel if true, marks as terminated rather than cleanly ended */ - async end(cancel:Boolean=false) { this.logS('Token('+this.id +').end: currentNode=' + this.currentNode.id +' status='+this.status); - //await pause(); - if (this.status ==TOKEN_STATUS.end || this.status==TOKEN_STATUS.terminated) return; this.status = TOKEN_STATUS.end; @@ -631,8 +653,7 @@ class Token implements IToken { const child = children[i]; if (this.type==TOKEN_TYPE.SubProcess || this.type==TOKEN_TYPE.AdHoc || this.type == TOKEN_TYPE.EventSubProcess || this.type == TOKEN_TYPE.Instance - || child.type==TOKEN_TYPE.Instance || child.type==TOKEN_TYPE.AdHoc) - //if (child.type == TOKEN_TYPE.EventSubProcess || child.type==TOKEN_TYPE.AdHoc || child.type==TOKEN_TYPE.Instance ) + || child.type==TOKEN_TYPE.Instance || child.type==TOKEN_TYPE.AdHoc) { await child.terminate(); } @@ -649,16 +670,19 @@ class Token implements IToken { this.logE('Token('+this.id +').end(): finished!'); } + /** Updates the token's current node pointer. */ setCurrentNode(newCurrentNode:Node){ this.log('Token('+this.id +').setCurrentNode(): newCurrentNode.id=' + newCurrentNode.id+' currentNode='+this.currentNode); this._currentNode = newCurrentNode; } - /* - * once node is completed the token will move to next action - * - */ + /** + * Advances the token to the next node(s) after the current node completes. + * + * If diverging (multiple outbounds), creates child tokens for each branch. + * If single outbound, continues on the same token. + */ async goNext() { await pause(); /** issue 186: token with empty path due to loop preceded by gateway */ @@ -684,8 +708,6 @@ class Token implements IToken { return await this.end(true); } this.log('Token('+this.id +').goNext(): currentNodeId=' + this.currentNode.id +' type='+this.currentNode.type+' currentItem.status='+this.currentItem.status); - //this.log(`..token.goNext from ${this.currentNode.id} ${this.currentNode.type}`); - if (!await this.preNext()) { this.logE('Token('+this.id +').goNext(): no more outbounds - ending this token '+this.id); return; @@ -739,7 +761,6 @@ class Token implements IToken { self.log('Token(' + self.id + ').goNext(): ... currentNodeId(' + self.currentNode.name + '|' + self.currentNode.id + ') processing Flow(' + flowItem.element.id + ") to " + nextNode.id); if (nextNode) { self._currentNode = nextNode; -// await self.execute(null); promises.push(self.execute(null)); } }); diff --git a/src/interfaces/DataObjects.ts b/src/interfaces/DataObjects.ts index 7c5ba99..55cc992 100644 --- a/src/interfaces/DataObjects.ts +++ b/src/interfaces/DataObjects.ts @@ -1,4 +1,64 @@ -import { ITEM_STATUS, } from './Enums'; +import { ITEM_STATUS, EXECUTION_STATUS, BPMN_TYPE } from './Enums'; + +/** MongoDB-style query operator for field-level conditions */ +type QueryOperator = T | { + $eq?: T; + $gt?: T; + $gte?: T; + $lt?: T; + $lte?: T; + $in?: T[]; + $exists?: boolean; +}; + +/** + * Query for locating process instances. + * Supports direct instance fields, dot-notation item fields, and data fields. + */ +interface InstanceQuery { + id?: QueryOperator; + name?: QueryOperator; + status?: QueryOperator; + startedAt?: QueryOperator; + endedAt?: QueryOperator; + version?: QueryOperator; + parentItemId?: QueryOperator; + /** Dot-notation item fields (e.g., 'items.status', 'items.elementId') */ + 'items.id'?: QueryOperator; + 'items.status'?: QueryOperator; + 'items.elementId'?: QueryOperator; + 'items.itemKey'?: QueryOperator; + 'items.type'?: QueryOperator; + 'items.assignee'?: QueryOperator; + 'items.candidateUsers'?: QueryOperator; + 'items.candidateGroups'?: QueryOperator; + 'items.messageId'?: QueryOperator; + 'items.signalId'?: QueryOperator; + 'items.timeDue'?: QueryOperator | { $exists: boolean }; + 'items.name'?: QueryOperator; + /** $or for complex queries (e.g., security qualification) */ + $or?: InstanceQuery[]; + /** Open for data.* fields and other custom filters */ + [key: string]: any; +} + +/** + * Query for locating specific items within instances. + * Same shape as InstanceQuery since items are queried via instance-level MongoDB queries. + */ +type ItemQuery = InstanceQuery; + +/** Input data passed to engine operations (start, invoke, signal, etc.) */ +type InputData = Record; + +/** Assignment data for user task assignment (assignee, candidateUsers, candidateGroups, etc.) */ +type AssignmentData = Record; + +/** Matching query for throwMessage / throwSignal */ +type MatchingQuery = Record; + +/** Options for findInstances projection */ +type FindOption = 'summary' | 'full'; interface IItemData { id: string; // System generated unique Id @@ -110,4 +170,5 @@ interface IProcessData { historyTimeToLive; isStartableInTasklist; } -export { IItemData, IInstanceData , IDefinitionData, IElementData, IFlowData , IBpmnModelData, IProcessData, IEventData } \ No newline at end of file +export { IItemData, IInstanceData , IDefinitionData, IElementData, IFlowData , IBpmnModelData, IProcessData, IEventData, + InstanceQuery, ItemQuery, InputData, AssignmentData, MatchingQuery, FindOption, QueryOperator } \ No newline at end of file diff --git a/src/interfaces/Enums.ts b/src/interfaces/Enums.ts index d88b806..5aabe74 100644 --- a/src/interfaces/Enums.ts +++ b/src/interfaces/Enums.ts @@ -84,9 +84,7 @@ enum ITEM_STATUS { } -//type ITEMSTATUS = 'enter' | 'start' | 'wait' | 'end' | 'terminated' | 'discard'; - -enum EXECUTION_STATUS { running='running',wait='wait', end = 'end' , terminated ='terminated' } +enum EXECUTION_STATUS { running='running',wait='wait', end = 'end' , terminated ='terminated', error='error' } enum TOKEN_STATUS { running = 'running', wait = 'wait', end = 'end', terminated = 'terminated' , queued = 'queued' } /* diff --git a/src/interfaces/datastore.ts b/src/interfaces/datastore.ts index 02a174b..002f8ff 100644 --- a/src/interfaces/datastore.ts +++ b/src/interfaces/datastore.ts @@ -1,5 +1,5 @@ import { IDefinition, IInstanceData } from './'; -import { IBpmnModelData, IItemData, IEventData } from './'; +import { IBpmnModelData, IItemData, IEventData, InstanceQuery, ItemQuery, FindOption } from './'; export interface FindParams { filter?: Record; @@ -24,16 +24,16 @@ interface IDataStore { db: any; logger: any; locker: any; - save(instance:any,options:any): Promise; - loadInstance(instanceId: any,options:any): Promise<{ + save(instance: any, options: any): Promise; + loadInstance(instanceId: string, options: any): Promise<{ instance: any; items: any[]; }>; - findItem(query: any): Promise; - findInstance(query: any, options: any): Promise; - findInstances(query: any, option: 'summary' | 'full'|any): Promise; - findItems(query: any): Promise; - deleteInstances(query?: any): Promise; + findItem(query: InstanceQuery): Promise; + findInstance(query: InstanceQuery, options: any): Promise; + findInstances(query: InstanceQuery, option?: FindOption): Promise; + findItems(query: InstanceQuery): Promise; + deleteInstances(query?: InstanceQuery): Promise; install(); archive(query); find(FindParams) : Promise; diff --git a/src/interfaces/server.ts b/src/interfaces/server.ts index 1e09d28..af2a5df 100644 --- a/src/interfaces/server.ts +++ b/src/interfaces/server.ts @@ -1,4 +1,5 @@ -import { IExecution , ILogger , IItem, IConfiguration, IAppDelegate, IDataStore,IModelsDatastore, IScriptHandler } from '../'; +import { IExecution , ILogger , IItem, IConfiguration, IAppDelegate, IDataStore,IModelsDatastore, IScriptHandler, + InstanceQuery, ItemQuery, InputData, AssignmentData, MatchingQuery } from '../'; import type { EventEmitter } from 'eventemitter3'; import type { EventEmitter as NodeEventEmitter } from 'events'; @@ -38,7 +39,7 @@ interface IEngine { * @param data input data * @param startNodeId in process has multiple start node; you need to specify which one */ - start(name: any, data?: any, startNodeId?: string, userName?: string, options?: any): Promise; + start(name: string, data?: InputData, startNodeId?: string, userName?: string, options?: any): Promise; /** * restores an instance into memeory or provides you access to a running instance * @@ -47,7 +48,7 @@ interface IEngine { * @param instanceQuery criteria to fetch the instance * * query example: - * + * * ```jsonl * { id: instanceId} * { data: {caseId: 1005}} @@ -56,14 +57,14 @@ interface IEngine { * ``` * */ - get(instanceQuery: any): Promise; + get(instanceQuery: InstanceQuery): Promise; /** * Continue an existing item that is in a wait state * * ------------------------------------------------- - * + * * scenario: - * + * * ``` * itemId {itemId: value } * itemKey {itemKey: value} @@ -73,12 +74,12 @@ interface IEngine { * @param itemQuery criteria to retrieve the item * @param data */ - invoke(itemQuery: any, data: {}, userName?: string, options?: {}): Promise; + invoke(itemQuery: ItemQuery, data: InputData, userName?: string, options?: any): Promise; - assign(itemQuery: any, data: {}, assignment: {}, userName: string,options?:{}): Promise; + assign(itemQuery: ItemQuery, data: InputData, assignment: AssignmentData, userName: string, options?: any): Promise; - startRepeatTimerEvent(instanceId, prevItem: IItem, data: {},options?:{}) : Promise; + startRepeatTimerEvent(instanceId: string, prevItem: IItem, data: InputData, options?: any) : Promise; /** * @@ -96,7 +97,7 @@ interface IEngine { * @param elementId * @param data */ - startEvent(instanceId: any, elementId: any, data?: {},userName?: string,options?:{}): Promise; + startEvent(instanceId: string, elementId: string, data?: InputData, userName?: string, options?: any): Promise; /** * * signal/message raise a signal or throw a message @@ -109,10 +110,9 @@ interface IEngine { * @param matchingKey should match the itemKey (if specified) * @param data message data */ - //signal(messageId: any, matchingKey: any, data?: {}): Promise; - throwMessage(messageId, data: {}, matchingQuery: {}): Promise; - throwSignal(signalId, data: {}, matchingQuery: {}); - restart(itemQuery, data:any,userName, options) :Promise; + throwMessage(messageId: string, data: InputData, matchingQuery: MatchingQuery): Promise; + throwSignal(signalId: string, data: InputData, matchingQuery: MatchingQuery); + restart(itemQuery: ItemQuery, data: InputData, userName: string, options?: any) :Promise; upgrade(model:string,afterNodeIds:string[]):Promise; diff --git a/src/server/Engine.ts b/src/server/Engine.ts index c38b5ea..554d66c 100644 --- a/src/server/Engine.ts +++ b/src/server/Engine.ts @@ -7,12 +7,24 @@ import { DataStore } from '../datastore'; import { exec } from 'child_process'; +/** + * Core orchestrator for BPMN process execution. + * + * Manages the full lifecycle of process instances: starting new processes, + * resuming waiting items, handling signals/messages, and timer events. + * All operations acquire an instance lock before modifying state and release it on completion. + */ class Engine extends ServerComponent implements IEngine{ runningCounter=0; callsCounter=0; + /** + * Creates an Engine instance bound to the given server context. + * + * @param server the server providing dataStore, definitions, cache, and appDelegate + */ constructor(server) { - + super(server); } @@ -78,8 +90,16 @@ class Engine extends ServerComponent implements IEngine{ } + /** + * Restarts a completed or terminated item, allowing re-execution from that point. + * + * @param itemQuery criteria to locate the item to restart + * @param data input data for the restarted item + * @param userName user performing the restart + * @param options execution options + */ public async restart(itemQuery, data:any,userName, options={}) :Promise { - + this.logger.log(`^Action:engine.restart`); let execution; this.runningCounter++; @@ -112,20 +132,17 @@ class Engine extends ServerComponent implements IEngine{ /** - * restores an instance into memeory or provides you access to a running instance - * - * this will also resume execution - * - * @param instanceQuery criteria to fetch the instance - * - * query example: - * - * ```jsonl - * { id: instanceId} - * { data: {caseId: 1005}} - * { items.id : 'abcc111322'} - * { items.itemKey : 'businesskey here'} - * ``` + * Restores an instance into memory and returns the execution. + * + * @param instanceQuery criteria to fetch the instance + * + * Query examples: + * ``` + * { id: instanceId } + * { data: { caseId: 1005 } } + * { "items.id": "abcc111322" } + * { "items.itemKey": "businesskey here" } + * ``` */ async get(instanceQuery): Promise { @@ -138,10 +155,9 @@ class Engine extends ServerComponent implements IEngine{ lock instance */ private async lock(executionId) { - this.logger.log('...locking ..'+executionId); - await this.server.dataStore.locker.lock(executionId); - - this.logger.log(' locking complete' + executionId); + this.logger.log('...locking ..'+executionId); + await this.server.dataStore.locker.lock(executionId); + this.logger.log(' locking complete' + executionId); } /** release instance lock @@ -150,29 +166,15 @@ class Engine extends ServerComponent implements IEngine{ if (id===null) id =execution.id; this.logger.log('...unlocking ..' + id); - await this.server.dataStore.locker.release(id); - if (execution) - execution.isLocked=false; + await this.server.dataStore.locker.release(id); + if (execution) + execution.isLocked=false; } /*** Loads instance into memory for purpose of execution Locks instance first if required check if in cache */ - /*static restorePromise = null; - private async restore(instanceId): Promise { - - if (Engine.restorePromise) - await Engine.restorePromise; - - Engine.restorePromise = this.doRestore(instanceId); - - let ret=await Engine.restorePromise; - - Engine.restorePromise = null; - return ret; - } - */ private async restore(instanceId,itemId=null): Promise { // need to load instance first @@ -191,12 +193,6 @@ class Engine extends ServerComponent implements IEngine{ execution = await Execution.restore(this.server,instance,itemId); execution.isLocked = true; - /* new dataStore for every execution to be monitored - const newDataStore = new DataStore(execution.server); - execution.server.dataStore = newDataStore; - - newDataStore.monitorExecution(execution); */ - this.cache.add(execution); this.logger.log("restore completed: "+instance.saved); @@ -205,18 +201,25 @@ class Engine extends ServerComponent implements IEngine{ return execution; } + /** + * Convenience wrapper around {@link invoke} for signaling a single waiting item. + * + * @param itemQuery criteria to retrieve the item + * @param data data to pass to the item + */ async invokeItem(itemQuery, data = {}): Promise { return await this.invoke(itemQuery, data); } /** - * update an existing item that is in a wait state with an assignment - * can modify data or assignment or both - * - * ------------------------------------------------- - * + * Updates an existing item that is in a wait state with an assignment. + * Can modify data, assignment, or both without completing the item. + * * @param itemQuery criteria to retrieve the item - * @param data + * @param data data to merge into the item + * @param assignment assignment fields (e.g. assignee, candidateGroups) + * @param userName user performing the assignment + * @param options execution options */ async assign(itemQuery, data = {}, assignment = {}, userName: string,options= {}): Promise { @@ -257,20 +260,23 @@ class Engine extends ServerComponent implements IEngine{ } /** - * Continue an existing item that is in a wait state - * - * ------------------------------------------------- - * - * scenario: - * - * ``` - * itemId {itemId: value } - * itemKey {itemKey: value} - * instance,task {instanceId: instanceId, elementId: value } - * ``` - * + * Continues a waiting item by signaling its token with input data. + * + * Uses a two-phase signal when `options.noWait` is true: + * - Phase 1 (`signalItem`): sets up context, passes data to the token, saves, and returns immediately. + * - Phase 2 (`signalItemContinue`): runs in the background to complete execution and release the lock. + * + * Query examples: + * ``` + * { itemId: value } + * { itemKey: value } + * { instanceId: instanceId, elementId: value } + * ``` + * * @param itemQuery criteria to retrieve the item - * @param data + * @param data input data to pass to the item + * @param userName user performing the invocation + * @param options execution options; set `noWait: true` for async two-phase signal */ async invoke(itemQuery, data = {}, userName: string = null, options = {}): Promise { @@ -293,46 +299,39 @@ class Engine extends ServerComponent implements IEngine{ if (item.status !== 'wait') { this.logger.log(`*****Item status is not in wait state ${item.status} ${item.elementId}-${item.processName}`) - //this.logger.error(`Item status is not in wait state`); } execution = await this.restore(item.instanceId); await execution.signalItem(item.id, this.sanitizeData(data),userName,options); let exeItem=execution.item; - try { + try { if (options['noWait'] == true) { this.logger.log(`.noWait`); let self=this; execution.save(); - execution.worker=execution.signalItem2(item.id); - execution.worker.then(async function (obj) { - // await execution.signalItem2(item.id); + execution.worker=execution.signalItemContinue(item.id); + execution.worker.then(async function (obj) { self.logger.log('after worker is done releasing ..'+item.instanceId); self.release(execution); - }); + }); return execution; } else { - // await execution.signalItem2(item.id); not needed since signal() issues goNext() - this.logger.log(`.engine.continue ended`); - await this.release(execution); return execution; } } - catch(exc) - { - await this.release(execution); - throw exc; + catch(exc) { + await this.release(execution); + throw exc; } - finally { if (execution && execution.isLocked) await this.release(execution); } - } + } catch (exc) { return await this.exception(exc,execution); } @@ -343,11 +342,14 @@ class Engine extends ServerComponent implements IEngine{ } } /** - * - * Repeat Timers need to create new Item - * @param instanceId - * @param elementId - * @param data + * Creates a new item for a repeating timer event on an existing instance. + * + * Called by the Cron scheduler when a repeat-cycle timer fires again. + * + * @param instanceId the instance to attach the new timer item to + * @param prevItem the previous timer item that triggered this repeat + * @param data input data for the new timer item + * @param options execution options */ async startRepeatTimerEvent(instanceId, prevItem, data = {},options={}) : Promise { @@ -377,19 +379,16 @@ class Engine extends ServerComponent implements IEngine{ } } /** - * - * Invoking an event (usually start event of a secondary process) against an existing instance - * or - * Invoking a start event (of a secondary process) against an existing instance - * ---------------------------------------------------------------------------- - * instance,task - *``` - * {instanceId: instanceId, elementId: value } - *``` - * - * @param instanceId - * @param elementId - * @param data + * Invokes a start event (typically from a secondary process) against an existing instance. + * + * Used when a signal or message triggers a new event subprocess or secondary + * start event within an already-running instance. + * + * @param instanceId the running instance to target + * @param elementId the start event element id to invoke + * @param data input data for the event + * @param userName user triggering the event + * @param options execution options */ async startEvent(instanceId, elementId, data = {},userName: string = null, options = {}) : Promise { @@ -419,6 +418,16 @@ class Engine extends ServerComponent implements IEngine{ } } + /** + * Throws a BPMN message event. + * + * First checks model definitions for a matching start event to launch a new instance. + * If none found, searches running instances for a waiting item with the given messageId. + * + * @param messageId the message id as defined in the BPMN model + * @param data message payload + * @param matchingQuery additional criteria to narrow down the target item + */ async throwMessage(messageId, data = {}, matchingQuery = {}): Promise { this.logger.log('..^Action:engine.throwMessage ', messageId,this.sanitizeData(data),matchingQuery); @@ -515,7 +524,6 @@ class Engine extends ServerComponent implements IEngine{ for (var i = 0; i < items.length; i++) { let item = items[i]; -// console.log(`Throw Signal ${signalId} found target: ${item.processName} ${item.id}`); this.logger.log('..^Action:engine.Throw Signal found target', item.processName,item.id ); var res=await this.invoke({ "items.id": item.id }, this.sanitizeData(data)); instances.push({instanceId:res.instance.id,itemId:item.id}); @@ -523,26 +531,32 @@ class Engine extends ServerComponent implements IEngine{ } return instances; } + /** + * Returns current engine status counters. + */ status() { return { running: this.runningCounter, calls: this.callsCounter }; } /** - * - * @param model - * @param afterNodeIds - */ - async upgrade(model:string,afterNodeIds:string[]):Promise { - - let ds=this.server.dataStore; + * Upgrades running instances to a new model source. + * + * Finds all instances of the given model that have not yet reached any of + * the specified nodes, then replaces their stored BPMN source with the latest version. + * + * @param model name of the BPMN model to upgrade + * @param afterNodeIds exclude instances that have already passed through these nodes + * @returns list of upgraded instance ids, or an error object + */ + async upgrade(model:string,afterNodeIds:string[]):Promise { - // {"name":"boundary-event","$nor":[{"items":{"$elemMatch":{"elementId":"Reminder-Timer"}}}]} + let ds=this.server.dataStore; let query={"name":model}; if (afterNodeIds.length>0) { let nors = []; afterNodeIds.forEach(node=>{ - nors.push({"items":{"elemMatch":{"elementId":node}}}); + nors.push({"items":{"elemMatch":{"elementId":node}}}); }); query["$nor"]=nors; } @@ -555,28 +569,30 @@ class Engine extends ServerComponent implements IEngine{ const resIds=[]; let self=this; for(let i=0;i