// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package vsync

import (
	"bytes"
	"reflect"
	"testing"

	wire "v.io/v23/services/syncbase/nosql"
	"v.io/x/ref/services/syncbase/server/interfaces"
	"v.io/x/ref/services/syncbase/server/util"
	"v.io/x/ref/services/syncbase/server/watchable"
	"v.io/x/ref/services/syncbase/store"
)

/*
Test setup:

Group1:
Oid: x, isConflict = true, has Local update, has remote update, has ancestor, resolution = pickLocal

Group2:
Oid: b, isConflict = true, has Local update, remote deleted, has ancestor, resolution = pickRemote

Group3:
Oid: c, isConflict = true, Local deleted, has remote update, has ancestor, resolution = pickRemote

Group4:
Oid: p, isConflict = true, has Local update, has remote update, has ancestor, resolution = createNew

Group5:
Oid: y, isConflict = true, has Local update, has remote update, has no ancestor, resolution = pickRemote
Oid: z, isConflict = false, no local update, has remote update, has unknown ancestor, resolution = pickRemote
Oid: e, isConflict = false, no local update, has remote update, has unknown ancestor, resolution = createNew
Oid: f, isConflict = false, no local update, has remote update, has unknown ancestor, resolution = pickLocal
Oid: g, isConflict = false, no local update, has remote update, has unknown ancestor, local head is deleted, resolution = pickLocal
Oid: a, local value rubberbanded in due to localBatch, resolution = createNew

localBatch: {y, a}
remoteBatch: {y, z, e, f}
*/

var (
	updObjectsAppResolves = map[string]*objConflictState{
		// group1
		x: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
		// group2
		b: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
		// group3
		c: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),
		// group4
		p: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, true /*hasAncestor*/),

		// group5
		y: createObjConflictState(true /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
		z: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
		a: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, false /*hasRemote*/, false /*hasAncestor*/),

		e: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
		f: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
		g: createObjConflictState(false /*isConflict*/, true /*hasLocal*/, true /*hasRemote*/, false /*hasAncestor*/),
	}

	localBatchId  uint64 = 34
	remoteBatchId uint64 = 58
)

func createGroupedCrTestData() *groupedCrData {
	groupedCrTestData := &groupedCrData{oids: map[string]bool{}}
	var group *crGroup
	// group1
	group = newGroup()
	addToGroup(group, x, NoBatchId, -1)
	groupedCrTestData.oids[x] = true
	groupedCrTestData.groups = append(groupedCrTestData.groups, group)

	// group2
	group = newGroup()
	addToGroup(group, b, NoBatchId, -1)
	groupedCrTestData.oids[b] = true
	groupedCrTestData.groups = append(groupedCrTestData.groups, group)

	// group3
	group = newGroup()
	addToGroup(group, c, NoBatchId, -1)
	groupedCrTestData.oids[c] = true
	groupedCrTestData.groups = append(groupedCrTestData.groups, group)

	// group4
	group = newGroup()
	addToGroup(group, p, NoBatchId, -1)
	groupedCrTestData.oids[p] = true
	groupedCrTestData.groups = append(groupedCrTestData.groups, group)

	// group5
	group = newGroup()
	addToGroup(group, y, localBatchId, wire.BatchSourceLocal)
	addToGroup(group, y, remoteBatchId, wire.BatchSourceRemote)
	addToGroup(group, z, remoteBatchId, wire.BatchSourceRemote)
	addToGroup(group, e, remoteBatchId, wire.BatchSourceRemote)
	addToGroup(group, f, remoteBatchId, wire.BatchSourceRemote)
	addToGroup(group, g, remoteBatchId, wire.BatchSourceRemote)
	addToGroup(group, a, localBatchId, wire.BatchSourceLocal)
	groupedCrTestData.oids[y] = true
	groupedCrTestData.oids[z] = true
	groupedCrTestData.oids[a] = true
	groupedCrTestData.oids[e] = true
	groupedCrTestData.oids[f] = true
	groupedCrTestData.groups = append(groupedCrTestData.groups, group)

	return groupedCrTestData
}

