// 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 syncbase defines the wire API for a structured store that supports
// peer-to-peer synchronization.
//
// TODO(sadovsky): Write a detailed package description, or provide a reference
// to the Syncbase documentation.
//
// Security notes:
// The Syncbase service uses permissions tags from v23/security/access.Tag,
// restricted on each hierarchy level to tags used at that level:
// - Valid Service permissions tags are all v23/security/access.Tag tags.
// - Valid Database permissions tags are Admin, Read, Write, Resolve.
// - Valid Collection permissions tags are Admin, Read, Write.
// - Valid Syncgroup permissions tags are Admin, Read.
// Other tags are not allowed and are reserved for future use.
// Unless stated otherwise, each permissions tag requirement on a method also
// implies requiring Resolve on all levels of hierarchy up to, but excluding,
// the level requiring the tag.
// ErrNoAccess, Err[No]Exist, ErrUnknownBatch are only returned if the caller
// is allowed to call Exists on the receiver of the RPC (or the first missing
// component of the hierarchy to the receiver); otherwise, the returned error
// is ErrNoExistOrNoAccess.
package syncbase

import (
	"time"

	"v.io/v23/security/access"
	"v.io/v23/services/permissions"
	"v.io/v23/services/watch"
)

const (
	// Access tags used in Syncbase database ACLs.
	AllDatabaseTags   = []access.Tag{access.Admin, access.Read, access.Write, access.Resolve}
	// Access tags used in Syncbase collection ACLs.
	AllCollectionTags = []access.Tag{access.Admin, access.Read, access.Write}
	// Access tags used in Syncbase syncgroup ACLs.
	AllSyncgroupTags  = []access.Tag{access.Admin, access.Read}
)

// NOTE(sadovsky): Various methods below may end up needing additional options.

// TODO(sadovsky): Move "DevMode" methods elsewhere, so that they are completely
// hidden from clients. Relatedly, configure the server to not even export these
// RPC methods if the --dev flag is not set.

// Service represents a Vanadium Syncbase service.
// Service.Glob operates over Database ids, requiring Read on Service, returning
// ids sorted by blessing, then by name.
type Service interface {
	// DevModeUpdateVClock updates various bits of Syncbase virtual clock and
	// clock daemon state based on the specified options.
	//
	// Requires: Admin on Service.
	// Also requires --dev flag to be set.
	DevModeUpdateVClock(uco DevModeUpdateVClockOpts) error {access.Admin}

	// DevModeGetTime returns the current time per the Syncbase clock.
	//
	// Requires: Admin on Service.
	// Also requires --dev flag to be set.
	DevModeGetTime() (time.Time | error) {access.Admin}

	// SetPermissions and GetPermissions are included from the Object interface.
	// Permissions must include at least one admin.
	//
	// Requires: Admin on Service.
	permissions.Object
}

