/*
 * © Copyright 1996-2012 ECMWF.
 * 
 * This software is licensed under the terms of the Apache Licence Version 2.0
 * which can be obtained at http://www.apache.org/licenses/LICENSE-2.0. 
 * In applying this licence, ECMWF does not waive the privileges and immunities 
 * granted to it by virtue of its status as an intergovernmental organisation nor
 * does it submit to any jurisdiction.
 */

/*

B.Raoult
ECMWF Oct-93

*/

#include "mars.h"

#ifdef CRAY
#include <rpcsvc/dbm.h>
#include <dirent.h>

#define DBM_INSERT     		 0
#define DBM_REPLACE    		 1
#define dbm_open(a,b,c)		(dbminit(a)<0?0:1)
#define dbm_close(a)		dbmclose()
#define dbm_fetch(a,b)		fetch(b)
#define dbm_delete(a,b)		delete(b)
#define	dbm_store(a,b,c,d)	store(b,c)

#else
#include <ndbm.h>
#endif
#include <errno.h>
#include <ctype.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <dirent.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>

static struct sembuf lock[] = {
    { 0, 0,  SEM_UNDO }, /* test */
    { 0, 1,  SEM_UNDO }, /* lock */
};

static struct sembuf unlock[] = {
    { 0, -1, SEM_UNDO }, /* ulck */
};


typedef struct field_entry {
	long	offset;
	long	length;
	ino_t   id;
} field_entry;

typedef struct file_entry {
	time_t  access;
	time_t  created;
	long	length;
	char    fname[1024];
} file_entry;

typedef struct dbmdata {

	char         *dir;

	char         *fname;
	char         *chigh;
	char         *clow;

	long         offset;
	ino_t        id;

	FILE         *f;

	request      *w;
	request      *u;

	int          cnt;
	int			 sem;
	key_t        key;
	boolean      locked;

#ifdef CRAY
	int        dbm_fields; /* dummy */
	int        dbm_files;  /* dummy */
#else
	DBM        *dbm_fields;
	DBM        *dbm_files;
#endif

    double         high;
    double         low;

	file_entry   file;
	field_entry  field;

}dbmdata;

static void dbmcache_init(void);
static void dbmcache_admin(void*);
static err  dbmcache_open(void *data,request*,request*,int);
static err  dbmcache_close(void *data);
static err  dbmcache_read(void *data,request *r,void *buffer,long *length);
static err  dbmcache_write(void *data,request *r,void *buffer,long *length);
static boolean dbmcache_check(void *data, request *r);

static int written = 0;

static option opts[] = {
	{"cachespace","MARS_CACHE_PATH","-cachepath","/tmp/database",
	t_str,sizeof(char*),OFFSET(dbmdata,dir) },
	
    {"cachehigh", "MARS_CACHE_HIGH","-cachehigh","1G",
	t_str,sizeof(char*),OFFSET(dbmdata,chigh)},

    {"cachelow", "MARS_CACHE_LOW","-cachelow","1G",
	t_str,sizeof(char*),OFFSET(dbmdata,clow)},
};

static base_class _cachebase = {

	NULL,                     /* parent class */
	"cachebase",             /* name         */

	false,                   /* inited       */

	sizeof(dbmdata),         /* private size */
	NUMBER(opts),            /* option count */
	opts,                    /* options      */

	dbmcache_init,           /* init         */

	dbmcache_open,           /* open         */
	dbmcache_close,          /* close        */

	dbmcache_read,           /* read         */
	dbmcache_write,          /* write        */
	NULL,                    /* control      */
	dbmcache_check,          /* check        */
	NULL,                    /* query */
	NULL,                    /* archive */

	dbmcache_admin,          /* admin */
};


typedef char *(*chkproc)(request *r,const char*,const char*,int idx);

static char *extchar(request *r,const char*,const char*,int idx);
static char *extlong(request *r,const char*,const char*,int idx);
static char *extreal(request *r,const char*,const char*,int idx);

typedef struct chktype{
	char    *name;
	char    *format;
	int     size;
	chkproc extract;
} chktype;

