pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

int task_failed = 0;

// Task variables

int user_id = -1;        // -1 means all users, > -1 == users.id
char date_from[30] = "";
char date_till[30] = "";

// Rerating variables

char buffer[4096] = "";
char update_query[BUFFER_SIZE] = "";

char update_client_query_beginning[256] = "INSERT INTO calls (id,calldate,user_price,user_rate,user_billsec,prefix) VALUES ";
char update_client_query_ending[512] = "ON DUPLICATE KEY UPDATE id=VALUES(id),calldate=VALUES(calldate),user_price=VALUES(user_price),user_rate=VALUES(user_rate),user_billsec=VALUES(user_billsec),prefix=VALUES(prefix);";

char update_supplier_query_beginning[256] = "INSERT INTO calls (id,calldate,provider_price,provider_rate,provider_billsec,prefix) VALUES ";
char update_supplier_query_ending[512] = "ON DUPLICATE KEY UPDATE id=VALUES(id),calldate=VALUES(calldate),provider_price=VALUES(provider_price),provider_rate=VALUES(provider_rate),provider_billsec=VALUES(provider_billsec),prefix=VALUES(prefix);";

double new_user_price = 0;
double old_user_price = 0;
double user_delta_price = 0;

double new_provider_price = 0;
double old_provider_price = 0;
double provider_delta_price = 0;

long long int user_match = 0;
long long int user_diff = 0;

long long int provider_match = 0;
long long int provider_diff = 0;

long long int old_user_billsec = 0;
long long int new_user_billsec = 0;

long long int old_provider_billsec = 0;
long long int new_provider_billsec = 0;

unsigned long long int last_call_id = 0;
unsigned long long int zero_calls = 0;
int rerate_batches = 0;
long long int total_calls = 0;        // total number of calls fetched from database
long long int rerated_calls = 0;      // number of calls rerated
long long int updated = 0;            // number of calls updated

int batch_counter = 0;
int first_iteration = 1;

char rerate_supplier_str[64] = "";
int rerate_supplier = 0;  // 0 - rerate client, 1 - rerate supplier

// elasticsearch
int elasticsearch_installed = 0;

// call data list

typedef struct call_data_struct {
    long long int call_id;
    char calldate[20];
    char prefix[256];
    char new_prefix[256];
    char dst[256];
    int billsec;
    double user_price;
    double user_new_price;
    double user_new_rate;
    double user_exchange_rate;
    int user_id;
    int user_billsec;
    int user_new_billsec;
    int user_tariff;
    int user_grace_time;
    double user_rate;
    int user_increment;
    int user_min_time;
    double user_connection_fee;
    int user_rate_found;
    double provider_price;
    double provider_new_price;
    double provider_exchange_rate;
    int provider_id;
    int provider_billsec;
    int provider_new_billsec;
    int provider_tariff;
    int provider_grace_time;
    double provider_rate;
    double provider_new_rate;
    int provider_increment;
    int provider_min_time;
    double provider_connection_fee;
    int provider_rate_found;
    int digits_used_for_price;
    char digits_rounding_method[20];
    int provider_digits_used_for_price;
    char provider_digits_rounding_method[20];
    struct call_data_struct *next;
} call_data;

call_data* call_data_start = NULL;

// time vars

struct tm tm;
struct timeval t0, t1;
time_t t, tt;
suseconds_t ut0, ut1;

struct timeval _t0, _t1;
time_t _t, _tt;
suseconds_t _ut0, _ut1;
char datetime[100];

// user balance

typedef struct user_balance_struct {
    int user_id;
    double old_price;
    double new_price;
    double diff_price;
} user_balance_t;

user_balance_t *user_balance = NULL;
int user_balance_count = 0;

/* DEBUG */

int DEBUG_RERATE = 0; // 0 - rerate, but do not update database; 1 - rerate and update database

int calls_get(int i);
void call_list_free();
void calls_rerate();
void *set_timer();
void update_record();
void error_handle();
void get_rate_details(call_data *node);
void calculate_call_price(call_data *node, int user);
void format_prefix_sql(char *prefixes, int prefixes_len, const char *number);
void get_calls_count();
void reset_globals();
int get_user_balance_index(int user_id);
void add_user_balance(int user_id, double user_price);
void update_user_old_balance(int index, double user_price);
void calculate_user_balance_diff();
void update_user_new_balance(int index, double user_price);
void update_user_balance();
void m2_round(double *price, int digits, char *method);