/* Copyright 2000-2006 The Apache Software Foundation or its licensors, as
 * applicable.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/*
This mod-cache-requester modules starts background threads which keeps monitoring soon to expire pages from cache.
and if such pages are 'popular' then those pages are re-requested using libcurl so that such pages are not removed from the cache.
 */

#define CORE_PRIVATE
#include "mod_cache_requester.h"
#include <curl/curl.h>

#if APR_HAVE_UNISTD_H
#include <unistd.h>
#endif

#if !APR_HAS_THREADS
#error This module does not currently compile unless you have a thread-capable APR. Sorry!
#endif

# define DEFAULT_NUMBER_OF_THREADS 2
# define DEFAULT_QUEUE_SIZE 10
# define DEFAULT_EXPIRING_SIGNAL_TIME_IN_SECONDS 1000
# define DEFAULT_SECRET_CODE "SECRET_CODE"

module AP_MODULE_DECLARE_DATA cache_requester_module;

static cache_req_conf_t *sconf;
static cache_requester *cache_req_obj;

/**
 * Priority queue management functions
 */
static long cache_req_get_priority(void *data) {
    cache_req_expiring_page *expiring_page  = (cache_req_expiring_page *) data;
    return expiring_page->priority;
}

static void cache_req_set_pos(void *a, apr_ssize_t pos)
{
    cache_req_expiring_page *obj = (cache_req_expiring_page *)a;

    apr_atomic_set32(&(obj->pos), pos);
}

static apr_ssize_t cache_req_get_pos(void *a)
{
    cache_req_expiring_page *obj = (cache_req_expiring_page *)a;
    return apr_atomic_read32(&(obj->pos));
} 

/**
 * Queue Management operations. all those operations must be thread safe. use cache_req_obj->lock to ensure mutual exclusion.
 */
static void create_expiring_page(cache_req_expiring_page **node, request_rec *r) {
    char *request_uri;

    *node = (cache_req_expiring_page *) malloc(sizeof(cache_req_expiring_page));
    if(!node) {
        return;
    }
    (*node)->priority = 0;

    request_uri = apr_pstrcat(r->pool, r->server->server_hostname, 
                                                    ":", 
                                                    apr_itoa(r->pool, r->server->port), 
                                                    r->uri, NULL);

    (*node)->request_uri = malloc(strlen(request_uri) + 1);
    strcpy((*node)->request_uri, request_uri);

    (*node)->uri = malloc(strlen(r->uri) + 1);
    strcpy((*node)->uri, r->uri);

    (*node)->next = NULL;
}

static void delete_expiring_page(cache_req_expiring_page *node) {
    if(node->request_uri) {
        free(node->request_uri);
    }
    if(node->uri) {
        free(node->uri);
    }

    if(node) {
        free(node);
    }
}

static cache_req_expiring_page *insert_into_queue(request_rec *r) {

    cache_req_expiring_page *value;
    cache_req_expiring_page *new_value;
    int queue_size;

    queue_size = cache_pq_size(cache_req_obj->cache_request_pq);

    if(queue_size >= sconf->queue_size) {
        ap_log_error(APLOG_MARK, APLOG_INFO, 0, r->server,
                     "cache_requester: cache size queue exceeded");
        return NULL;
    }

    /*check if the entry for this page already exists or not.*/
    value = apr_hash_get(cache_req_obj->expiring_page_hash, r->uri, APR_HASH_KEY_STRING);

    /* if not, then insert it into the queue, initialize the priority*/
    if(value == NULL) {
        create_expiring_page(&new_value, r);
        apr_hash_set(cache_req_obj->expiring_page_hash, new_value->uri, APR_HASH_KEY_STRING, new_value);
        cache_pq_insert(cache_req_obj->cache_request_pq, new_value);

        // if this is the first entry added in the queue, then lets wake up one of the threads.
        if(queue_size == 0) {
            apr_thread_cond_signal(cache_req_obj->cond);
        }
        return new_value;
    }
    /* if its already there, then increment the priority by one.*/
    else {
        value->priority++;
        cache_pq_change_priority(cache_req_obj->cache_request_pq, (value->priority - 1), value->priority, value);
        return value;
    }
}

/* delete the entry from the queue. */
static int delete_from_queue(cache_req_expiring_page *expiring_page) {
    cache_req_expiring_page *value;

    if(!expiring_page || !(expiring_page->uri)) {
        return DECLINED;
    }

    value = apr_hash_get(cache_req_obj->expiring_page_hash, expiring_page->uri, APR_HASH_KEY_STRING);
    if(value == NULL) {
        return DECLINED;
    }
    else {
        apr_hash_set(cache_req_obj->expiring_page_hash, expiring_page->uri, APR_HASH_KEY_STRING, NULL);
        delete_expiring_page(value);
        return OK;
    }
}

/*
 * insert the soon-to-expire page into the queue.
 */