static chktype checks_grib[] = {
	{"TYPE"         ,"%s",2, 		extchar, },
	{"CLASS"        ,"%s",2, 		extchar, },
	{"STREAM"       ,"%s",4, 		extchar, },
	{"LEVTYPE"      ,"%s",3,     extchar, },
	{"DOMAIN"       ,"%s",1,      extchar, },
	{"PARAM"        ,"%s",3,      extchar, },
	{"LEVELIST",     "%s",4,      extchar, },
	{"TIME",         "%04d",4, extlong, },
	{"STEP",         "%06d",6, extlong, },
	{"ITERATION",    "%d",2, extlong, },
	{"DIAGNOSTIC",   "%d",2, extlong, },

	{"STEP",         "%d",6, extlong, },
	{"NUMBER",      "%d",4, extlong, },
	{"_EXPVER",      "%X",8, extlong, },
	{"REPRES",       "%s",2, extchar, },

/*
	{"_AREA_N",      "%.6f",11, extreal, },
	{"_AREA_W",      "%.6f",11, extreal, },
	{"_AREA_S",      "%.6f",11, extreal, },
	{"_AREA_E",      "%.6f",11, extreal, },
	{"_GRID_NS",     "%.6f",11, extreal, },
	{"_GRID_EW",     "%.6f",11, extreal, },
	{"_ROTATION_LAT","%.6f",11, extreal, },
	{"_ROTATION_LON","%.6f",11, extreal, },
	{"_GAUSSIAN",    "%d",4, extlong, },
	{"_TRUNCATION",  "%d",4, extlong, },
*/
	{"DATE"         ,"%s",8, 		extchar, },

};

static chktype checks_image[] = {
	{"TYPE"         ,"%s",2, 		extchar, },
	{"CLASS"        ,"%s",2, 		extchar, },
	{"STREAM"       ,"%s",4, 		extchar, },
	{"TIME",         "%04d",4, extlong, },
	{"OBSTYPE",      "%04d",4, extlong, },
	{"IDENT",        "%s",8, extchar, },
	{"_EXPVER",      "%X",8, extlong, },
	{"DATE"         ,"%s",8, 		extchar, },
};

static chktype checks_obs[] = {
	{"TYPE"         ,"%s",2, 		extchar, },
	{"TIME",         "%04d",4, extlong, },
	{"OBSTYPE",      "%04d",4, extlong, },
	{"IDENT",        "%s",8, extchar, },
	{"DATE"         ,"%s",8, 		extchar, },
};


static char *indexname(request *r)
{
	static char buf[1024];
	int i;

	chktype *checks = checks_grib;
	int n = NUMBER(checks_grib);

	if(image(r))
	{
		checks = checks_image;
		n = NUMBER(checks_image);
	}

	if(observation(r))
	{
		checks = checks_obs;
		n = NUMBER(checks_obs);
	}

	buf[0] = 0;
	for(i=0;i<n;i++)
	{
		strcat(buf,checks[i].extract(r,checks[i].name,
			checks[i].format,checks[i].size));
		strcat(buf,"/");
	}

	marslog(LOG_DBUG,"%s",buf);

	return buf;
}

static void complete(char *p,int len)
{
	len -= strlen(p);
	while(len--) strcat(p,"x");
}

static char *extchar(request *r,const char* name,const char* format,int size)
{
	static char buf[20];
	const char *x = get_value(r,name,0);
	buf[0] = 0;
	if(x) sprintf(buf,format,x);
	complete(buf,size);
	return buf;
}

static char *extlong(request *r,const char* name,const char* format,int size)
{
	static char buf[20];
	const char *x = get_value(r,name,0);
	buf[0] = 0;
	if(x) sprintf(buf,format,atol(x));
	complete(buf,size);
	return buf;
}

static char *extreal(request *r,const char* name,const char* format,int size)
{
	static char buf[20];
	const char *x = get_value(r,name,0);
	buf[0] = 0;
	if(x) sprintf(buf,format,atof(x));
	complete(buf,size);
	return buf;
}

/* the only 'public' variable ... */

base_class *cachebase = &_cachebase;

static void dbmcache_init(void)
{
}

