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

import (
	"bytes"

	"v.io/v23/context"
	"v.io/v23/naming"
	"v.io/v23/rpc"
	"v.io/v23/security/access"
	wire "v.io/v23/services/syncbase"
	"v.io/v23/services/watch"
	pubutil "v.io/v23/syncbase/util"
	"v.io/v23/verror"
	"v.io/v23/vom"
	"v.io/x/ref/services/syncbase/common"
	"v.io/x/ref/services/syncbase/server/filter"
	"v.io/x/ref/services/syncbase/server/interfaces"
	"v.io/x/ref/services/syncbase/store"
	"v.io/x/ref/services/syncbase/store/watchable"
)

// GetResumeMarker implements the wire.DatabaseWatcher interface.
func (d *database) GetResumeMarker(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle) (watch.ResumeMarker, error) {
	allowGetResumeMarker := wire.AllDatabaseTags

	if !d.exists {
		return nil, verror.New(verror.ErrNoExist, ctx, d.id)
	}
	var res watch.ResumeMarker
	impl := func(sntx store.SnapshotOrTransaction) (err error) {
		// Check permissions on Database.
		if _, err := common.GetPermsWithAuth(ctx, call, d, allowGetResumeMarker, sntx); err != nil {
			return err
		}
		res, err = watchable.GetResumeMarker(sntx)
		return err
	}
	if err := d.runWithExistingBatchOrNewSnapshot(ctx, bh, impl); err != nil {
		return nil, err
	}
	return res, nil
}

// WatchGlob implements the wire.DatabaseWatcher interface.
func (d *database) WatchGlob(ctx *context.T, call watch.GlobWatcherWatchGlobServerCall, req watch.GlobRequest) error {
	// Database permissions checked in d.watchWithFilter and d.processLogBatch.
	sender := &watchBatchSender{
		send: call.SendStream().Send,
	}
	gf, err := filter.NewGlobFilter(req.Pattern)
	if err != nil {
		return verror.New(verror.ErrBadArg, ctx, err)
	}
	return d.watchWithFilter(ctx, call, sender, req.ResumeMarker, gf)
}

// WatchPatterns implements the wire.DatabaseWatcher interface.
func (d *database) WatchPatterns(ctx *context.T, call wire.DatabaseWatcherWatchPatternsServerCall, resumeMarker watch.ResumeMarker, patterns []wire.CollectionRowPattern) error {
	// Database permissions checked in d.watchWithFilter and d.processLogBatch.
	sender := &watchBatchSender{
		send: call.SendStream().Send,
	}
	mpf, err := filter.NewMultiPatternFilter(patterns)
	if err != nil {
		return verror.New(verror.ErrBadArg, ctx, err)
	}
	return d.watchWithFilter(ctx, call, sender, resumeMarker, mpf)
}

// watchWithFilter sends the initial state (if necessary) and watch events,
// filtered using watchFilter, to the caller using sender.
func (d *database) watchWithFilter(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, resumeMarker watch.ResumeMarker, watchFilter filter.CollectionRowFilter) error {
	allowWatchDbStart := []access.Tag{access.Read}

	if !d.exists {
		return verror.New(verror.ErrNoExist, ctx, d.id)
	}
	initImpl := func(sntx store.SnapshotOrTransaction) error {
		// Check permissions on Database.
		if _, err := common.GetPermsWithAuth(ctx, call, d, allowWatchDbStart, sntx); err != nil {
			return err
		}
		needInitialState := len(resumeMarker) == 0
		needResumeMarker := needInitialState || bytes.Equal(resumeMarker, []byte("now"))
		// Get the resume marker if necessary.
		if needResumeMarker {
			var err error
			if resumeMarker, err = watchable.GetResumeMarker(sntx); err != nil {
				return err
			}
		}
		// Send the root update to notify the client that watch has started.
		rootChangeState := watch.InitialStateSkipped
		if needInitialState {
			rootChangeState = watch.Exists
		}
		if err := sender.addChange(
			"",
			rootChangeState,
			&wire.StoreChange{
				FromSync: false,
			}); err != nil {
			return err
		}
		// Send initial state if necessary.
		if needInitialState {
			if err := d.scanInitialState(ctx, call, sender, sntx, watchFilter); err != nil {
				return err
			}
		}
		// Finalize initial state or root update batch.
		return sender.finishBatch(resumeMarker)
	}
	if err := store.RunWithSnapshot(d.st, initImpl); err != nil {
		return err
	}
	return d.watchUpdates(ctx, call, sender, resumeMarker, watchFilter)
}

