#!/usr/bin/perl
use DBI;
use strict;
use POSIX;
#use ESL;
use Switch;
use Time::HiRes;
use WWW::Curl::Easy;
use JSON;
use Config::Tiny;
use Data::Dumper;
use Cwd;
use URI;
use threads;
use Thread::Semaphore;
use threads::shared;
use POSIX qw(strftime);


# ----------------------------------------
# Need to Install Config-Tiny Perl Module
# ----------------------------------------
my $app_folder = getcwd;
chdir "/var/www/html/PortalSMS";
my $filename = "./SmsInterface.ini";
my $config = Config::Tiny->read( $filename, 'utf8' );
my $daemon = 0;
my $dbh;
my $logfile = "./SmsInterface.log";
my $limiteDiario= 0;
my $enableLimites = 0;
my $threadCount = 10;
my $site = 1;
my $route = 4;

if( $config->{core}{daemon} eq "Yes" ) {
   $daemon = 1;
}

# ---------------------------
# Is program already running
# ---------------------------
writeToLog( $logfile, "", 1 );
my $progname = "SmsInterface.pl";
my $pidx = `ps -C $progname -o pid=`;
#my $pidx = `ps -C $progname -o command=`;
if( $pidx != "" ) {
#  writeToLog( $logfile, sprintf( "smsMasivos.pl daemon PID: %d IS RUNNING", $pidx ),0 );
   if( $pidx != getpid() ) {
      writeToLog( $logfile, sprintf( "SmsInterface PID: %d is ALREADY RUNNING", $pidx ) ,0 );
      exit( 0 );
   }
   else {
      writeToLog( $logfile, sprintf( "SmsInterface.pl daemon PID: %d IS NOT RUNNING", $pidx ),0 ); 
   }
}
else {
   writeToLog( $logfile, sprintf( "SmsInterface.pl daemon PID: %d IS NOT RUNNING", $pidx ),0 ); 
}

if( $daemon == 1 ) {
   fork and exit;
   POSIX::setsid();
   fork and exit;
   umask 0;

   writeToLog( $logfile, sprintf( "Start 'SmsInterface' daemon PID: %d", getpid() ),0 );
   if( open(my $fp, '>', "./SmsInterface.pid") ) {
      printf $fp "%d", getpid();    
      close $fp;
   }

   close STDIN;
   close STDOUT;
   close STDERR;
}
else {
   writeToLog( $logfile, sprintf( "Start 'SmsInterface' as Foreground" ) ,0 );
}   

$SIG{TERM} = sub {
   if( $dbh ) {
      $dbh->disconnect();
   }   
   writeToLog( $logfile, sprintf( "Terminate 'SmsInterface' daemon PID: %d", getpid() ) ,0 );
   unlink("./SmsInterface.pid");
   die "Program Terminate!" 
};

# --------------------------------
# MariaDB
# --------------------------------
my $driver = "mysql"; 
my $database = "TestSMS-2";
my $dsn = "DBI:$driver:database=smpp:127.0.0.1";
my $userid = "root";
my $password = "dialer19";
my $dbhost = "10.0.66.119";

my $errorCount = 300;
my $debug = 0;
my $dialerName = "";
my $error = "";


# --------------------------------
# MariaDB Configuration
# --------------------------------
if( $config->{database}{name} ne "" ) {
   $database = $config->{database}{name};
}
if( $config->{database}{host} ne "" ) {
   $dbhost = $config->{database}{host};
}
if( $config->{database}{host} ne "" ) {
   $userid = $config->{database}{user};
}
if( $config->{database}{password} ne "" ) {
   $password = $config->{database}{password};
}

$dsn = "DBI:$driver:database=$database:$dbhost";

if( $config->{core}{debug} eq "Yes" ) {
   $debug = 1;
}

if( $config->{core}{limiteDiario} ne "" ) {
   $limiteDiario = $config->{core}{limiteDiario}; 
}

if( $config->{core}{enableLimites} eq "Yes" ) {
   $enableLimites = 1;
}

if( $config->{core}{threads} ne "" ) {
   $threadCount = $config->{core}{threads}; 
}

if( $config->{core}{site} ne "" ) {
   $site = $config->{core}{site}; 
}

