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

#include <sp.h>

typedef struct {
	uint32_t count;
	uint32_t psize;
	uint32_t bsize;
} spupdate0;

static inline void
sp_mergeget0(spii *pos, uint32_t n, spupdate0 *u)
{
	memset(u, 0, sizeof(*u));
	/* 
	 * collect n or less versions for scheduled page write,
	 * not marked as delete, calculate page size and the
	 * block size.
	*/
	spii i = *pos;
	while (u->count < n && sp_ihas(&i)) {
		spv *v = sp_ival(&i);
		if (v->flags & SPDEL) {
			sp_inext(&i);
			continue;
		}
		if (v->size > u->bsize)
			u->bsize = v->size;
		sp_inext(&i);
		u->count++;
		u->psize += sp_vvsize(v);
	}
	u->bsize += sizeof(spvh);
	u->psize += sizeof(sppageh) + u->bsize * u->count;
}

static inline int sp_merge0(sp *s, spepoch *x, spi *index)
{
	spv *max = NULL;
	spv *min = NULL;
	int rc;
	spii i;
	sp_iopen(&i, index);

	while (sp_active(s))
	{
		/* get the new page properties and a data */
		spupdate0 u;
		sp_mergeget0(&i, s->env->page, &u);
		if (spunlikely(u.count == 0))
			 break;

		/* ensure enough space for the page in the file */
		sp_lock(&x->lock);
		rc = sp_mapensure(&x->db, u.psize, s->env->dbgrow);
		if (spunlikely(rc == -1)) {
			sp_unlock(&x->lock);
			sp_em(s, SPEIO|SPEF, x->epoch, "failed to remap db file");
			goto err;
		}
		sp_unlock(&x->lock);

		/* write the page.
		 *
		 * [header] [keys (block sized)] [values]
		 *
		 * Use partly precalculated crc for a version. 
		*/
		sppageh *h = (sppageh*)(x->db.map + x->db.used);
		h->id    = ++s->psn;
		h->count = u.count;
		h->bsize = u.bsize;
		h->size  = u.psize - sizeof(sppageh);
		h->crc   = sp_crc32c(0, &h->id, sizeof(sppageh) - sizeof(h->crc));

		char *ph = x->db.map + x->db.used + sizeof(sppageh);
		char *pv = ph + u.count * u.bsize;

		uint32_t current = 0;
		spv *last = NULL;
		while (sp_active(s) && current < u.count)
		{
			spv *v = sp_ival(&i);
			if (v->flags & SPDEL) {
				sp_inext(&i);
				continue;
			}
			if (spunlikely(min == NULL)) {
				min = sp_vdup(s, v);
				if (spunlikely(min == NULL)) {
					sp_em(s, SPEOOM|SPEF, "failed to allocate key");
					goto err;
				}
			}
			assert(v->size <= u.bsize);
			spvh *vh = (spvh*)(ph);
			vh->size    = v->size;
			vh->flags   = v->flags;
			vh->vsize   = sp_vvsize(v);
			vh->voffset = pv - (char*)h;
			vh->crc     = sp_crc32c(v->crc, &vh->size, sizeof(spvh) - sizeof(vh->crc));
			memcpy(vh->key, v->key, v->size);
			memcpy(pv, sp_vv(v), vh->vsize);

			ph += u.bsize;
			pv += vh->vsize;
			last = v;
			current++;
			sp_inext(&i);
		}

		/* cancellation point check */
		if (! sp_active(s))
			goto err;

		/* create in-memory page */
		sppage *page = sp_pagenew(s, x);
		if (spunlikely(page == NULL)) {
			sp_em(s, SPEOOM|SPEF, "failed to allocate page");
			goto err;
		}
		max = sp_vdup(s, last);
		if (spunlikely(max == NULL)) {
			sp_em(s, SPEOOM|SPEF, "failed to allocate key");
			goto err;
		}
		assert(min != NULL);
		page->id     = s->psn;
		page->offset = x->db.used;
		page->size   = u.psize;
		page->min    = min;
		page->max    = max;

		/* insert page to the index */
		sp_lock(&s->locks);
		sppage *o = NULL;
		rc = sp_catset(&s->s, page, &o);
		if (spunlikely(rc == -1)) {
			sp_unlock(&s->locks);
			sp_pagefree(s, page);
			sp_em(s, SPEOOM|SPEF, "failed to allocate page index page");
			goto err;
		}
		sp_unlock(&s->locks);

		/* attach page to the epoch list */
		sp_pageattach(page);

		/* advance file buffer */
		sp_mapuse(&x->db, u.psize);

		min = NULL;
		max = NULL;
	}
	return 0;
err:
	if (min)
		sp_free(&s->a, min);
	if (max)
		sp_free(&s->a, max);
	return -1;
}

