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

// New log records are created when objects in the local store are created,
// updated or deleted. Local log records are also replayed to keep the
// per-object dags consistent with the local store state. Sync module assigns
// each log record created within a Database a unique sequence number, called
// the generation number. Locally on each device, the position of each log
// record is also recorded relative to other local and remote log records.
//
// When a device receives a request to send log records, it first computes the
// missing generations between itself and the incoming request on a per-prefix
// basis. It then sends all the log records belonging to the missing generations
// in the order they occur locally (using the local log position). A device that
// receives log records over the network replays all the records received from
// another device in a single batch. Each replayed log record adds a new version
// to the dag of the object contained in the log record. At the end of replaying
// all the log records, conflict detection and resolution is carried out for all
// the objects learned during this iteration. Conflict detection and resolution
// is carried out after a batch of log records are replayed, instead of
// incrementally after each record is replayed, to avoid repeating conflict
// resolution already performed by other devices.
//
// Sync module tracks the current generation number and the current local log
// position for each Database. In addition, it also tracks the current
// generation vectors for a Database. Log records are indexed such that they can
// be selectively retrieved from the store for any missing generation from any
// device.
//
// Sync also tracks the current generation number and the current local log
// position for each mutation of a syncgroup, created on a Database. Similar to
// the data log records, these log records are used to sync syncgroup metadata.
//
// The generations for the data mutations and mutations for each syncgroup are
// in separate spaces. Data mutations in a Database start at gen 1, and grow.
// Mutations for each syncgroup start at gen 1, and grow. Thus, for the local
// data log records, the keys are of the form y:l:d:<devid>:<gen>, and the keys
// for local syncgroup log record are of the form y:l:<sgoid>:<devid>:<gen>.

// TODO(hpucha): Should this space be separate from the data or not? If it is
// not, it can provide consistency between data and syncgroup metadata. For
// example, lets say we mutate the data in a syncgroup and soon after change the
// syncgroup ACL to prevent syncing with a device. This device may not get the
// last batch of updates since the next time it will try to sync, it will be
// rejected. However implementing consistency is not straightforward. Even if we
// had syncgroup updates in the same space as the data, we need to switch to the
// right syncgroup ACL at the responder based on the requested generations.

import (
	"fmt"
	"time"

	"v.io/v23/context"
	wire "v.io/v23/services/syncbase"
	"v.io/v23/verror"
	"v.io/v23/vom"
	"v.io/x/lib/vlog"
	"v.io/x/ref/services/syncbase/common"
	"v.io/x/ref/services/syncbase/server/interfaces"
	"v.io/x/ref/services/syncbase/store"
	"v.io/x/ref/services/syncbase/store/watchable"
)

// localGenInfoInMem represents the state corresponding to local generations.
type localGenInfoInMem struct {
	gen        uint64
	pos        uint64
	checkptGen uint64
}

func (in *localGenInfoInMem) deepCopy() *localGenInfoInMem {
	out := &localGenInfoInMem{
		gen:        in.gen,
		pos:        in.pos,
		checkptGen: in.checkptGen,
	}
	return out
}

// dbSyncStateInMem represents the in-memory sync state of a Database and all
// its syncgroups.
type dbSyncStateInMem struct {
	data *localGenInfoInMem // info for data.

	// Info for syncgroups. The key here is the syncgroup oid of the form
	// y:s:<groupId>. More details in syncgroup.go.
	sgs map[string]*localGenInfoInMem

	// Note: Generation vectors contain state from remote devices only.
	genvecs   interfaces.Knowledge
	sggenvecs interfaces.Knowledge

	// Tracks if sync is paused for this database.
	isPaused bool
}

func (in *dbSyncStateInMem) deepCopy() *dbSyncStateInMem {
	out := &dbSyncStateInMem{}
	out.data = in.data.deepCopy()

	out.sgs = make(map[string]*localGenInfoInMem)
	for oid, info := range in.sgs {
		out.sgs[oid] = info.deepCopy()
	}

	out.genvecs = in.genvecs.DeepCopy()
	out.sggenvecs = in.sggenvecs.DeepCopy()
	out.isPaused = in.isPaused

	return out
}