// Database represents a set of Collections. Batches, queries, syncgroups, and
// watch all operate at the Database level.
// Database.Glob operates over Collection ids, requiring Read on Database,
// returning ids sorted by blessing, then by name.
type Database interface {
	// Create creates this Database. Permissions must be non-nil and include at
	// least one admin.
	//
	// Requires: Write on Service.
	// Also requires the creator's blessing to match the pattern in the newly
	// created Database's id. This requirement is waived for Admin on Service.
	Create(metadata ?SchemaMetadata, perms access.Permissions) error {access.Write}

	// Destroy destroys this Database, permanently removing all of its data.
	// TODO(sadovsky): Specify what happens to syncgroups.
	//
	// Requires: Admin on Database or Service.
	Destroy() error {access.Admin}

	// Exists returns true only if this Database exists.
	//
	// Requires: at least one tag on Database, or Read or Write on Service.
	// Otherwise, ErrNoExistOrNoAccess is returned.
	Exists() (bool | error)

	// ListCollections returns a list of ids of all Collections in this Database.
	// The list is sorted by blessing, then by name.
	// This method exists on Database but not on Service because for the latter
	// we can simply use glob, while for the former glob lists only Collections
	// visible in a new snapshot of the Database, ignoring user batches.
	// (Note that the same issue is present in glob on Collection, where Scan can
	// be used instead if batch awareness is required.)
	// TODO(sadovsky): Maybe switch to streaming RPC.
	//
	// Requires: Read on Database.
	ListCollections(bh BatchHandle) ([]Id | error) {access.Read}

	// Exec executes a syncQL query with positional parameters and returns all
	// results as specified by the query's select/delete statement.
	// Concurrency semantics are documented in model.go.
	//
	// Requires: Read and/or Write on Collection, depending on the query:
	// - Read for select
	// - Read and Write for delete
	// TODO(ivanpi): Write should suffice for delete without v in WHERE clause.
	Exec(bh BatchHandle, query string, params []any) stream<_, []any> error

	// BeginBatch creates a new batch. It returns a batch handle to pass in when
	// calling batch-aware RPCs.
	// Concurrency semantics are documented in model.go.
	// All batch-aware RPCs can also be called outside a batch (with an empty
	// handle), with the exception of Commit and Abort which only make sense on
	// a batch. Note that glob RPCs are not batch-aware.
	// TODO(sadovsky): Maybe make BatchOptions optional. Also, rename 'bo' to
	// 'opts' once v.io/i/912 is resolved for Java.
	//
	// Requires: at least one tag on Database.
	BeginBatch(bo BatchOptions) (BatchHandle | error)

	// Commit persists the pending changes to the database.
	// If the batch is readonly, Commit() will fail with ErrReadOnlyBatch; Abort()
	// should be used instead.
	// If the BatchHandle is empty, Commit() will fail with ErrNotBoundToBatch.
	//
	// Requires: at least one tag on Database.
	// Also verifies that any changes to data and ACLs are allowed for the caller,
	// since the batch is signed by the committer. Since only the final value for
	// each key is committed and synced, changes to data need to be allowed by
	// the ACL before or after the batch. Specifically, adding Write permission,
	// changing a value based on it, then removing Write permission within a batch
	// is not allowed because it cannot be verified by remote peers.
	Commit(bh BatchHandle) error

	// Abort notifies the server that any pending changes can be discarded.
	// It is not strictly required, but it may allow the server to release locks
	// or other resources sooner than if it was not called.
	// If the BatchHandle is empty, Abort() will fail with ErrNotBoundToBatch.
	//
	// Requires: at least one tag on Database.
	Abort(bh BatchHandle) error

	// PauseSync pauses sync for this database. Incoming sync, as well as outgoing
	// sync of subsequent writes, will be disabled until ResumeSync is called.
	// PauseSync is idempotent.
	//
	// Requires: Admin on Database.
	PauseSync() error {access.Admin}

	// ResumeSync resumes sync for this database. ResumeSync is idempotent.
	//
	// Requires: Admin on Database.
	ResumeSync() error {access.Admin}

	// SetPermissions and GetPermissions are included from the Object interface.
	// Permissions must include at least one admin.
	//
	// Requires: Admin on Database.
	permissions.Object

	// DatabaseWatcher implements the API to watch for updates in the database.
	DatabaseWatcher

	// SyncgroupManager implements the API for managing syncgroups attached to a
	// Database.
	SyncgroupManager

	// BlobManager implements the API for managing blobs attached to rows in
	// a Database.
	BlobManager

	// SchemaManager implements the API for managing schema metadata attached
	// to a Database.
	SchemaManager

	// ConflictManager implements the API for registering resolvers, receiving
	// conflicts and sending resolutions.
	ConflictManager
}