static err closeindex(dbmdata *g)
{
	if(g->dbm_files)  dbm_close(g->dbm_files);
	if(g->dbm_fields) dbm_close(g->dbm_fields);
	g->dbm_files  = NULL;
	g->dbm_fields = NULL;


	if(g->locked)
	{
		if(semop(g->sem,unlock,NUMBER(unlock)) == -1)
		{
			marslog(LOG_EROR|LOG_PERR,"semop");
			return -1;
		}
		g->locked = false;
	}

	return 0;
}

static err openindex(dbmdata *g)
{
	char buffer[1024];

	if(g->dbm_files) return 0; /* already opened */

	if(g->sem == -1)
	{
		key_t key;

		sprintf(buffer,"%s/.fields.pag",g->dir);
		if(access(buffer,F_OK) != 0)
			close(creat(buffer,0777));

		sprintf(buffer,"%s/.fields.dir",g->dir);
		if(access(buffer,F_OK) != 0)
			close(creat(buffer,0777));

		sprintf(buffer,"%s/.files.pag",g->dir);
		if(access(buffer,F_OK) != 0)
			close(creat(buffer,0777));

		sprintf(buffer,"%s/.files.dir",g->dir);
		if(access(buffer,F_OK) != 0)
			close(creat(buffer,0777));

		sprintf(buffer,"%s/.lock.lck",g->dir);
		if(access(buffer,F_OK) != 0)
			close(creat(buffer,0777));


		 key = ftok(buffer,1);
		 if(key == (key_t)-1)
		 {
			marslog(LOG_EROR|LOG_PERR,"ftok(%s)",buffer);
			return -1;
		}

		g->sem = semget(key,1,0666 | IPC_CREAT);
		if(g->sem == -1)
		{
			marslog(LOG_EROR|LOG_PERR,"semget(%s)",buffer);
			return -1;
		}
	}

	
	/* Lock file */

	if(semop(g->sem,lock,NUMBER(lock)) == -1)
	{
		marslog(LOG_EROR|LOG_PERR,"semop");
		return -1;
	}

	g->locked = true;

	sprintf(buffer,"%s/.fields",g->dir);

    if(!(g->dbm_fields = dbm_open(buffer,O_RDWR,0777)))
	{
		marslog(LOG_EROR|LOG_PERR,"dbm_open:%s",buffer);
		return -1;
	}

	sprintf(buffer,"%s/.files",g->dir);

    if(!(g->dbm_files = dbm_open(buffer,O_RDWR,0777)))
	{
		marslog(LOG_EROR|LOG_PERR,"dbm_open:%s",buffer);
		return -1;
	}

	return 0;

}


static double parse(const char* p)
{
	char *last;
	double n = strtod(p,&last);
	switch(*last)
	{
		case 0  : 
			break;

		case 'k':
		case 'K':
			n *= 1024.0;
			break;

		case 'm':
		case 'M':
			n *= 1024.0 * 1014.0;
			break;

		case 'g':
		case 'G':
			n *= 1024.0 * 1024.0 * 1024.0;
			break;

		default:
			marslog(LOG_WARN,"Bad size suffix %c, only K, M and G are valid",*last);
			marslog(LOG_WARN,"Cache size is set to %sbytes",bytename(n));
			break;
	}

	return n;
}

static err  dbmcache_open(void *data,request *r,request *e,int mode)
{
	dbmdata *g = (dbmdata*)data;

	g->sem = -1;

	if(access(g->dir,X_OK|R_OK|W_OK) != 0)
	{
		mkdirp(g->dir,0777);

		if(access(g->dir,X_OK|R_OK|W_OK) != 0)
		{
			marslog(LOG_EROR|LOG_PERR,"Cannot access dbm directory %s", g->dir);
			return -1;
		}
	}

	if(mode == READ_MODE)
	{
		g->u = g->w = unwind_one_request(r);
	}
	else {
		g->high = parse(g->chigh);
		g->low  = parse(g->clow );
	}

	return 0;

}


static err  dbmcache_close(void *data)
{
	dbmdata *g = (dbmdata*)data;

	if(g->f) fclose(g->f);
	free_all_requests(g->w);
	closeindex(g);
	return 0;
}


