######## KEEPALIVE PINGING ########

#!trydef KEEPALIVE_NAT_ONLY 0
#!trydef KEEPALIVE_UDP_ONLY 0
#!trydef KEEPALIVE_TIMERS 8
#!trydef KEEPALIVE_INTERVAL 30
#!trydef KEEPALIVE_TIMEOUT 5000
#!trydef KEEPALIVE_FAILED_THRESHOLD 2
#!trydef KEEPALIVE_EXPIRE_SUBSCRIPTIONS 1
#!trydef KEEPALIVE_EXPIRE_REGISTRATIONS 1
#!trydef KEEPALIVE_FAILED_LOG_LEVEL 0

#!substdef "!KEEPALIVE_S_FROM_URI!sip:keepalive@MY_HOSTNAME!g"
#!substdef "!KEEPALIVE_S_TIMERS!$def(KEEPALIVE_TIMERS)!g"

kazoo.keepalive_udp_only = KEEPALIVE_UDP_ONLY descr "should we send keepalive for udp only"
kazoo.keepalive_nat_only = KEEPALIVE_NAT_ONLY descr "should we send keepalive for nat phones only"
kazoo.keepalive_timeout = KEEPALIVE_TIMEOUT descr "timeout in ms for keepalive transaction"
kazoo.keepalive_failed_threshold = KEEPALIVE_FAILED_THRESHOLD descr "how many times can a device fail to respond to OPTIONS"
kazoo.keepalive_expire_subscriptions = KEEPALIVE_EXPIRE_SUBSCRIPTIONS descr "expires subscriptions that do not respond to OPTIONS"
kazoo.keepalive_expire_registrations = KEEPALIVE_EXPIRE_REGISTRATIONS descr "expires registrations that do not respond to OPTIONS"
kazoo.keepalive_failed_log_level = KEEPALIVE_FAILED_LOG_LEVEL descr "loglevel for keepalive failed reply"

modparam("rtimer", "timer", "name=keepalive_timer;interval=1;mode=KEEPALIVE_S_TIMERS;")
modparam("rtimer", "exec", "timer=keepalive_timer;route=KEEPALIVE_TIMER")

modparam("rtimer", "timer", "name=keepalive_db_timer;interval=1;mode=1;")
modparam("rtimer", "exec", "timer=keepalive_db_timer;route=KEEPALIVE_DB_TIMER")

modparam("mqueue","mqueue", "name=keepalive_db_queue")

route[KEEPALIVE_DB_TIMER]
{
  $var(runloop) = 1;
  while(mq_fetch("keepalive_db_queue") == 1 && $var(runloop) < MAX_WHILE_LOOPS) {
     $var(ci) = $mqk(keepalive_db_queue);
     xlog("L_DEBUG", "Query : $mqv(keepalive_db_queue)\n");
     lock("keepalive");
     $var(sqlres) = sql_query("cb", "$mqv(keepalive_db_queue)");
     xlog("L_DEBUG", "Query result : $var(sqlres)\n");
     if($var(sqlres) < 0) {
        xlog("L_ERROR", "$var(ci)|log|error running query : $mqv(keepalive_db_queue)\n");
     } else {
       $var(nrows) = $sqlrows(cb);
       xlog("L_DEBUG", "$var(ci)|log|end UPDATED $var(nrows)\n");
       if($var(nrows) == 0) {
           xlog("L_DEBUG", "$var(ci)|log|error no rows affected when running query\n");
       }
     }
     unlock("keepalive");
     $var(runloop) = $var(runloop) + 1;
  }
}

route[KEEPALIVE_CLEANUP]
{
   lock("keepalive");
   if($sel(cfg_get.kazoo.keepalive_expire_registrations) == 1) {
       $var(Query) = $_s(update location set expires = last_modified where id in(select b.id from w_keepalive_contact a inner join w_location_contact b on a.contact = b.contact where slot = $rtimer_worker AND failed > $sel(cfg_get.kazoo.keepalive_failed_threshold)));
       sql_query("cb", "$var(Query)");
   }
   if($sel(cfg_get.kazoo.keepalive_expire_subscriptions) == 1) {
       $var(Query) = $_s(DELETE FROM active_watchers where id in(select b.id from w_keepalive_contact a inner join w_watchers_contact b on a.contact = b.contact where slot = $rtimer_worker and failed > $sel(cfg_get.kazoo.keepalive_failed_threshold)));
       sql_query("cb", "$var(Query)");
   }
   $var(Query) = $_s(DELETE FROM keepalive where slot = $rtimer_worker and failed > $sel(cfg_get.kazoo.keepalive_failed_threshold));
   sql_query("cb", "$var(Query)");
   unlock("keepalive");
}