if( $config->{core}{ruta} ne "" ) {
   $route = $config->{core}{ruta}; 
}

writeToLog( $logfile, sprintf( "Max Threads: %d", $threadCount ) );


# --------------------------------
# Variables
# --------------------------------
my $sth;
my $rs;
my $decoded;
my $forever = 1;
my $dbConnect = 1;
my $i;
my @thr;
my @Used:shared;
my $globalThreadCount:shared = 0;
my $semaphore = Thread::Semaphore->new();


# --------------------------------
# Connect to MariaDB
# --------------------------------
while( $dbConnect ) {

   if( ( $dbh = DBI->connect($dsn, $userid, $password ) ) == 0 ) {
      writeToLog( $logfile, sprintf( "Cannot connect to MariaDB: $DBI::errstr", 0 ) );
      Time::HiRes::sleep(10);
   }
   else {
      $dbConnect = 0;
      writeToLog( $logfile, sprintf( "Connected to MariaDB @%s", $dbhost ) );
      last;
   }
}

for( $i= 0; $i < $threadCount; $i++ ) {
   $Used[$i] = 0;
}

=begin
# ------------------------------------
# Cuantos mensajes colgados quedaron
# ------------------------------------
$sth = $dbh->prepare("select count(id) as total from outbound_request where pending=200 and received > DATE_SUB(now(),INTERVAL 5 MINUTE) and ruta=$route;");
if ( !$sth->err ) {
   if( $sth->execute() ) {
      if (my @row = $sth->fetchrow_array()) {
         my ($total ) = @row;
         writeToLog( $logfile, sprintf( "Total messages to recover: %d", $total ), 0 );
      }
   }
   $sth->finish();
}         

# ---------------------
# Marco para reenviar
# ---------------------
$sth = $dbh->prepare("update outbound_request set pending=1 where pending=200 and received > DATE_SUB(now(),INTERVAL 5 MINUTE);");
if ( !$sth->err ) {
   $sth->execute();
   $sth->finish();
}     
=cut        


