<commit_before>/**
 * Created by ron.yadgar on 17/07/2016.
 */
const Q = require('q');
const util = require('util');
const events = require("events");
const config = require('../../common/Configuration');
const qio = require('q-io/fs');
const path = require('path');
const _ = require('underscore');
const glob = require('glob');
const ErrorUtils = require('../utils/error-utils');
const PlaylistGenerator = require('../playlistGenerator/PlaylistGenerator');
const persistenceFormat = require('../../common/PersistenceFormat.js');
const RecordingEntrySession = require('./RecordingEntrySession.js')
const logger = require('../../common/logger').getLogger("RecordingManager");
const mathUtils = require('./../utils/math-utils');
const configRecording = config.get("recording");
const kalturaTypes = require('../kaltura-client-lib/KalturaTypes');
const unknownEntry = "UNKNOWN";
const backendClient = require('../BackendClientFactory.js').getBackendClient();
const maxAllowedMp4FilesQueueSize = config.get("recording").recordingMaxAllowedMp4FilesQueueSize;
const TaskManager = require('../TaskManager').TaskManager;
const completedRecordingFolderPath = configRecording.completedRecordingFolderPath;
const clippingTaskTemplateName = config.get('newClippingTaskPushNotificationTemplateName');

class RecordingManager {
//todo what to do if we stop stream and start, but other machine has start to get it
//todo move all path join baseName etc to persistent format
// should also recording for primary ?
//change then style, and fix indentation
// todo Assumptions:
    /*
     1. No change configuration!
     2. recording window in not more that entry session window -  check it
     3. All chunks are comming at the same time
     recordingRootPath -
     recordingEntryPath
     recordingSessionPath
     recordingSessionSourcePath
     */
    constructor() {

        const recordingTimeInterval = configRecording.recordingTimeIntervalInSec * 1000;
        this.completedRecordingFolderPath = configRecording.completedRecordingFolderPath;
        //verify that recordingMaxDurationInHours is no longer that
        this.recordingMaxDurationInMs = config.get("recording").recordingMaxDurationInHours * 60 * 60 * 1000;
        this.recordingMinDurationInMS = config.get("recording").recordingMinDurationInSeconds * 1000;
        this.recordingList = {};

        logger.info("Initializing recording manager");
        TaskManager.register(kalturaTypes.KalturaEntryServerNodeType.LIVE_CLIPPING_TASK, clippingTaskTemplateName, this.handleNewLiveClippingTask);
        
        setInterval(()=> {
            this.handleRecordingEntries()
        }, recordingTimeInterval, this);
        this.synchronizeAddNewChunks=Q.resolve();
    }