func createAndSaveNodeAndLogRecDataAppResolves(iSt *initiationState) {
	// group1
	saveNodeAndLogRec(iSt.tx, x, updObjectsAppResolves[x].oldHead, 24, false)
	saveNodeAndLogRec(iSt.tx, x, updObjectsAppResolves[x].newHead, 25, false)
	saveNodeAndLogRec(iSt.tx, x, updObjectsAppResolves[x].ancestor, 20, false)

	// group2
	saveNodeAndLogRec(iSt.tx, b, updObjectsAppResolves[b].oldHead, 56, false)
	saveNodeAndLogRec(iSt.tx, b, updObjectsAppResolves[b].newHead, 23, true)
	saveNodeAndLogRec(iSt.tx, b, updObjectsAppResolves[b].ancestor, 15, false)

	// group3
	saveNodeAndLogRec(iSt.tx, c, updObjectsAppResolves[c].oldHead, 56, true)
	saveNodeAndLogRec(iSt.tx, c, updObjectsAppResolves[c].newHead, 23, false)
	saveNodeAndLogRec(iSt.tx, c, updObjectsAppResolves[c].ancestor, 15, false)

	// group4
	saveNodeAndLogRec(iSt.tx, p, updObjectsAppResolves[p].oldHead, 56, false)
	saveNodeAndLogRec(iSt.tx, p, updObjectsAppResolves[p].newHead, 23, false)
	saveNodeAndLogRec(iSt.tx, p, updObjectsAppResolves[p].ancestor, 15, false)

	//group5
	saveNodeAndLogRec(iSt.tx, y, updObjectsAppResolves[y].oldHead, 56, false)
	saveNodeAndLogRec(iSt.tx, y, updObjectsAppResolves[y].newHead, 23, false)

	saveNodeAndLogRec(iSt.tx, z, updObjectsAppResolves[z].oldHead, 56, false)
	saveNodeAndLogRec(iSt.tx, z, updObjectsAppResolves[z].newHead, 23, false)

	saveNodeAndLogRec(iSt.tx, e, updObjectsAppResolves[e].oldHead, 56, false)
	saveNodeAndLogRec(iSt.tx, e, updObjectsAppResolves[e].newHead, 23, false)

	saveNodeAndLogRec(iSt.tx, f, updObjectsAppResolves[f].oldHead, 56, false)
	saveNodeAndLogRec(iSt.tx, f, updObjectsAppResolves[f].newHead, 23, false)

	saveNodeAndLogRec(iSt.tx, g, updObjectsAppResolves[g].oldHead, 56, true)
	saveNodeAndLogRec(iSt.tx, g, updObjectsAppResolves[g].newHead, 23, false)

	saveNodeAndLogRec(iSt.tx, a, updObjectsAppResolves[a].oldHead, 56, false)
}

func writeVersionedValues(t *testing.T, iSt *initiationState) {
	// group1
	saveValue(t, iSt.tx, x, updObjectsAppResolves[x].oldHead)
	saveValue(t, iSt.tx, x, updObjectsAppResolves[x].newHead)
	saveValue(t, iSt.tx, x, updObjectsAppResolves[x].ancestor)

	// group2
	saveValue(t, iSt.tx, b, updObjectsAppResolves[b].oldHead)
	saveValue(t, iSt.tx, b, updObjectsAppResolves[b].ancestor)

	// group3
	saveValue(t, iSt.tx, c, updObjectsAppResolves[c].newHead)
	saveValue(t, iSt.tx, c, updObjectsAppResolves[c].ancestor)

	// group4
	saveValue(t, iSt.tx, p, updObjectsAppResolves[p].oldHead)
	saveValue(t, iSt.tx, p, updObjectsAppResolves[p].newHead)
	saveValue(t, iSt.tx, p, updObjectsAppResolves[p].ancestor)

	// group5
	saveValue(t, iSt.tx, y, updObjectsAppResolves[y].oldHead)
	saveValue(t, iSt.tx, y, updObjectsAppResolves[y].newHead)

	saveValue(t, iSt.tx, z, updObjectsAppResolves[z].oldHead)
	saveValue(t, iSt.tx, z, updObjectsAppResolves[z].newHead)

	saveValue(t, iSt.tx, e, updObjectsAppResolves[e].oldHead)
	saveValue(t, iSt.tx, e, updObjectsAppResolves[e].newHead)

	saveValue(t, iSt.tx, f, updObjectsAppResolves[f].oldHead)
	saveValue(t, iSt.tx, f, updObjectsAppResolves[f].newHead)

	// No value for oldHead for g as g was deleted on local.
	saveValue(t, iSt.tx, g, updObjectsAppResolves[g].newHead)

	saveValue(t, iSt.tx, a, updObjectsAppResolves[a].oldHead)
}

