// 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

// Tests for the sync watcher in Syncbase.

import (
	"bytes"
	"reflect"
	"testing"
	"time"

	wire "v.io/v23/services/syncbase"
	"v.io/v23/vom"
	_ "v.io/x/ref/runtime/factories/generic"
	"v.io/x/ref/services/syncbase/common"
	"v.io/x/ref/services/syncbase/server/interfaces"
	"v.io/x/ref/services/syncbase/store/watchable"
	sbwatchable "v.io/x/ref/services/syncbase/watchable"
)

// signpostApproxEq() returns whether Signpost objects *a and *b are approximately equal,
// differing by no more than 60s in their timestamps.
func signpostApproxEq(a *interfaces.Signpost, b *interfaces.Signpost) bool {
	// We avoid reflect.DeepEqual() because LocationData contains a
	// time.Time which is routinely given a different time zone
	// (time.Location).
	if len(a.Locations) != len(b.Locations) || len(a.SgIds) != len(b.SgIds) {
		return false
	}
	for sg := range a.SgIds {
		if _, exists := b.SgIds[sg]; !exists {
			return false
		}
	}
	for peer, aLocData := range a.Locations {
		bLocData, exists := b.Locations[peer]
		if !exists ||
			aLocData.WhenSeen.After(bLocData.WhenSeen.Add(time.Minute)) ||
			bLocData.WhenSeen.After(aLocData.WhenSeen.Add(time.Minute)) {
			return false
		}
	}
	return true
}

// TestSetResmark tests setting and getting a resume marker.
func TestSetResmark(t *testing.T) {
	svc := createService(t)
	defer destroyService(t, svc)
	st := svc.St()

	resmark, err := getResMark(nil, st)
	if err == nil || resmark != nil {
		t.Errorf("found non-existent resume marker: %s, %v", resmark, err)
	}

	wantResmark := watchable.MakeResumeMarker(1234567890)
	tx := st.NewTransaction()
	if err := setResMark(nil, tx, wantResmark); err != nil {
		t.Errorf("cannot set resume marker: %v", err)
	}
	tx.Commit()

	resmark, err = getResMark(nil, st)
	if err != nil {
		t.Errorf("cannot get new resume marker: %v", err)
	}
	if !bytes.Equal(resmark, wantResmark) {
		t.Errorf("invalid new resume: got %s instead of %s", resmark, wantResmark)
	}
}