    getRecordingEntrySession(entryObj, mp4FilesQueue = null) {
        let entryId = entryObj.entryId;
        if (mp4FilesQueue != null && mp4FilesQueue.length > 0) {
            logger.debug(`[${entryObj.entryId}-${entryObj.recordedEntryId}] getRecordingEntrySession called with mp4 files queue size [${mp4FilesQueue.length}]`);
        }
        if (!entryObj.recordedEntryId) {
            entryObj.recordedEntryId = unknownEntry;
            logger.warn(`[${entryObj.entryId}] Entry has no recordedEntryId`);
        }
        if (!this.recordingList[entryId]) {
            logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Create new recording session`);
            let recordingEntrySession = new RecordingEntrySession(entryObj, mp4FilesQueue);
            this.recordingList[entryId] = recordingEntrySession;
            return Q.resolve({session:recordingEntrySession, sessionStarted:false});
        }
        else if (this.recordingList[entryId] && this.recordingList[entryId].disableRecording) {
            let msg = `disableRecording is set. Entry won't be recording anymore`;
            logger.error(msg);
            delete this.recordingList[entryId];
            return Q.reject(msg);
        }    
        else if (this.recordingList[entryId].recordedEntryId != entryObj.recordedEntryId) {
            let msg = `[${entryId}-${this.recordingList[entryId].recordedEntryId}] unexpected state: session already exist for same liveEntry!!! [NEW ${entryObj.recordedEntryId}]`
            logger.error(msg);
            return Q.reject(msg);
        }
        else {
            logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Found a previous recording session, update time stamp`);
            this.recordingList[entryId].lastUpdateTime = new Date();
            return Q.resolve({session:this.recordingList[entryId], sessionStarted:true});
        }
    }

    startRecording(entryObj) {
        return Q.resolve().then(()=> {
            let startRecordingTime = new Date();
            logger.debug(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Start recording`);
            return this.stopSessionIfNeeded(entryObj)
                .then((mp4FilesQueue)=> {
                    return this.getRecordingEntrySession(entryObj, mp4FilesQueue)
                })
                .then((sessionObj)=> {
                      return this.handleFlavorsMismatch(entryObj, sessionObj)
                        .then((sessionObj) => {
                            if (!sessionObj.sessionStarted) {
                                return sessionObj.session.startRecording()
                                    .then(()=> {
                                        // Once start process is complete, release lock on files queue
                                        logger.info(`[startRecording] release lock mp4FilesQueueLock`)
                                        sessionObj.session.mp4FilesQueueLock = false;
                                        let startRecordingTimeDuration = new Date() - startRecordingTime;
                                        logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Start recording completed, time : ${startRecordingTimeDuration} ms`);
                                    })
                            } else {
                                logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] session running already, no reason to start it again.`);
                                if (sessionObj.session.mp4FilesQueueLock) {
                                    logger.warn(`[startRecording] new session was not started. Check log for reason. Release lock mp4FilesQueueLock`);
                                    sessionObj.session.mp4FilesQueueLock = false;
                                }
                            }
                        });
                });
        }).catch((err)=> {
            logger.error(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Error while trying to start recording : ${ErrorUtils.error2string(err)}, delete session`);
            if (this.recordingList[entryObj.entryId]) {
                delete this.recordingList[entryObj.entryId];
                logger.warn(`[startRecording] new session was not started due to error.`);
            }
            return Q.reject(`[startRecording] start recording failed`);
        })
    }

    stopLive(entryId) {
        let recordingEntrySession = this.recordingList[entryId];
        if (!recordingEntrySession) {
            logger.warn("[%s] Can't updateServer recording, entry does not exist on recording list", entryId);
            return
        }
        if (this.isRecordedEntryUnknown(recordingEntrySession.recordedEntryId)) {
            this.replaceUnknownRecordedEntry(recordingEntrySession);
        }
        recordingEntrySession.updateEntryDuration(true)
    }

    splitRecording(entryId, reason) {
        let recordingSession = this.recordingList[entryId];
        if (!recordingSession) {
            logger.warn(`[${entryId}] Error splitting recording for entry, recordingEntrySession not found`);
            return Q.resolve();
        }
        if (recordingSession.disableRecording) {
            logger.warn(`[${entryId}] recording is disabled. Look for reason in log`);
            return Q.resolve();
        }
        return recordingSession.isUnlimitedRecordingDurationEnabled()
            .then((isEnabled)=> {
                if (!isEnabled) {
                    return this.stopRecording(entryId);
                }
                else {
                    let entryObject = recordingSession.entryObject;
                    logger.info(`[${entryId}-${recordingSession.entryObject.recordedEntryId}] Recording entry ${this.recordingList[entryId].recordedEntryId} has passed the limit of recording duration - ${this.recordingMaxDurationInMs / 1000} seconds. Split session`);
                    return recordingSession.splitRecordedEntry(recordingSession.entryObject, reason)
                        .then(()=> {
                            // Pass startRecording the queue of previous session in case there are files waiting to be written
                            return this.startRecording(entryObject);
                        })
                        .then(()=> {
                            logger.info(`[${entryId}-${entryObject.recordedEntryId}] Split process completed successfully. Replaced recordedEntryId: [${recordingSession.recordedEntryId}] by: [${entryObject.recordedEntryId}]`);
                        })
                        .catch((err)=> {
                            logger.error(`[${entryId}-${entryObject.recordedEntryId}] Error splitting recording for entry. Error: ${ErrorUtils.error2string(err)}`);
                        });
                }

            })
    }

    isRecordedEntryUnknown(recordedEntryId) {
        return recordedEntryId === unknownEntry
    }

    addNewChunks(Mp4Files, entryId, flavorId) {
        let chunksNameChained = _.reduce(Mp4Files, function (str, obj) {
            return str + obj.chunkName + " ";
        }, "");

        logger.info(`[${entryId}] Receiving new chunks: ${chunksNameChained}`);

        this.synchronizeAddNewChunks = this.synchronizeAddNewChunks.finally( ()=>
        {
            return this.handleMaxDuration(entryId)
                .then(()=>
                {
                    let recordingEntrySession = this.recordingList[entryId];
                    if (!recordingEntrySession)
                    {
                        logger.info(`[${entryId}] failed to add chunks. RecordedEntrySession doesn't exist in RecordingList. Maybe entry has passed the limit of max duration`);
                        return Q.resolve();
                    }
                    if (recordingEntrySession.mp4FilesQueue.length > maxAllowedMp4FilesQueueSize)
                    {
                        logger.fatal(`[${entryId}] recording queue reached [${maxAllowedMp4FilesQueueSize}] size. Dropping chunks!!!! CALL DEVELOPER`);
                    }
                    else
                    {
                        // update lastUpdateTime
                        recordingEntrySession.lastUpdateTime = new Date();

                        let Mp4FilesClone = _.clone(Mp4Files);
                        return recordingEntrySession.updateRecordingFiles(Mp4FilesClone, flavorId);
                    }
                })
                .catch((err)=>
                {
                    logger.error(`[${entryId}] failed adding new chunks. Error [${ErrorUtils.error2string(err)}]`);
                });
        });
        return this.synchronizeAddNewChunks;
    }

    handleRecordingEntries() {
        let now = new Date();
        // Function is not used anywhere in the project!

        function isRecordingFinish(recordingSession) {
            if (now - recordingSession.lastUpdateTime > recordingSession.recordingSessionDuration) {
                logger.info("[%s-%s] recording  didn't get new chunks for %d - stop recording", recordingSession.entryId, recordingSession.recordedEntryId, recordingSession.recordingSessionDuration);
                return true
            }
            if (recordingSession.playlistGenerator.MaxClipCountReached()) {
                logger.warn("[%s-%s] recording  has passed num of clips - stop recording", recordingSession.entryId, recordingSession.recordedEntryId);
                return true
            }

            if (recordingSession.entryObject.recordingStatus == kalturaTypes.KalturaRecordingStatus.PAUSED)
            {
                logger.info("[%s-%s] recording is paused", recordingSession.entryId, recordingSession.recordedEntryId);
                return false;
            }
            else if (recordingSession.entryObject.recordingStatus == kalturaTypes.KalturaRecordingStatus.STOPPED)
            {
                logger.info("[%s-%s] recording is stopped", recordingSession.entryId, recordingSession.recordedEntryId);
                return true;
            }

            return false
        }

        try {
            logger.debug("Check for end recording session");
            _.each(this.recordingList, (element, entryId) => {
                logger.debug("[%s] handleRecordingEntries-  now:[%s], duration: [%s], element: [%j]",
                    element.recordingSessionId, now, element.getDurationMSec() / 1000, element);

                if (this.isRecordedEntryUnknown(element.recordedEntryId)) {
                    this.replaceUnknownRecordedEntry(element)
                }
                if (isRecordingFinish(element)) {
                    this.stopRecording(entryId);
                }
            });
            logger.info("Available recording entry: %j", _.keys(this.recordingList));
        } catch (err) {
            logger.error("Error while trying to run handleRecordingEntries: %s ", ErrorUtils.error2string(err))
        }<commit_msg>SUP-22058: reset recording if per session is used (#621)<commit_after>/**
 * Created by ron.yadgar on 17/07/2016.
 */
const Q = require('q');
const util = require('util');
const events = require("events");
const config = require('../../common/Configuration');
const qio = require('q-io/fs');
const path = require('path');
const _ = require('underscore');
const glob = require('glob');
const ErrorUtils = require('../utils/error-utils');
const PlaylistGenerator = require('../playlistGenerator/PlaylistGenerator');
const persistenceFormat = require('../../common/PersistenceFormat.js');
const RecordingEntrySession = require('./RecordingEntrySession.js')
const logger = require('../../common/logger').getLogger("RecordingManager");
const mathUtils = require('./../utils/math-utils');
const configRecording = config.get("recording");
const kalturaTypes = require('../kaltura-client-lib/KalturaTypes');
const unknownEntry = "UNKNOWN";
const backendClient = require('../BackendClientFactory.js').getBackendClient();
const maxAllowedMp4FilesQueueSize = config.get("recording").recordingMaxAllowedMp4FilesQueueSize;
const TaskManager = require('../TaskManager').TaskManager;
const completedRecordingFolderPath = configRecording.completedRecordingFolderPath;
const clippingTaskTemplateName = config.get('newClippingTaskPushNotificationTemplateName');

class RecordingManager {
//todo what to do if we stop stream and start, but other machine has start to get it
//todo move all path join baseName etc to persistent format
// should also recording for primary ?
//change then style, and fix indentation
// todo Assumptions:
    /*
     1. No change configuration!
     2. recording window in not more that entry session window -  check it
     3. All chunks are comming at the same time
     recordingRootPath -
     recordingEntryPath
     recordingSessionPath
     recordingSessionSourcePath
     */
    constructor() {

        const recordingTimeInterval = configRecording.recordingTimeIntervalInSec * 1000;
        this.completedRecordingFolderPath = configRecording.completedRecordingFolderPath;
        //verify that recordingMaxDurationInHours is no longer that
        this.recordingMaxDurationInMs = config.get("recording").recordingMaxDurationInHours * 60 * 60 * 1000;
        this.recordingMinDurationInMS = config.get("recording").recordingMinDurationInSeconds * 1000;
        this.recordingList = {};

        logger.info("Initializing recording manager");
        TaskManager.register(kalturaTypes.KalturaEntryServerNodeType.LIVE_CLIPPING_TASK, clippingTaskTemplateName, this.handleNewLiveClippingTask);
        
        setInterval(()=> {
            this.handleRecordingEntries()
        }, recordingTimeInterval, this);
        this.synchronizeAddNewChunks=Q.resolve();
    }

    getRecordingEntrySession(entryObj, mp4FilesQueue = null) {
        let entryId = entryObj.entryId;
        if (mp4FilesQueue != null && mp4FilesQueue.length > 0) {
            logger.debug(`[${entryObj.entryId}-${entryObj.recordedEntryId}] getRecordingEntrySession called with mp4 files queue size [${mp4FilesQueue.length}]`);
        }
        if (!entryObj.recordedEntryId) {
            entryObj.recordedEntryId = unknownEntry;
            logger.warn(`[${entryObj.entryId}] Entry has no recordedEntryId`);
        }
        if (!this.recordingList[entryId]) {
            logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Create new recording session`);
            let recordingEntrySession = new RecordingEntrySession(entryObj, mp4FilesQueue);
            this.recordingList[entryId] = recordingEntrySession;
            return Q.resolve({session:recordingEntrySession, sessionStarted:false});
        }
        else if (this.recordingList[entryId] && this.recordingList[entryId].disableRecording) {
            let msg = `disableRecording is set. Entry won't be recording anymore`;
            logger.error(msg);
            delete this.recordingList[entryId];
            return Q.reject(msg);
        }    
        else if (this.recordingList[entryId].recordedEntryId != entryObj.recordedEntryId) {
            let msg = `[${entryId}-${this.recordingList[entryId].recordedEntryId}] unexpected state: session already exist for same liveEntry!!! [NEW ${entryObj.recordedEntryId}]`
            logger.error(msg);
            return Q.reject(msg);
        }
        else {
            logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Found a previous recording session, update time stamp`);
            this.recordingList[entryId].lastUpdateTime = new Date();
            return Q.resolve({session:this.recordingList[entryId], sessionStarted:true});
        }
    }

    startRecording(entryObj) {
        return Q.resolve().then(()=> {
            let startRecordingTime = new Date();
            logger.debug(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Start recording`);
            return this.stopSessionIfNeeded(entryObj)
                .then((mp4FilesQueue)=> {
                    return this.getRecordingEntrySession(entryObj, mp4FilesQueue)
                })
                .then((sessionObj)=> {
                      return this.handleFlavorsMismatch(entryObj, sessionObj)
                        .then((sessionObj) => {
                            if (!sessionObj.sessionStarted) {
                                return sessionObj.session.startRecording()
                                    .then(()=> {
                                        // Once start process is complete, release lock on files queue
                                        logger.info(`[startRecording] release lock mp4FilesQueueLock`)
                                        sessionObj.session.mp4FilesQueueLock = false;
                                        let startRecordingTimeDuration = new Date() - startRecordingTime;
                                        logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Start recording completed, time : ${startRecordingTimeDuration} ms`);
                                    })
                            } else {
                                logger.info(`[${entryObj.entryId}-${entryObj.recordedEntryId}] session running already, no reason to start it again.`);
                                if (sessionObj.session.mp4FilesQueueLock) {
                                    logger.warn(`[startRecording] new session was not started. Check log for reason. Release lock mp4FilesQueueLock`);
                                    sessionObj.session.mp4FilesQueueLock = false;
                                }
                            }
                        });
                });
        }).catch((err)=> {
            logger.error(`[${entryObj.entryId}-${entryObj.recordedEntryId}] Error while trying to start recording : ${ErrorUtils.error2string(err)}, delete session`);
            if (this.recordingList[entryObj.entryId]) {
                delete this.recordingList[entryObj.entryId];
                logger.warn(`[startRecording] new session was not started due to error.`);
            }
            return Q.reject(`[startRecording] start recording failed`);
        })
    }

    stopLive(entryId) {
        let recordingEntrySession = this.recordingList[entryId];
        if (!recordingEntrySession) {
            logger.warn("[%s] Can't updateServer recording, entry does not exist on recording list", entryId);
            return
        }
        if (this.isRecordedEntryUnknown(recordingEntrySession.recordedEntryId)) {
            this.replaceUnknownRecordedEntry(recordingEntrySession);
        }
        recordingEntrySession.updateEntryDuration(true)
    }

    splitRecording(entryId, reason) {
        let recordingSession = this.recordingList[entryId];
        if (!recordingSession) {
            logger.warn(`[${entryId}] Error splitting recording for entry, recordingEntrySession not found`);
            return Q.resolve();
        }
        if (recordingSession.disableRecording) {
            logger.warn(`[${entryId}] recording is disabled. Look for reason in log`);
            return Q.resolve();
        }
        return recordingSession.isUnlimitedRecordingDurationEnabled()
            .then((isEnabled)=> {
                if (!isEnabled) {
                    return this.stopRecording(entryId);
                }
                else {
                    let entryObject = recordingSession.entryObject;
                    logger.info(`[${entryId}-${recordingSession.entryObject.recordedEntryId}] Recording entry ${this.recordingList[entryId].recordedEntryId} has passed the limit of recording duration - ${this.recordingMaxDurationInMs / 1000} seconds. Split session`);
                    return recordingSession.splitRecordedEntry(recordingSession.entryObject, reason)
                        .then(()=> {
                            // Pass startRecording the queue of previous session in case there are files waiting to be written
                            return this.startRecording(entryObject);
                        })
                        .then(()=> {
                            logger.info(`[${entryId}-${entryObject.recordedEntryId}] Split process completed successfully. Replaced recordedEntryId: [${recordingSession.recordedEntryId}] by: [${entryObject.recordedEntryId}]`);
                        })
                        .catch((err)=> {
                            logger.error(`[${entryId}-${entryObject.recordedEntryId}] Error splitting recording for entry. Error: ${ErrorUtils.error2string(err)}`);
                        });
                }

            })
    }

    isRecordedEntryUnknown(recordedEntryId) {
        return recordedEntryId === unknownEntry
    }

    addNewChunks(Mp4Files, entryId, flavorId) {
        let chunksNameChained = _.reduce(Mp4Files, function (str, obj) {
            return str + obj.chunkName + " ";
        }, "");

        logger.info(`[${entryId}] Receiving new chunks: ${chunksNameChained}`);

        this.synchronizeAddNewChunks = this.synchronizeAddNewChunks.finally( ()=>
        {
            return this.handleMaxDuration(entryId)
                .then(()=>
                {
                    let recordingEntrySession = this.recordingList[entryId];
                    if (!recordingEntrySession)
                    {
                        logger.info(`[${entryId}] failed to add chunks. RecordedEntrySession doesn't exist in RecordingList. Maybe entry has passed the limit of max duration`);
                        return Q.resolve();
                    }
                    if (recordingEntrySession.mp4FilesQueue.length > maxAllowedMp4FilesQueueSize)
                    {
                        logger.fatal(`[${entryId}] recording queue reached [${maxAllowedMp4FilesQueueSize}] size. Dropping chunks!!!! CALL DEVELOPER`);
                    }
                    else
                    {
                        // update lastUpdateTime
                        recordingEntrySession.lastUpdateTime = new Date();

                        let Mp4FilesClone = _.clone(Mp4Files);
                        return recordingEntrySession.updateRecordingFiles(Mp4FilesClone, flavorId);
                    }
                })
                .catch((err)=>
                {
                    logger.error(`[${entryId}] failed adding new chunks. Error [${ErrorUtils.error2string(err)}]`);
                });
        });
        return this.synchronizeAddNewChunks;
    }

    handleRecordingEntries() {
        let now = new Date();
        // Function is not used anywhere in the project!

        function isRecordingFinish(recordingSession) {
            if (now - recordingSession.lastUpdateTime > recordingSession.recordingSessionDuration) {
                logger.info("[%s-%s] recording  didn't get new chunks for %d - stop recording", recordingSession.entryId, recordingSession.recordedEntryId, recordingSession.recordingSessionDuration);
                return true
            }
            if (recordingSession.playlistGenerator.MaxClipCountReached()) {
                logger.warn("[%s-%s] recording  has passed num of clips - stop recording", recordingSession.entryId, recordingSession.recordedEntryId);
                return true
            }

            if (recordingSession.entryObject.recordingStatus == kalturaTypes.KalturaRecordingStatus.PAUSED)
            {
                logger.info("[%s-%s] recording is paused", recordingSession.entryId, recordingSession.recordedEntryId);
                return false;
            }
            else if (recordingSession.entryObject.recordingStatus == kalturaTypes.KalturaRecordingStatus.STOPPED)
            {
                logger.info("[%s-%s] recording is stopped", recordingSession.entryId, recordingSession.recordedEntryId);
                return true;
            }

            return false
        }

        try {
            logger.debug("Check for end recording session");
            _.each(this.recordingList, (element, entryId) => {
                logger.debug("[%s] handleRecordingEntries-  now:[%s], duration: [%s], element: [%j]",
                    element.recordingSessionId, now, element.getDurationMSec() / 1000, element);

                if (this.isRecordedEntryUnknown(element.recordedEntryId)) {
                    this.replaceUnknownRecordedEntry(element)
                }
                if (isRecordingFinish(element)) {
                    const resetRecordedEntryId = (element.entryObject.recordStatus != kalturaTypes.KalturaRecordStatus.APPENDED);
                    this.stopRecording(entryId, resetRecordedEntryId);
                }
            });
            logger.info("Available recording entry: %j", _.keys(this.recordingList));
        } catch (err) {
            logger.error("Error while trying to run handleRecordingEntries: %s ", ErrorUtils.error2string(err))
        }
{'fraction_non_alphanumeric': 0.10335711440263644, 'fraction_numerical': 0.004852786730887625, 'mean_word_length': 2.1743878606736406, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1059270', 'n_tokens_mistral': 6923, 'n_tokens_neox': 6661, 'n_words': 1443}