// sgPublishInfo holds information on a syncgroup waiting to be published to a
// remote peer.  It is an in-memory entry in a queue of pending syncgroups.
type sgPublishInfo struct {
	sgId    wire.Id
	dbId    wire.Id
	queued  time.Time
	lastTry time.Time
}

// initSync initializes the sync module during startup. It scans all the
// databases to initialize the following:
// a) in-memory sync state of a Database and all its syncgroups consisting of
// the current generation number, log position and generation vectors.
// b) watcher map of prefixes currently being synced.
// c) republish names in mount tables for all syncgroups.
// d) in-memory queue of syncgroups to be published.
func (s *syncService) initSync(ctx *context.T) error {
	vlog.VI(2).Infof("sync: initSync: begin")
	defer vlog.VI(2).Infof("sync: initSync: end")
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()

	var errFinal error
	s.syncState = make(map[wire.Id]*dbSyncStateInMem)
	newMembers := make(map[string]*memberInfo)

	s.forEachDatabaseStore(ctx, func(dbId wire.Id, st *watchable.Store) bool {
		// Fetch the sync state for data and syncgroups.
		ds, err := getDbSyncState(ctx, st)
		if err != nil && verror.ErrorID(err) != verror.ErrNoExist.ID {
			errFinal = err
			return false
		}

		dsInMem := &dbSyncStateInMem{
			data: &localGenInfoInMem{},
			sgs:  make(map[string]*localGenInfoInMem),
		}

		if err == nil {
			// Initialize in memory state from the persistent state.
			dsInMem.genvecs = ds.GenVecs
			dsInMem.sggenvecs = ds.SgGenVecs
			dsInMem.isPaused = ds.IsPaused
		}

		vlog.VI(2).Infof("sync: initSync: initing db %v, dsInMem %v", dbId, dsInMem)

		sgCount := 0

		// Scan the syncgroups and init relevant metadata.
		forEachSyncgroup(st, func(gid interfaces.GroupId, sg *interfaces.Syncgroup) bool {
			sgCount++

			// Only use syncgroups that have been marked as
			// "watchable" by the sync watcher thread. This is to
			// handle the case of a syncgroup being created but
			// Syncbase restarting before the watcher processed the
			// SyncgroupOp entry in the watch queue. It should not
			// be syncing that syncgroup's data after restart, but
			// wait until the watcher processes the entry as would
			// have happened without a restart.
			state, err := getSGIdEntry(ctx, st, gid)
			if err != nil {
				errFinal = err
				return false
			}
			if state.Watched {
				for _, c := range sg.Spec.Collections {
					addWatchPrefixSyncgroup(dbId, toCollectionPrefixStr(c), gid)
				}
			}

			if sg.Status == interfaces.SyncgroupStatusPublishPending {
				s.enqueuePublishSyncgroup(sg.Id, dbId, false)
			}

			// Refresh membership view.
			refreshSyncgroupMembers(gid, sg, dbId, newMembers)

			sgoid := sgOID(gid)
			info := &localGenInfoInMem{}
			dsInMem.sgs[sgoid] = info

			// Adjust the gen and pos for the sgoid.
			info.gen, info.pos, err = s.computeCurGenAndPos(ctx, st, sgoid, dsInMem.sggenvecs[sgoid])
			if err != nil {
				errFinal = err
				return false
			}
			info.checkptGen = info.gen - 1

			vlog.VI(4).Infof("sync: initSync: initing db %v sg %v info %v", dbId, sgoid, info)

			return false
		})

		if sgCount == 0 {
			vlog.VI(2).Infof("sync: initSync: initing db %v done (no sgs found)", dbId)
			return false
		}

		// Compute the max known data generation for each known device.
		maxgenvec := interfaces.GenVector{}
		for _, gv := range dsInMem.genvecs {
			for dev, gen := range gv {
				if gen > maxgenvec[dev] {
					maxgenvec[dev] = gen
				}
			}
		}

		// Adjust the gen and pos for the data.
		dsInMem.data.gen, dsInMem.data.pos, err = s.computeCurGenAndPos(ctx, st, logDataPrefix, maxgenvec)
		if err != nil {
			errFinal = err
			return false
		}
		dsInMem.data.checkptGen = dsInMem.data.gen - 1

		s.syncState[dbId] = dsInMem

		vlog.VI(2).Infof("sync: initSync: initing db %v done dsInMem %v (data %v)", dbId, dsInMem, dsInMem.data)

		return false
	})

	s.allMembersLock.Lock()
	s.allMembers = &memberView{expiration: time.Now().Add(memberViewTTL), members: newMembers}
	s.allMembersLock.Unlock()

	return errFinal
}