static boolean fetch_field(dbmdata *g,char* index)
{
	datum  idx,dat;
	idx.dsize = strlen(index)+1;
	idx.dptr  = index;
	dat = dbm_fetch(g->dbm_fields,idx);
	if(dat.dptr == NULL) return false;
	memcpy(&g->field,dat.dptr,dat.dsize);
	return true;
}

static err store_field(dbmdata *g,char* index)
{
	datum  idx,dat;
	idx.dsize = strlen(index)+1;
	idx.dptr  = index;

	dat.dsize = sizeof(field_entry);
	dat.dptr  = (char*)&g->field;

	if((dbm_store(g->dbm_fields,idx,dat,DBM_REPLACE)) < 0)
	{
		marslog(LOG_EROR|LOG_PERR,"dbmstore");
		return -2;
	}

	return 0;
}

static boolean fetch_file(dbmdata *g,ino_t index)
{
	datum  idx,dat;
	idx.dsize = sizeof(index);
	idx.dptr  = (char*)&index;
	dat = dbm_fetch(g->dbm_files,idx);
	if(dat.dptr == NULL) return false;
	memcpy(&g->file,dat.dptr,dat.dsize);
	return true;
}

static err store_file(dbmdata *g,ino_t index)
{
	datum  idx,dat;
	idx.dsize = sizeof(index);
	idx.dptr  = (char*)&index;

	dat.dsize = sizeof(file_entry) - 1024 + strlen(g->file.fname) + 1;
	dat.dptr  = (char*)&g->file;

	if(dbm_store(g->dbm_files,idx,dat,DBM_REPLACE) < 0)
	{
		marslog(LOG_EROR|LOG_PERR,"dbmstore");
		return -2;
	}

	return 0;
}


static void remove_fields(dbmdata *g,ino_t index)
{
	datum  idx,dat;
	boolean more;
	idx.dsize = sizeof(index);
	idx.dptr  = (char*)&index;

	dbm_delete(g->dbm_files,idx);

	more = true;

	while(more)
	{
		more = false;

		for(idx = dbm_firstkey(g->dbm_fields); 
			idx.dptr != NULL; idx = dbm_nextkey(g->dbm_fields))
		{
			field_entry f;

			dat = dbm_fetch(g->dbm_fields,idx);
			memcpy(&f,dat.dptr,dat.dsize);
			if(f.id == index) {
				dbm_delete(g->dbm_fields,idx);
				more = true;
			}
		}
	}
}

static err  dbmcache_read(void *data,request *r,void *buffer,long *length)
{
	dbmdata *g = (dbmdata*)data;
	char    *iname;
	err e;

	/* last request */

	if(g->u == NULL) return EOF;

	if(openindex(g))
		return -1;

	/* build index name */

	iname     = indexname(g->u);

	/* fetch the field */

	if(!fetch_field(g,iname))
		return EOF;

	/* open a file if not the same */

	if(g->id != g->field.id || g->f == NULL)
	{
		struct stat s;

		g->id = g->field.id;

		if(g->f)  fclose(g->f);
		g->f = NULL;

		/* find file */

		if(fetch_file(g,g->id))
		{
			g->file.access = time(0);
			store_file(g,g->id);
		}
		else {
			/* something is wrong... */
			marslog(LOG_EROR,"Cannot find file %d",g->id);
			remove_fields(g,g->id);
			return -2;
		}

		if(stat(g->file.fname,&s)<0)
		{
			marslog(LOG_EROR|LOG_PERR,"stat: %s",g->file.fname);
			remove_fields(g,g->id);
			return -1;
		}

		if( g->id != s.st_ino )
		{
			marslog(LOG_EROR,"File have changed %d",g->id);
			remove_fields(g,g->id);
			return -2;
		}

		if((g->f = fopen(g->file.fname,"r")) == NULL)
		{
			if(errno == ENOENT) /* the file is lost or purged */
			{
				marslog(LOG_EROR,"Cannot find file %d",g->id);
				remove_fields(g,g->id);
				return EOF;
			}
			else
			{
				marslog(LOG_EROR|LOG_PERR,"dbm : open %s",g->file.fname);
				return -2;
			}
		}
	} 

	fseek(g->f,g->field.offset,SEEK_SET);
	e = _readany(g->f,buffer,length);

	if(r && (e == 0))
	{
		grib_to_request(r,buffer,*length);
		set_value(r,"_ORIGINAL_FIELD","1");
	}

	g->u = g->u->next;

	closeindex(g);

	return e;
}

