#include "autoconfig.h"

#ifdef RCL_MONITOR
#ifndef lint
static char rcsid[] = "@(#$Id: rclmonprc.cpp,v 1.9 2006-12-23 13:07:21 dockes Exp $ (C) 2006 J.F.Dockes";
#endif
/*
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the
 *   Free Software Foundation, Inc.,
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 */

/**
 * Recoll real time monitor processing. This file has the code to retrieve
 * event from the event queue and do the database-side processing, and the 
 * initialization function.
 */

#include <pthread.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <signal.h>

#include "debuglog.h"
#include "rclmon.h"
#include "debuglog.h"
#include "indexer.h"
#include "pathut.h"
#include "x11mon.h"

typedef map<string, RclMonEvent> queue_type;

/** Private part of RclEQ: things that we don't wish to exist in the interface
 *  include file.
 */
class RclEQData {
public:
    queue_type m_queue;
    RclConfig *m_config;
    bool       m_ok;
    pthread_mutex_t m_mutex;
    pthread_cond_t m_cond;
    RclEQData() 
	: m_config(0), m_ok(false)
    {
	if (!pthread_mutex_init(&m_mutex, 0) && !pthread_cond_init(&m_cond, 0))
	    m_ok = true;
    }
};

static RclMonEventQueue rclEQ;

RclMonEventQueue::RclMonEventQueue()
{
    m_data = new RclEQData;
}

RclMonEventQueue::~RclMonEventQueue()
{
    delete m_data;
}

bool RclMonEventQueue::empty()
{
    return m_data == 0 ? true : m_data->m_queue.empty();
}

// Must be called with the queue locked
RclMonEvent RclMonEventQueue::pop()
{
    RclMonEvent ev;
    if (!empty()) {
	ev = m_data->m_queue.begin()->second;
	m_data->m_queue.erase(m_data->m_queue.begin());
    }
    return ev;
}

/** Wait until there is something to process on the queue.
 *  Must be called with the queue locked 
 */
bool RclMonEventQueue::wait(int seconds, bool *top)
{
    MONDEB(("RclMonEventQueue::wait\n"));
    if (!empty()) {
	MONDEB(("RclMonEventQueue:: imm return\n"));
	return true;
    }

    int err;
    if (seconds > 0) {
	struct timespec to;
	to.tv_sec = time(0L) + seconds;
	to.tv_nsec = 0;
	if (top)
	    *top = false;
	if ((err = 
	     pthread_cond_timedwait(&m_data->m_cond, &m_data->m_mutex, &to))) {
	    if (err == ETIMEDOUT) {
		*top = true;
		MONDEB(("RclMonEventQueue:: timeout\n"));
		return true;
	    }
	    LOGERR(("RclMonEventQueue::wait:pthread_cond_timedwait failed"
		    "with err %d\n", err));
	    return false;
	}
    } else {
	if ((err = pthread_cond_wait(&m_data->m_cond, &m_data->m_mutex))) {
	    LOGERR(("RclMonEventQueue::wait: pthread_cond_wait failed"
		    "with err %d\n", err));
	    return false;
	}
    }
    MONDEB(("RclMonEventQueue:: normal return\n"));
    return true;
}

bool RclMonEventQueue::lock()
{
    MONDEB(("RclMonEventQueue:: lock\n"));
    if (pthread_mutex_lock(&m_data->m_mutex)) {
	LOGERR(("RclMonEventQueue::lock: pthread_mutex_lock failed\n"));
	return false;
    }
    MONDEB(("RclMonEventQueue:: lock return\n"));
    return true;
}
bool RclMonEventQueue::unlock()
{
    MONDEB(("RclMonEventQueue:: unlock\n"));
    if (pthread_mutex_unlock(&m_data->m_mutex)) {
	LOGERR(("RclMonEventQueue::lock: pthread_mutex_unlock failed\n"));
	return false;
    }
    return true;
}

void RclMonEventQueue::setConfig(RclConfig *cnf)
{
    m_data->m_config = cnf;
}

RclConfig *RclMonEventQueue::getConfig()
{
    return m_data->m_config;
}

extern int stopindexing;

bool RclMonEventQueue::ok()
{
    if (m_data == 0) {
	LOGDEB(("RclMonEventQueue: not ok: bad state\n"));
	return false;
    }
    if (!x11IsAlive()) {
	LOGDEB(("RclMonEventQueue: not ok: x11\n"));
	return false;
    }
    if (stopindexing) {
	LOGDEB(("RclMonEventQueue: not ok: stop request\n"));
	return false;
    }
    if (!m_data->m_ok) {
	LOGDEB(("RclMonEventQueue: not ok: end from rcv\n"));
	return false;
    }
    return true;
}

void RclMonEventQueue::setTerminate()
{
    MONDEB(("RclMonEventQueue:: setTerminate\n"));
    lock();
    m_data->m_ok = false;
    pthread_cond_broadcast(&m_data->m_cond);
    unlock();
}