func setResInfoData(mockCrs *conflictResolverStream) {
	// group1
	addResInfo(mockCrs, x, wire.ValueSelectionLocal, nil, false)
	// group2
	addResInfo(mockCrs, b, wire.ValueSelectionRemote, nil, false)
	// group3
	addResInfo(mockCrs, c, wire.ValueSelectionRemote, nil, false)
	// group4
	addResInfo(mockCrs, p, wire.ValueSelectionOther, []byte("newValue"), false)
	// group5
	addResInfo(mockCrs, y, wire.ValueSelectionRemote, nil, true)
	addResInfo(mockCrs, z, wire.ValueSelectionRemote, nil, true)
	addResInfo(mockCrs, e, wire.ValueSelectionOther, []byte("newValue"), true)
	addResInfo(mockCrs, f, wire.ValueSelectionLocal, nil, true)
	addResInfo(mockCrs, g, wire.ValueSelectionLocal, nil, true)
	addResInfo(mockCrs, a, wire.ValueSelectionOther, []byte("newValue"), false)
	setMockCRStream(mockCrs)
}

func TestResolveViaApp(t *testing.T) {
	service := createService(t)
	defer destroyService(t, service)

	iSt := &initiationState{
		updObjects: updObjectsAppResolves,
		tx:         service.St().NewTransaction(),
		config: &initiationConfig{
			sync: service.sync,
			db:   newDb(t, service.sync),
		},
	}
	createAndSaveNodeAndLogRecDataAppResolves(iSt)
	writeVersionedValues(t, iSt)
	mockCrs := &conflictResolverStream{}
	setResInfoData(mockCrs)
	groupedCrTestData := createGroupedCrTestData()

	if err := iSt.resolveViaApp(nil, groupedCrTestData); err != nil {
		t.Errorf("Error returned by resolveViaApp: %v", err)
	}

	// verify
	verifyConflictInfos(t, mockCrs)
	// group1
	verifyResolution(t, iSt.updObjects, x, pickLocal)
	// group2
	verifyResolution(t, iSt.updObjects, b, pickRemote)
	// group3
	verifyResolution(t, iSt.updObjects, c, pickRemote)
	// group4
	verifyCreateNew(t, iSt.updObjects, p, false)
	// group5
	verifyResolution(t, iSt.updObjects, y, pickRemote)
	verifyResolution(t, iSt.updObjects, z, pickRemote)
	verifyCreateNew(t, iSt.updObjects, e, false)
	verifyCreateNew(t, iSt.updObjects, f, false)
	verifyCreateNew(t, iSt.updObjects, g, true)
	verifyCreateNew(t, iSt.updObjects, a, false)

	// verify batch ids
	verifyBatchId(t, iSt.updObjects, NoBatchId, x, b, c, p)
	bid := iSt.updObjects[y].res.batchId
	if bid == NoBatchId {
		t.Errorf("BatchId for group5 should not be NoBatchId")
	}
	verifyBatchId(t, iSt.updObjects, bid, y, z, e, f, a)
}