// Collection represents a set of Rows.
// Collection.Glob operates over keys of Rows in the Collection, requiring Read
// on Collection, returning keys in a lexicographically sorted order.
type Collection interface {
	// Create creates this Collection. Permissions must be non-nil and include at
	// least one admin.
	//
	// Requires: Write on Database.
	// Also requires the creator's blessing to match the pattern in the newly
	// created Collection's id.
	Create(bh BatchHandle, perms access.Permissions) error {access.Write}

	// Destroy destroys this Collection, permanently removing all of its data.
	//
	// Requires: Admin on Collection or on Database.
	// TODO(ivanpi): Prevent for synced Collections.
	Destroy(bh BatchHandle) error {access.Admin}

	// Exists returns true only if this Collection exists.
	//
	// Requires: at least one tag on Collection, or Read or Write on Database.
	// Otherwise, ErrNoExistOrNoAccess is returned.
	// If Database does not exist, returned value is identical to
	// Database.Exists().
	Exists(bh BatchHandle) (bool | error)

	// GetPermissions returns the current Permissions for the Collection.
	//
	// Requires: Admin on Collection.
	GetPermissions(bh BatchHandle) (access.Permissions | error) {access.Admin}

	// SetPermissions replaces the current Permissions for the Collection.
	// Permissions must include at least one admin.
	//
	// Requires: Admin on Collection.
	SetPermissions(bh BatchHandle, perms access.Permissions) error {access.Admin}

	// DeleteRange deletes all rows in the given half-open range [start, limit).
	// If limit is "", all rows with keys >= start are included.
	//
	// Requires: Write on Collection.
	DeleteRange(bh BatchHandle, start, limit []byte) error {access.Write}

	// Scan returns all rows in the given half-open range [start, limit). If limit
	// is "", all rows with keys >= start are included.
	// Concurrency semantics are documented in model.go.
	// Note, we use []byte rather than string for start and limit because they
	// need not be valid UTF-8; VDL expects strings to be valid UTF-8.
	//
	// Requires: Read on Collection.
	Scan(bh BatchHandle, start, limit []byte) stream<_, KeyValue> error {access.Read}
}

// Row represents a single row in a Collection.
// All access checks are performed against the Collection ACL.
type Row interface {
	// Exists returns true only if this Row exists.
	//
	// Requires: Read or Write on Collection.
	// Otherwise, ErrNoExistOrNoAccess is returned.
	// If Collection does not exist, returned value is identical to
	// Collection.Exists().
	// Note, write methods on Row do not leak information whether the Row existed
	// before, but Write is sufficient to call Exists. Therefore, Read protects
	// Row data and listing, but not Row existence.
	Exists(bh BatchHandle) (bool | error)

	// Get returns the value for this Row.
	//
	// Requires: Read on Collection.
	Get(bh BatchHandle) (any | error) {access.Read}

	// Put writes the given value for this Row.
	//
	// Requires: Write on Collection.
	Put(bh BatchHandle, value any) error {access.Write}

	// Delete deletes this Row.
	//
	// Requires: Write on Collection.
	Delete(bh BatchHandle) error {access.Write}
}