typedef struct {
	uint32_t pi;
	sppage *p;
	spepoch *s; /* p->epoch */
	uint32_t count;
	uint32_t bsize;
} spupdate;

typedef struct {
	/* a is an original page version
	   b is in-memory version */
	int a_bsize, b_bsize;
	int a_count, b_count;
	int A, B;
	spvh *a;
	spv *b;
	spref last;
	spii i;
	spepoch *x;
} spmerge;

typedef struct {
	splist split;
	int count;
} spsplit;

static inline int
sp_mergeget(sp *s, spii *from, spupdate *u)
{
	spii i = *from;
	if (spunlikely(! sp_ihas(&i)))
		return 0;
	memset(u, 0, sizeof(spupdate));
	/* match the origin page and a associated
	 * range of keys. */
	sppage *origin = NULL;
	uint32_t origin_idx = 0;
	uint32_t n = 0;
	while (sp_ihas(&i)) {
		spv *v = sp_ival(&i);
		if (splikely(origin)) {
			if (! sp_catown(&s->s, origin_idx, v))
				break;
		}  else {
			origin = sp_catroute(&s->s, v->key, v->size, &origin_idx);
			assert(((spepoch*)origin->epoch)->type == SPDB);
		}
		if (v->size > u->bsize)
			u->bsize = v->size;
		sp_inext(&i);
		n++;
	}
	assert(n > 0);
	u->count = n;
	u->bsize += sizeof(spvh);
	u->pi = origin_idx;
	u->p = origin;
	u->s = origin->epoch;
	return 1;
}

static inline void
sp_mergeinit(spepoch *x, spmerge *m, spupdate *u, spii *from)
{
	sppageh *h = (sppageh*)(u->s->db.map + u->p->offset);
	uint32_t bsize = u->bsize;
	if (h->bsize > bsize)
		bsize = h->bsize;
	m->a_bsize = h->bsize;
	m->b_bsize = bsize;
	memset(&m->last, 0, sizeof(m->last));
	m->i = *from;
	m->A = 0;
	m->B = 0;
	m->a_count = h->count;
	m->b_count = u->count;
	m->a = (spvh*)((char*)h + sizeof(sppageh));
	m->b = sp_ival(from);
	m->x = x;
}

static inline int sp_mergenext(sp *s, spmerge *m)
{
	if (m->A < m->a_count && m->B < m->b_count)
	{
		register int cmp =
			s->env->cmp(m->a->key, m->a->size,
		                m->b->key,
		                m->b->size, s->env->cmparg);
		switch (cmp) {
		case  0:
			/* use updated key B */
			m->last.type = SPREFM;
			m->last.v.v = m->b;
			m->A++;
			m->a = (spvh*)((char*)m->a + m->a_bsize);
			m->B++;
			sp_inext(&m->i);
			m->b = sp_ival(&m->i);
			return 1;
		case -1:
			/* use A */
			m->last.type = SPREFD;
			m->last.v.vh = m->a;
			m->A++;
			m->a = (spvh*)((char*)m->a + m->a_bsize);
			return 1;
		case  1:
			/* use B */
			m->last.type = SPREFM;
			m->last.v.v = m->b;
			m->B++;
			sp_inext(&m->i);
			m->b = sp_ival(&m->i);
			return 1;
		}
	}
	if (m->A < m->a_count) {
		/* use A */
		m->last.type = SPREFD;
		m->last.v.vh = m->a;
		m->A++;
		m->a = (spvh*)((char*)m->a + m->a_bsize);
		return 1;
	}
	if (m->B < m->b_count) {
		/* use B */
		m->last.type = SPREFM;
		m->last.v.v = m->b;
		m->B++;
		sp_inext(&m->i);
		m->b = sp_ival(&m->i);
		return 1;
	}
	return 0;
}