// computeCurGenAndPos computes the current local generation count and local log
// position for data or a specified syncgroup.
func (s *syncService) computeCurGenAndPos(ctx *context.T, st store.Store, pfx string, genvec interfaces.GenVector) (uint64, uint64, error) {
	found := false

	// Scan the local log records to determine latest gen and its pos.
	stream := st.Scan(common.ScanPrefixArgs(logRecsPerDeviceScanPrefix(pfx, s.id), ""))
	defer stream.Cancel()

	// Get the last value.
	var val []byte
	for stream.Advance() {
		val = stream.Value(val)
		found = true
	}

	if err := stream.Err(); err != nil {
		return 0, 0, err
	}

	var maxpos, maxgen uint64
	if found {
		var lrec LocalLogRec
		if err := vom.Decode(val, &lrec); err != nil {
			return 0, 0, err
		}
		maxpos = lrec.Pos
		maxgen = lrec.Metadata.Gen
	}

	for id, gen := range genvec {
		if gen == 0 {
			continue
		}
		// Since log records may be filtered, we search for the last
		// available log record going backwards from the generation up
		// to which a device is caught up.
		lrec, err := getPrevLogRec(ctx, st, pfx, id, gen)
		if err != nil {
			return 0, 0, err
		}
		if lrec != nil && lrec.Pos > maxpos {
			found = true
			maxpos = lrec.Pos
		}
	}

	if found {
		maxpos++
	}

	return maxgen + 1, maxpos, nil
}

// TODO(hpucha): This can be optimized using a backwards scan or a better
// search.
func getPrevLogRec(ctx *context.T, st store.Store, pfx string, dev, gen uint64) (*LocalLogRec, error) {
	for i := gen; i > 0; i-- {
		rec, err := getLogRec(ctx, st, pfx, dev, i)
		if err == nil {
			return rec, nil
		}
		if verror.ErrorID(err) != verror.ErrNoExist.ID {
			return nil, err
		}
	}
	return nil, nil
}

// enqueuePublishSyncgroup appends the given syncgroup to the publish queue.
func (s *syncService) enqueuePublishSyncgroup(sgId, dbId wire.Id, attempted bool) {
	s.sgPublishQueueLock.Lock()
	defer s.sgPublishQueueLock.Unlock()

	entry := &sgPublishInfo{
		sgId:   sgId,
		dbId:   dbId,
		queued: time.Now(),
	}
	if attempted {
		entry.lastTry = entry.queued
	}
	s.sgPublishQueue.PushBack(entry)
}

// Note: For all the utilities below, if the sgid parameter is non-nil, the
// operation is performed in the syncgroup space. If nil, it is performed in the
// data space for the Database.

// reserveGenAndPosInDbLog reserves a chunk of generation numbers and log
// positions in a Database's log. Used when local updates result in log
// entries.
func (s *syncService) reserveGenAndPosInDbLog(ctx *context.T, dbId wire.Id, sgoid string, count uint64) (uint64, uint64) {
	return s.reserveGenAndPosInternal(dbId, sgoid, count, count)
}

// reservePosInDbLog reserves a chunk of log positions in a Database's log. Used
// when remote log records are received.
func (s *syncService) reservePosInDbLog(ctx *context.T, dbId wire.Id, sgoid string, count uint64) uint64 {
	_, pos := s.reserveGenAndPosInternal(dbId, sgoid, 0, count)
	return pos
}

