/**
 * Manage psql operations like insert new user update old user, Add new event update eventstring
 * @author Semusi Technologies Private Limited (Developer Group)
 * @copyright Semusi Technologies PVT LTD
 * @Dated 2018-10-04 15:14:40
 */
 const Promise = require('bluebird'),
 common = require('./common'),
 logger = require('../../logger'),
 semusiConfig = require('./../config'),
 Sequelize = require('sequelize'),
 uuid = require('uuid'),
 cacheApi = require('./semusi.cache.js'),
 {QueueProviderFactory } = require('src_typescript_appice-core/dist/queue/index.js'),
 moment = require('moment'),
 sequelize = new Sequelize(semusiConfig.postgreSQL.productionCitus.database, semusiConfig.postgreSQL.productionCitus.user, semusiConfig.postgreSQL.productionCitus.password,{
     host: semusiConfig.postgreSQL.productionCitus.host,
     dialect: 'postgres',
     port:semusiConfig.postgreSQL.productionCitus.port,
     pool: {
         max: semusiConfig.postgreSQL.productionCitus.max,
         min: semusiConfig.postgreSQL.productionCitus.min,
         acquire: semusiConfig.postgreSQL.productionCitus.acquire,
         idle: semusiConfig.postgreSQL.productionCitus.idle
     },
 });
 
 sequelizePgQueue = new Sequelize(semusiConfig.PgQueue.database, semusiConfig.PgQueue.user, semusiConfig.PgQueue.pgQueuepassword, {
     host: semusiConfig.PgQueue.host,
     dialect: semusiConfig.PgQueue.dialect,
     port: semusiConfig.PgQueue.port,
     ssl: semusiConfig.PgQueue.sslconnection,
     pool: {
         max: semusiConfig.PgQueue.maxConnection,
         min: semusiConfig.PgQueue.minConnection,
         acquire: semusiConfig.PgQueue.acquire,
         idle: semusiConfig.PgQueue.idle
     },
 });
 
 
 
 /**
  *
  * @param appid is object type that has all required information
  */
 exports.setAppTables = function(appid){
    let queryCitusAppUser = `CREATE TABLE app_users_${appid} 
                (
                 inittime  int8 NULL,
                 fs        int4 NULL,
                 ls        int4 NULL,
                 did       VARCHAR(250) NULL,
                 cty       VARCHAR(100) NULL,
                 cc        VARCHAR(100) NULL,
                 country   VARCHAR(100) NULL,
                 region    VARCHAR(100) NULL,
                 active    BOOLEAN NULL,
                 sdkv      VARCHAR(50) NULL,
                 sdkiv     VARCHAR(50) NULL,
                 p         VARCHAR(50) NULL,
                 pv        VARCHAR(50) NULL,
                 av        VARCHAR(50) NULL,
                 ac        VARCHAR(50) NULL,
                 d         VARCHAR(150) NULL,
                 c         VARCHAR(50) NULL,
                 res       VARCHAR(100) NULL,
                 l         VARCHAR(50) NULL,
                 tz        int4 NULL,
                 _app_pack VARCHAR(250) NULL,
                 sc        int4 NULL,
                 tsd       int4 NULL,
                 lsd       int4 NULL,
                 ls_st     int4 NULL,
                 history   JSONB NULL,
                 gcmid     VARCHAR(1000) NULL,
                 fcmid     VARCHAR(1000) NULL,
                 playservice JSONB NULL,
                 permissions JSONB NULL,
                 competingapps JSONB NULL,
                 interests JSONB NULL,
                 user_info JSONB NULL,
                 _custom   JSONB NULL,
                 le        JSONB NULL,
                 el        JSONB NULL,
                 eventstring VARCHAR NULL,
                 createdat int4 NULL,
                 updatedat int4 NULL
             );`;
 
         // run citus query
         citusQuery(queryCitusAppUser)
             .then(res =>{
                 logger.info(`done in citus => ${appid} and res => ${res}`);
             })
             .catch(error =>{
                 logger.error(`error =>  ${error}`);
             });
             
         //AppUser Distribution
         let queryCitusDistributedAppUser = `SELECT create_distributed_table('app_users_${appid}', 'did');`;
         // run citus query
         pgQuery(queryCitusDistributedAppUser)
                     .then(res =>{
                         return resolve(true);
                     })
                     .catch(error =>{
                         logger.error(`error => ${error}`);  
                     });
 
         //App_events Table
         let queryCitusAppEvents = `CREATE TABLE app_events_${appid}
             (
             id        SERIAL PRIMARY KEY,
             name      VARCHAR NULL,
             list      JSONB NULL,
             clist     JSONB NULL,
             createdat int4 NULL,
             updatedat int4 NULL
         );`;
 
         // run citus query
         citusQuery(queryCitusAppEvents)
             .then(res =>{
                  logger.info(`done in citus => ${appid} and res => ${res}`);
             })
             .catch(error =>{
                 logger.error(`error => ${error}`);
             });
             
         //Events Table 
         let queryCitusEvents = `CREATE TABLE events_${appid}
             (
             appid     VARCHAR(150) NULL,
             did       VARCHAR(200) NULL,
             KEY       VARCHAR(200) NULL,
             sid       VARCHAR(200) NULL,
             rtime     INTEGER NULL,
             utime     INTEGER NULL,
             segment   JSONB NULL,
             context   JSONB NULL,
             eventtime INTEGER NULL,
             dt        date NULL
             );`;
 
         // run citus query
         citusQuery(queryCitusEvents)
             .then(res =>{
                 logger.info(`done in citus => ${appid} and res => ${res}`);
             })
             .catch(error =>{
                 logger.error(`error => ${error}`);
             });
 
 }
 
 /**
  *
  * @param userObj is object type that has all required information
  */
 exports.addNewUser = function(userObj, app){
     return new Promise( (resolve, reject) =>{
         try{
             // get custom variables
             let customs = [];
             for(let key of Object.keys(userObj)){
                 if(key.indexOf('_custom') >= 0){
                     customs.push({
                         [key] : user[key]
                     })
                 }
             }
 
             let history = userObj.history;
             // manage history fr
             history.forEach(function(his){
                 if(his.fr){
                     his.fr = escape(his.fr)
                 }
             });
 
             // data validation and default value
             let sc = (userObj.sc)? parseInt(userObj.sc) : 0;
             let tsd = (userObj.tsd)? parseInt(userObj.tsd) : 0;
             let lsd = (userObj.lsd)? parseInt(userObj.lsd) : 0;
             let ls_st = (userObj.ls_st)? parseInt(userObj.ls_st) : 0;
             history = JSON.stringify(history);
             let gcmid = (userObj.gcmid)? userObj.gcmid : '';
             let fcmid = (userObj.fcmid)? userObj.fcmid : '';
             let playservice = (userObj.playservice)? JSON.stringify([userObj.playservice]) : '[]';
             let permissions = (userObj.permissions)? JSON.stringify([userObj.permissions]) : '[]';
             let competingapps = (userObj.competingapps)? JSON.stringify([userObj.competingapps]) : '[]';
             let Interests = (userObj.Interests)? JSON.stringify([userObj.Interests]) : '[]';
             let user_info = (userObj.user_info)? JSON.stringify([userObj.user_info]) : '[]';
             let le = (userObj.le)? JSON.stringify(userObj.le) : '[]';
             let el = (userObj.el)? JSON.stringify(userObj.el) : '[]';
             let createdAt = (userObj.createdAt)? parseInt(userObj.createdAt) : 0;
             let updatedAt = (userObj.updatedAt)? parseInt(userObj.updatedAt) : 0;
 
             // insert into citus
             //let queryCitus = format('INSERT INTO app_users_'+app+' (inittime,fs,ls,did,cty,cc,country,region,active,sdkv,sdkiv,p,pv,av,ac,d,c,res,l,tz,_app_pack,sc,tsd,lsd,ls_st,history,gcmid,fcmid,playservice,permissions,competingapps,interests,user_info,_custom,le,el,eventstring,createdat,updatedat) VALUES %L', valArray);
             let query = ` (inittime,fs,ls,did,cty,cc,country,region,active,sdkv,sdkiv,p,pv,av,ac,d,c,res,l,tz,_app_pack,sc,tsd,lsd,ls_st,history,gcmid,fcmid,playservice,permissions,competingapps,interests,user_info,_custom,le,el,eventstring,createdat,updatedat) VALUES (
               `+parseInt(userObj.inittime)+`,
               `+parseInt(userObj.fs)+`,
               `+parseInt(userObj.ls)+`,
               '`+userObj.did+`',
               '`+userObj.cty+`',
               '`+userObj.cc+`',
               '`+userObj.country+`',
               '`+userObj.region+`',
               '`+userObj.active+`',
               '`+userObj.sdkv+`',
               '`+userObj.sdkiv+`',
               '`+userObj.p+`',
               '`+userObj.pv+`',
               '`+userObj.av+`',
               '`+userObj.ac+`',
               '`+userObj.d+`',
               '`+userObj.c+`',
               '`+userObj.res+`',
               '`+userObj.l+`',
               `+parseInt(userObj.tz)+`,
               '`+userObj._app_pack+`',
               `+sc+`,
               `+tsd+`,
               `+lsd+`,
               `+ls_st+`,
               '`+history+`',
               '`+gcmid+`',
               '`+fcmid+`',
               '`+playservice+`',
               '`+permissions+`',
               '`+competingapps+`',
               '`+Interests+`',
               '`+user_info+`',
               '`+JSON.stringify(customs)+`',
               '`+le+`',
               '`+el+`',
               '`+''+`',
               `+createdAt+`,
               `+updatedAt+`
             )`;
 
             let queryCitus = `INSERT INTO app_users_`+app+ query;
             citusQuery(queryCitus)
                 .then(res =>{
                     logger.info(`done in citus => ${appid} and res => ${res}`);
                 })
                 .catch(error =>{
                     logger.error(`error => ${error}`);
                 });
 
             // insert into PG
             if(common.config.azureCitus.indexOf(app) >= 0) {
                 let queryPG = `INSERT INTO app_users_`+app +` `+ query;
                 pgQuery(queryPG)
                     .then(res =>{
                          
                         return resolve(true);
                     })
                     .catch(error =>{
                         logger.error(`error=> ${error}`);
                     });
             }
 
 
         }
         catch(e){
             return reject(e)
         }
     });
 }
 
 /**
  *
  * @param eventObj is object type that has all required information
  */
 exports.addNewEvent = function(eventObj, app){
     return new Promise( (resolve, reject) =>{
         try{
             // insert into citus
             let queryCitus = "INSERT INTO app_events_"+app+" (name, list, clist, createdat, updatedat) VALUES('"+eventObj.key+"','"+JSON.stringify(eventObj.list)+"','"+JSON.stringify(eventObj.clist)+"',"+parseInt(moment().valueOf()/1000)+", "+parseInt(moment().valueOf()/1000)+")";
             citusQuery(queryCitus)
                 .then(res =>{
                     if(res && res.rowCount){
                         let sQueryCitus = "SELECT id FROM app_events_"+app+" WHERE name='"+eventObj.key+"'";
                         citusQuery(sQueryCitus)
                           .then(res =>{
                               if(res && res.rowCount){
                                 // update event information into cache
                                 cacheApi.getKey('app_evnets'+app, function(error, eventData){
                                     if(!error && (eventData != null && eventData != '')) {
                                         // parse event data string
                                         eventData = JSON.parse(eventData);
                                         eventData[eventObj.key] = res.rows[0].id;
                                         // update event information into cache
                                         cacheApi.setKey('app_evnets'+app, JSON.stringify(eventData));
                                          
                                     }
                                     else{
                                         let eventData = {
                                             [eventObj.key]:res.rows[0].id
                                         }
                                         cacheApi.setKey('app_evnets'+app, JSON.stringify(eventData));
                                          
                                     }
                                 });
                               }
                           })
                           .catch(erorr =>{
                             logger.error(`error=> ${error}`);
                           });
                     }
                 })
                 .catch(error =>{
                     logger.error(`error=> ${error}`);
                 });
 
             if(common.config.azureCitus.indexOf(app) >= 0) {
                 // insert into PG
                 let queryPG = "INSERT INTO app_events_"+app+" (name, list, clist, createdat, updatedat) VALUES('"+eventObj.key+"','"+JSON.stringify(eventObj.list)+"','"+JSON.stringify(eventObj.clist)+"',"+parseInt(moment().valueOf()/1000)+", "+parseInt(moment().valueOf()/1000)+")";
                 pgQuery(queryPG)
                     .then(res =>{
                         return resolve(true);
                     })
                     .catch(error =>{
                         logger.error(`error=> ${error}`); 
                     });
             }
         }
         catch(e){
             return reject(e)
         }
     });
 }
 
 /**
 * Insert event into citus and pg
 */
 exports.addEvent = function(e, appid, eventUtils){
       // validate session id and add blank if does not exits
       if(!e.sid || e.sid == undefined){
           e.sid = 'null';
       }
 
       let segment = JSON.stringify({});
       // validate event segment and add blank if does not exits
       if(e.segment){
           segment = JSON.stringify(e.segment)
           // removing special char that break pg query
           segment = segment.replace(/[`~%!#$%^*()\?;']/gi, "");
       }
 
       if(!e.utime || e.utime == undefined){
           e.utime = e.rtime;
       }
 
       let context = "";
       if(e.context){
         context = JSON.stringify(e.context);
         // removing special char that break pg query
         context = context.replace(/[`~%!#$%^*()\?;']/gi, "");
       }
 
       // insert campaign information into segment
       if(e.campid){
           segment['campid'] = e.campid;
           // set campaign type
           if(e.camp_type){
             segment['camp_type'] = e.camp_type;
           }
       }
 
       let sql = `INSERT INTO events_` +appid+ ` (appid, did, KEY, sid, rtime, utime, segment, context, eventtime, dt) VALUES ('` +e.appid+ `','` +e.did+ `','` +e.key+ `','` +e.sid+ `','` +e.rtime+ `','` +e.utime+ `','` +segment+ `','` +context+ `','` +e.eventTime+ `', to_timestamp(`+e.eventTime+`)::date)`;
 
       // insert data into citus pg
       let sqlCitus = `INSERT INTO events_` +appid+ ` (appid, did, KEY, sid, rtime, utime, segment, context, eventtime, dt) VALUES ('` +e.appid+ `','` +e.did+ `','` +e.key+ `','` +e.sid+ `','` +e.rtime+ `','` +e.utime+ `','` +segment+ `','` +context+ `','` +e.eventTime+ `', to_timestamp(`+e.eventTime+`)::date)`;
       citusQuery(sqlCitus)
         .then((result) =>{
             logger.info(`result=> ${result}`);
         })
         .catch((error) =>{
             logger.error(`CITUS INSERT query error => ${error}`);
             
            eventUtils.errorQueue(sqlCitus, 'CITUS', '', 'processPG', function(status){ });
         });
 }
 
 /**
  *
  * @param userObj is object type that has all required information
  */
 exports.updateUser = function(eventObj, app){
     return new Promise( (resolve, reject) =>{
         try{
             let userObj = eventObj.user;
             let query = " SET did = '" + eventObj.did + "'";
             // get custom variables
             let customs = [];
             for(let key of Object.keys(userObj)){
                 if(key.indexOf('_custom') >= 0){
                     customs.push({
                         [key] : user[key]
                     })
                 }
             }
 
             // update custom information
             if(customs.length > 0){
                 query += ", _custom = '" + JSON.stringify(customs) + "'"
             }
 
             // update last event information
             if(userObj.le){
                 query += ", le = '" + JSON.stringify(userObj.le) + "'"
             }
 
             // update session count information
             if(userObj.sc){
                 query += ", sc = '" + parseInt(userObj.sc) + "'"
             }
 
             // update last session duration information
             if(userObj.lsd){
                 query += ", lsd = '" + parseInt(userObj.lsd) + "'"
             }
 
             // update last session duration time
             if(userObj.ls_st){
                 query += ", ls_st = '" + parseInt(userObj.ls_st) + "'"
             }
 
             // update competing apps
             if(userObj.competingapps){
                 query += ", competingapps = '" + JSON.stringify(userObj.competingapps) + "'"
             }
 
             // update User interest
             if(userObj.Interests){
                 query += ", interests = '" + JSON.stringify(userObj.Interests) + "'"
             }
 
             // update User permissions
             if(userObj.permissions){
                 query += ", permissions = '" + JSON.stringify(userObj.permissions) + "'"
             }
 
             // update User playservice
             if(userObj.playservice){
                 query += ", playservice = '" + JSON.stringify(userObj.playservice) + "'"
             }
 
             // update User user_info
             if(userObj.user_info){
                 query += ", user_info = '" + JSON.stringify(userObj.user_info) + "'"
             }
 
             // update GCM token
             if(userObj.gcmid){
                 query += ", gcmid = '" + userObj.gcmid + "'"
             }
 
             // update FCM token
             if(userObj.fcmid){
                 query += ", fcmid = '" + userObj.fcmid + "'"
             }
 
             query += " WHERE did = '"+ eventObj.did +"'";
 
             // citus query
             let queryCitus = "UPDATE app_users_" + app + query;
             citusQuery(queryCitus)
                 .then(res =>{
                     logger.info(`error=> ${res}`);
                 })
                 .catch(error =>{
                     logger.error(`error=> ${error}`);
                 });
 
             if(common.config.azureCitus.indexOf(app) >= 0) {
                 // PG query
                 let queryPG = "UPDATE app_users_" + app + " " + query;
                 pgQuery(queryPG)
                     .then(res =>{
                         logger.info(`error=> ${res}`);
                         return resolve(true);
                     })
                     .catch(error =>{
                         logger.error(`error=> ${error}`);
                     });
             }
         }
         catch(e){
             return reject(e)
         }
     });
 }
 
 /**
  *
  * @param userObj is object type that has all required information
  */
 
 exports.updateUserEventString = function(userObj, app){
     return new Promise( (resolve, reject) =>{
         try{
             // get event string from cache
             cacheApi.getKey('eventString'+userObj.appid+userObj.did+userObj.sid, function(error, eventStringData){
                  
                 if(!error && (eventStringData != null && eventStringData != '')) {
                     // citus query
                     let queryCitus = "SELECT eventstring FROM app_users_" + app + " WHERE did = '" + userObj.did + "'";
                     citusQuery(queryCitus)
                         .then(res => {
                             let updateQuery = '';
                             if(res && res.rowCount && res.rows[0].eventstring != ''){
                                 if(res.rows[0].eventstring.indexOf('#SS'+userObj.sid) > 0){
                                       // get database substring position
                                       let [dstart, dstrlen] = getSubStringPosition(res.rows[0].eventstring, '#SS'+userObj.sid, '#SE'+userObj.sid);
                                       let substrDb = res.rows[0].eventstring.substr(dstart, dstrlen);
                                       substrDb += '#SE'+userObj.sid;
 
                                       // get cache substring position
                                       let [cstart, cstrlen] = getSubStringPosition(eventStringData, '#SS'+userObj.sid, '#SE'+userObj.sid);
                                       let substrCache = eventStringData.substr(cstart, cstrlen);
                                       substrCache += '#SE'+userObj.sid;
 
                                        
                                       if(substrCache && substrDb){
                                           res.rows[0].eventstring = res.rows[0].eventstring.replace(substrDb, substrCache);
                                       }
                                 }
                                 else {
                                     res.rows[0].eventstring += eventStringData;
                                 }
                                 updateQuery = "UPDATE app_users_" + app + " SET eventstring = '" + res.rows[0].eventstring + "' WHERE did = '" + userObj.did + "'";
                             }
                             else {
                                 updateQuery = "UPDATE app_users_" + app + " SET eventstring = '" + eventStringData + "' WHERE did = '" + userObj.did + "'";
                             }
                              
                             // update query
                             citusQuery(updateQuery)
                         })
                         .catch(error => {
                             logger.error(`error=> ${error}`);
                         });
 
                     if(common.config.azureCitus.indexOf(app) >= 0) {
                         // PG query
                         let queryPG = "SELECT eventstring FROM app_users_" + app + " WHERE did = '" + userObj.did + "'";
                         pgQuery(queryPG)
                             .then(res => {
                                 let updateQuery = '';
                                 if(res && res.rowCount && res.rows[0].eventstring != ''){
                                     if(res.rows[0].eventstring.indexOf('#SS'+userObj.sid) > 0){
                                         // get database substring position
                                         let [dstart, dstrlen] = getSubStringPosition(res.rows[0].eventstring, '#SS'+userObj.sid, '#SE'+userObj.sid);
                                         let substrDb = res.rows[0].eventstring.substr(dstart, dstrlen);
                                         substrDb += '#SE'+userObj.sid;
 
                                         // get cache substring position
                                         let [cstart, cstrlen] = getSubStringPosition(eventStringData, '#SS'+userObj.sid, '#SE'+userObj.sid);
                                         let substrCache = eventStringData.substr(cstart, cstrlen);
                                         substrCache += '#SE'+userObj.sid;
 
                                          
                                         if(substrCache && substrDb){
                                             res.rows[0].eventstring = res.rows[0].eventstring.replace(substrDb, substrCache);
                                         }
 
                                     }
                                     else {
                                         res.rows[0].eventstring += eventStringData;
                                     }
                                     updateQuery = "UPDATE app_users_" + app + " SET eventstring = '" + res.rows[0].eventstring + "' WHERE did = '" + userObj.did + "'";
                                 }
                                 else {
                                     updateQuery = "UPDATE app_users_" + app + " SET eventstring = '" + eventStringData + "' WHERE did = '" + userObj.did + "'";
                                 }
 
                                 // update query
                                 pgQuery(updateQuery)
                             })
                             .catch(error => {
                                 logger.error(`error=> ${error}`);
                             });
                     }
                 }
             });
         }
         catch(e){
             return reject(e)
         }
     });
 }
 
 // get substring start and end position
 let getSubStringPosition = function(str, substrstart, substrend){
     let start = parseInt(str.indexOf(substrstart))+(substrstart).length;
     let strlen = parseInt(str.indexOf(substrend)) - start;
     return [start, strlen];
 }
 
 exports.executeQuery = function(query){
     return citusQuery(query);
 }
 
  // parameterized Query 
  exports.executeParameterizedQuery = function( parameterizedquery ){
    return citusParameterizedQuery(parameterizedquery)
 }

 // run citus query
 let citusQuery = function(query){
     return new Promise((resolve, reject) =>{
         try{
             logger.info(`query => ${query}`) 
             sequelize.query(query)
               .then(rows => {
                    
                   return resolve(rows);
               })
               .catch((error) =>{
                 logger.error(`error=> ${error}`);
                   return reject(error);
               })
         }
         catch(e) {
             logger.error(`error=> ${e.message}`);
             return reject(e.message);
         }
     });
 }

  // parameterized Query executing 
  let citusParameterizedQuery = function(parameterizedquery) {

  
    return new Promise((resolve, reject) => {
      try {
            // executing query with binding parameters 
        sequelize.query(parameterizedquery.text, {
            type: Sequelize.QueryTypes.SELECT,
            replacements: parameterizedquery.values
        }).then(rows => {
            logger.info(`Parameterized Query Executed sucessfully`)
          return resolve(rows);
        }).catch((error) => {
            logger.info(`Error While Executing Parameterized Query${error}`)
          return reject(error);
        })
      } catch (e) {
        logger.info(`Error While citusParameterizedQuery${e}`)
        return reject(e.message);
      }
    });
  }

 
 // run pg query
 let pgQuery = function(query){
     return new Promise((resolve, reject) =>{
         try{
             // pg query
             common.pgdbClient.query(query, function(error, res){
                 if(error){
                     logger.error(`error in pg => ${error}`);
                     logger.info(`queryPG => ${query}`);
                     return reject(error);
                 }
                 else {
                     return resolve(res)
                 }
             });
         }
         catch(e) {
             return reject(e.message);
         }
     });
 }
 
  // function for insert cmp data into pg table
  exports.addCampaignSchedule = function(appid, apikey, appKey, cid, jobid, sd, ed, type, time, s) {
    let sJson = "";
    if (s !== undefined && s !== "") {
        sJson = `,"s":${JSON.stringify(s)}`;
    }
    let datatypeset = `{"appid": "${appid}","apiKey": "${apikey}","appKey": "${appKey}","type": "${type}"${sJson}}`;
    logger.info(`Data set =>" ${datatypeset}`);
     logger.info(`sd" ${sd}`);
     let momentSd = moment.unix(sd/1000);
     let momentEd = moment.unix(ed/1000);


     let queryCITUSCampaignSchedule = "INSERT INTO campaign_schedule (cid, payload, jobid, start_time, end_time, created_at, expire_at) VALUES ('" + cid + "', '" + datatypeset + "','"+ jobid +"', '"+ momentSd.format('YYYY-MM-DD HH:mm:ss') +"', '"+momentEd.format('YYYY-MM-DD HH:mm:ss')+"', now(), now())";
     let queryCITUSCampaignScheduleStatus = "insert into campaign_schedule_status_"+appid+" (cid, jobid, status, created_at) values ('" + cid + "','"+ jobid +"', 'active', now())";
     pgqueueQuery(queryCITUSCampaignSchedule)
         .then(res => {
            pgqueueQuery(queryCITUSCampaignScheduleStatus)
                 .then(res => {
                     logger.info(`Campaign Scheule Status =>" ${res}`)
                 })
                 .catch(error => {
                     logger.error(`Campaign error =>", ${error}`)
                 })
             logger.info(`Campaign Schedule =>" ${res}`)
         })
         .catch(error => {
             logger.error(`Campaign error =>" ${error}`)
         })
 }

 /**
 * Writes uploads CSV file data to pg queue for processing.
 * 
 * @param {object} data - The CSV data to be written to the queue.
 * @param {string} data.appid - The ID of the application.
 * This function updates the information of csv file in two tables 
 * 1. segmentupload_APPID, sets its payload and jobID
 * 2.segmentuploadstatus_APPID , sets the status as unprocessed as default with the same jobId
 * @returns {void}
 */
 exports.writeCsvDataToPgQueue = function(data){

        // Generate a unique job ID using UUID
    const jobid = uuid.v4();                                // generate UUID
    const status = 'unprocessed';                            // default status as unprocessed

    //define Table name 
    const segmentTableName = 'segmentupload_'+ data.appid;          
    const segmentStatusTableName = 'segmentupload_status_'+ data.appid
    

     // Create parameterized queries for inserting data into tables
    const segmentUploadInsertQuery = `INSERT INTO ${segmentTableName} (jobid, payload, created_at) VALUES ('${jobid}', '${JSON.stringify(data)}', current_timestamp AT TIME ZONE 'IST')`;
    const segmentUploadStatusInsertOrUpdateQuery = `INSERT INTO ${segmentStatusTableName} (jobid, status, created_at) VALUES ('${jobid}', '${status}', current_timestamp AT TIME ZONE 'IST')`;
    
    
    // Insert into segmentupload_APPID
    pgqueueQuery(segmentUploadInsertQuery)
    .then((insertResult) => {
        logger.info(`segmentUploadInsertQuery Inserted`)
        
        // Insert into segmentupload_APPID
        pgqueueQuery(segmentUploadStatusInsertOrUpdateQuery)
        .then((statusResult) => {
            logger.info(`segmentUploadStatusInsertOrUpdateQuery Inserted`)
        })
        .catch((statusError) => {
            logger.error(`segmentUploadStatusInsertOrUpdateQuery error =>", ${statusError}`)
        });
    })
    .catch((insertError) => {
        logger.error(`segmentUploadInsertQuery error =>", ${insertError}`)
    });
};

/**
 * Updates the status of a campaign to 'Processed' in campaign_schedule_appid upon deactivation in the database.
 * 
 * @param {string} appid - The _Id of the App.
 * @param {string} cid - The campaign ID of the schedule to be updated.
 * @returns {void}
 */

 exports.updateCampaignSchedule = function(appid, cid){

        // Construct the SQL query to update the status of the campaign schedule
     let updateQuery = `UPDATE campaign_schedule_status_${appid} set status= 'Processed' where cid='${cid}'`;

         // Execute the query using pgqueueQuery function, which returns a promise
     pgqueueQuery(updateQuery)
         .then(res => {
                // If the query is successful
             logger.info(`campaign_schedule_status_ Updated Successfully During Deactivation Process =>" ${res}`)
         })
         .catch(error => {
              logger.error(`Error during campaign_schedule_status_ Updated During Deactivation Process =>" ${error}`)
         })
 }
 
 //pgqueue code for data save in pg queue
 exports.runQuery = function(query){
    return pgqueueQuery(query);
   }
   
/**
 * Enqueues a payload into a queue for processing.
 * 
 * @param {Object} payload - The  payload to be enqueued.
 * @param {string} appid - The app_Id associated with the payload.
 * @param {Object} qconfig - queue config .
 * 
 * @returns {} - A boolean  that resolves when the payload is successfully enqueued.
 */
exports.enqueuePayload = async function (payload, appid, qconfig) {


    // Create an instance of QueueProviderFactory to get the queue provider
    const queue = new QueueProviderFactory()

    // Create a queue provider instance with the given appid and config
    const queueProvider =await queue.createProvider(appid, qconfig)

    // Wrap the payload in an array to match the expected format
    const payloads = [{ payload }]

    // default partition number
    const partitionNumber = -1;

    try {
        // Enqueue the payloads into the queue using the created provider
        const enqueueResult =  await queueProvider.enqueue(payloads, partitionNumber);
        
        // Log a message indicating the payload was successfully enqueued
        logger.info(`Enqueued successfully: ${enqueueResult}`);
    } catch (error) {

        // Log an error message if something goes wrong during enqueuing
        logger.error(`Failed to enqueue payload: ${error}`);
    }

};   

   // run PG Queue query
   let pgqueueQuery = function(query) {
    return new Promise((resolve, reject) => {
         try {
            logger.info(`=============== query try block ========>" ${query}`)
             // EXECUTE MAIN USER QUERY
             sequelizePgQueue.query(query)
                 .then(rows => {
                    logger.info(`=============== query try rows ========>" ${rows}`)
                    return resolve(rows);
                 })
                 .catch((error) => {
                    logger.info(`=============== query try error ========>" ${error}`)
                    return reject(error);
                 });
         } catch (e) {
            return reject(e.message);
         }
     });
   }
 //End of Document