static inline void
sp_splitinit(spsplit *l) {
	sp_listinit(&l->split);
	l->count = 0;
}

static inline void
sp_splitfree(sp *s, spsplit *l) {
	splist *i, *n;
	sp_listforeach_safe(&l->split, i, n) {
		sppage *p = spcast(i, sppage, link);
		sp_pagefree(s, p);
	}
}

static inline int sp_split(sp *s, spupdate *u, spmerge *m, spsplit *l)
{
	int rc;
	int bsize = m->b_bsize;
	uint32_t pagesize = sizeof(sppageh);
	uint32_t count = 0;

	/*
	 * merge in-memory keys with the origin page keys,
	 * skip any deletes and calculate result
	 * page size.
	*/
	sp_refsetreset(&s->refs);
	while (count < s->env->page && sp_mergenext(s, m)) {
		if (sp_refisdel(&m->last))
			continue;
		sp_refsetadd(&s->refs, &m->last);
		pagesize += bsize + sp_refvsize(&m->last);
		count++;
	}
	if (spunlikely(count == 0 && l->count > 0))
		return 0;

	/*
	 * set the origin page id for a first spitted page
	*/
	uint32_t psn = (l->count == 0) ? u->p->id : ++s->psn;

	/* ensure enough space for the page in the file */
	sp_lock(&m->x->lock);
	rc = sp_mapensure(&m->x->db, pagesize, s->env->dbgrow);
	if (spunlikely(rc == -1)) {
		sp_unlock(&m->x->lock);
		return sp_em(s, SPEIO|SPEF, m->x->epoch, "failed to remap db file");
	}
	sp_unlock(&m->x->lock);

	/* in case if all origin page keys are deleted.
	 *
	 * write special page header without any data, indicating
	 * that page should be skipped during recovery
	 * and not being added to the index.
	 */
	if (spunlikely(count == 0 && l->count == 0)) {
		sppageh *h = (sppageh*)(m->x->db.map + m->x->db.used);
		h->id    = psn;
		h->count = 0;
		h->bsize = 0;
		h->size  = 0;
		h->crc   = sp_crc32c(0, &h->id, sizeof(sppageh) - sizeof(h->crc));
		sp_mapuse(&m->x->db, pagesize);
		return 0;
	}

	spref *r = s->refs.r;
	spref *min = r;
	spref *max = r + (count - 1);

	/*
	 * write the page
	*/
	sppageh *h = (sppageh*)(m->x->db.map + m->x->db.used);
	h->id    = psn;
	h->count = count;
	h->bsize = bsize;
	h->size  = pagesize - sizeof(sppageh);
	h->crc   = sp_crc32c(0, &h->id, sizeof(sppageh) - sizeof(h->crc));

	spvh *ptr  = (spvh*)(m->x->db.map + m->x->db.used + sizeof(sppageh));
	char *ptrv = (char*)ptr + count * bsize;

	uint32_t i = 0;
	while (i < count)
	{
		uint32_t voffset = ptrv - (char*)h;
		switch (r->type) {
		case SPREFD:
			memcpy(ptr, r->v.vh, sizeof(spvh) + r->v.vh->size);
			memcpy(ptrv, u->s->db.map + u->p->offset + r->v.vh->voffset,
			       r->v.vh->vsize);
			ptr->voffset = voffset;
			uint32_t crc;
			crc = sp_crc32c(0, ptr->key, ptr->size);
			crc = sp_crc32c(crc, ptrv, r->v.vh->vsize);
			crc = sp_crc32c(crc, &ptr->size, sizeof(spvh) - sizeof(ptr->crc));
			ptr->crc = crc;
			ptrv += r->v.vh->vsize;
			break;
		case SPREFM:
			ptr->size    = r->v.v->size;
			ptr->flags   = r->v.v->flags;
			ptr->voffset = voffset;
			ptr->vsize   = sp_vvsize(r->v.v);
			ptr->crc = sp_crc32c(r->v.v->crc, &ptr->size, sizeof(spvh) -
			                     sizeof(ptr->crc));
			memcpy(ptr->key, r->v.v->key, r->v.v->size);
			memcpy(ptrv, sp_vv(r->v.v), ptr->vsize);
			ptrv += ptr->vsize;
			break;
		}
		assert((uint32_t)(ptrv - (char*)h) <= pagesize);
		ptr = (spvh*)((char*)ptr + bsize);
		r++;
		i++;
	}

	/* create in-memory page */
	sppage *p = sp_pagenew(s, m->x);
	if (spunlikely(p == NULL))
		return sp_em(s, SPEOOM|SPEF, "failed to allocate page");
	p->id     = psn;
	p->offset = m->x->db.used;
	p->size   = pagesize;
	p->min    = sp_vdupref(s, min, m->x->epoch);
	if (spunlikely(p->min == NULL)) {
		sp_free(&s->a, p);
		return sp_em(s, SPEOOM|SPEF, "failed to allocate key");
	}
	p->max    = sp_vdupref(s, max, m->x->epoch);
	if (spunlikely(p->max == NULL)) {
		sp_free(&s->a, p->min);
		sp_free(&s->a, p);
		return sp_em(s, SPEOOM|SPEF, "failed to allocate key");
	}

	/* add page to split list */
	sp_listappend(&l->split, &p->link);
	l->count++;

	/* advance buffer */
	sp_mapuse(&m->x->db, pagesize);
	return 1;
}