while( $forever ) {

   if ( not $dbh->ping ) {
      if( ( $dbh = DBI->connect($dsn, $userid, $password ) ) == 0 ) {
         writeToLog( $logfile, sprintf( "Cannot connect to MariaDB $DBI::errstr", 0 ) );
         Time::HiRes::sleep(10);
         next;
      }
   }

   my $message_id = 0;
   my $sql = "";
   my $idPort;
   my $descripcion;
   my $max_daily;
   my $por_hora;
   my $active;
   my $fecha;
   my $ip;
   my $port;
   my $gw;
   my $slot;
   my $auth_user;
   my $auth_pswd;
   my $plan;
   my $user_id;


   # -------------------------------------
   # Me fijo si hay mensajes para enviar
   # -------------------------------------
#   $sql = sprintf( "select sql_no_cache id, received, sender_id, recipient, message, customer_id from outbound_request where pending=1 order by received;" );
   $sql = sprintf( "select sql_no_cache id, received, sender_id, recipient, message, customer_id from outbound_request where pending=1 order by received limit 1;" );
   $sth = $dbh->prepare($sql);
   if ( $sth->err ) {
      writeToLog( $logfile, sprintf( "DBI prepare Error: Count: $errorCount => $sth->err : $sth->errstr" ), 0 );
      Time::HiRes::sleep(10);
      next;
   }
   my $foundSms = 0;
   my $foundThread = 0;
   if( $sth->execute() ) {

      # -------------------------------------------------------
      # Por cada campaign verifico cuantos llamados debo hacer
      # -------------------------------------------------------
      my $sqlHandle;
      my $countTest = 0;

      while (my @row = $sth->fetchrow_array()) {

         my ($id, $received_date, $sender_id, $recipient, $message, $customer_id ) = @row;
         $foundSms = 1;

         my $message_text = URI->new($message);
         my $auth = "";

         # ------------------------------------------------
         # Verifico si hay puertos activos disponibles?
         # ------------------------------------------------
         my $testFlag = 0;
         my $found = 0;

         $semaphore->down();
         $sql = sprintf( "select sql_no_cache gateway_ports.id as id_port, descripcion, maximo_diarios, mensajes_por_hora, enabled, fecha_ultimo, ip, gateway_ports.port as gwport,gw,slot,api_user,api_password,planes.id as plan_id from gateway_ports left join planes on planes.id=gateway_ports.plan left join gateways on gateways.id=gateway_ports.gw where enabled=1 and gateways.active=1 and date_add( fecha_ultimo, interval 3600/mensajes_por_hora second ) < now() and gateway_ports.status=1 and registered=1 order by fecha_ultimo;" );
         $sqlHandle = $dbh->prepare($sql);
         if ( $sqlHandle->err ) {
            writeToLog( $logfile, sprintf( "DBI prepare Error: $sqlHandle->err : $sqlHandle->errstr" ), 0 );
            Time::HiRes::sleep(2);
            next;
         }
         if( $sqlHandle->execute() ) {

            $semaphore->up();
            while (my @portRow = $sqlHandle->fetchrow_array()) {

               ($idPort, $descripcion, $max_daily, $por_hora, $active, $fecha, $ip, $port, $gw, $slot, $auth_user, $auth_pswd, $plan ) = @portRow;
#               printf( "Puerto: %d, %s, %s, %d:%s Max Daily: %s, Max Hora: %d\n", $idPort, $descripcion, $ip, $port, $slot, $max_daily, $por_hora );
               $auth = $auth_user . ":" . $auth_pswd;

               # -----------
               # TESTING
               # -----------
               my $testing = 1;
               if( $testing ) {
                  $max_daily = 300;
                  $por_hora = 45;
                  $active = 1;
#                 $fecha = date("Y-m-d");
                  $ip = "10.0.66.104";
                  $port = 23;
                  $gw = 1;
                  $slot = 1;
                  $auth = "admin:pepito20";
                  $found = 0;
                  $plan = 2;
                  $descripcion = "Dinstar-1";
                  
                  my $code = 10000 + int(rand(99999 - 10000));
                  $message = "Tu codigo es " . $code;
                  $message = "Tu clave es " . $code;

                  ++$countTest;
                  if( $countTest == 19 ) {
                     $recipient = "1154080199";
                     $testFlag = 1;
                  }
                  if( $countTest == 20 ) {
                     $recipient = "1144255139";
                     $message = "Tu clave es " . $code;
                     $countTest = 0;
                     $testFlag = 1;
                  }
               }
               # --------------
               # TESTING END
               # --------------

               # ------------------------------------------------------------------
               # Me fijo si tengo registro de estadisticas del puerto/slot activo
               # ------------------------------------------------------------------
               $sql = sprintf( "select sql_no_cache enviados, disabled_day from port_statistics where gateway=%d and port=%s and slot='%s' and date(fecha)=date(now());", $gw, $port, $slot, $max_daily );
               my $sqlPort = $dbh->prepare($sql);
               if ( $sqlPort->err ) {
                  writeToLog( $logfile, sprintf( "DBI prepare Error: $sqlPort->err : $sqlPort->errstr" ), 0 );
                  Time::HiRes::sleep(2);
                  next;
               }
               if( $sqlPort->execute() ) {

                  if( $sqlPort->rows() > 0 ) {
                     if (my @statistics = $sqlPort->fetchrow_array()) {
                        my ($enviados, $disabled_day ) = @statistics;

                        if( $enviados < $max_daily and $disabled_day == 0 ) {
#                          printf( "FOUND: %s, %d, %d, %d, %s, %s, %d, %d, %s, %s\n", $descripcion, $max_daily, $por_hora, $active, $fecha, $ip, $port, $gw, $slot, $auth );
                           $found = 1;
                           last;
                        }
                     }
                  }                                                                                                                                                              
                  else {
                     $sqlPort->finish();
                     $sql = sprintf( "insert into `port_statistics` (`id`,`fecha`,`gateway`,`port`,`slot`,`enviados`,`plan`,`status`,`enabled`) VALUES (0,now(),%d,%d,'%s',%d, 0,1,1);", int($gw), int( $port ), $slot, int($plan) );
                     $sqlPort = $dbh->prepare($sql);
                     if ( $sqlPort->err ) {
                        writeToLog( $logfile, sprintf( "DBI prepare Error: $sqlPort->err : $sqlPort->errstr" ), 0 );
                        Time::HiRes::sleep(2);
                        next;
                     }
                     if( $sqlPort->execute() ) {
                        $found = 1;
#                       printf( "FOUND: %s, %d, %d, %d, %s, %s, %d, %d, %s, %s\n", $descripcion, $max_daily, $por_hora, $active, $fecha, $ip, $port, $gw, $slot, $auth );
                        last;
                     }
                  }
               }
               $sqlPort->finish();
            }

            # -------------------------------------------------------
            # Si estan todos disponibles me fijo cual usar cambiando
            # -------------------------------------------------------
            if( $found ) {
               $sql = sprintf( "update port_statistics set enviados=enviados+1, fecha=now() where gateway=%d and port=%s and slot='%s';", $gw, $port, $slot );
               my $sqlStat = $dbh->prepare($sql);
               if ( $sqlStat->err ) {
                  writeToLog( $logfile, sprintf( "DBI prepare Error: $sqlStat->err : $sqlStat->errstr" ), 0 );
                  Time::HiRes::sleep(2);
                  next;
               }
               $sqlStat->execute();
               $sqlStat->finish();

               my $str = sprintf( "Message: %s, %s, %s, %s, %s, %s, %s", $descripcion, $ip, $recipient, $message, $port, $auth, $user_id ); 
               writeToLog( $logfile, $str, 0 ); 
            
#               my $sndCode = sendSms( $ip, $recipient, $message, $port, $auth, $user_id );
#               my $errCode = $sndCode["error_code"];

               $semaphore->down();
               $foundThread = 0;
               for( $i = 0; $i < $threadCount; $i++ ) {
                  if( $Used[$i] == 0 ) {
                     $Used[$i] = 1;
                     $foundThread = 1;
                     last;
                  }
               }

               # -----------------------------
               # Actualizo Thread en Uso
               # -----------------------------
               UsedThreads();
               $semaphore->up();

               my $sqlt = sprintf( "update config set used_threads=%d where id=%d", $globalThreadCount +1, $site );
               my $stdt = $dbh->prepare($sqlt);
               if ( !$stdt->err ) {
                  $stdt->execute();
                  $stdt->finish();
               }             

               # ---------------
               # Ejecuto thread 
               # ---------------
               if( $foundThread == 1 ) {
                  $thr[$i] = threads->create(\&sendSmsThread, $ip, $recipient, $message, $port, $auth, $user_id, $i, $testFlag, $id, $gw, $slot, $customer_id, $idPort );
               }
            }  
            else {
               printf( "No hay puertos disponibles\n" );
               Time::HiRes::sleep(3);

               $sql = sprintf( "select sql_no_cache * from config where id=%d and stop=1;", $site );
               my $result = $dbh->prepare($sql);
               if ( $result->err ) {
                  writeToLog( $logfile, sprintf( "DBI prepare Error: $result->err : $result->errstr" ), 0 );
                  Time::HiRes::sleep(2);
                  next;
               }
               $result->execute();
               my $rows = $result->rows;
               if( $rows > 0 ) {
                  $forever = 0;
                  $result->finish();
                  last;
               }
               $result->finish();
            }
         }
         else {
            $semaphore->up(); 
         }            
#        Time::HiRes::sleep(10);


=begin
         # ----------------
         # Rutas
         # ----------------
         switch( $provider ) {


            case "5"  {
               my $message_text = $message;
               my $send_error = 1;
               my $auth_name = "AC67857d17eda55402d2eb69431c9f14c4:f4cb245d7df5a7e9d5c054212151794a";

               $sendMessage = { 
                  To => $recipient,
                  Channel => "sms",
                  message => $message_text
               };
               my $jsonEncoded = encode_json( $sendMessage );

               # ----------------------------
               # Me fio Slot de Thread libre 
               # ----------------------------
               $semaphore->down();
               my $found = 0;
               for( $i = 0; $i < $threadCount; $i++ ) {
                  if( $Used[$i] == 0 ) {
                     $Used[$i] = 1;
                     $found = 1;
                     last;
                  }
               }

               # -----------------------------
               # Actualizo Thread en Uso
               # -----------------------------
               UsedThreads();
               $semaphore->up();

               my $sqlt = sprintf( "update config set used_threads=%d where id=%d", $globalThreadCount +1, $site );
               my $stdt = $dbh->prepare($sqlt);
               if ( !$stdt->err ) {
                  $stdt->execute();
                  $stdt->finish();
               }             

               # ---------------
               # Ejecuto thread 
               # ---------------
               if( $found == 1 ) {
                  $thr[$i] = threads->create(\&sendSms, $jsonEncoded, $dbh, $id, $customer_id, $recipient, $message_text, $i, $auth_name, $provider );
               }
            }
               
                        
            # --------
            # Peru
            # --------
            case "4" {

               my $message_text = $message;
               my $send_error = 1;
               
               $sendMessage = { 
                  username => "ycli-2439",
                  password => "L4lrpp9(",
                  phone => $recipient,
                  message => $message_text
               };
               my $jsonEncoded = encode_json( $sendMessage );
#               writeToLog( "./debugRuta4.log", sprintf( "%s", $jsonEncoded), 0 );

               # ----------------------------
               # Me fio Slot de Thread libre 
               # ----------------------------
               $semaphore->down();
               my $found = 0;
               for( $i = 0; $i < $threadCount; $i++ ) {
                  if( $Used[$i] == 0 ) {
                     $Used[$i] = 1;
                     $found = 1;
                     last;
                  }
               }

               # -----------------------------
               # Actualizo Thread en Uso
               # -----------------------------
               UsedThreads();
               $semaphore->up();

               my $sqlt = sprintf( "update config set used_threads=%d where id=%d", $globalThreadCount +1, $site );
               my $stdt = $dbh->prepare($sqlt);
               if ( !$stdt->err ) {
                  $stdt->execute();
                  $stdt->finish();
               }             

               # ---------------
               # Ejecuto thread 
               # ---------------
               if( $found == 1 ) {
                  $thr[$i] = threads->create(\&sendSms, $jsonEncoded, $dbh, $id, $customer_id, $recipient, $message_text, $i, "", $provider );
               }
            }
         }
=cut

         while( $globalThreadCount >= $threadCount ) {
            printf( "waiting...\n" );
            Time::HiRes::sleep(1);
         }
      }           # while
      $sqlHandle->finish();
#     Time::HiRes::sleep(2);
   }
   $sth->finish();         # Mensaje a enviar
}  
writeToLog( $logfile, sprintf( "End 'SmsInterface' daemon PID: %d", getpid() ),0 );
    

