src/controllers/ws_manager.js

/*
 * Copyright 2019-present Open Networking Foundation

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 * http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


(function () {
    'use strict';

    const _ = require('lodash');
    const Workflow = require('../types/workflow.js');
    const logger = require('../config/logger.js');

    let serviceEvents = {
        // manager -> controller -> manager
        WORKFLOW_REGISTER: 'cord.workflow.ctlsvc.workflow.register',
        WORKFLOW_REGISTER_ESSENCE: 'cord.workflow.ctlsvc.workflow.register_essence',
        WORKFLOW_LIST: 'cord.workflow.ctlsvc.workflow.list',
        WORKFLOW_LIST_RUN: 'cord.workflow.ctlsvc.workflow.run.list',
        WORKFLOW_CHECK: 'cord.workflow.ctlsvc.workflow.check',
        WORKFLOW_REMOVE: 'cord.workflow.ctlsvc.workflow.remove',
        WORKFLOW_REMOVE_RUN: 'cord.workflow.ctlsvc.workflow.run.remove',
        WORKFLOW_REPORT_NEW_RUN: 'cord.workflow.ctlsvc.workflow.report_new_run',
        WORKFLOW_REPORT_RUN_STATE: 'cord.workflow.ctlsvc.workflow.report_run_state',
        // controller -> manager
        WORKFLOW_KICKSTART: 'cord.workflow.ctlsvc.workflow.kickstart',
        WORKFLOW_CHECK_STATE: 'cord.workflow.ctlsvc.workflow.check.state'
    };

    // WebSocket interface for workflow registration
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.reg',
    //     message: {
    //         req_id: <req_id>, // optional
    //         workflow: <workflow>
    //     }
    // }
    const registerWorkflow = (topic, message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let errorMessage;
        if(!message) {
            // error
            errorMessage = `Message body for topic ${topic} is null or empty`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow' in message)) {
            // error
            errorMessage = `field 'workflow' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        let workflow = message.workflow;

        logger.log('debug', `workflow - ${JSON.stringify(workflow)}`);

        let result = eventrouter.addWorkflow(workflow);
        if(!result) {
            errorMessage = `failed to register a workflow ${workflow.getId()}`;
            cb(errorMessage, false);
        }
        else {
            cb(null, true);
        }
        return;
    };

    // WebSocket interface for workflow registration (via essence)
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.reg_essence',
    //     message: {
    //         req_id: <req_id> // optional
    //         essence: <workflow essence>
    //     }
    // }
    const registerWorkflowEssence = (topic, message, cb) => {
        const eventrouter = require('./eventrouter.js');
        let errorMessage;
        if(!message) {
            // error
            errorMessage = `Message body for topic ${topic} is null or empty`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('essence' in message)) {
            // error
            errorMessage = `field 'essence' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        let essence = message.essence;
        let result = true;
        let errorResults = [];

        logger.log('debug', `workflow essence - ${JSON.stringify(essence)}`);

        let workflows = Workflow.loadWorkflowsFromEssence(essence);
        workflows.forEach((workflow) => {
            if(workflow) {
                let localResult = eventrouter.addWorkflow(workflow);
                errorResults.push(localResult);
                result = result && localResult; // false if any of registrations fails
            }
        });

        if(!result) {
            errorMessage = `failed to register workflows ${errorResults}`;
            cb(errorMessage, false);
        }
        else {
            cb(null, true);
        }
        return;
    };

    // WebSocket interface for workflow listing
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.list',
    //     message: {
    //         req_id: <req_id> // optional
    //     }
    // }
    const listWorkflows = (_topic, _message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let result = eventrouter.listWorkflows();
        cb(null, result);
        return;
    };

    // WebSocket interface for workflow run listing
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.list',
    //     message: {
    //         req_id: <req_id> // optional
    //     }
    // }
    const listWorkflowRuns = (_topic, _message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let result = eventrouter.listWorkflowRuns();
        cb(null, result);
        return;
    };

    // WebSocket interface for workflow check
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.check',
    //     message: {
    //         req_id: <req_id> // optional
    //         workflow_id: <workflow_id>
    //     }
    // }
    const checkWorkflow = (topic, message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let errorMessage;
        if(!message) {
            // error
            errorMessage = `Message body for topic ${topic} is null or empty`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_id' in message)) {
            // error
            errorMessage = `field 'workflow_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        let workflowId = message.workflow_id;
        let result = eventrouter.checkWorkflow(workflowId);
        cb(null, result);
        return;
    };

    // WebSocket interface for workflow removal
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.remove',
    //     message: {
    //         req_id: <req_id> // optional
    //         workflow_id: <workflow_id>
    //     }
    // }
    const removeWorkflow = (topic, message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let errorMessage;
        if(!message) {
            // error
            errorMessage = `Message body for topic ${topic} is null or empty`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_id' in message)) {
            // error
            errorMessage = `field 'workflow_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        let workflowId = message.workflow_id;
        let result = eventrouter.removeWorkflow(workflowId);
        cb(null, result);
        return;
    };

    // WebSocket interface for workflow run removal
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.run.remove',
    //     message: {
    //         req_id: <req_id> // optional
    //         workflow_id: <workflow_id>,
    //         workflow_run_id: <workflow_run_id>
    //     }
    // }
    const removeWorkflowRun = (topic, message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let errorMessage;
        if(!message) {
            // error
            errorMessage = `Message body for topic ${topic} is null or empty`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_id' in message)) {
            // error
            errorMessage = `field 'workflow_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_run_id' in message)) {
            // error
            errorMessage = `field 'workflow_run_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        let workflowRunId = message.workflow_run_id;

        let result = eventrouter.removeWorkflowRun(workflowRunId);
        cb(null, result);
        return;
    };

    // WebSocket interface for reporting a new workflow run
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.report_new_run',
    //     message: {
    //         req_id: <req_id> // optional
    //         workflow_id: <workflow_id>,
    //         workflow_run_id: <workflow_run_id>
    //     }
    // }
    const reportNewWorkflowRun = (topic, message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let errorMessage;
        if(!message) {
            // error
            errorMessage = `Message body for topic ${topic} is null or empty`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_id' in message)) {
            // error
            errorMessage = `field 'workflow_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_run_id' in message)) {
            // error
            errorMessage = `field 'workflow_run_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        let workflowRunId = message.workflow_run_id;

        // there must be a workflow matching
        // set the workflow kickstarted
        let result = eventrouter.setWorkflowRunKickstarted(workflowRunId);
        cb(null, result);
        return;
    };

    // WebSocket interface for reporting workflow run state
    // Message format:
    // {
    //     topic: 'cord.workflow.ctlsvc.workflow.report_run_state',
    //     message: {
    //         req_id: <req_id> // optional
    //         workflow_id: <workflow_id>,
    //         workflow_run_id: <workflow_run_id>,
    //         state: one of ['success', 'running', 'failed', 'unknown']
    //     }
    // }
    const reportWorkflowRunState = (topic, message, cb) => {
        const eventrouter = require('./eventrouter.js');

        let errorMessage;
        if(!message) {
            // error
            errorMessage = `Message body for topic ${topic} is null or empty`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_id' in message)) {
            // error
            errorMessage = `field 'workflow_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('workflow_run_id' in message)) {
            // error
            errorMessage = `field 'workflow_run_id' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        if(!('state' in message)) {
            // error
            errorMessage = `field 'state' does not exist in message body - ${JSON.stringify(message)}`;
            logger.log('warn', `Return error - ${errorMessage}`);
            cb(errorMessage, false);
            return;
        }

        let workflowRunId = message.workflow_run_id;
        let state = message.state;

        // there must be a workflow matching
        // set workflow state
        let result = eventrouter.setWorkflowRunState(workflowRunId, state);
        cb(null, result);
        return;
    }

    const getRouter = () => {
        return {
            registerWorkflow: {
                topic: serviceEvents.WORKFLOW_REGISTER,
                handler: registerWorkflow
            },
            registerWorkflowEssence: {
                topic: serviceEvents.WORKFLOW_REGISTER_ESSENCE,
                handler: registerWorkflowEssence
            },
            listWorkflows: {
                topic: serviceEvents.WORKFLOW_LIST,
                handler: listWorkflows
            },
            listWorkflowRuns: {
                topic: serviceEvents.WORKFLOW_LIST_RUN,
                handler: listWorkflowRuns
            },
            checkWorkflow: {
                topic: serviceEvents.WORKFLOW_CHECK,
                handler: checkWorkflow
            },
            removeWorkflow: {
                topic: serviceEvents.WORKFLOW_REMOVE,
                handler: removeWorkflow
            },
            removeWorkflowRun: {
                topic: serviceEvents.WORKFLOW_REMOVE_RUN,
                handler: removeWorkflowRun
            },
            reportNewWorkflowRun: {
                topic: serviceEvents.WORKFLOW_REPORT_NEW_RUN,
                handler: reportNewWorkflowRun
            },
            reportWorkflowRunState: {
                topic: serviceEvents.WORKFLOW_REPORT_RUN_STATE,
                handler: reportWorkflowRunState
            }
        };
    };

    // out-going commands
    const kickstartWorkflow = (workflowId, workflowRunId) => {
        const eventrouter = require('./eventrouter.js');

        let clients = eventrouter.getWorkflowManagerClients();
        _.forOwn(clients, (client, _clientId) => {
            let socket = client.getSocket();
            if(socket) {
                socket.emit(serviceEvents.WORKFLOW_KICKSTART, {
                    workflow_id: workflowId,
                    workflow_run_id: workflowRunId
                });
            }
        });
        return;
    };

    const checkWorkflowState = (workflowId, workflowRunId) => {
        const eventrouter = require('./eventrouter.js');

        let clients = eventrouter.getWorkflowManagerClients();
        _.forOwn(clients, (client, _clientId) => {
            let socket = client.getSocket();
            if(socket) {
                socket.emit(serviceEvents.WORKFLOW_CHECK_STATE, {
                    workflow_id: workflowId,
                    workflow_run_id: workflowRunId
                });
            }
        });
        return;
    };

    module.exports = {
        serviceEvents: serviceEvents,
        getRouter: getRouter,
        kickstartWorkflow: kickstartWorkflow,
        checkWorkflowState: checkWorkflowState
    };
})();