package streaming

import (
	"context"
	"time"

	"github.com/cockroachdb/errors"

	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
	"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
	"github.com/milvus-io/milvus/internal/streamingnode/server/wal"
	kvfactory "github.com/milvus-io/milvus/internal/util/dependency/kv"
	"github.com/milvus-io/milvus/internal/util/hookutil"
	"github.com/milvus-io/milvus/internal/util/streamingutil/util"
	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
	"github.com/milvus-io/milvus/pkg/v3/streaming/util/options"
	"github.com/milvus-io/milvus/pkg/v3/streaming/util/types"
)

var singleton WALAccesser = nil

var (
	ErrWALReleaseTimeout = errors.New("wal release timeout")
	releaseTimeout       = 5 * time.Second
)

// Init initializes the wal accesser with the given etcd client.
// should be called before any other operations.
func Init() {
	c, _ := kvfactory.GetEtcdAndPath()
	// init and select wal name
	util.InitAndSelectWALName()
	// register cipher for cipher message
	if hookutil.IsClusterEncryptionEnabled() {
		message.RegisterCipher(hookutil.GetCipher())
	}
	singleton = newWALAccesser(c)
}

// Release releases the resources of the wal accesser.
func Release() error {
	if w, ok := singleton.(*walAccesserImpl); ok && w != nil {
		done := make(chan struct{})
		go func() {
			defer close(done)
			w.Close()
		}()

		if releaseTimeout <= 0 {
			<-done
			return nil
		}
		select {
		case <-done:
			return nil
		case <-time.After(releaseTimeout):
			return errors.Wrapf(ErrWALReleaseTimeout, "timeout=%s", releaseTimeout)
		}
	}
	return nil
}

// WAL is the entrance to interact with the milvus write ahead log.
func WAL() WALAccesser {
	return singleton
}

// AppendOption is the option for append operation.
type AppendOption struct {
	BarrierTimeTick uint64 // BarrierTimeTick is the barrier time tick of the message.
	// Must be allocated from tso, otherwise undetermined behavior.
}

type ReadOption struct {
	// PChannel is the target pchannel to read, if the pchannel is not set.
	// It will be parsed from setted `VChannel`.
	PChannel string

	// VChannel is the target vchannel to read.
	// It must be set to read message from a vchannel.
	// If VChannel is empty, the PChannel must be set, and all message of pchannel will be read.
	VChannel string

	// DeliverPolicy is the deliver policy of the consumer.
	DeliverPolicy options.DeliverPolicy

	// DeliverFilters is the deliver filters of the consumer.
	DeliverFilters []options.DeliverFilter

	// Handler is the message handler used to handle message after recv from consumer.
	MessageHandler message.Handler

	// IgnorePauseConsumption is the flag to ignore the consumption pause of the scanner.
	IgnorePauseConsumption bool
}

// Scanner is the interface for reading records from the wal.
type Scanner interface {
	// Done returns a channel which will be closed when scanner is finished or closed.
	Done() <-chan struct{}

	// Error returns the error of the scanner.
	Error() error

	// Close the scanner, release the underlying resources.
	Close()
}

// ReplicateService is the interface for the replicate service.
type ReplicateService interface {
	// Append appends the message into current cluster.
	Append(ctx context.Context, msg message.ReplicateMutableMessage) (*types.AppendResult, error)

	// UpdateReplicateConfiguration updates the replicate configuration to the milvus cluster.
	UpdateReplicateConfiguration(ctx context.Context, req *milvuspb.UpdateReplicateConfigurationRequest) error

	// GetReplicateConfiguration returns the current replication configuration
	// with sensitive fields (tokens) sanitized.
	GetReplicateConfiguration(ctx context.Context) (*commonpb.ReplicateConfiguration, error)

	// GetReplicateCheckpoint returns the WAL checkpoint that will be used to create scanner
	// from the correct position, ensuring no duplicate or missing messages.
	GetReplicateCheckpoint(ctx context.Context, channelName string) (*wal.ReplicateCheckpoint, error)

	// GetSalvageCheckpoint returns all salvage checkpoints captured during force promote.
	// Returns an empty slice if no force promote has occurred.
	GetSalvageCheckpoint(ctx context.Context, channelName string) ([]*wal.ReplicateCheckpoint, error)
}