route[KEEPALIVE_TIMER]
{
   route(KEEPALIVE_CLEANUP);

   lock("keepalive");
   $var(Query) = $_s(UPDATE keepalive SET selected = 1 WHERE slot = $rtimer_worker AND selected = 0 AND time_sent < datetime('now', '-$def(KEEPALIVE_INTERVAL) seconds'));
   $var(sqlres) = sql_query("cb", "$var(Query)");
   if($var(sqlres) < 0) {
       xlog("L_ERROR", "$rtimer_worker|log|error running query : $var(Query)\n");
   } else {
       $var(nrows) = $sqlrows(cb);
       xlog("L_DEBUG", "$rtimer_worker|log|selected $var(nrows) endpoints to ping\n");
   }
   unlock("keepalive");
   
   lock("keepalive");   
   $var(Query) = $_s(SELECT id, contact, sockinfo from keepalive WHERE slot = $rtimer_worker AND selected = 1);
   xlog("L_DEBUG", "$rtimer_worker|timer|SQL => $var(Query)\n");
   if (sql_xquery("cb", "$var(Query)", "ra") == 1)
   {
        while($xavp(ra) != $null) {
            $var(loop) = 0;
            while($xavp(ra) != $null && $var(loop) < MAX_WHILE_LOOPS) {
                route(KEEPALIVE_SEND_PING);
                pv_unset("$xavp(ra)");
                $var(loop) = $var(loop) + 1;
            }
        }
   }
   unlock("keepalive");
   
   lock("keepalive");
   $var(Query) = $_s(UPDATE keepalive SET selected = 2 WHERE slot = $rtimer_worker AND selected = 1);
   $var(sqlres) = sql_query("cb", "$var(Query)");
   if($var(sqlres) < 0) {
       xlog("L_ERROR", "$rtimer_worker|log|error running query : $var(Query)\n");
   }
   unlock("keepalive");
   
   
}

route[KEEPALIVE_SEND_PING]
{
    $var(CallId) = $uuid(g);
    xlog("L_DEBUG", "$var(CallId)|$rtimer_worker|timer|SENDING PING FROM $xavp(ra=>local_contact) TO => $xavp(ra=>contact)\n");
    
    $uac_req(method)="OPTIONS";
    $uac_req(hdrs) = "X-TM-Local: KEEPALIVE_PING\r\nX-TM-SockInfo: " + $xavp(ra=>sockinfo) + "\r\n";
    $uac_req(turi) = $xavp(ra=>contact);
    $uac_req(ruri) = $xavp(ra=>contact);
    $uac_req(furi) = $_s(KEEPALIVE_S_FROM_URI;nat_id=$xavp(ra=>id));
    $uac_req(ouri) = "sip:127.0.0.1:5090;transport=tcp";
    $uac_req(callid) = $var(CallId);
    uac_req_send();
    
}

onreply_route[KEEPALIVE_REPLY]
{
   xlog("L_DEBUG", "$ci|keepalive|KEEPALIVE REPLY $(tu{nameaddr.uri})\n");
   $var(Query) = $_s(UPDATE keepalive SET selected = 0, failed = 0, time_sent = datetime('now') WHERE id = $(fu{uri.param,nat_id}););
   xlog("L_DEBUG", "$ci|keepalive|KEEPALIVE UPDATE SQL => '$var(Query)'\n");
   mq_add("keepalive_db_queue", "$uuid(g)", "$var(Query)");
   t_drop();
}