// SyncgroupManager is the interface for syncgroup operations. The Database is
// the parent of its syncgroups for permissions checking purposes.
// TODO(hpucha): Add blessings to create/join and add a refresh method.
type SyncgroupManager interface {
	// ListSyncgroups returns a list of ids of all syncgroups attached to this
	// Database. The list is sorted by blessing, then by name.
	//
	// Requires: Read on Database.
	ListSyncgroups() ([]Id | error) {access.Read}

	// CreateSyncgroup creates a new syncgroup with the given spec.
	//
	// Requires: Write on Database.
	// Also requires the creator's blessing to match the pattern in the newly
	// created syncgroup's id.
	// Permissions in spec must allow the creator at least Read access.
	// All Collections in spec must exist and the creator must have Read access
	// on them.
	// For each Collection in spec that isn't already part of another syncgroup,
	// its permissions must be signed by a blessing matching the pattern in the
	// Collection id, and all data must be signed by a blessing currently allowed
	// to Write.
	// TODO(ivanpi): Since signatures are currently not enforced, we only check
	// that the Write permissions are not empty.
	CreateSyncgroup(sgId Id, spec SyncgroupSpec, myInfo SyncgroupMemberInfo) error

	// JoinSyncgroup joins the syncgroup.
	//
	// Requires: Write on Database and Read (no Resolve required) on syncgroup.
	// For each locally existing Collection in spec, as well as each Collection
	// in spec on the remote (joinee) Syncbase, the joiner must have Read access
	// on it.
	// For each locally existing Collection in spec that isn't already part of
	// another syncgroup, its permissions must be signed by a blessing matching
	// the pattern in the Collection id, and all data must be signed by a blessing
	// currently allowed to Write.
	// TODO(ivanpi): Since signatures are currently not enforced, we only check
	// that the Write permissions are not empty.
	JoinSyncgroup(remoteSyncbaseName string, expectedSyncbaseBlessings []string, sgId Id, myInfo SyncgroupMemberInfo) (spec SyncgroupSpec | error)

	// LeaveSyncgroup leaves the syncgroup. Previously synced data will continue
	// to be available. If the last syncgroup on a Collection is left, the data
	// will become read-only and the Collection must be destroyed before joining
	// a syncgroup that includes it.
	//
	// Requires: Write on Database.
	LeaveSyncgroup(sgId Id) error {access.Write}

	// DestroySyncgroup destroys the syncgroup. Previously synced data will
	// continue to be available to all members, equivalent to all members
	// leaving the syncgroup.
	//
	// Requires: Write on Database and Admin (no Resolve required) on syncgroup.
	DestroySyncgroup(sgId Id) error

	// EjectFromSyncgroup ejects a member from the syncgroup. The ejected member
	// will not be able to sync further, but will retain any data it has already
	// synced, equivalent to having left the syncgroup.
	//
	// Requires: Admin on syncgroup.
	// The caller cannot eject themselves.
	EjectFromSyncgroup(sgId Id, member string) error

	// GetSyncgroupSpec gets the syncgroup spec. version allows for atomic
	// read-modify-write of the spec - see comment for SetSyncgroupSpec.
	//
	// Requires: Read on syncgroup.
	GetSyncgroupSpec(sgId Id) (spec SyncgroupSpec, version string | error)

	// SetSyncgroupSpec sets the syncgroup spec. version may be either empty or
	// the value from a previous Get. If not empty, Set will only succeed if the
	// current version matches the specified one.
	//
	// Requires: Admin on syncgroup.
	// The caller must continue to have Read access.
	SetSyncgroupSpec(sgId Id, spec SyncgroupSpec, version string) error

	// GetSyncgroupMembers gets the info objects for members of the syncgroup.
	//
	// Requires: Read on syncgroup.
	GetSyncgroupMembers(sgId Id) (members map[string]SyncgroupMemberInfo | error)

	// TODO(hpucha): Allow clients to tune the behavior of sync.
	// - Suspend/ResumeSync per syncgroup instead of database
	// - Get/SetSyncPolicy with policies such as "sync only via wifi", "sync
	//   aggressively", "sync once per day"
}

// SchemaManager implements the API for managing schema metadata attached
// to a Database.
type SchemaManager interface {
	// GetSchemaMetadata retrieves schema metadata for this database.
	//
	// Requires: Read on Database.
	GetSchemaMetadata() (SchemaMetadata | error) {access.Read}

	// SetSchemaMetadata stores schema metadata for this database.
	//
	// Requires: Admin on Database.
	SetSchemaMetadata(metadata SchemaMetadata) error {access.Admin}
}