print ( "-------------------------------------------------------------------------------------------------------------------------------------------------------------------\n" );
$dbh->disconnect();

   # ------------------------------------
   # sendSms - THREAD
   # ------------------------------------

   sub sendSmsThread
   {
      my $LocalCopy;
      my $dbh;
      my $ip = $_[0];
      my $recipient = $_[1];
      my $message = $_[2];
      my $port = $_[3];
      my $auth = $_[4];
      my $user_id = $_[5];
      my $index = $_[6];
      my $testFlag = $_[7];
      my $id = $_[8];
      my $gw = $_[9];
      my $slot = $_[10];
      my $customer_id = $_[11];
      my $idPort = $_[12];

=begin
      my $driver = "mysql"; 
      my $database = "TestSMS-2";
      my $dsn = "DBI:$driver:database=Sms:127.0.0.1";
      my $userid = "root";
      my $password = "dialer19";
      my $dbhost = "10.0.66.119";
=cut

      threads->detach();
      my $dsn = "DBI:$driver:database=$database:$dbhost";
      if( ( $dbh = DBI->connect($dsn, $userid, $password ) ) == 0 ) {
         $semaphore->down();
         $Used[$index] = 0;
         $semaphore->up();
         return;
      }
   
      $semaphore->down();
      $LocalCopy = $globalThreadCount; 
      $Used[$index] = 1;
      $LocalCopy++;
      $globalThreadCount = $LocalCopy; 

      # -------------------------
      # marco como en proceso
      # -------------------------
      my $sql = sprintf( "update outbound_request set pending=201 where id=%d", $id );
      my $stdx = $dbh->prepare($sql);
      if ( $stdx->err ) {
         writeToLog( $logfile, sprintf( "DBI prepare Error: $stdx->err : $stdx->errstr"), 0 );
      }
      else {
         $stdx->execute();
         $stdx->finish();
      }             
      $semaphore->up();

#      my $sendCode = sendSms( $ip, $recipient, $message, $port, $auth, $user_id );
#      my $errCode = $sndCode["error_code"];
      my $errCode = 202;
      my $resultUpdate;
      if( $errCode == 202 ) {

         $semaphore->down();
#        my $smsInQueue = $sendCode["sms_in_queue"];

         if( $testFlag == 0 ) {
            $sql = sprintf( "update outbound_request set pending=%d, gw=%d, port=%d, slot='%s', forwarded=now() where id=%d;", $errCode, $gw, $port, $slot, $user_id );
            $resultUpdate = $dbh->prepare($sql);
            if ( !$resultUpdate->err ) {
               $resultUpdate->execute();
            }
            $resultUpdate->finish();
         }
            
         # ------------------------
         # Actualizo Estadisticas
         # ------------------------
         $sql = sprintf( "update statistics set sended_count=sended_count+1 where actual_date=date(now());" );
         $resultUpdate = $dbh->prepare($sql);
         if ( !$resultUpdate->err ) {
            $resultUpdate->execute();
         }
         $resultUpdate->finish();
 
         # --------------------
         # Actualizo Customer
         # --------------------
         $sql = sprintf( "update customers set credit_limit=credit_limit-1 where id=%d;", $customer_id );
         $resultUpdate = $dbh->prepare($sql);
         if ( !$resultUpdate->err ) {
            $resultUpdate->execute();
         }
         $resultUpdate->finish();
         $semaphore->up();
      }
      else {
         if( $testFlag == 0 ) {
            $semaphore->down();
            $sql = sprintf( "update outbound_request set pending=%d, port=%d, slot='%s', forwarded=now()  where id=%d;", $errCode, $gw, $port, $slot, $user_id );
            $resultUpdate = $dbh->prepare($sql);
            if ( !$resultUpdate->err ) {
               $resultUpdate->execute();
            }
            $resultUpdate->finish();
            $semaphore->up();
         }
      }

      $semaphore->down();
      $sql = sprintf( "update gateway_ports set fecha_ultimo=now() where id=%d;", $idPort );
      my $sqlStat = $dbh->prepare($sql);
      if ( $sqlStat->err ) {
         writeToLog( $logfile, sprintf( "DBI prepare Error: $sqlStat->err : $sqlStat->errstr" ), 0 );
         Time::HiRes::sleep(2);
         next;
      }
      $sqlStat->execute();
      $sqlStat->finish();
      
      $LocalCopy = $globalThreadCount; 
      $LocalCopy--;
      $globalThreadCount = $LocalCopy; 
      $Used[$index] = 0;
      $dbh->disconnect();
      $semaphore->up();
      threads->exit();
   }   

   # --------------------------------------
   # writeToLog
   # --------------------------------------

   sub writeToLog 
   {  
      my $filename = $_[0];
      my $text = $_[1];
      my $mode = $_[2];
   
      if( open(FH, '>>', $filename) ) {
         if( $mode == 1 ) {
            $text = "+--------------------------------------------------------------------------------------------------+";
         }
         my ($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst) = localtime();
         my $hoy = sprintf( "%02d.%02d.%04d %02d:%02d:%02d", $mday, $mon + 1, $year + 1900, $hour, $min, $sec );
         printf( FH sprintf( "%s: %s\n", $hoy, $text ) );
         printf( sprintf( "%s: %s\n", $hoy, $text ) );
         close(FH);
      }      
   }


   # ---------------------------
   # Cuantos Thread hay en uso
   # ---------------------------

   sub UsedThreads
   {
      my $str = "";   
      for( my $x = 0; $x < $threadCount; $x++ ) {
         if( $x == 0 ) {
            if( $Used[$x] == 0 ) {
               $str .= "0";
            }
            else {
               $str .= "1";
            }
         }
         else {
            if( $Used[$x] == 0 ) {
               $str .= ", 0";
            }
            else {
               $str .= ", 1";
            }
         }

      }
#      writeToLog( $logfile, sprintf( "Threads %d: (%s)", $globalThreadCount +1, $str ), 0 );
#      writeToLog( $logfile, "", 1 );
   }