failure_route[KEEPALIVE_FAULT]
{
   xlog("$(sel(cfg_get.kazoo.keepalive_failed_log_level){s.int})", "$ci|keepalive|received error $T_reply_code $T_reply_reason from $(tu{nameaddr.uri})\n");
   $var(Query) = $_s(UPDATE keepalive SET selected = 0, failed = failed + 1, time_sent = datetime('now') WHERE id = $(fu{uri.param,nat_id}););
   xlog("L_DEBUG", "$ci|keepalive|KEEPALIVE REMOVE SQL => '$var(Query)'\n");
   mq_add("keepalive_db_queue", "$uuid(g)", "$var(Query)");
   t_drop();
}

route[KEEPALIVE_PING]
{
   $fs = $hdr(X-TM-SockInfo);
   remove_hf_re("^X-TM-SockInfo");

   force_rport();
   handle_ruri_alias();
   record_route();
   xlog("L_DEBUG", "$ci|local|sending $proto keepalive using $fs to $ru => $du => $tu\n");

   t_on_reply("KEEPALIVE_REPLY");
   t_on_failure("KEEPALIVE_FAULT");
   t_set_fr(0, $sel(cfg_get.kazoo.keepalive_timeout));

   t_relay();
}

route[KEEPALIVE_ON_REGISTRATION]
{
   
   if($proto == "ws" || $proto == "wss") {
      return;
   }
   
   if (isbflagset(FLB_NATB)) {
        if(!isbflagset(FLB_NATSIPPING)) {
            return;
        }
   } else {
        if($sel(cfg_get.kazoo.keepalive_nat_only) == 1) {
            return;
        }
   }

   $var(alias) = $(avp(AVP_RECV_PARAM){uri.host}) + "~" + $(avp(AVP_RECV_PARAM){uri.port}) + "~" + $prid;
   $var(contact) = $(ct{nameaddr.uri}) + ";alias=" + $var(alias);
   $var(local_contact) = "sip:" + $Ri + ":" + $Rp + ";transport=" + $proto;
   xlog("L_DEBUG", "$ci|keepalive|KEEPALIVE ON REG $var(save_result) $proto $RAut $var(contact) $var(alias) $(ct{nameaddr.uri}) $ct $avp(AVP_RECV_PARAM) $tu $xavp(ulrcd=>ruid) , $xavp(ulrcd=>contact) , $xavp(ulrcd=>expires)\n");

   if($var(save_result) == 3) {
        $var(sql) = $_s(DELETE FROM keepalive WHERE contact = "$var(contact)");
   } else {
        $var(slot) = $(var(contact){s.corehash, KEEPALIVE_S_TIMERS});
        $var(sql) = $_s(INSERT OR IGNORE INTO keepalive (contact, sockinfo, slot) values("$var(contact)", "$(RAut{uri.tosocket})", $var(slot)));
   }
   mq_add("keepalive_db_queue", "$ci", "$var(sql)");

   return;
}

route[KEEPALIVE_ON_SUBSCRIBE]
{
   if(!( ($sel(cfg_get.kazoo.keepalive_udp_only) == 1 && $proto != "udp")
      || (!isflagset(FLT_NATS) && $sel(cfg_get.kazoo.keepalive_nat_only) == 1)
      || ($proto == "ws" || $proto == "wss")
     )) {
     $var(slot) = $(subs(contact){s.corehash, KEEPALIVE_S_TIMERS});
     $var(sql) = $_s(INSERT OR IGNORE INTO keepalive (contact, sockinfo, slot) values("$subs(contact)", "$subs(sockinfo)", $var(slot)));
     mq_add("keepalive_db_queue", "$ci", "$var(sql)");
   }
}

route[KEEPALIVE_ON_EXPIRED_REGISTRATION]
{
   $var(Query) = $_s(DELETE FROM keepalive where contact like "$ulc(exp=>addr)%");
   mq_add("keepalive_db_queue", "$uuid(g)", "$var(Query)");
}

route[KEEPALIVE_ON_OPTIONS]
{
   $var(Query) = $_s(UPDATE keepalive set selected = 3 where contact like "%alias=$si~$sp~$prid%");
   mq_add("keepalive_db_queue", "$uuid(g)", "$var(Query)");
}

route[KEEPALIVE_ON_NOTIFY]
{
   $var(Query) = $_s(UPDATE keepalive set selected = 4 where contact like "%alias=$si~$sp~$prid%");
   mq_add("keepalive_db_queue", "$uuid(g)", "$var(Query)");
}