// ConflictManager interface provides all the methods necessary to handle
// conflict resolution for a given database.
type ConflictManager interface {
	// StartConflictResolver registers a resolver for the database that is
	// associated with this ConflictManager and creates a stream to receive
	// conflicts and send resolutions.
	// Batches of ConflictInfos will be sent over with the Continued field
	// within the ConflictInfo representing the batch boundary. Client must
	// respond with a batch of ResolutionInfos in the same fashion.
	// A key is under conflict if two different values were written to it
	// concurrently (in logical time), i.e. neither value is an ancestor of the
	// other in the history graph.
	// A key under conflict can be a part of a batch committed on local or
	// remote or both syncbases. ConflictInfos for all keys in these two batches
	// are grouped together. These keys may themselves be under conflict; the
	// presented batch is a transitive closure of all batches containing keys
	// under conflict.
	// For example, for local batch {key1, key2} and remote batch {key1, key3},
	// the batch sent for conflict resolution will be {key1, key2, key3}.
	// If there was another concurrent batch {key2, key4}, then the batch sent
	// for conflict resolution will be {key1, key2, key3, key4}.
	//
	// Requires: Admin on Database.
	StartConflictResolver() stream<ResolutionInfo, ConflictInfo> error {access.Admin}
}

// BlobManager is the interface for blob operations.
//
// Description of API for resumable blob creation (append-only):
// - Up until commit, a BlobRef may be used with PutBlob, GetBlobSize,
//   DeleteBlob, and CommitBlob. Blob creation may be resumed by obtaining the
//   current blob size via GetBlobSize and appending to the blob via PutBlob.
// - After commit, a blob is immutable, at which point PutBlob and CommitBlob
//   may no longer be used.
// - All other methods (GetBlob, FetchBlob, PinBlob, etc.) may only be used
//   after commit.
type BlobManager interface {
	// CreateBlob returns a BlobRef for a newly created blob.
	//
	// Requires: Write on Database.
	CreateBlob() (br BlobRef | error) {access.Write}

	// PutBlob appends the byte stream to the blob.
	//
	// Requires: Write on Database and valid BlobRef.
	PutBlob(br BlobRef) stream<[]byte, _> error {access.Write}

	// CommitBlob marks the blob as immutable.
	//
	// Requires: Write on Database and valid BlobRef.
	CommitBlob(br BlobRef) error {access.Write}

	// GetBlobSize returns the count of bytes written as part of the blob
	// (committed or uncommitted).
	//
	// Requires: at least one tag on Database and valid BlobRef.
	GetBlobSize(br BlobRef) (int64 | error)

	// DeleteBlob locally deletes the blob (committed or uncommitted).
	//
	// Requires: at least one tag on Database and valid BlobRef.
	DeleteBlob(br BlobRef) error

	// GetBlob returns the byte stream from a committed blob starting at offset.
	//
	// Requires: at least one tag on Database and valid BlobRef.
	GetBlob(br BlobRef, offset int64) stream<_, []byte> error

	// FetchBlob initiates fetching a blob if not locally found. priority
	// controls the network priority of the blob. Higher priority blobs are
	// fetched before the lower priority ones. However, an ongoing blob
	// transfer is not interrupted. Status updates are streamed back to the
	// client as fetch is in progress.
	//
	// Requires: at least one tag on Database and valid BlobRef.
	FetchBlob(br BlobRef, priority uint64) stream<_, BlobFetchStatus> error

	// PinBlob locally pins the blob so that it is not evicted.
	//
	// Requires: Write on Database and valid BlobRef.
	PinBlob(br BlobRef) error {access.Write}

	// UnpinBlob locally unpins the blob so that it can be evicted if needed.
	//
	// Requires: at least one tag on Database and valid BlobRef.
	UnpinBlob(br BlobRef) error

	// KeepBlob locally caches the blob with the specified rank. Lower
	// ranked blobs are more eagerly evicted.
	//
	// Requires: at least one tag on Database and valid BlobRef.
	KeepBlob(br BlobRef, rank uint64) error

	// TODO(hpucha): Clarify how to pick priority and rank. Add API for
	// efficient blob cloning. Options include: (1) CloneBlob RPC with an
	// array of mods that specify the offset and len for the new bytes. This
	// might need two len fields to support growing a blob in the middle
	// instead of just replacing byte for byte in the src blob. Or perhaps
	// Offset>=0 to mean "read from old blob at this offset for Length
	// bytes", and Offset<0 to mean "read the next Length Bytes from the
	// PutBlob() stream". (2) We could adopt API similar to the local blob
	// store with BlockOrFile segments, giving a more flexible way to clone
	// blobs. Also provide support for parallel blob upload.
}