static inline int sp_splitcommit(sp *s, spupdate *u, spmerge *m, spsplit *l)
{
	sp_lock(&s->locks);
	/* remove origin page, if there were no page
	 * updates after split */
	if (spunlikely(l->count == 0)) {
		sp_pagefree(s, u->p);
		u->s->ngc++;
		u->p = NULL;
		sp_catdel(&s->s, u->pi);
		sp_unlock(&s->locks);
		return 0;
	}
	splist *i, *n;
	sp_listforeach_safe(&l->split, i, n)
	{
		sppage *p = spcast(i, sppage, link);
		/* update origin page first */
		if (spunlikely(p->id == u->p->id)) {
			sp_listunlink(&p->link);
			/* relink origin page to new epoch */
			sppage *origin = u->p;
			assert(origin->epoch != m->x);
			sp_listunlink(&origin->link);
			u->s->ngc++; /* origin db epoch */
			m->x->n++; /* current db epoch */
			sp_listappend(&m->x->pages, &origin->link);
			/* update origin page */
			origin->offset = p->offset;
			assert(p->epoch == m->x);
			origin->epoch = m->x;
			origin->size  = p->size;
			sp_free(&s->a, origin->min);
			sp_free(&s->a, origin->max);
			origin->min   = p->min;
			origin->max   = p->max;
			sp_free(&s->a, p);
			continue;
		}
		/* insert split page */
		sppage *o = NULL;
		int rc = sp_catset(&s->s, p, &o);
		if (spunlikely(rc == -1)) {
			sp_unlock(&s->locks);
			return sp_em(s, SPEOOM|SPEF, "failed to allocate page index page");
		}
		assert(o == NULL);
		sp_pageattach(p);
		m->x->n++;
	}
	sp_unlock(&s->locks);
	return 0;
}

static inline int sp_mergeN(sp *s, spepoch *x, spi *index)
{
	int rc;
	spii i;
	sp_iopen(&i, index);
	spupdate u;
	while (sp_mergeget(s, &i, &u))
	{
		spmerge m;
		sp_mergeinit(x, &m, &u, &i);
		spsplit l;
		sp_splitinit(&l);
		while (sp_active(s)) {
			rc = sp_split(s, &u, &m, &l);
			if (spunlikely(rc == 0))
				break;
			else
			if (spunlikely(rc == -1)) {
				sp_splitfree(s, &l);
				return -1;
			}
		}
		if (spunlikely(! sp_active(s)))
			return 0;
		rc = sp_splitcommit(s, &u, &m, &l);
		if (spunlikely(rc == -1)) {
			sp_splitfree(s, &l);
			return -1;
		}
		i = m.i;
	}
	return 0;
}