static err  dbmcache_write(void *data,request *r,void *buffer,long *length)
{
	dbmdata *g = (dbmdata*)data;
	long len;
	char *iname = indexname(r);
	const char *s = get_value(r,"_ORIGINAL_FIELD",0);

	if(s == NULL || atol(s) != 1)
		return EOF;

	if(openindex(g))
		return EOF;

	if(g->f == NULL)
	{

		 /* Find a file name */
		struct stat s;
		char fname[1024];

		do {
			sprintf(fname,"%s/%s.%d.%d.%d", g->dir, iname, g->cnt++,time(0),getpid());
			mkdirp(mdirname(fname),0777);
		}while(access(fname,F_OK) == 0);

		g->f = fopen(fname,"w");
		if(g->f == NULL)
		{
			marslog(LOG_EROR|LOG_PERR,"dbm : %s",fname);
			return -1;
		}

		/* get inode */

		if(stat(fname,&s)<0)
		{
			marslog(LOG_EROR|LOG_PERR,"stat: %s",fname);
			return -1;
		}

		g->id = s.st_ino;

		/* check that we don't have any thing with the same inode */

		if(fetch_file(g,g->id))
		{
			marslog(LOG_EROR,"Duplicate file id %d",g->id);
			remove_fields(g,g->id);
		}

		strcpy(g->file.fname,fname);
		g->file.created = g->file.access = time(0);
		g->file.length  = 0;
		g->offset = 0;

	}

	len = *length;

	g->field.id     = g->id;
	g->field.offset = g->offset;
	g->field.length = len;

	if((*length = fwrite(buffer,1,len,g->f)) != len)
	{
		marslog(LOG_EROR|LOG_PERR,"dbm : write error");
		return -1;
	}

	g->offset      += len;
	g->file.length += len;

	if(store_field(g,iname) != 0)
		return -1;

	if(store_file(g,g->id) != 0)
		return -1;

	written++;

	return closeindex(g);
}

static boolean dbmcache_check(void *data, request *r)
{
	return !observation(r);
}

static void dbmcache_admin(void *data)
{
	dbmdata *g = (dbmdata*)data;
	struct stat s;

	datum  idx,dat;

	boolean more = (written != 0);
	long    max  = g->high;

	while(more)
	{
		boolean found = false;
		time_t min    = time(0);
		double   total  = 0;
		ino_t id;
		char fname[1024];


		more          = false;

		if(openindex(g) != 0)
			return;

		for(idx = dbm_firstkey(g->dbm_files); idx.dptr != NULL; 
			idx = dbm_nextkey(g->dbm_files))
		{
			file_entry f;
			ino_t i;

			dat = dbm_fetch(g->dbm_files,idx);

			memcpy(&f,dat.dptr,dat.dsize);
			memcpy(&i,idx.dptr,idx.dsize);

			if(stat(f.fname,&s)<0)
			{
				marslog(LOG_EROR|LOG_PERR,"stat: %s",f.fname);
				remove_fields(g,i);
				more = true;
				break;
			}

			if(s.st_atime < min)
			{
				min   = s.st_atime;
				id    = i;
				found = true;
				strcpy(fname,f.fname);
			}

			total += s.st_size;
		}

		marslog(LOG_INFO,"Cache size is %sbytes",bytename(total));
		marslog(LOG_INFO,"Cache high is %sbytes",bytename(g->high));
		marslog(LOG_INFO,"Cache low is %sbytes",bytename(g->low));

		if(total > max && found)
		{
			marslog(LOG_INFO,"Removing fields from: %s",fname);
			more = true;
			max  = g->low;
			remove_fields(g,id);
			if(unlink(fname) < 0)
				marslog(LOG_EROR|LOG_PERR,"remove %s",fname);
		}


		closeindex(g);
	}

	written = 0;
}