func (s *syncService) reserveGenAndPosInternal(dbId wire.Id, sgoid string, genCount, posCount uint64) (uint64, uint64) {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()

	ds := s.getOrCreateSyncStateInternal(dbId)
	var info *localGenInfoInMem
	if sgoid != "" {
		var ok bool
		info, ok = ds.sgs[sgoid]
		if !ok {
			info = &localGenInfoInMem{gen: 1}
			ds.sgs[sgoid] = info
		}
	} else {
		info = ds.data
	}
	gen := info.gen
	pos := info.pos

	info.gen += genCount
	info.pos += posCount

	return gen, pos
}

// checkptLocalGen freezes the local generation number for the responder's use.
func (s *syncService) checkptLocalGen(ctx *context.T, dbId wire.Id, sgs sgSet) error {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()

	ds, ok := s.syncState[dbId]
	if !ok {
		return verror.New(verror.ErrInternal, ctx, "db state not found", dbId)
	}

	// The frozen generation is the last generation number used, i.e. one
	// below the next available one to use.
	if len(sgs) > 0 {
		// Checkpoint requested syncgroups.
		for id := range sgs {
			info, ok := ds.sgs[sgOID(id)]
			if !ok {
				return verror.New(verror.ErrInternal, ctx, "sg state not found", dbId, id)
			}
			info.checkptGen = info.gen - 1
		}
	} else {
		ds.data.checkptGen = ds.data.gen - 1
	}
	return nil
}

// initSyncStateInMem initializes the in memory sync state of the
// database/syncgroup if needed.
func (s *syncService) initSyncStateInMem(ctx *context.T, dbId wire.Id, sgoid string) {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()

	if s.syncState[dbId] == nil {
		s.syncState[dbId] = &dbSyncStateInMem{
			data: &localGenInfoInMem{gen: 1},
			sgs:  make(map[string]*localGenInfoInMem),
		}
	}
	if sgoid != "" {
		ds := s.syncState[dbId]
		if _, ok := ds.sgs[sgoid]; !ok {
			ds.sgs[sgoid] = &localGenInfoInMem{gen: 1}
		}
	}
	return
}

// copyDbSyncStateInMem returns a copy of the current in memory sync state of the Database.
func (s *syncService) copyDbSyncStateInMem(ctx *context.T, dbId wire.Id) (*dbSyncStateInMem, error) {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()

	ds, ok := s.syncState[dbId]
	if !ok {
		return nil, verror.New(verror.ErrInternal, ctx, "db state not found", dbId)
	}
	return ds.deepCopy(), nil
}

// copyDbGenInfo returns a copy of the current generation information of the Database.
func (s *syncService) copyDbGenInfo(ctx *context.T, dbId wire.Id, sgs sgSet) (interfaces.Knowledge, uint64, error) {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()

	ds, ok := s.syncState[dbId]
	if !ok {
		return nil, 0, verror.New(verror.ErrInternal, ctx, "db state not found", dbId)
	}

	var genvecs interfaces.Knowledge
	var gen uint64
	if len(sgs) > 0 {
		genvecs = make(interfaces.Knowledge)
		for id := range sgs {
			sgoid := sgOID(id)
			gv := ds.sggenvecs[sgoid]
			genvecs[sgoid] = gv.DeepCopy()
			var ckpt uint64 = 0
			if info, ok := ds.sgs[sgoid]; ok {
				ckpt = info.checkptGen
			}
			genvecs[sgoid][s.id] = ckpt
		}
	} else {
		genvecs = ds.genvecs.DeepCopy()

		// Add local generation information to the genvec.
		for _, gv := range genvecs {
			gv[s.id] = ds.data.checkptGen
		}
		gen = ds.data.checkptGen
	}
	return genvecs, gen, nil
}

// putDbGenInfoRemote puts the current remote generation information of the Database.
func (s *syncService) putDbGenInfoRemote(ctx *context.T, dbId wire.Id, sg bool, genvecs interfaces.Knowledge) error {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()

	ds, ok := s.syncState[dbId]
	if !ok {
		return verror.New(verror.ErrInternal, ctx, "db state not found", dbId)
	}

	if sg {
		ds.sggenvecs = genvecs.DeepCopy()
	} else {
		ds.genvecs = genvecs.DeepCopy()
	}

	return nil
}