int sp_merge(sp *s)
{
	sp_lock(&s->lockr);
	sp_lock(&s->locki);

	spepoch *x = sp_replive(&s->rep);
	/* rotate current live epoch */
	sp_repset(&s->rep, x, SPXFER);
	int rc = sp_rotate(s, &s->em);
	if (spunlikely(rc == -1)) {
		sp_esetfatal(&s->em);
		sp_lock(&s->lockr);
		sp_lock(&s->locki);
		return -1;
	}
	/* swap index */
	spi *index = sp_iswap(s);

	sp_unlock(&s->lockr);
	sp_unlock(&s->locki);

	/* complete old live epoch log */ 
	rc = sp_logeof(&x->log);
	if (spunlikely(rc == -1))
		return sp_em(s, SPEIO|SPEF, x->epoch, "failed to write eof marker");
	rc = sp_logcomplete(&x->log);
	if (spunlikely(rc == -1))
		return sp_em(s, SPEIO|SPEF, x->epoch, "failed to complete log file");

	/* create db file */
	rc = sp_mapepochnew(&x->db, s->env->dbnewsize, s->env->dir,
	                    x->epoch, "db");
	if (spunlikely(rc == -1))
		return sp_em(s, SPEIO|SPEF, x->epoch, "failed to create db file");

	/* merge index */
	if (splikely(s->s.count > 0))
		rc = sp_mergeN(s, x, index);
	else
		rc = sp_merge0(s, x, index);

	/* check cancellation point */
	if (! sp_active(s)) {
		sp_mapunlink(&x->db);
		sp_mapclose(&x->db);
		return rc;
	}
	if (spunlikely(rc == -1))
		return -1;

	/* gc */
	if (s->env->gc) {
		rc = sp_gc(s, x);
		if (spunlikely(rc == -1))
			return -1;
	}

	/* sync/truncate db file and remap read-only only if
	 * database file is not empty. */
	if (splikely(x->db.used > 0)) {
		sp_lock(&x->lock);
		rc = sp_mapcomplete(&x->db);
		if (spunlikely(rc == -1)) {
			sp_unlock(&x->lock);
			return sp_em(s, SPEIO|SPEF, x->epoch, "failed to complete db file");
		}
		sp_unlock(&x->lock);
		/* set epoch as db */
		sp_lock(&s->lockr);
		sp_repset(&s->rep, x, SPDB);
		sp_unlock(&s->lockr);
		/* remove log file */
		rc = sp_logunlink(&x->log);
		if (spunlikely(rc == -1))
			return sp_em(s, SPEIO|SPEF, x->epoch, "failed to unlink log file");
		rc = sp_logclose(&x->log);
		if (spunlikely(rc == -1))
			return sp_em(s, SPEIO|SPEF, x->epoch, "failed to close log file");
	} else {
		/* there are possible situation when all keys has
		 * been deleted. */
		rc = sp_mapunlink(&x->db);
		if (spunlikely(rc == -1))
			return sp_em(s, SPEIO|SPEF, x->epoch, "failed to unlink db file");
		rc = sp_mapclose(&x->db);
		if (spunlikely(rc == -1))
			return sp_em(s, SPEIO|SPEF, x->epoch, "failed to close db file");
	}

	/* remove all xfer epochs that took part in the merge
	 * including current, if it's database file
	 * is empty. */
	while (sp_active(s)) {
		sp_lock(&s->lockr);
		spepoch *e = sp_repxfer(&s->rep);
		sp_unlock(&s->lockr);
		if (e == NULL)
			break;
		rc = sp_logunlink(&e->log);
		if (spunlikely(rc == -1))
			return sp_em(s, SPEIO|SPEF, e->epoch, "failed to unlink log file");
		rc = sp_logclose(&e->log);
		if (spunlikely(rc == -1))
			return sp_em(s, SPEIO|SPEF, e->epoch, "failed to close log file");
		sp_lock(&s->lockr);
		sp_repdetach(&s->rep, e);
		sp_free(&s->a, e);
		sp_unlock(&s->lockr);
	}

	/* truncate the index (skip index on a read) */
	sp_iskipset(s, 1);
	rc = sp_itruncate(index);
	if (spunlikely(rc == -1)) {
		sp_iskipset(s, 0);
		return sp_em(s, SPE|SPEF, "failed create index");
	}
	sp_iskipset(s, 0);
	return 0;
}