/*
 * sophia database
 * sphia.org
 *
 * Copyright (c) Dmitry Simonenko
 * BSD License
*/

#include <sp.h>

static inline void
sp_pageopen(spc *c, uint32_t page)
{
	assert(page < c->s->s.count);
	sppage *p = c->s->s.i[page];
	spepoch *e = p->epoch;
	c->pi = page;
	c->p = p;
	c->ph = (sppageh*)(e->db.map + p->offset);
	/* validate header */
	assert(c->ph->id > 0);
	c->pvi = 0 ;
	c->pv  = (spvh*)((char*)c->ph + sizeof(sppageh));
	c->dup = 0;
}

static inline void sp_pageclose(spc *c) {
	c->p  = NULL;
	c->ph = NULL;
	c->pv = NULL;
}

static inline void
sp_pagesetlast(spc *c)
{
	c->pvi = c->ph->count - 1;
	c->pv  = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * c->pvi);
}

static inline int
sp_pageseek(spc *c, char *rkey, int size, int *minp, int *midp, int *maxp)
{
	int min = 0;
	int max = c->ph->count - 1;
	int mid = 0;
	while (max >= min)
	{
		mid = min + ((max - min) >> 1);
		spvh *a = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * mid);
		register int rc =
			c->s->e->cmp(a->key, a->size, rkey, size, c->s->e->cmparg);
		switch (rc) {
		case -1: min = mid + 1;
			continue;
		case  1: max = mid - 1;
			continue;
		default:
			*minp = min;
			*midp = mid;
			*maxp = max;
			return 1;
		}
	}
	*minp = min;
	*midp = mid;
	*maxp = max;
	return 0;
}

static inline int
sp_pagesetlte(spc *c, char *rkey, int size)
{
	int mid, min, max;
	int match;
	int eq = 0;
	if (sp_pageseek(c, rkey, size, &min, &mid, &max)) {
		eq = 1;
		match = mid;
	} else {
		match = min;
	}
	/* not found, set minimal */
	if (match >= c->ph->count) {
		c->pv  = (spvh*)((char*)c->ph + sizeof(sppageh));
		c->pvi = 0;
	} else {
		c->pv  = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * match);
		c->pvi = match;
	}
	return eq;
}

static inline int
sp_pagesetgte(spc *c, char *rkey, int size)
{
	int mid, min, max;
	int match;
	int eq = 0;
	if (sp_pageseek(c, rkey, size, &min, &mid, &max)) {
		eq = 1;
		match = mid;
	} else {
		match = max;
	}
	/* not found, set max */
	if (match >= c->ph->count)
		match = c->ph->count - 1;
	c->pv  = (spvh*)((char*)c->ph + sizeof(sppageh) + c->ph->bsize * match);
	c->pvi = match;
	return eq;
}

static inline void
sp_pagenext(spc *c)
{
	c->pvi++;
	c->pv = (spvh*)((char*)c->pv + c->ph->bsize);
	if (spunlikely(c->pvi == c->ph->count)) {
		sp_pageclose(c);
		uint32_t next_page = c->pi + 1;
		if (splikely(next_page < c->s->s.count))
			sp_pageopen(c, next_page);
	}
}

static inline void
sp_pageprev(spc *c)
{
	c->pvi--;
	c->pv = (spvh*)((char*)c->pv - c->ph->bsize);
	if (spunlikely(c->pvi < 0)) {
		sp_pageclose(c);
		int prev_page = c->pi - 1;
		if (splikely(prev_page >= 0)) {
			sp_pageopen(c, prev_page);
			sp_pagesetlast(c);
		}
	}
}

static inline void sp_first(spc *c)
{
	sp_iopen(&c->i0, c->s->i);
	if (! c->s->iskip)
		sp_iopen(&c->i1, sp_ipair(c->s));
	if (spunlikely(c->s->s.count == 0))
		return;
	sp_pageopen(c, 0);
}

static inline int
sp_firstkey(spc *c, char *rkey, int size)
{
	/* do lte search on all indexes for the key */
	int eq = sp_ilte(c->s->i, &c->i0, rkey, size);
	if (! c->s->iskip)
		eq = eq + sp_ilte(sp_ipair(c->s), &c->i1, rkey, size);
	/* match page for the key */
	if (spunlikely(c->s->s.count == 0))
		return eq;
	/* open page and do lte search of key */
	uint32_t idx = 0;
	sppage *p spunused;
	p = sp_catroute(&c->s->s, rkey, size, &idx);
	assert(p != NULL);
	sp_pageopen(c, idx);
	eq = eq + sp_pagesetlte(c, rkey, size);
	return eq;
}

static inline void sp_last(spc *c)
{
	sp_iopen(&c->i0, c->s->i);
	sp_ilast(&c->i0);
	if (! c->s->iskip) {
		sp_iopen(&c->i1, sp_ipair(c->s));
		sp_ilast(&c->i1);
	}
	if (spunlikely(c->s->s.count == 0))
		return;
	sp_pageopen(c, c->s->s.count-1);
	sp_pagesetlast(c);
}

