#include "tra.h"

static void		triage(Syncpath*);
static int		triagedecision(Syncpath*);
ulong start;

static void
sendwork(Syncpath *s)
{
/*
	if(s->state==SyncCopy && s->t->state==SFile && s->f->state==SFile
	&& memcmp(s->t->sha1, s->f->sha1, sizeof s->t->sha1) == 0){
		syncfinish(s, 1);
		return;
	}
*/
	qsend(s->sync->workq, s);
}

void
syncthread0(void *a)
{
	Queue *q;

	startclient();
	q = a;
	for(;;)
		triage(qrecv(q));
}

void
syncthread(void *a)
{
	int i;

	for(i=0; i<32; i++)
		threadcreate(syncthread0, a);
}

/*
 * It's important to run the stats in parallel, because the
 * most common case where we don't have s->f and s->t
 * is the root, and statting the root takes a long time (it has
 * to check the whole tree for changes).
 */
typedef struct Statarg Statarg;
struct Statarg
{
	Replica *repl;
	Stat *s;
	Path *p;
	Chan *c;
	char *err;
};
static void
statthread(void *a)
{
	Statarg *sa;

	sa = a;
	startclient();
	sa->s = rpcstat(sa->repl, sa->p);
	if(sa->s == nil)
		sa->err = rpcerror();
	endclient();
	sendp(sa->c, sa);
}
static void
triage(Syncpath *s)
{
	Chan *c;
	Statarg *sa;

	dbg(DbgSync, "%P triage\n", s->p);
	if(s->state != SyncStart){
		fprint(2, "%P: state is not Start\n", s->p);
		return;
	}

	if(s->f==nil || s->t==nil){
		sa = emalloc(2*sizeof(Statarg));
		c = chan(Statarg*);
		if(s->f == nil){
			sa[0].repl = s->sync->from;
			sa[0].p = s->p;
			sa[0].c = c;
			threadcreate(statthread, &sa[0]);
		}
		if(s->t == nil){
			sa[1].repl = s->sync->to;
			sa[1].p = s->p;
			sa[1].c = c;
			threadcreate(statthread, &sa[1]);
		}
		if(s->f == nil)
			recvp(c);
		if(s->t == nil)
			recvp(c);
		free(c);
		if(s->f == nil){
			s->f = sa[0].s;
			if(s->f == nil)
				s->err = sa[0].err;
		}
		if(s->t == nil){
			s->t = sa[1].s;
			if(s->t == nil){
				if(s->err == nil)
					s->err = sa[1].err;
				else
					free(sa[1].err);
			}
		}
		free(sa);
		if(s->f == nil || s->t == nil){
			if(s->err != nil)
				s->err = estrdup("triage couldn't stat file");
			qsend(s->sync->eventq, s);
			return;
		}
	}

if(start == 0)
	start = nsec()/1000000;

	dbg(DbgSync, "%P triage decision\n\t%$\n\t%$\n", s->p, s->f, s->t);
	s->state = triagedecision(s);
	tlog("triage %P %$ %$ => %s\n", s->p, s->f, s->t,
	     syncpathstate(s->state));
	switch(s->state){
	default:
		fprint(2, "%P: bad triage decision %d\n", s->p, s->state);
		break;

	case SyncComplete:
		syncfinish(s, 1);
		break;

	case SyncIncomplete:
		syncfinish(s, 0);
		break;

	case SyncCopy:
	case SyncRemove:
		sendwork(s);
		break;

	case SyncKids:
		queuesynckids(s);
		break;

	case SyncError:
		qsend(s->sync->eventq, s);
		break;

	case SyncConflict:
		qsend(s->sync->eventq, s);
		break;
	}
}

typedef struct Kids Kids;
struct Kids
{
	Replica *repl;
	Path *p;
	Kid *k;
	int nk;
	Chan *c;
	char *err;
};
static void
kidthread(void *a)
{
	Kids *k;

	k = a;
	startclient();
	k->nk = rpckids(k->repl, k->p, &k->k);
	if(k->nk < 0)
		k->err = rpcerror();
	endclient();
	send(k->c, k);
}
static int
mergekids(Syncpath *s, Kid *f, int nf, Kid *t, int nt, int n)
{	
	int i, j, k, c;
	Syncpath *w;

	if(n){
		s->nkid = n;
		s->kid = emalloc(n*sizeof(Syncpath));
		w = s->kid;
	}else
		w = nil;

	for(i=j=k=0; i<nf || j<nt; k++){
		if(w){
			w->sync = s->sync;
			w->state = SyncStart;
			w->parent = s;
		}
		if(i>=nf)
			goto UseT;
		if(j>=nt)
			goto UseF;
		c = strcmp(f[i].name, t[j].name);
		if(c < 0){
		UseF:
			if(w){
				w->p = mkpath(s->p, f[i].name);
				w->f = f[i].stat;
				f[i].stat = nil;
				w++;
			}
			i++;
			continue;
		}
		if(c == 0){
			if(w){
				w->p = mkpath(s->p, f[i].name);
				w->f = f[i].stat;
				f[i].stat = nil;
				w->t = t[j].stat;
				t[j].stat = nil;
				w++;
			}
			i++;
			j++;
			continue;
		}
		if(c > 0){
		UseT:
			if(w){
				w->p = mkpath(s->p, t[j].name);
				w->t = t[j].stat;
				t[j].stat = nil;
				w++;
			}
			j++;
			continue;
		}
		abort();	/* not reached */
	}
	return k;
}
static int
kidnamecmp(const void *a, const void *b)
{
	Kid *ka, *kb;

	ka = (Kid*)a;
	kb = (Kid*)b;
	return strcmp(ka->name, kb->name);
}
void
queuesynckids(Syncpath *s)
{
	int i, n;
	Chan *c;
	Kids *fk, *tk;

	c = chan(Kids*);
	fk = emalloc(sizeof(*fk));
	fk->repl = s->sync->from;
	fk->p = s->p;
	fk->c = c;
	threadcreate(kidthread, fk);

	tk = emalloc(sizeof(*tk));
	tk->repl = s->sync->to;
	tk->p = s->p;
	tk->c = c;
	threadcreate(kidthread, tk);

	recvp(c);
	recvp(c);
	free(c);

	if(fk->nk)
		qsort(fk->k, fk->nk, sizeof(fk->k[0]), kidnamecmp);
	if(tk->nk)
		qsort(tk->k, tk->nk, sizeof(tk->k[0]), kidnamecmp);
	n = mergekids(s, fk->k, fk->nk, tk->k, tk->nk, 0);
	if(n == 0){
		syncfinish(s, 1);
		goto End;
	}
	mergekids(s, fk->k, fk->nk, tk->k, tk->nk, n);
	s->npend = s->nkid;
	for(i=0; i<n; i++)
		qsend(s->sync->syncq, &s->kid[i]);

End:
	freekids(fk->k, fk->nk);
	freekids(tk->k, tk->nk);
	free(fk);
	free(tk);
}