// TestWatchPrefixes tests setting and updating the watch prefixes map.
func TestWatchPrefixes(t *testing.T) {
	svc := createService(t)
	defer destroyService(t, svc)

	if len(watchPrefixes) != 0 {
		t.Errorf("watch prefixes not empty: %v", watchPrefixes)
	}

	gid1 := interfaces.GroupId(1234)
	gid2 := interfaces.GroupId(5678)

	watchPrefixOps := []struct {
		dbId wire.Id
		key  string
		add  bool
		gid  interfaces.GroupId
	}{
		{wire.Id{"app1", "db1"}, "foo", true, gid1},
		{wire.Id{"app1", "db1"}, "bar", true, gid1},
		{wire.Id{"app2", "db1"}, "xyz", true, gid1},
		{wire.Id{"app3", "db1"}, "haha", true, gid2},
		{wire.Id{"app1", "db1"}, "foo", true, gid2},
		{wire.Id{"app1", "db1"}, "foo", false, gid1},
		{wire.Id{"app2", "db1"}, "ttt", true, gid2},
		{wire.Id{"app2", "db1"}, "ttt", true, gid1},
		{wire.Id{"app2", "db1"}, "ttt", false, gid2},
		{wire.Id{"app2", "db1"}, "ttt", false, gid1},
		{wire.Id{"app2", "db2"}, "qwerty", true, gid1},
		{wire.Id{"app3", "db1"}, "haha", true, gid1},
		{wire.Id{"app2", "db2"}, "qwerty", false, gid1},
		{wire.Id{"app3", "db1"}, "haha", false, gid2},
	}

	for _, op := range watchPrefixOps {
		if op.add {
			addWatchPrefixSyncgroup(op.dbId, op.key, op.gid)
		} else {
			rmWatchPrefixSyncgroup(op.dbId, op.key, op.gid)
		}
	}

	expPrefixes := map[wire.Id]sgPrefixes{
		wire.Id{"app1", "db1"}: sgPrefixes{"foo": sgSet{gid2: struct{}{}}, "bar": sgSet{gid1: struct{}{}}},
		wire.Id{"app2", "db1"}: sgPrefixes{"xyz": sgSet{gid1: struct{}{}}},
		wire.Id{"app3", "db1"}: sgPrefixes{"haha": sgSet{gid1: struct{}{}}},
	}
	if !reflect.DeepEqual(watchPrefixes, expPrefixes) {
		t.Errorf("invalid watch prefixes: got %v instead of %v", watchPrefixes, expPrefixes)
	}

	checkSyncableTests := []struct {
		dbId   wire.Id
		key    string
		result bool
	}{
		{wire.Id{"app1", "db1"}, "foo", true},
		{wire.Id{"app1", "db1"}, "foobar", true},
		{wire.Id{"app1", "db1"}, "bar", true},
		{wire.Id{"app1", "db1"}, "bar123", true},
		{wire.Id{"app1", "db1"}, "f", false},
		{wire.Id{"app1", "db1"}, "ba", false},
		{wire.Id{"app1", "db1"}, "xyz", false},
		{wire.Id{"app1", "db555"}, "foo", false},
		{wire.Id{"app555", "db1"}, "foo", false},
		{wire.Id{"app2", "db1"}, "xyz123", true},
		{wire.Id{"app2", "db1"}, "ttt123", false},
		{wire.Id{"app2", "db2"}, "qwerty", false},
		{wire.Id{"app3", "db1"}, "hahahoho", true},
		{wire.Id{"app3", "db1"}, "hoho", false},
		{wire.Id{"app3", "db1"}, "h", false},
	}

	for _, test := range checkSyncableTests {
		res := syncable(test.dbId, makeRowKey(test.key))
		if res != test.result {
			t.Errorf("checkSyncable: invalid output: %v, %s: got %t instead of %t", test.dbId, test.key, res, test.result)
		}
	}
}

// newLog creates a Put or Delete watch log entry.
func newLog(key, version string, delete bool) *watchable.LogEntry {
	k, v := []byte(key), []byte(version)
	log := &watchable.LogEntry{}
	if delete {
		log.Op = vom.RawBytesOf(&watchable.DeleteOp{Key: k})
	} else {
		log.Op = vom.RawBytesOf(&watchable.PutOp{Key: k, Version: v})
	}
	return log
}

// newSGLog creates a syncgroup watch log entry.
func newSGLog(gid interfaces.GroupId, prefixes []string, remove bool) *watchable.LogEntry {
	return &watchable.LogEntry{
		// Note that the type information will be written in each log entry. This
		// is inefficient. Ideally we'd split the type information from the log
		// entries themselves (we only have 6 types of them). The issue tracking
		// this is http://v.io/i/1242.
		Op: vom.RawBytesOf(&sbwatchable.SyncgroupOp{SgId: gid, Prefixes: prefixes, Remove: remove}),
	}
}

// putBlobRefData inserts data containing blob refs for a key at a version.
func putBlobRefData(t *testing.T, tx *watchable.Transaction, key, version string) {
	data := struct {
		Msg  string
		Blob wire.BlobRef
	}{
		"hello world",
		wire.BlobRef("br_" + key + "_" + version),
	}

	buf, err := vom.Encode(&data)
	if err != nil {
		t.Errorf("putBlobRefData: cannot vom-encode data: key %s, version %s: %v", key, version, err)
	}
	if err := watchable.PutAtVersion(nil, tx, []byte(key), buf, []byte(version)); err != nil {
		t.Errorf("putBlobRefData: cannot PutAtVersion: key %s, version %s: %v", key, version, err)
	}
}