// Balancer is the interface for managing the balancer of the wal.
type Balancer interface {
	// ListStreamingNode lists the streaming node.
	ListStreamingNode(ctx context.Context) ([]types.StreamingNodeInfo, error)

	// GetWALDistribution returns the wal distribution of the streaming node.
	GetWALDistribution(ctx context.Context, nodeID int64) (*types.StreamingNodeAssignment, error)

	// GetFrozenNodeIDs returns the frozen node ids.
	GetFrozenNodeIDs(ctx context.Context) ([]int64, error)

	// IsRebalanceSuspended returns whether the rebalance of the wal is suspended.
	IsRebalanceSuspended(ctx context.Context) (bool, error)

	// SuspendRebalance suspends the rebalance of the wal.
	SuspendRebalance(ctx context.Context) error

	// ResumeRebalance resumes the rebalance of the wal.
	ResumeRebalance(ctx context.Context) error

	// FreezeNodeIDs freezes the streaming node.
	// The wal will not be assigned to the frozen nodes and the wal will be removed from the frozen nodes.
	FreezeNodeIDs(ctx context.Context, nodeIDs []int64) error

	// DefreezeNodeIDs defreezes the streaming node.
	DefreezeNodeIDs(ctx context.Context, nodeIDs []int64) error
}

// WALAccesser is the interfaces to interact with the milvus write ahead log.
type WALAccesser interface {
	// ForwardService returns the forward service of the wal.
	ForwardService() ForwardService

	// Replicate returns the replicate service of the wal.
	Replicate() ReplicateService

	// ControlChannel returns the control channel name of the wal.
	// It will return the channel name of the control channel of the wal.
	ControlChannel() string

	// Balancer returns the balancer management of the wal.
	Balancer() Balancer

	// Local returns the local services.
	Local() Local

	// PrepareReleaseManualFlush prepares process-local release handoff.
	// Returns false when the current process is not the local flush owner or
	// the channel does not need growing-source retention.
	PrepareReleaseManualFlush(ctx context.Context, collectionID int64, vchannel string, releaseSegmentIDs []int64) (bool, error)

	// RawAppend writes a records to the log.
	RawAppend(ctx context.Context, msgs message.MutableMessage, opts ...AppendOption) (*types.AppendResult, error)

	// Broadcast returns a broadcast service of wal.
	// Broadcast support cross-vchannel message broadcast.
	// It promises the atomicity written of the messages and eventual consistency.
	// And the broadcasted message must be acked cat consuming-side, otherwise resource leak on broadcast.
	// Broadcast also support the resource-key to achieve a resource-exclusive acquirsion.
	Broadcast() Broadcast

	// Read returns a scanner for reading records from the wal.
	Read(ctx context.Context, opts ReadOption) Scanner

	// AppendMessages appends messages to the wal.
	// It it a helper utility function to append messages to the wal.
	// If the messages is belong to one vchannel, it will be sent as a transaction.
	// Otherwise, it will be sent as individual messages.
	// !!! This function do not promise the atomicity and deliver order of the messages appending.
	AppendMessages(ctx context.Context, msgs ...message.MutableMessage) AppendResponses
}

type Local interface {
	// GetLatestMVCCTimestampIfLocal gets the latest mvcc timestamp of the vchannel.
	// If the wal is located at remote, it will return 0, error.
	GetLatestMVCCTimestampIfLocal(ctx context.Context, vchannel string) (uint64, error)

	// GetMetricsIfLocal gets the metrics of the local wal.
	// It will only return the metrics of the local wal but not the remote wal.
	GetMetricsIfLocal(ctx context.Context) (*types.StreamingNodeMetrics, error)
}

// Broadcast is the interface for writing broadcast message into the wal.
type Broadcast interface {
	// Ack acknowledges a broadcast message at the specified vchannel.
	// It must be called after the message is comsumed by the unique-consumer.
	// It will only return error when the ctx is canceled.
	Ack(ctx context.Context, msg message.ImmutableMessage) error
}