kazoo.presence_notify = 1 descr "enable/disable sending notify callback to omnipresence"
kazoo.presence_notify_timeout = 3000 descr "timeout in ms waiting for notify reply"
kazoo.presence_notify_log_body = 0 descr "logs the body sent in the notification"
kazoo.presence_notify_log_resp_body = 0 descr "logs the body received from notification"
kazoo.presence_notify_log_to_table = 1 descr "logs notify/reply to active_watchers_log table"
kazoo.presence_notify_log_to_amqp = 0 descr "logs notify/reply to amqp"
kazoo.presence_notify_record_route = 0 descr "add record route header to notify msg sent"
kazoo.presence_notify_log_init_body = 0 descr "logs the body before its sent"
kazoo.presence_notify_force_send_socket = 0 descr "forces the send socket to the contact"

######## Generic Hash Table container in shared memory ########
modparam("htable", "htable", "notify=>size=16;autoexpire=3600;updateexpire=1;initval=0")

#!trydef PRESENCE_NOTIFY_INIT
#!trydef MAX_NOTIFY_ERROR 5

route[PRESENCE_LOCAL_NOTIFY]
{
    if($rm != "NOTIFY") {
       return;
    }
    t_set_fr(@cfg_get.kazoo.presence_notify_timeout, @cfg_get.kazoo.presence_notify_timeout);
    xlog("L_INFO", "$ci|log|init preparing $subs(event) notify to $subs(watcher_username)@$subs(watcher_domain) on behalf of $subs(pres_uri) : $du\n");
    if(@cfg_get.kazoo.presence_notify_log_init_body == 1) {
       xlog("L_INFO", "$ci|log|init|body $(mb{s.escape.common}{s.replace,\','}{s.replace,$$,})\n");
    }
    if(@cfg_get.kazoo.presence_notify_force_send_socket == 1) {
       $fs = $_s($(pr{s.tolower}):$(hdr(Contact){nameaddr.uri}{uri.host}):$(hdr(Contact){nameaddr.uri}{uri.port}));
       xlog("L_INFO", "$ci|log|init|forcing socket to $fs, $(pr{s.tolower}):$(hdr(Contact){nameaddr.uri}{uri.host}):$(hdr(Contact){nameaddr.uri}{uri.port}) , $ct\n");
    }
    if(@cfg_get.kazoo.presence_notify_record_route == 1) {
       record_route();
    }
}



####### SQL OPS module ##########
#!ifndef SQLOPS_LOADED
loadmodule "sqlops.so"
modparam("sqlops","sqlcon", "cb=>KAZOO_DB_URL")
#!trydef SQLOPS_LOADED
#!endif


####### MQUEUE module ##########
#!ifndef MQUEUE_LOADED
loadmodule "mqueue.so"
#!trydef MQUEUE_LOADED
#!endif
modparam("mqueue","mqueue", "name=presence_last_notity")

#!ifndef TIMER_LOADED
loadmodule "timer.so"
#!trydef TIMER_LOADED
#!endif
modparam("timer", "declare_timer", "PRESENCE_LOG_TIMER=PRESENCE_LOG_TIMER_ROUTE,500,fast,enable");


event_route[presence:notify-reply]
{
    if(@cfg_get.kazoo.presence_notify != 1)
       return;

    $xavp(pres=>delete_subscription) = 0;

    if($notify_reply($rs) == 200) {
       $sht(notify=>$ci) = $null;
       xlog("L_INFO", "$ci|end|notified $subs(watcher_username)@$subs(watcher_domain) on behalf of $subs(pres_uri)\n");
    } else if($notify_reply($rs) == 481 && $subs(reason) == "timeout") {
       xlog("L_INFO","$ci|end|sent subscription $hdr(Subscription-State)\n");
    } else {
       xlog("L_ERROR", "$ci|error|received $notify_reply($rs) when notifying $subs(watcher_username)@$subs(watcher_domain) on behalf of $subs(pres_uri)\n");
       if($rP != "UDP") {
         $xavp(pres=>delete_subscription) = 1;
         xlog("L_ERROR", "$ci|error|removing $rP watcher $subs(watcher_username)@$subs(watcher_domain) for $subs(pres_uri)\n");
      } else {
         $var(shtinc) = $shtinc(notify=>$ci::count);
         if($var(shtinc) > MAX_NOTIFY_ERROR) {
            $xavp(pres=>delete_subscription) = 1;
            xlog("L_ERROR", "$ci|error|removing $rP watcher $subs(watcher_username)@$subs(watcher_domain) for $subs(pres_uri)\n");
         }
       }
    }
    if(@cfg_get.kazoo.presence_notify_log_body == 1)
       xlog("L_INFO", "$ci|log|sent|body $(mb{s.escape.common}{s.replace,\','}{s.replace,$$,})\n");
    if(@cfg_get.kazoo.presence_notify_log_resp_body == 1)
       xlog("L_INFO", "$ci|log|resp|body $(notify_reply($mb){s.escape.common}{s.replace,\','}{s.replace,$$,})\n");

    if(@cfg_get.kazoo.presence_notify_log_to_amqp == 1) {
       route(PRESENCE_NOTIFY_AMQP);
    }

    if(@cfg_get.kazoo.presence_notify_log_to_table == 1) {
       $var(Query) = $_s(REPLACE active_watchers_log SET sent_msg="$(mb{s.escape.common}{s.replace,\','}{s.replace,$$,})", received_msg="$(notify_reply($mb){s.escape.common}{s.replace,\','}{s.replace,$$,})", time=$TS, result=$notify_reply($rs), user_agent="$(subs(user_agent){s.escape.common}{s.replace,\','}{s.replace,$$,})", callid="$subs(callid)", to_user="$subs(to_user)", to_domain="$subs(to_domain)" where presentity_uri="$subs(uri)" and watcher_username="$subs(watcher_username)" and watcher_domain="$subs(watcher_domain)" and event="$subs(event)");
       mq_add("presence_last_notity", "$subs(callid)", "$var(Query)");
    }
}


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

route[PRESENCE_NOTIFY_AMQP]
{
    $var(amqp_payload_request) = $_s({"Event-Category" : "presence", "Event-Name" : "notify", "Event-Package" : "$subs(event)", "Timestamp" : $TS, "Call-ID" : "$subs(callid)", "From" : "$fu", "To" : "$subs(to_user)@$subs(to_domain)", "Sent" : "$(TS{s.ftime,%Y-%m-%d %H:%M:%S})", "Body" : "Hostname : MY_HOSTNAME\r\nTimestamp : $(TS{s.ftime,%Y-%m-%d %H:%M:%S})\r\n$(mb{s.escape.common}{s.replace,\','}{s.replace,$$,})\r\nResponse\r\n$(notify_reply($mb){s.escape.common}{s.replace,\','}{s.replace,$$,})","Remote-CSeq" : $subs(remote_cseq), "Local-CSeq" : $subs(local_cseq), "Sequence" : $cs, "Version" : $subs(version), "Reply" : $notify_reply($rs) });
    $var(rk) = "notify." + $(subs(to_domain){kz.encode}) + "." + $(subs(to_user){kz.encode});
    kazoo_publish("omnipresence", "$var(rk)", $var(amqp_payload_request));
    xlog("L_INFO", "$ci|log|sent notify callback for event $subs(event) : $tu\n");
}