func verifyConflictInfos(t *testing.T, mockCrs *conflictResolverStream) {
	var ci wire.ConflictInfo
	if len(mockCrs.sendQ) != 12 {
		t.Errorf("ConflictInfo count expected: %v, actual: %v", 9, len(mockCrs.sendQ))
	}

	// group1
	ci = mockCrs.sendQ[0]
	checkConflictRow(t, x, ci, []uint64{}, false /*localDeleted*/, false /*remoteDeleted*/)
	checkContinued(t, mockCrs.sendQ[0:1])

	// group2
	ci = mockCrs.sendQ[1]
	checkConflictRow(t, b, ci, []uint64{}, false /*localDeleted*/, true /*remoteDeleted*/)
	checkContinued(t, mockCrs.sendQ[1:2])

	// group3
	ci = mockCrs.sendQ[2]
	checkConflictRow(t, c, ci, []uint64{}, true /*localDeleted*/, false /*remoteDeleted*/)
	checkContinued(t, mockCrs.sendQ[2:3])

	// group4
	ci = mockCrs.sendQ[3]
	checkConflictRow(t, p, ci, []uint64{}, false /*localDeleted*/, false /*remoteDeleted*/)
	checkContinued(t, mockCrs.sendQ[3:4])

	// group5
	batchMap := map[uint64]wire.ConflictInfo{}
	batchMap[getBid(mockCrs.sendQ[4])] = mockCrs.sendQ[4]
	batchMap[getBid(mockCrs.sendQ[5])] = mockCrs.sendQ[5]
	ci = batchMap[localBatchId]
	checkConflictBatch(t, localBatchId, ci, wire.BatchSourceLocal)
	ci = batchMap[remoteBatchId]
	checkConflictBatch(t, remoteBatchId, ci, wire.BatchSourceRemote)

	rowMap := map[string]wire.ConflictInfo{}
	rowMap[getOid(mockCrs.sendQ[6])] = mockCrs.sendQ[6]
	rowMap[getOid(mockCrs.sendQ[7])] = mockCrs.sendQ[7]
	rowMap[getOid(mockCrs.sendQ[8])] = mockCrs.sendQ[8]
	rowMap[getOid(mockCrs.sendQ[9])] = mockCrs.sendQ[9]
	rowMap[getOid(mockCrs.sendQ[10])] = mockCrs.sendQ[10]
	rowMap[getOid(mockCrs.sendQ[11])] = mockCrs.sendQ[11]
	ci = rowMap[y]
	checkConflictRow(t, y, ci, []uint64{localBatchId, remoteBatchId}, false /*localDeleted*/, false /*remoteDeleted*/)
	ci = rowMap[z]
	checkConflictRow(t, z, ci, []uint64{remoteBatchId}, false /*localDeleted*/, false /*remoteDeleted*/)
	ci = rowMap[a]
	checkConflictRow(t, a, ci, []uint64{localBatchId}, false /*localDeleted*/, false /*remoteDeleted*/)
	ci = rowMap[e]
	checkConflictRow(t, e, ci, []uint64{remoteBatchId}, false /*localDeleted*/, false /*remoteDeleted*/)
	ci = rowMap[f]
	checkConflictRow(t, f, ci, []uint64{remoteBatchId}, false /*localDeleted*/, false /*remoteDeleted*/)
	ci = rowMap[g]
	checkConflictRow(t, g, ci, []uint64{remoteBatchId}, true /*localDeleted*/, false /*remoteDeleted*/)

	checkContinued(t, mockCrs.sendQ[4:])
}

func checkContinued(t *testing.T, infoGroup []wire.ConflictInfo) {
	lastIndex := len(infoGroup) - 1
	for i := range infoGroup {
		if infoGroup[i].Continued != (i != lastIndex) {
			t.Errorf("Wrong value for continued field in %#v", infoGroup[i])
		}
	}
}

func getOid(ci wire.ConflictInfo) string {
	ciData := ci.Data.(wire.ConflictDataRow).Value
	writeOp := ciData.Op.(wire.OperationWrite).Value
	return toRowKey(writeOp.Key)
}

func getBid(ci wire.ConflictInfo) uint64 {
	ciData := ci.Data.(wire.ConflictDataBatch).Value
	return ciData.Id
}