bool RclMonEventQueue::pushEvent(const RclMonEvent &ev)
{
    MONDEB(("RclMonEventQueue::pushEvent for %s\n", ev.m_path.c_str()));
    lock();
    // It seems that a newer event is always correct to override any
    // older. TBVerified ?
    m_data->m_queue[ev.m_path] = ev;
    pthread_cond_broadcast(&m_data->m_cond);
    unlock();
    return true;
}

static string pidfile;
/** There can only be one real time indexing process running */
static bool processlock(const string &confdir) 
{
    pidfile = path_cat(confdir, RCL_MONITOR_PIDFILENAME);
    for (int i = 0; i < 2; i++) {
	int fd = open(pidfile.c_str(), O_RDWR|O_CREAT|O_EXCL, 0644);
	if (fd < 0) {
	    if (errno != EEXIST) {
		LOGERR(("processlock: cant create %s, errno %d\n",
			pidfile.c_str(), errno));
		return false;
	    }
	    if ((fd = open(pidfile.c_str(), O_RDONLY)) < 0) {
		LOGERR(("processlock: cant open existing %s, errno %d\n",
			pidfile.c_str(), errno));
		return false;
	    }
	    char buf[20];
	    memset(buf, 0, sizeof(buf));
	    if (read(fd, buf, 19) < 0) {
		LOGERR(("processlock: cant read existing %s, errno %d\n",
			pidfile.c_str(), errno));
		close(fd);
		return false;
	    }
	    close(fd);
	    int pid = atoi(buf);
	    if (pid <= 0 || (kill(pid, 0) < 0 && errno == ESRCH)) {
		// File exists but no process
		if (unlink(pidfile.c_str()) < 0) {
		    LOGERR(("processlock: cant unlink existing %s, errno %d\n",
			    pidfile.c_str(), errno));
		    return false;
		}
		// Let's retry
		continue;
	    } else {
		// Other process running
		return false;
	    }
	}

	// File could be created, write my pid in there
	char buf[20];
	sprintf(buf, "%d\n", getpid());
	if (write(fd, buf, strlen(buf)+1) != int(strlen(buf)+1)) {
	    LOGERR(("processlock: cant write to %s, errno %d\n",
		    pidfile.c_str(), errno));
	    close(fd);
	    return false;
	}
	close(fd);
	// Ok
	break;
    }
    return true;
}

static void processunlock()
{
    unlink(pidfile.c_str());
}

pthread_t rcv_thrid;

bool startMonitor(RclConfig *conf, bool nofork)
{
    if (!processlock(conf->getConfDir())) {
	LOGERR(("startMonitor: lock error. Other process running ?\n"));
	return false;
    }
    atexit(processunlock);

    rclEQ.setConfig(conf);
    if (pthread_create(&rcv_thrid, 0, &rclMonRcvRun, &rclEQ) != 0) {
	LOGERR(("startMonitor: cant create event-receiving thread\n"));
	return false;
    }

    if (!rclEQ.lock()) {
	return false;
    }
    LOGDEB(("start_monitoring: entering main loop\n"));
    bool timedout;
    time_t lastauxtime = time(0);
    bool didsomething = false;

    // Set a relatively short timeout for better monitoring of exit requests
    while (rclEQ.wait(2, &timedout)) {
	// Queue is locked.

	if (!rclEQ.ok())
	    break;
	list<string> modified;
	list<string> deleted;

	// Process event queue
	while (!rclEQ.empty()) {
	    // Retrieve event
	    RclMonEvent ev = rclEQ.pop();
	    switch (ev.m_etyp) {
	    case RclMonEvent::RCLEVT_MODIFY:
		LOGDEB(("Monitor: Modify/Check on %s\n", ev.m_path.c_str()));
		modified.push_back(ev.m_path);
		break;
	    case RclMonEvent::RCLEVT_DELETE:
		LOGDEB(("Monitor: Delete on %s\n", ev.m_path.c_str()));
		deleted.push_back(ev.m_path);
		break;
	    default:
		LOGDEB(("Monitor: got Other on %s\n", ev.m_path.c_str()));
	    }
	}
	// Unlock queue before processing lists
	rclEQ.unlock();

	// Process
	if (!modified.empty()) {
	    if (!indexfiles(conf, modified))
		break;
	    didsomething = true;
	}
	if (!deleted.empty()) {
	    if (!purgefiles(conf, deleted))
		break;
	    didsomething = true;
	}

	// Recreate the auxiliary dbs every hour.
	if (time(0) - lastauxtime > 60 *60) {
	    lastauxtime = time(0);
	    didsomething = false;
	    if (!createAuxDbs(conf))
		break;
	}

	// Lock queue before waiting again
	rclEQ.lock();
    }
    LOGDEB(("Monitor: returning\n"));
    return true;
}
#endif // RCL_MONITOR