// scanInitialState sends the initial state of all matching and accessible
// collections and rows in the database. Checks access on collections, but
// not on database.
// Note: Assumes Read perms on database. Careful if supporting RPCs requiring
// only Resolve.
// TODO(ivanpi): Abstract out multi-scan for scan and possibly query support.
// TODO(ivanpi): Use watch pattern prefixes to optimize scan ranges.
func (d *database) scanInitialState(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, sntx store.SnapshotOrTransaction, watchFilter filter.CollectionRowFilter) error {
	// Scan matching and accessible collections.
	// TODO(ivanpi): Collection scan order not alphabetical.
	cxIt := sntx.Scan(common.ScanPrefixArgs(common.CollectionPermsPrefix, ""))
	defer cxIt.Cancel()
	cxKey, cxPermsValue := []byte{}, []byte{}
	for cxIt.Advance() {
		cxKey, cxPermsValue = cxIt.Key(cxKey), cxIt.Value(cxPermsValue)
		// Database permissions for Watch ensure that the user is always allowed
		// to know that a collection exists.
		cxId, err := common.ParseCollectionPermsKey(string(cxKey))
		if err != nil {
			return verror.New(verror.ErrInternal, ctx, err)
		}
		// Filter out unnecessary collections.
		if !watchFilter.CollectionMatches(cxId) {
			continue
		}
		// Send collection info.
		var cxPerms interfaces.CollectionPerms
		if err := vom.Decode(cxPermsValue, &cxPerms); err != nil {
			return verror.NewErrInternal(ctx) // no detailed error for cxPerms before filtering cxInfo
		}
		cxInfo := collectionInfoFromPerms(ctx, call, cxPerms.GetPerms())
		cxInfoAsRawBytes, err := vom.RawBytesFromValue(cxInfo)
		if err != nil {
			return verror.New(verror.ErrInternal, ctx, err)
		}
		if err := sender.addChange(
			pubutil.EncodeId(cxId),
			watch.Exists,
			&wire.StoreChange{
				Value: cxInfoAsRawBytes,
				// Note: FromSync cannot be reconstructed from scan.
				FromSync: false,
			}); err != nil {
			return err
		}
		// Filter out rows with no read access.
		// TODO(ivanpi): Collection scan already gets perms, optimize?
		// TODO(ivanpi): Check service and database resolve only once.
		c := &collectionReq{
			id: cxId,
			d:  d,
		}
		if _, err := common.GetPermsWithAuth(ctx, call, c, []access.Tag{access.Read}, sntx); err != nil {
			if verror.ErrorID(err) == verror.ErrNoAccess.ID {
				// Skip sending rows if the collection is inaccessible. Caller can see
				// from collection info that they have no read access and may therefore
				// have missing rows.
				// TODO(ivanpi): If read access is regained, should skipped rows be sent
				// retroactively?
				continue
			}
			return err
		}
		// Send matching rows.
		if err := c.scanInitialState(ctx, call, sender, sntx, watchFilter); err != nil {
			return err
		}
	}
	if err := cxIt.Err(); err != nil {
		return verror.New(verror.ErrInternal, ctx, err)
	}
	return nil
}

// scanInitialState sends the initial state of all matching rows in the
// collection. Does not check access.
// TODO(ivanpi): Abstract out multi-scan for scan and possibly query support.
// TODO(ivanpi): Use watch pattern prefixes to optimize scan ranges.
func (c *collectionReq) scanInitialState(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, sntx store.SnapshotOrTransaction, watchFilter filter.CollectionRowFilter) error {
	// Scan matching rows.
	rIt := sntx.Scan(common.ScanPrefixArgs(common.JoinKeyParts(common.RowPrefix, c.stKeyPart()), ""))
	defer rIt.Cancel()
	key, value := []byte{}, []byte{}
	for rIt.Advance() {
		key, value = rIt.Key(key), rIt.Value(value)
		// See comment in util/constants.go for why we use SplitNKeyParts.
		parts := common.SplitNKeyParts(string(key), 3)
		externalKey := parts[2]
		// Filter out unnecessary rows.
		if !watchFilter.RowMatches(c.id, externalKey) {
			continue
		}
		// Send row.
		var valueAsRawBytes *vom.RawBytes
		if err := vom.Decode(value, &valueAsRawBytes); err != nil {
			return verror.New(verror.ErrInternal, ctx, err)
		}
		if err := sender.addChange(
			naming.Join(pubutil.EncodeId(c.id), externalKey),
			watch.Exists,
			&wire.StoreChange{
				Value: valueAsRawBytes,
				// Note: FromSync cannot be reconstructed from scan.
				FromSync: false,
			}); err != nil {
			return err
		}
	}
	if err := rIt.Err(); err != nil {
		return verror.New(verror.ErrInternal, ctx, err)
	}
	return nil
}