static inline int sp_lastkey(spc *c, char *rkey, int size)
{
	/* do gte search the index for the key */
	int eq = sp_igte(c->s->i, &c->i0, rkey, size);
	if (! c->s->iskip)
		eq = eq + sp_igte(sp_ipair(c->s), &c->i1, rkey, size);
	/* match page for the key */
	if (spunlikely(c->s->s.count == 0))
		return eq;
	/* open page and do gte search of key */
	uint32_t idx = 0;
	sppage *p spunused;
	p = sp_catroute(&c->s->s, rkey, size, &idx);
	assert(p != NULL);
	sp_pageopen(c, idx);
	eq = eq + sp_pagesetgte(c, rkey, size);
	return eq;
}

void sp_cursoropen(spc *c, sp *s, sporder o, char *rkey, int size)
{
	/* lock all */
	sp_glock(s);

	c->m = SPMCUR;
	c->o = o;
	c->s = s;
	c->dup = 0;
	c->p = NULL;
	c->ph = NULL;
	c->pi = 0;
	c->pvi = 0;
	c->pv = NULL;
	c->vsrc = SPCNONE;

	sp_iinv(c->s->i, &c->i0);
	sp_iinv(sp_ipair(c->s), &c->i1);

	c->r.type = SPREFNONE;
	switch (o) {
	case SPGTE:
	case SPGT:
		if (rkey) {
			/* init first key and skip on gt */
			if (sp_firstkey(c, rkey, size) && o == SPGT)
				sp_iterate(c);
			return;
		}
		sp_first(c);
		break;
	case SPLTE:
	case SPLT:
		if (rkey) {
			/* init last key and skip on lt */
			if (sp_lastkey(c, rkey, size) && o == SPLT)
				sp_iterate(c);
			return;
		}
		sp_last(c);
		break;
	}
}

void sp_cursorclose(spc *c)
{
	/* unlock all */
	sp_gunlock(c->s);
}

static inline void sp_reset(spc *c) {
	c->vsrc = SPCNONE;
	c->r.type = SPREFNONE;
	c->dup = 0;
}

static inline int sp_next(spc *c)
{
	switch (c->vsrc) {
	case SPCI0:
		sp_inext(&c->i0);
		if (c->dup & SPCPDUP)
			sp_pagenext(c);
		if (c->dup & SPCVDUP)
			sp_inext(&c->i1);
		break;
	case SPCI1:
		sp_inext(&c->i1);
		if (c->dup & SPCPDUP)
			sp_pagenext(c);
		if (c->dup & SPCVDUP)
			sp_inext(&c->i0);
		break;
	case SPCP:
		assert(c->p != NULL);
		assert(c->r.v.vh == c->pv);
		sp_pagenext(c);
		break;
	case SPCNONE:
		break;
	}
	sp_reset(c);
	
	int rc;
	spcsrc src = SPCNONE;
	spv *v0 = sp_ival(&c->i0);
	spv *v1 = sp_ival(&c->i1);
	spv *v  = NULL;

	/* END */
	if (v0 == NULL && v1 == NULL && c->pv == NULL) {
		c->vsrc = SPCNONE;
		return 0;
	}

	if (v0 && v1) {
		rc = c->s->e->cmp(v0->key, v0->size, v1->key, v1->size,
						  c->s->e->cmparg);
		switch (rc) {
		case  0: c->dup |= SPCVDUP;
		case -1: v = v0, src = SPCI0;
			break;
		case  1: v = v1, src = SPCI1;
			break;
		}
	} else if (v0) {
		v = v0, src = SPCI0;
	} else if (v1) {
		v = v1, src = SPCI1;
	}

	/* no page key */
	if (c->pv == NULL) {
		c->vsrc = src;
		c->r.type = SPREFM;
		c->r.v.v = v; 
		return 1;
	}
	/* no index key */
	if (v == NULL) {
		c->vsrc = SPCP;
		c->r.type = SPREFD;
		c->r.v.vh = c->pv;
		return 1;
	}

	/* compare in-memory key and page one's */
	rc = c->s->e->cmp(v->key, v->size, c->pv->key, c->pv->size,
	                  c->s->e->cmparg);
	switch (rc) {
	case 0: c->dup |= SPCPDUP;
	case -1:
		c->vsrc = src;
		c->r.type = SPREFM;
		c->r.v.v = v; 
		break;
	case 1:
		c->vsrc = SPCP;
		c->r.type = SPREFD;
		c->r.v.vh = c->pv;
		break;
	}
	return 1;
}