// TestProcessWatchLogBatch tests the processing of a batch of log records.
func TestProcessWatchLogBatch(t *testing.T) {
	svc := createService(t)
	defer destroyService(t, svc)
	st := createDatabase(t, svc).St()
	s := svc.sync

	fooKey := makeRowKey("foo")
	barKey := makeRowKey("bar")
	fooxyzKey := makeRowKey("fooxyz")

	// Empty logs does not fail.
	s.processWatchLogBatch(nil, mockDbId, st, nil, nil)

	// Non-syncable logs.
	batch := []*watchable.LogEntry{
		newLog(fooKey, "123", false),
		newLog(barKey, "555", false),
	}

	resmark := watchable.MakeResumeMarker(1234)
	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)

	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
	}
	if ok, err := hasNode(nil, st, fooKey, "123"); err != nil || ok {
		t.Error("hasNode() found DAG entry for non-syncable log on foo")
	}
	if ok, err := hasNode(nil, st, barKey, "555"); err != nil || ok {
		t.Error("hasNode() found DAG entry for non-syncable log on bar")
	}

	// Partially syncable logs.
	gid := interfaces.GroupId(1234)
	state := &SgLocalState{}
	tx := st.NewWatchableTransaction()
	if err := setSGIdEntry(nil, tx, gid, state); err != nil {
		t.Fatalf("setSGIdEntry() failed for gid %v", gid)
	}
	putBlobRefData(t, tx, fooKey, "333")
	putBlobRefData(t, tx, fooxyzKey, "444")
	if err := tx.Commit(); err != nil {
		t.Fatalf("cannot commit putting sg local state and some blob refs, err %v", err)
	}

	batch = []*watchable.LogEntry{
		newSGLog(gid, []string{"f", "x"}, false),
		newLog(fooKey, "333", false),
		newLog(fooxyzKey, "444", false),
		newLog(barKey, "222", false),
	}

	resmark = watchable.MakeResumeMarker(3456)
	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)

	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
	}
	if head, err := getHead(nil, st, fooKey); err != nil && head != "333" {
		t.Errorf("getHead() did not find foo: %s, %v", head, err)
	}
	node, err := getNode(nil, st, fooKey, "333")
	if err != nil {
		t.Errorf("getNode() did not find foo: %v", err)
	}
	if node.Level != 0 || node.Parents != nil || node.Logrec == "" || node.BatchId != NoBatchId {
		t.Errorf("invalid DAG node for foo: %v", node)
	}
	node2, err := getNode(nil, st, fooxyzKey, "444")
	if err != nil {
		t.Errorf("getNode() did not find fooxyz: %v", err)
	}
	if node2.Level != 0 || node2.Parents != nil || node2.Logrec == "" || node2.BatchId != NoBatchId {
		t.Errorf("invalid DAG node for fooxyz: %v", node2)
	}
	if ok, err := hasNode(nil, st, barKey, "222"); err != nil || ok {
		t.Error("hasNode() found DAG entry for non-syncable log on bar")
	}

	tx = st.NewWatchableTransaction()
	putBlobRefData(t, tx, fooKey, "1")
	if err := tx.Commit(); err != nil {
		t.Fatalf("cannot commit putting more blob refs, err %v", err)
	}

	// More partially syncable logs updating existing ones.
	batch = []*watchable.LogEntry{
		newLog(fooKey, "1", false),
		newLog(fooxyzKey, "", true),
		newLog(barKey, "7", false),
	}

	resmark = watchable.MakeResumeMarker(7890)
	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)

	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
	}
	if head, err := getHead(nil, st, fooKey); err != nil && head != "1" {
		t.Errorf("getHead() did not find foo: %s, %v", head, err)
	}
	node, err = getNode(nil, st, fooKey, "1")
	if err != nil {
		t.Errorf("getNode() did not find foo: %v", err)
	}
	expParents := []string{"333"}
	if node.Level != 1 || !reflect.DeepEqual(node.Parents, expParents) ||
		node.Logrec == "" || node.BatchId == NoBatchId {
		t.Errorf("invalid DAG node for foo: %v", node)
	}
	head2, err := getHead(nil, st, fooxyzKey)
	if err != nil {
		t.Errorf("getHead() did not find fooxyz: %v", err)
	}
	node2, err = getNode(nil, st, fooxyzKey, head2)
	if err != nil {
		t.Errorf("getNode() did not find fooxyz: %v", err)
	}
	expParents = []string{"444"}
	if node2.Level != 1 || !reflect.DeepEqual(node2.Parents, expParents) ||
		node2.Logrec == "" || node2.BatchId == NoBatchId {
		t.Errorf("invalid DAG node for fooxyz: %v", node2)
	}
	if ok, err := hasNode(nil, st, barKey, "7"); err != nil || ok {
		t.Error("hasNode() found DAG entry for non-syncable log on bar")
	}

	// Back to non-syncable logs (remove "f" prefix).
	batch = []*watchable.LogEntry{
		newSGLog(gid, []string{"f"}, true),
		newLog(fooKey, "99", false),
		newLog(fooxyzKey, "888", true),
		newLog(barKey, "007", false),
	}

	resmark = watchable.MakeResumeMarker(20212223)
	s.processWatchLogBatch(nil, mockDbId, st, batch, resmark)

	if res, err := getResMark(nil, st); err != nil && !bytes.Equal(res, resmark) {
		t.Errorf("invalid resmark batch processing: got %s instead of %s", res, resmark)
	}
	// No changes to "foo".
	if head, err := getHead(nil, st, fooKey); err != nil && head != "333" {
		t.Errorf("getHead() did not find foo: %s, %v", head, err)
	}
	if node, err := getNode(nil, st, fooKey, "99"); err == nil {
		t.Errorf("getNode() should not have found foo @ 99: %v", node)
	}
	if node, err := getNode(nil, st, fooxyzKey, "888"); err == nil {
		t.Errorf("getNode() should not have found fooxyz @ 888: %v", node)
	}
	if ok, err := hasNode(nil, st, barKey, "007"); err != nil || ok {
		t.Error("hasNode() found DAG entry for non-syncable log on bar")
	}

	// Verify the blob ref metadata.
	expBlobDir := make(map[wire.BlobRef]*interfaces.Signpost)
	expSp := &interfaces.Signpost{
		Locations: interfaces.PeerToLocationDataMap{s.name: interfaces.LocationData{WhenSeen: time.Now()}},
		SgIds:     sgSet{gid: struct{}{}},
	}
	expBlobDir[wire.BlobRef("br_"+fooKey+"_333")] = expSp
	expBlobDir[wire.BlobRef("br_"+fooxyzKey+"_444")] = expSp
	expBlobDir[wire.BlobRef("br_"+fooKey+"_1")] = expSp

	sps := s.bst.NewSignpostStream(svc.sync.ctx)
	spsCount := 0
	for sps.Advance() {
		spsCount++
		if expSp, ok := expBlobDir[sps.BlobId()]; !ok {
			t.Errorf("unexpectedly got unexpected signpost for blob %v", sps.BlobId())
		} else if sp := sps.Signpost(); !signpostApproxEq(&sp, expSp) {
			t.Errorf("signpost for blob %v: got %v, want %v", sps.BlobId(), sps.Signpost(), *expSp)
		}
	}
	if err := sps.Err(); err != nil {
		t.Errorf("NewSignpostStream yielded error  %v\n", err)
	}
	if spsCount != len(expBlobDir) {
		t.Errorf("got %d signposts, expected %d", spsCount, len(expBlobDir))
	}

	// Scan the batch records and verify that there is only 1 DAG batch
	// stored, with a total count of 3 and a map of 2 syncable entries.
	// This is because the 1st batch, while containing syncable keys, is a
	// syncgroup snapshot that does not get assigned a batch ID.  The 2nd
	// batch is an application batch with 3 keys of which 2 are syncable.
	// The 3rd batch is also a syncgroup snapshot.
	count := 0
	start, limit := common.ScanPrefixArgs(dagBatchPrefix, "")
	stream := st.Scan(start, limit)
	for stream.Advance() {
		count++
		key := string(stream.Key(nil))
		var info BatchInfo
		if err := vom.Decode(stream.Value(nil), &info); err != nil {
			t.Errorf("cannot decode batch %s: %v", key, err)
		}
		if info.Count != 3 {
			t.Errorf("wrong total count in batch %s: got %d instead of 3", key, info.Count)
		}
		if n := len(info.Objects); n != 2 {
			t.Errorf("wrong object count in batch %s: got %d instead of 2", key, n)
		}
	}
	if count != 1 {
		t.Errorf("wrong count of batches: got %d instead of 1", count)
	}
}