int notify_cache_requester(request_rec *r, cache_handle_t *handle) {
    const char *is_from_cache_requester;
    cache_req_expiring_page *value;

    is_from_cache_requester = apr_table_get(r->headers_in, "Cache-Requester");
    if(is_from_cache_requester) {

        /* verify the request is actually made by mod-cache-requester. */
        if(!strcmp(is_from_cache_requester, sconf->secret_code)) {
            return DECLINED;
        }
        return OK;
    }

     /*
    See the expiry time of the page. If its soon to expire, 
    then notify add this page in the queue (if already there then increment the counter) to re-request
    */
    if(!is_soon_to_expire(handle)) {
        apr_thread_mutex_lock(cache_req_obj->lock);
        value = insert_into_queue(r);
        apr_thread_mutex_unlock(cache_req_obj->lock);

    }
    return OK;
}

/*
 * Checks whether this page is soon-to-expire. returns 0 if its the case
 */
int is_soon_to_expire(cache_handle_t *handle) {
    cache_info *info;
    if(!handle) {
        return -1;
    }

    info = &(handle->cache_obj->info);
    if((apr_time_sec(info->expire) - apr_time_sec(apr_time_now())) < sconf->expiring_signal_time_in_seconds) {
        return 0;
    }
    return -1;
}

/*
 *  Process the most popular page, remove it from the queue (in request_page) 
 *  - re request it using libcurl - make_curl_request
 */
static size_t process_received_data(void *buffer, size_t size, size_t nmemb, void *userp);

void make_curl_request(char *request_uri, char *secret_code_header) {
    CURL *curl;
    struct curl_slist *headers=NULL;
    int returnValue;

    curl = curl_easy_init();
    headers = curl_slist_append(headers, secret_code_header);

    curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
    curl_easy_setopt(curl, CURLOPT_URL, request_uri);
    curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, process_received_data);

    returnValue = curl_easy_perform(curl);

    ap_log_error(APLOG_MARK, APLOG_INFO, 0, NULL, "cache_requester: URI: %s re-requested, with return value: %d", request_uri, returnValue);

}

void *APR_THREAD_FUNC request_page(apr_thread_t *thd, void * dummy) {
    cache_req_expiring_page *curr_obj;
    request_rec *r;
    request_rec *rnew;
    char *secret_code_header;

    secret_code_header = (char *) malloc (sizeof("Cache-Requester: ") + sizeof(sconf->secret_code) + 1);
    strcpy(secret_code_header, "Cache-Requester: ");
    strcat(secret_code_header, sconf->secret_code);

    for(;;) {
        apr_thread_mutex_lock(cache_req_obj->lock);

        /* If no entry in the queue => sleep */
        while(cache_pq_size(cache_req_obj->cache_request_pq) == 0) {
            apr_thread_cond_wait(cache_req_obj->cond, cache_req_obj->lock);
        }

        /*Pop the most popular page from the queue*/
        curr_obj = cache_pq_pop(cache_req_obj->cache_request_pq);

        if(!curr_obj) {
            apr_thread_mutex_unlock(cache_req_obj->lock);
            continue;
        }

        /*Remove entry from the expiring_page and insert it into expiring_page hashmap.*/
        apr_hash_set(cache_req_obj->expiring_page_hash, curr_obj->uri, APR_HASH_KEY_STRING, NULL);
        apr_thread_mutex_unlock(cache_req_obj->lock);

        //Re-request the page.
        make_curl_request(curr_obj->request_uri, secret_code_header);

        //Delete the object.
        delete_expiring_page(curr_obj);
    }
    free(secret_code_header);
    return NULL;
}

static size_t process_received_data(void *buffer, size_t size, size_t nmemb, void *userp) {
    /*
     * throw away data received as a part of curl request.
     */
}

/**
 * Directive Configurations.
 */
static void *create_cache_requester_config(apr_pool_t *p, server_rec *s)
{
    sconf = apr_pcalloc(p, sizeof(cache_req_conf_t));
    sconf->number_of_threads = DEFAULT_NUMBER_OF_THREADS;
    sconf->queue_size = DEFAULT_QUEUE_SIZE;
    sconf->expiring_signal_time_in_seconds = DEFAULT_EXPIRING_SIGNAL_TIME_IN_SECONDS;
    sconf->secret_code = DEFAULT_SECRET_CODE;
    return sconf;
}

static const char *set_no_of_threads(cmd_parms *parms, void *in_struct_ptr, const char *arg) {
    apr_size_t val;

    if (sscanf(arg, "%" APR_SIZE_T_FMT, &val) != 1) {
        return "CacheRequesterThreads argument must be an integer representing number of therads.";
    }
    if(val < 1) {
        return "CacheRequesterThreads argument must not be less than 1.";
    }
    sconf->number_of_threads = val;
    return NULL;
}