static inline int sp_prev(spc *c)
{
	switch (c->vsrc) {
	case SPCI0:
		sp_iprev(&c->i0);
		if (c->dup & SPCPDUP)
			sp_pageprev(c);
		if (c->dup & SPCVDUP)
			sp_iprev(&c->i1);
		break;
	case SPCI1:
		sp_iprev(&c->i1);
		if (c->dup & SPCPDUP)
			sp_pageprev(c);
		if (c->dup & SPCVDUP)
			sp_iprev(&c->i0);
		break;
	case SPCP:
		assert(c->p != NULL);
		assert(c->r.v.vh == c->pv);
		sp_pageprev(c);
		break;
	case SPCNONE:
		break;
	}
	sp_reset(c);
	
	int rc;
	spcsrc src = SPCNONE;
	spv *v0 = sp_ival(&c->i0);
	spv *v1 = sp_ival(&c->i1);
	spv *v  = NULL;

	/* END */
	if (v0 == NULL && v1 == NULL && c->pv == NULL) {
		c->vsrc = SPCNONE;
		return 0;
	}

	if (v0 && v1) {
		rc = c->s->e->cmp(v0->key, v0->size, v1->key, v1->size,
						  c->s->e->cmparg);
		switch (rc) {
		case  0: c->dup |= SPCVDUP;
		case -1: v = v1, src = SPCI1;
			break;
		case  1: v = v0, src = SPCI0;
			break;
		}
	} else if (v0) {
		v = v0, src = SPCI0;
	} else if (v1) {
		v = v1, src = SPCI1;
	}

	/* no page key */
	if (c->pv == NULL) {
		c->vsrc = src;
		c->r.type = SPREFM;
		c->r.v.v = v; 
		return 1;
	}
	/* no index key */
	if (v == NULL) {
		c->vsrc = SPCP;
		c->r.type = SPREFD;
		c->r.v.vh = c->pv;
		return 1;
	}

	/* compare in-memory key and page one's */
	rc = c->s->e->cmp(v->key, v->size, c->pv->key, c->pv->size,
	                  c->s->e->cmparg);
	switch (rc) {
	case  0: c->dup |= SPCPDUP;
	case  1:
		c->vsrc = src;
		c->r.type = SPREFM;
		c->r.v.v = v; 
		break;
	case -1:
		c->vsrc = SPCP;
		c->r.type = SPREFD;
		c->r.v.vh = c->pv;
		break;
	}
	return 1;
}

int sp_iterate(spc *c) {
	int rc = 0;
	do {
		switch (c->o) {
		case SPGTE:
		case SPGT: rc = sp_next(c);
			break;
		case SPLTE:
		case SPLT: rc = sp_prev(c);
			break;
		}
	} while (rc > 0 && sp_refisdel(&c->r));
	
	return rc;
}

static inline int
sp_matchi(sp *s, spi *i, void *key, size_t size, void **v, size_t *vsize)
{
	spv *a = sp_igetraw(i, key, size);
	if (splikely(a)) {
		if (a->flags & SPDEL)
			return 0;
		if (v) {
			*vsize = sp_vvsize(a);
			*v = NULL;
			if (*vsize > 0) {
				*v = sp_memdup(s, sp_vv(a), sp_vvsize(a));
				if (spunlikely(*v == NULL))
					return sp_e(s, SPEOOM, "failed to allocate key");
			}
		}
		return 1;
	}
	return 0;
}

int sp_match(sp *s, void *k, size_t ksize, void **v, size_t *vsize)
{
	/* I. match both in-memory index'es for the key */
	sp_lock(&s->locki);
	int rc = sp_matchi(s, s->i, k, ksize, v, vsize);
	if (rc == -1 || rc == 1) {
		sp_unlock(&s->locki);
		return rc;
	}
	/*
	 * skip the second index if it is been truncating at the
	 * somement, all updates are on disk pages.
	*/
	if (! s->iskip) {
		int rc = sp_matchi(s, sp_ipair(s), k, ksize, v, vsize);
		if (rc == -1 || rc == 1) {
			sp_unlock(&s->locki);
			return rc;
		}
	}
	sp_unlock(&s->locki);

	/* II. match the page */
	sp_lock(&s->locks);
	uint32_t page = 0;
	sppage *p = sp_catfind(&s->s, k, ksize, &page);
	if (p == NULL) {
		sp_unlock(&s->locks);
		return 0;
	}

	/* III. match the key in the page */
	spepoch *e = (spepoch*)p->epoch;
	sp_lock(&e->lock);

	sppageh *ph = (sppageh*)(e->db.map + p->offset);
	rc = 0;
	register uint32_t min = 0;
	register uint32_t max = ph->count - 1;
	spvh *match = NULL;
	while (max >= min) {
		register uint32_t mid = min + ((max - min) >> 1);
		register spvh *v =
			(spvh*)(e->db.map + p->offset + sizeof(sppageh) +
			        ph->bsize * mid);
		switch (s->e->cmp(v->key, v->size, k, ksize, s->e->cmparg)) {
		case -1: min = mid + 1;
			continue;
		case  1: max = mid - 1;
			continue;
		default:
			match = v;
			goto done;
		}
	}
done:
	if (match == NULL)
		goto ret;
	if (v) {
		*vsize = match->vsize;
		*v = NULL;
		if (match->vsize > 0) {
			*v = sp_memdup(s, e->db.map + p->offset + match->voffset, match->vsize);
			if (spunlikely(*v == NULL)) {
				sp_e(s, SPEOOM, "failed to allocate key");
				rc = -1;
				goto ret;
			}
		}
	}
	rc = 1;
ret:
	sp_unlock(&e->lock);
	sp_unlock(&s->locks);
	return rc;
}