// DatabaseWatcher allows a client to watch for updates to the database. For
// each watch request, the client will receive a reliable stream of watch events
// without re-ordering. Only rows and collections matching at least one of the
// patterns are returned. Rows in collections with no Read access are also
// filtered out.
//
// Watching is done by starting a streaming RPC. The RPC takes a ResumeMarker
// argument that points to a particular place in the database event log. If an
// empty ResumeMarker is provided, the WatchStream will begin with a Change
// batch containing the initial state, always starting with an empty update for
// the root entity. Otherwise, the WatchStream will contain only changes since
// the provided ResumeMarker.
// See watch.GlobWatcher for a detailed explanation of the behavior.
//
// The result stream consists of a never-ending sequence of Change messages
// (until the call fails or is canceled). Each Change contains the Name field
// with the Vanadium name of the watched entity relative to the database:
// - "<encCxId>/<rowKey>" for row updates
// - "<encCxId>" for collection updates
// - "" for the initial root entity update
// The Value field is a StoreChange.
// If the client has no access to a row specified in a change, that change is
// excluded from the result stream. Collection updates are always sent and can
// be used to determine that access to a collection is denied, potentially
// skipping rows.
//
// Note: A single Watch Change batch may contain changes from more than one
// batch as originally committed on a remote Syncbase or obtained from conflict
// resolution. However, changes from a single original batch will always appear
// in the same Change batch.
type DatabaseWatcher interface {
	// GetResumeMarker returns the ResumeMarker that points to the current end
	// of the event log. GetResumeMarker() can be called on a batch.
	//
	// Requires: at least one tag on Database.
	GetResumeMarker(bh BatchHandle) (watch.ResumeMarker | error)

	// WatchPatterns returns a stream of changes that match any of the specified
	// patterns. At least one pattern must be specified.
	//
	// Requires: Read on Database.
	WatchPatterns(resumeMarker watch.ResumeMarker, patterns []CollectionRowPattern) stream<_, watch.Change> error {access.Read}

	// Requires: Read on Database (instead of Resolve).
	watch.GlobWatcher
}

error (
	NotInDevMode() {"en": "not running with --dev=true"}
	InvalidName(name string) {"en": "invalid name: '{name}'{:_}"}
	CorruptDatabase(path string) {"en": "database corrupt, moved to '{path}'; client must create a new database"}
	UnknownBatch() {"en": "unknown batch, perhaps the server restarted"}
	NotBoundToBatch() {"en": "not bound to batch"}
	ReadOnlyBatch() {"en": "batch is read-only"}
	ConcurrentBatch() {"en": "concurrent batch"}
	BlobNotCommitted() {"en": "blob is not yet committed"}
	SyncgroupJoinFailed() {"en": "syncgroup join failed{:_}"}
	BadExecStreamHeader() {"en": "Exec stream header improperly formatted"}
	InvalidPermissionsChange() {"en": "the sequence of permission changes is invalid"}
	UnauthorizedCreateId(blessing, name string) {"en": "not authorized to create object with id blessing '{blessing}' (name '{name}'){:_}"}
	InferAppBlessingFailed(entity, name string) {"en": "failed to infer app blessing pattern for {entity} '{name}'{:_}"}
	InferUserBlessingFailed(entity, name string) {"en": "failed to infer user blessing pattern for {entity} '{name}'{:_}"}
	InferDefaultPermsFailed(entity, id string) {"en": "failed to infer default perms for user for {entity} '{id}'{:_}"}
)