static const char *set_max_queue_size(cmd_parms *parms, void *in_struct_ptr, const char *arg) {
    apr_size_t val;

    if (sscanf(arg, "%" APR_SIZE_T_FMT, &val) != 1) {
        return "CacheRequesterMaxQueueSize argument must be an integer representing number of therads.";
    }

    if(val < 1) {
        return "CacheRequesterMaxQueueSize argument must be greater than 1.";
    }
    sconf->queue_size = val;
    return NULL;
}

static const char *set_soon_to_expire_time(cmd_parms *parms, void *in_struct_ptr, const char *arg) {
    apr_size_t val;

    if (sscanf(arg, "%" APR_SIZE_T_FMT, &val) != 1) {
        return "CacheRequesterSoonToExpireTime value must be a positive integer";
    }
    if(val < 1) {
        return "CacheRequesterSoonToExpireTime argument must be greater than 1.";
    }

    sconf->expiring_signal_time_in_seconds = val;
    return NULL;
}

static const char *set_secret_code(cmd_parms *parms, void *name, const char *arg) {

    sconf->secret_code = arg;
    return NULL;
}

static const command_rec cache_req_cmds[] =
{
    AP_INIT_TAKE1("CacheRequesterThreads", set_no_of_threads, NULL, RSRC_CONF,
                  "Number of cache requester threads"),
    AP_INIT_TAKE1("CacheRequesterMaxQueueSize", set_max_queue_size, NULL, RSRC_CONF,
                  "The maximum number of objects allowed to be placed in the cache requester queue."),
    AP_INIT_TAKE1("CacheRequesterSoonToExpireTime", set_soon_to_expire_time, NULL, RSRC_CONF,
                  "The required time remaining to expire a page to be able to be re-requested (in seconds)"),
    AP_INIT_TAKE1("CacheRequesterSecretCode", set_secret_code, NULL, RSRC_CONF,
                  "the secret code to authenticate requests made by mod-cache-requester"),

    {NULL}
};
 
/**
 * initialize and start the threads.
 */
static int init_cache_req_obj_threads(apr_pool_t *p, server_rec *s) {
    int i;
    cache_req_thread *current_thread;

    current_thread = cache_req_obj->thread_pool->queue;
    apr_thread_create(&(current_thread->handle), NULL, request_page, NULL, p);
    current_thread->next = NULL;
    for(i=0; i < sconf->number_of_threads-1; i++) {
        current_thread->next = (cache_req_thread *) apr_palloc(p, sizeof(cache_req_thread));
        apr_thread_create(&(current_thread->next->handle), NULL, request_page, NULL, p);
        current_thread->next->next = NULL;
        current_thread = current_thread->next;
    }
}

/**
 * module configuration stuff.
 */

static int cache_requester_child_init(apr_pool_t *p, apr_pool_t *plog,
                                       apr_pool_t *ptemp, server_rec *s) {
    //Here Threads would be instantiated.
    init_cache_req_obj_threads(p, s);
}

static int cache_requester_post_config(apr_pool_t *p, apr_pool_t *plog,
                                 apr_pool_t *ptemp, server_rec *s) {
    // Initialize the curl lib.
    curl_global_init(CURL_GLOBAL_ALL);

    //Create main object first.
    cache_req_obj = (cache_requester *) apr_palloc(p, sizeof(cache_requester));

    // instantiate the thread pool.
    cache_req_obj->thread_pool = (cache_req_thread_pool *) apr_palloc(p, sizeof(cache_req_thread_pool));

    // create the lock, condition variable and priority queue.
    apr_thread_mutex_create(&(cache_req_obj->lock), APR_THREAD_MUTEX_DEFAULT, p);
    apr_thread_cond_create(&(cache_req_obj->cond), p);
    cache_req_obj->cache_request_pq = cache_pq_init(sconf->queue_size, 
                                                    cache_req_get_priority, 
                                                    cache_req_get_pos, 
                                                    cache_req_set_pos);

    // instantiate the hashmaps.
    cache_req_obj->expiring_page_hash = apr_hash_make(p);

    //Instantiate the thread queue.
    cache_req_obj->thread_pool->queue = (cache_req_thread *) apr_palloc(p, sizeof(cache_req_thread));

    return OK;
}

static const cache_req_provider cache_requester_provider =
{
    &notify_cache_requester,
};

static void register_hooks(apr_pool_t *p)
{
    ap_hook_post_config(cache_requester_post_config, NULL, NULL, APR_HOOK_MIDDLE);
    ap_hook_child_init(cache_requester_child_init, NULL, NULL, APR_HOOK_MIDDLE);
    ap_register_provider(p, CACHE_REQUESTER_PROVIDER_GROUP, "cache_req", "0",
                         &cache_requester_provider);
}

module AP_MODULE_DECLARE_DATA cache_requester_module =
{
    STANDARD20_MODULE_STUFF,
    NULL,                          /* create per-directory config structure */
    NULL,                          /* merge per-directory config structures */
    create_cache_requester_config, /* create per-server config structure */
    NULL,                          /* merge per-server config structures */
    cache_req_cmds,                /* command apr_table_t */
    register_hooks
};