static int
triagedecision(Syncpath *s)
{
	Stat *f, *t;

	f = s->f;
	t = s->t;
	if((f->state&SNonreplicated) && (t->state&SNonreplicated)){
		/* neither cares, but the enclosing directories aren't complete */
		coverage();
		return SyncIncomplete;
	}

	if(f->state&SNonreplicated){
		/* from does not store path; to is necessarily up to date */
		coverage();
		return SyncComplete;
	}

	if(t->state&SNonreplicated){
		/* to does not store path; from is necessarily out of date */
		coverage();
		return SyncIncomplete;
	}

	/* path is replicated on both from and to */

	if(f->state==SNonexistent && t->state==SNonexistent){
		/* just propagate ghost information; success will add sync time */
		coverage();
		return SyncComplete;
	}

	if(f->state==SNonexistent){
		coverage();
		if(intersectvtime(t->ctime, f->synctime)){
			/* from knew about to's copy but deleted it ... */
			coverage();
			if(leqvtime(t->mtime, f->synctime)){
				/* ... and from's copy was the same as or newer than to's */
				coverage();
				return SyncRemove;
			}else{
				/* ... and from's copy wasn't the same as or newer than to's */
				coverage();
				s->conflict = DeleteUpdate;
				return SyncConflict;
			}
		}else{
			/* from has never seen this file; do nothing */
			coverage();
			return SyncComplete;
		}
	}

	if(t->state==SNonexistent){
		coverage();
		if(intersectvtime(f->ctime, t->synctime)){
			/* to knew about from's copy but deleted it ... */
			coverage();
			if(leqvtime(f->mtime, t->synctime)){
				/* ... and to's copy was the same as or newer than from's */
				coverage();
				return SyncComplete;
			}else{
				/* ... and to's copy wasn't the same as or newer than from's */
				coverage();
				s->conflict = UpdateDelete;
				return SyncConflict;
			}
		}else{
			/* to has never seen this file; copy it */
			coverage();
			return SyncCopy;
		}
	}

	/* path exists on both from and to */

	if(leqvtime(f->mtime, t->synctime)){
		/* to's copy is the same as or newer than from's */
		coverage();
		return SyncComplete;
	}

	if(f->state==SDir && t->state==SDir){
		/* there are differences, and we've got directories, recur */
		coverage();
		return SyncKids;
	}

	if(leqvtime(t->mtime, f->synctime)){
		/* from's copy is newer than to's copy */
		coverage();
		return SyncCopy;
	}

	/* from's copy and to's copy are different; neither is newer than the other */
	s->conflict = UpdateUpdate;
	return SyncConflict;
}

void
syncfinish(Syncpath *s, int iscomplete)
{
	int i;
	Syncpath *up;
	Vtime *m;

	dbg(DbgSync, "syncfinish %P: %s\n", s->p, iscomplete ? "complete" : "incomplete");
	if(iscomplete){
		s->state = SyncComplete;
		m = nil;
		if(s->f->state==SDir && s->t->state==SDir)
			m = s->f->mtime;
		if(rpcaddtime(s->sync->to, s->p, s->f->synctime, m) < 0){
			s->state = SyncError;
			s->err = rpcerror();
			qsend(s->sync->eventq, s);
			return;
		}
	}else
		s->state = SyncIncomplete;

	up = s->parent;
	if(up == nil){
		qsend(s->sync->eventq, s);
		return;
	}

	if(!iscomplete)
		up->badkids++;

	--up->npend;
	if(up->npend > 0)
		return;

	for(i=0; i<up->nkid; i++){
		freestat(up->kid[i].f);
		freestat(up->kid[i].t);
		freepath(up->kid[i].p);
		free(up->kid[i].err);
	}
	free(up->kid);
	up->kid = nil;
	syncfinish(up, !up->badkids);
}

void
resolve(Syncpath *s, int dir)
{
	if(dir=='t'){
		syncfinish(s, 1);
		return;
	}
	switch(s->conflict){
	default:
		abort();
	case UpdateUpdate:
	case UpdateDelete:
		s->state = SyncCopy;
		break;
	case DeleteUpdate:
		s->state = SyncRemove;
		break;
	}
	sendwork(s);
}