func checkConflictBatch(t *testing.T, bid uint64, ci wire.ConflictInfo, source wire.BatchSource) {
	ciData := ci.Data.(wire.ConflictDataBatch).Value
	if ciData.Source != source {
		t.Errorf("Source for batchid %v expected: %#v, actual: %#v", bid, source, ciData.Source)
	}
	if ci.Continued != true {
		t.Errorf("Bid: %v, Unexpected value for continued: %v", bid, ci.Continued)
	}
}

func checkConflictRow(t *testing.T, oid string, ci wire.ConflictInfo, batchIds []uint64, localDeleted, remoteDeleted bool) {
	ciData := ci.Data.(wire.ConflictDataRow).Value
	writeOp := ciData.Op.(wire.OperationWrite).Value
	st, _ := updObjectsAppResolves[oid]
	if (st.ancestor != NoVersion) && !bytes.Equal(makeValue(oid, st.ancestor), writeOp.AncestorValue.Bytes) {
		t.Errorf("Oid: %v, Ancestor value expected: %v, actual: %v", oid, string(makeValue(oid, st.ancestor)), string(writeOp.AncestorValue.Bytes))
	}
	if remoteDeleted && writeOp.RemoteValue == nil {
		t.Errorf("Oid: %v, for remote deleted remote value is expected to have an instance with no bytes")
	}
	if !remoteDeleted && (st.newHead != NoVersion) && !bytes.Equal(makeValue(oid, st.newHead), writeOp.RemoteValue.Bytes) {
		t.Errorf("Oid: %v, Remote value expected: %v, actual: %v", oid, string(makeValue(oid, st.newHead)), string(writeOp.RemoteValue.Bytes))
	}
	if localDeleted && writeOp.LocalValue == nil {
		t.Errorf("Oid: %v, for local deleted local value is expected to have an instance with no bytes")
	}
	if !localDeleted && (st.oldHead != NoVersion) && !bytes.Equal(makeValue(oid, st.oldHead), writeOp.LocalValue.Bytes) {
		t.Errorf("Oid: %v, Local value expected: %v, actual: %v", oid, string(makeValue(oid, st.oldHead)), string(writeOp.LocalValue.Bytes))
	}
	if !reflect.DeepEqual(ciData.BatchIds, batchIds) {
		t.Errorf("Oid: %v, BatchIds expected: %v, actual: %v", oid, batchIds, ciData.BatchIds)
	}
}

func addResInfo(crs *conflictResolverStream, oid string, sel wire.ValueSelection, val []byte, cntd bool) {
	var valRes *wire.Value
	if val != nil {
		valRes = &wire.Value{
			Bytes: val,
		}
	}
	rInfo := wire.ResolutionInfo{
		Key:       util.StripFirstKeyPartOrDie(oid),
		Selection: sel,
		Result:    valRes,
		Continued: cntd,
	}
	crs.recvQ = append(crs.recvQ, rInfo)
}

func saveValue(t *testing.T, tx store.Transaction, oid, version string) {
	if err := watchable.PutAtVersion(nil, tx, []byte(oid), makeValue(oid, version), []byte(version)); err != nil {
		t.Errorf("Failed to write versioned value for oid,ver: %s,%s", oid, version)
		t.FailNow()
	}
}

func makeValue(oid, ver string) []byte {
	return []byte(oid + ver)
}

func addToGroup(group *crGroup, oid string, bid uint64, source wire.BatchSource) {
	if bid != NoBatchId {
		group.batchSource[bid] = source
		group.batchesByOid[oid] = append(group.batchesByOid[oid], bid)
	} else {
		group.batchesByOid[oid] = []uint64{}
	}
}

func newGroup() *crGroup {
	return &crGroup{
		batchSource:  map[uint64]wire.BatchSource{},
		batchesByOid: map[string][]uint64{},
	}
}

func newDb(t *testing.T, s *syncService) interfaces.Database {
	app, err := s.sv.App(nil, nil, "mockApp")
	if err != nil {
		t.Errorf("Error while creating App: %v", err)
	}
	db, err := app.NoSQLDatabase(nil, nil, "mockDB")
	if err != nil {
		t.Errorf("Error while creating Database: %v", err)
	}
	return db
}