// watchUpdates waits for database updates and sends them to the client.
// This function does two steps in a for loop:
// - scan through the watch log until the end, sending all updates to the client
// - wait for one of two signals: new updates available or the call is canceled
// The 'new updates' signal is sent by the watcher via a Go channel.
func (d *database) watchUpdates(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, resumeMarker watch.ResumeMarker, watchFilter filter.CollectionRowFilter) error {
	watcher, cancelWatch := d.st.WatchUpdates(resumeMarker)
	defer cancelWatch()
	for {
		// Drain the log queue.
		for {
			// TODO(ivanpi): Switch to streaming log batch entries? Since sync and
			// conflict resolution merge batches, very large batches may not be
			// unrealistic. However, sync currently also processes an entire batch at
			// a time, and would need to be updated as well.
			logs, nextResumeMarker, err := watcher.NextBatchFromLog(d.st)
			if err != nil {
				// TODO(ivanpi): Log all internal errors, especially ones not returned.
				return verror.NewErrInternal(ctx) // no detailed error before access check
			}
			if logs == nil {
				// No new log records available at this time.
				break
			}
			resumeMarker = nextResumeMarker
			if err := d.processLogBatch(ctx, call, sender, watchFilter, logs); err != nil {
				return err
			}
			if err := sender.finishBatch(resumeMarker); err != nil {
				return err
			}
		}
		// Wait for new updates or cancel.
		select {
		case _, ok := <-watcher.Wait():
			if !ok {
				return watcher.Err()
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

// processLogBatch converts []*watchable.LogEntry to a watch.Change stream,
// filtering out unnecessary or inaccessible log records.
// Note: Since the governing ACL for each change is no longer tracked, the
// permissions check uses the ACLs in effect at the time processLogBatch is
// called.
// Note: Assumes Read perms on database. Careful if supporting RPCs requiring
// only Resolve.
func (d *database) processLogBatch(ctx *context.T, call rpc.ServerCall, sender *watchBatchSender, watchFilter filter.CollectionRowFilter, logs []*watchable.LogEntry) error {
	allowWatchDbContinue := []access.Tag{access.Read}

	sn := d.st.NewSnapshot()
	defer sn.Abort()
	// Check permissions on Database.
	if _, err := common.GetPermsWithAuth(ctx, call, d, allowWatchDbContinue, sn); err != nil {
		return err
	}
	valueBytes := []byte{}
	for _, logEntry := range logs {
		var opKey string
		var op interface{}
		if err := logEntry.Op.ToValue(&op); err != nil {
			return verror.NewErrInternal(ctx) // no detailed error before access check
		}
		switch op := op.(type) {
		case *watchable.PutOp:
			opKey = string(op.Key)
		case *watchable.DeleteOp:
			opKey = string(op.Key)
		default:
			continue
		}
		// TODO(rogulenko,ivanpi): Currently we only process rows and collection
		// perms. Consider making watchable and processing other keys.
		switch common.FirstKeyPart(opKey) {
		case common.RowPrefix:
			cxId, row, err := common.ParseRowKey(opKey)
			if err != nil {
				return verror.NewErrInternal(ctx) // no detailed error before access check
			}
			// Filter out unnecessary rows.
			if !watchFilter.RowMatches(cxId, row) {
				continue
			}
			// Filter out rows with no read access.
			// TODO(ivanpi): Check only once per collection per batch.
			// TODO(ivanpi): Check service and database resolve only once per batch.
			c := &collectionReq{
				id: cxId,
				d:  d,
			}
			if _, err := common.GetPermsWithAuth(ctx, call, c, []access.Tag{access.Read}, sn); err != nil {
				if verror.ErrorID(err) == verror.ErrNoAccess.ID || verror.ErrorID(err) == verror.ErrNoExist.ID {
					// Skip sending rows if the collection is inaccessible. Caller can see
					// from collection info that they have no read access and may therefore
					// have missing rows.
					// Note, the collection may not exist anymore, in which case permissions
					// cannot be retrieved. This case is treated the same as ErrNoAccess, by
					// skipping the row.
					// TODO(ivanpi): Consider using the implicit ACL instead for nonexistent
					// collections.
					// TODO(ivanpi): If read access is regained, should skipped rows be sent
					// retroactively?
					continue
				}
				return err
			}
			switch op := op.(type) {
			case *watchable.PutOp:
				// Note, valueBytes is reused on each iteration, so the reference must not
				// be used beyond this case block. The code below is safe since only the
				// VOM-decoded copy is used after the call to vom.Decode.
				if valueBytes, err = watchable.GetAtVersion(ctx, sn, op.Key, valueBytes, op.Version); err != nil {
					return verror.New(verror.ErrInternal, ctx, err)
				}
				var rowValueAsRawBytes *vom.RawBytes
				if err := vom.Decode(valueBytes, &rowValueAsRawBytes); err != nil {
					return verror.New(verror.ErrInternal, ctx, err)
				}
				if err := sender.addChange(
					naming.Join(pubutil.EncodeId(cxId), row),
					watch.Exists,
					&wire.StoreChange{
						Value:    rowValueAsRawBytes,
						FromSync: logEntry.FromSync,
					}); err != nil {
					return err
				}
			case *watchable.DeleteOp:
				if err := sender.addChange(
					naming.Join(pubutil.EncodeId(cxId), row),
					watch.DoesNotExist,
					&wire.StoreChange{
						FromSync: logEntry.FromSync,
					}); err != nil {
					return err
				}
			}

		case common.CollectionPermsPrefix:
			// Database permissions for Watch ensure that the user is always allowed
			// to know that a collection exists.
			cxId, err := common.ParseCollectionPermsKey(opKey)
			if err != nil {
				return verror.New(verror.ErrInternal, ctx, err)
			}
			// Filter out unnecessary collections.
			if !watchFilter.CollectionMatches(cxId) {
				continue
			}
			switch op := op.(type) {
			case *watchable.PutOp:
				if valueBytes, err = watchable.GetAtVersion(ctx, sn, op.Key, valueBytes, op.Version); err != nil {
					return verror.NewErrInternal(ctx) // no detailed error for cxPerms before filtering cxInfo
				}
				var cxPerms interfaces.CollectionPerms
				if err := vom.Decode(valueBytes, &cxPerms); err != nil {
					return verror.NewErrInternal(ctx) // no detailed error for cxPerms before filtering cxInfo
				}
				cxInfo := collectionInfoFromPerms(ctx, call, cxPerms.GetPerms())
				cxInfoAsRawBytes, err := vom.RawBytesFromValue(cxInfo)
				if err != nil {
					return verror.New(verror.ErrInternal, ctx, err)
				}
				if err := sender.addChange(
					pubutil.EncodeId(cxId),
					watch.Exists,
					&wire.StoreChange{
						Value:    cxInfoAsRawBytes,
						FromSync: logEntry.FromSync,
					}); err != nil {
					return err
				}
			case *watchable.DeleteOp:
				if err := sender.addChange(
					pubutil.EncodeId(cxId),
					watch.DoesNotExist,
					&wire.StoreChange{
						FromSync: logEntry.FromSync,
					}); err != nil {
					return err
				}
			}

		default:
			continue
		}
	}
	return nil
}

// collectionInfoFromPerms converts a collection permissions object into a
// StoreChangeCollectionInfo tailored to the caller. The returned collection
// info is safe to send to the caller, assuming the caller is allowed to know
// the collection exists. It includes a set listing all access tags that the
// caller has on the collection. The collection permissions object itself is
// included only if the caller is allowed to see it (has Admin permissions).
func collectionInfoFromPerms(ctx *context.T, call rpc.ServerCall, cxPerms access.Permissions) *wire.StoreChangeCollectionInfo {
	ci := &wire.StoreChangeCollectionInfo{
		Allowed: make(map[access.Tag]struct{}),
	}
	for tag, acl := range cxPerms {
		if acl.Authorize(ctx, call.Security()) == nil {
			ci.Allowed[access.Tag(tag)] = struct{}{}
		}
	}
	if _, isAdmin := ci.Allowed[access.Admin]; isAdmin {
		ci.Perms = cxPerms
	}
	return ci
}

// watchBatchSender sends a sequence of watch changes forming a batch, delaying
// sends to allow setting the Continued flag on the last change.
type watchBatchSender struct {
	// Function for sending changes to the stream. Must be set.
	send func(item watch.Change) error

	// Change set by previous addChange, sent by next addChange or finishBatch.
	staged *watch.Change
}

// addChange sends the previously added change (if any) with Continued set to
// true and stages the new one to be sent by the next addChange or finishBatch.
func (w *watchBatchSender) addChange(name string, state int32, storeChange *wire.StoreChange) error {
	// Encode the StoreChange for sending.
	storeChangeAsRawBytes, err := vom.RawBytesFromValue(*storeChange)
	if err != nil {
		return verror.New(verror.ErrInternal, nil, err)
	}
	// Send previously staged change now that we know the batch is continuing.
	if w.staged != nil {
		w.staged.Continued = true
		if err := w.send(*w.staged); err != nil {
			return err
		}
	}
	// Stage new change.
	w.staged = &watch.Change{
		Name:  name,
		State: state,
		Value: storeChangeAsRawBytes,
	}
	return nil
}

// finishBatch sends the previously added change (if any) with Continued set to
// false, finishing the batch.
func (w *watchBatchSender) finishBatch(resumeMarker watch.ResumeMarker) error {
	// Send previously staged change as last in batch.
	if w.staged != nil {
		w.staged.Continued = false
		w.staged.ResumeMarker = resumeMarker
		if err := w.send(*w.staged); err != nil {
			return err
		}
	}
	// Clear staged change.
	w.staged = nil
	return nil
}