// isDbSyncable checks if the given database is currently syncable.
func (s *syncService) isDbSyncable(ctx *context.T, dbId wire.Id) bool {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()
	ds := s.getOrCreateSyncStateInternal(dbId)
	return !ds.isPaused
}

// updateInMemoryPauseSt updates the in-memory state with the given isPaused state.
func (s *syncService) updateInMemoryPauseSt(ctx *context.T, dbId wire.Id, isPaused bool) {
	s.syncStateLock.Lock()
	defer s.syncStateLock.Unlock()
	ds := s.getOrCreateSyncStateInternal(dbId)
	ds.isPaused = isPaused
}

// updateDbPauseSt updates the db with the given isPaused state.
func (s *syncService) updateDbPauseSt(ctx *context.T, tx store.Transaction, dbId wire.Id, isPaused bool) error {
	vlog.VI(3).Infof("sync: updateDbPauseSt: updating sync paused for db %v with value %t", dbId, isPaused)
	ss, err := getDbSyncState(ctx, tx)
	if err != nil {
		if verror.ErrorID(err) != verror.ErrNoExist.ID {
			return err
		}
		ss = &DbSyncState{}
	}
	ss.IsPaused = isPaused
	return putDbSyncState(ctx, tx, ss)
}

func (s *syncService) getOrCreateSyncStateInternal(dbId wire.Id) *dbSyncStateInMem {
	ds, ok := s.syncState[dbId]
	if !ok {
		ds = &dbSyncStateInMem{
			data: &localGenInfoInMem{gen: 1},
			sgs:  make(map[string]*localGenInfoInMem),
		}
		s.syncState[dbId] = ds
	}
	return s.syncState[dbId]
}

////////////////////////////////////////////////////////////
// Low-level utility functions to access sync state.

// putDbSyncState persists the sync state object for a given Database.
func putDbSyncState(ctx *context.T, tx store.Transaction, ds *DbSyncState) error {
	return store.Put(ctx, tx, dbssKey, ds)
}

// getDbSyncState retrieves the sync state object for a given Database.
func getDbSyncState(ctx *context.T, st store.StoreReader) (*DbSyncState, error) {
	var ds DbSyncState
	if err := store.Get(ctx, st, dbssKey, &ds); err != nil {
		return nil, err
	}
	return &ds, nil
}

////////////////////////////////////////////////////////////
// Low-level utility functions to access log records.

// logRecsPerDeviceScanPrefix returns the prefix used to scan log records for a particular device.
func logRecsPerDeviceScanPrefix(pfx string, id uint64) string {
	return common.JoinKeyParts(logPrefix, pfx, fmt.Sprintf("%d", id))
}

// logRecKey returns the key used to access a specific log record.
func logRecKey(pfx string, id, gen uint64) string {
	return common.JoinKeyParts(logPrefix, pfx, fmt.Sprintf("%d", id), fmt.Sprintf("%016x", gen))
}

// hasLogRec returns true if the log record for (devid, gen) exists.
func hasLogRec(st store.StoreReader, pfx string, id, gen uint64) (bool, error) {
	return store.Exists(nil, st, logRecKey(pfx, id, gen))
}

// putLogRec stores the log record.
func putLogRec(ctx *context.T, tx store.Transaction, pfx string, rec *LocalLogRec) error {
	return store.Put(ctx, tx, logRecKey(pfx, rec.Metadata.Id, rec.Metadata.Gen), rec)
}

// getLogRec retrieves the log record for a given (devid, gen).
func getLogRec(ctx *context.T, st store.StoreReader, pfx string, id, gen uint64) (*LocalLogRec, error) {
	return getLogRecByKey(ctx, st, logRecKey(pfx, id, gen))
}

// getLogRecByKey retrieves the log record for a given log record key.
func getLogRecByKey(ctx *context.T, st store.StoreReader, key string) (*LocalLogRec, error) {
	var rec LocalLogRec
	if err := store.Get(ctx, st, key, &rec); err != nil {
		return nil, err
	}
	return &rec, nil
}

// delLogRec deletes the log record for a given (devid, gen).
func delLogRec(ctx *context.T, tx store.Transaction, pfx string, id, gen uint64) error {
	return store.Delete(ctx, tx, logRecKey(pfx, id, gen))
}