// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.

package remotecluster

import (
	"context"
	"net"
	"net/http"
	"sync"
	"sync/atomic"
	"time"

	"github.com/mattermost/mattermost/server/public/model"
	"github.com/mattermost/mattermost/server/public/shared/mlog"
	"github.com/mattermost/mattermost/server/v8/channels/store"
	"github.com/mattermost/mattermost/server/v8/einterfaces"
)

const (
	SendChanBuffer                = 50
	RecvChanBuffer                = 50
	ResultsChanBuffer             = 50
	ResultQueueDrainTimeoutMillis = 10000
	MaxConcurrentSends            = 10
	SendMsgURL                    = "api/v4/remotecluster/msg"
	SendTimeout                   = time.Minute
	SendFileTimeout               = time.Minute * 5
	PingURL                       = "api/v4/remotecluster/ping"
	// Raising PingFreq also raises the transport's IdleConnTimeout (PingFreq / 2),
	// which must stay below peer idle timeouts (60s default). See MM-69982.
	PingFreq           = time.Minute
	PingTimeout        = time.Second * 15
	ConfirmInviteURL   = "api/v4/remotecluster/confirm_invite"
	InvitationTopic    = "invitation"
	PingTopic          = "ping"
	ResponseStatusOK   = model.StatusOk
	ResponseStatusFail = model.StatusFail
	InviteExpiresAfter = time.Hour * 48
)

type ServerIface interface {
	Config() *model.Config
	IsLeader() bool
	AddClusterLeaderChangedListener(listener func()) string
	RemoveClusterLeaderChangedListener(id string)
	GetStore() store.Store
	Log() *mlog.Logger
	GetMetrics() einterfaces.MetricsInterface
}

type AppIface interface {
	OnSharedChannelsPing(rc *model.RemoteCluster) bool
}

// RemoteClusterServiceIFace is used to allow mocking where a remote cluster service is used (for testing).
// Unfortunately it lives here because the shared channel service, app layer, and server interface all need it.
// Putting it in app layer means shared channel service must import app package.
type RemoteClusterServiceIFace interface {
	Shutdown() error
	Start() error
	Active() bool
	AddTopicListener(topic string, listener TopicListener) string
	RemoveTopicListener(listenerId string)
	AddConnectionStateListener(listener ConnectionStateListener) string
	RemoveConnectionStateListener(listenerId string)
	SendMsg(ctx context.Context, msg model.RemoteClusterMsg, rc *model.RemoteCluster, f SendMsgResultFunc) error
	SendFile(ctx context.Context, us *model.UploadSession, fi *model.FileInfo, rc *model.RemoteCluster, rp ReaderProvider, f SendFileResultFunc) error
	SendProfileImage(ctx context.Context, userID string, rc *model.RemoteCluster, provider ProfileImageProvider, f SendProfileImageResultFunc) error
	AcceptInvitation(invite *model.RemoteClusterInvite, name string, displayName string, creatorId string, siteURL string, defaultTeamId string) (*model.RemoteCluster, error)
	ReceiveIncomingMsg(rc *model.RemoteCluster, msg model.RemoteClusterMsg) Response
	ReceiveInviteConfirmation(invite model.RemoteClusterInvite) (*model.RemoteCluster, error)
	PingNow(rc *model.RemoteCluster)
	NotifySyncFailed(remoteId string)
}

// TopicListener is a callback signature used to listen for incoming messages for
// a specific topic.
type TopicListener func(msg model.RemoteClusterMsg, rc *model.RemoteCluster, resp *Response) error

// ConnectionStateListener is used to listen to remote cluster connection state changes.
type ConnectionStateListener func(rc *model.RemoteCluster, online bool)

// Service provides inter-cluster communication via topic based messages. In product these are called "Secured Connections".
type Service struct {
	server      ServerIface
	app         AppIface
	httpClient  *http.Client
	send        []chan any
	done        chan struct{} // signals sendLoop workers to stop; lifecycle: Start -> Shutdown
	active      atomic.Bool   // true after Start(), false after Shutdown()
	disablePing bool          // when true, pingStart skips launching the ping loop; for testing

	syncFailedSinceLastPing sync.Map // set of remoteIds (string) with a pending sync failure; presence is the marker

	// everything below guarded by `mux`
	mux                      sync.RWMutex
	leaderListenerId         string
	topicListeners           map[string]map[string]TopicListener // maps topic id to a map of listenerid->listener
	connectionStateListeners map[string]ConnectionStateListener  // maps listener id to listener
	pingDone                 chan struct{}                       // signals ping workers to stop; lifecycle: pingStart -> pingStop
	pingFreq                 time.Duration
}

// NotifySyncFailed records that a sync has failed for the given remote. On the next
// successful ping, PingNow will fire a connection-state-change event so the shared
// channel service re-syncs any affected channels.
func (rcs *Service) NotifySyncFailed(remoteId string) {
	rcs.syncFailedSinceLastPing.Store(remoteId, struct{}{})
}

// NewRemoteClusterService creates a RemoteClusterService instance. In product this is called a "Secured Connection".
func NewRemoteClusterService(server ServerIface, app AppIface) (*Service, error) {
	transport := &http.Transport{
		Proxy: http.ProxyFromEnvironment,
		DialContext: (&net.Dialer{
			Timeout:   30 * time.Second,
			KeepAlive: 30 * time.Second,
			DualStack: true,
		}).DialContext,
		ForceAttemptHTTP2:   true,
		MaxIdleConns:        200,
		MaxIdleConnsPerHost: 2,
		// Must stay strictly below PingFreq so the pool always discards a connection
		// before the next ping reuses it. Otherwise pings race the peer reaping its own
		// idle keep-alive connections (ServiceSettings.IdleTimeout, default 60s) and fail
		// intermittently with stale-connection errors. See MM-69982.
		IdleConnTimeout:       PingFreq / 2,
		TLSHandshakeTimeout:   10 * time.Second,
		ExpectContinueTimeout: 1 * time.Second,
		DisableCompression:    false,
	}

	client := &http.Client{
		Transport: transport,
		Timeout:   SendTimeout,
	}

	service := &Service{
		server:                   server,
		app:                      app,
		httpClient:               client,
		topicListeners:           make(map[string]map[string]TopicListener),
		connectionStateListeners: make(map[string]ConnectionStateListener),
	}

	service.send = make([]chan any, MaxConcurrentSends)
	for i := range service.send {
		service.send[i] = make(chan any, SendChanBuffer)
	}
	service.pingFreq = PingFreq

	return service, nil
}

// Start is called by the server on server start-up.
func (rcs *Service) Start() error {
	if !rcs.active.CompareAndSwap(false, true) {
		return nil
	}
	rcs.done = make(chan struct{})
	for i := range rcs.send {
		go rcs.sendLoop(i, rcs.done)
	}

	rcs.mux.Lock()
	rcs.leaderListenerId = rcs.server.AddClusterLeaderChangedListener(rcs.onClusterLeaderChange)
	rcs.mux.Unlock()

	rcs.onClusterLeaderChange()

	return nil
}

// Shutdown is called by the server on server shutdown.
func (rcs *Service) Shutdown() error {
	if !rcs.active.CompareAndSwap(true, false) {
		return nil
	}
	rcs.mux.Lock()
	id := rcs.leaderListenerId
	rcs.mux.Unlock()

	rcs.server.RemoveClusterLeaderChangedListener(id)
	rcs.pingStop()
	close(rcs.done)
	return nil
}

// Active returns true if this instance of the remote cluster service has been started.
func (rcs *Service) Active() bool {
	return rcs.active.Load()
}

// GetPingFreq gets the frequency of pings to each remote.
func (rcs *Service) GetPingFreq() time.Duration {
	rcs.mux.Lock()
	defer rcs.mux.Unlock()
	return rcs.pingFreq
}

// SetPingFreq sets the frequency of pings to each remote. Defaults to `PingFreq`.
// This is typically used to set a higher frequency for testing.
func (rcs *Service) SetPingFreq(freq time.Duration) {
	rcs.mux.Lock()
	defer rcs.mux.Unlock()
	rcs.pingFreq = freq
}

// AddTopicListener registers a callback
func (rcs *Service) AddTopicListener(topic string, listener TopicListener) string {
	rcs.mux.Lock()
	defer rcs.mux.Unlock()

	id := model.NewId()

	listeners, ok := rcs.topicListeners[topic]
	if !ok || listeners == nil {
		rcs.topicListeners[topic] = make(map[string]TopicListener)
	}
	rcs.topicListeners[topic][id] = listener
	return id
}

func (rcs *Service) RemoveTopicListener(listenerId string) {
	rcs.mux.Lock()
	defer rcs.mux.Unlock()

	for topic, listeners := range rcs.topicListeners {
		if _, ok := listeners[listenerId]; ok {
			delete(listeners, listenerId)
			if len(listeners) == 0 {
				delete(rcs.topicListeners, topic)
			}
			break
		}
	}
}

func (rcs *Service) getTopicListeners(topic string) []TopicListener {
	rcs.mux.RLock()
	defer rcs.mux.RUnlock()

	listeners, ok := rcs.topicListeners[topic]
	if !ok {
		return nil
	}

	listenersCopy := make([]TopicListener, 0, len(listeners))
	for _, l := range listeners {
		listenersCopy = append(listenersCopy, l)
	}
	return listenersCopy
}

func (rcs *Service) AddConnectionStateListener(listener ConnectionStateListener) string {
	id := model.NewId()

	rcs.mux.Lock()
	defer rcs.mux.Unlock()

	rcs.connectionStateListeners[id] = listener
	return id
}

func (rcs *Service) RemoveConnectionStateListener(listenerId string) {
	rcs.mux.Lock()
	defer rcs.mux.Unlock()
	delete(rcs.connectionStateListeners, listenerId)
}

// onClusterLeaderChange is called whenever the cluster leader may have changed.
func (rcs *Service) onClusterLeaderChange() {
	if rcs.server.IsLeader() {
		rcs.pingStart()
	} else {
		rcs.pingStop()
	}
}

func (rcs *Service) pingStart() {
	rcs.mux.Lock()
	defer rcs.mux.Unlock()

	if rcs.pingDone != nil {
		return
	}
	rcs.pingDone = make(chan struct{})

	if !rcs.disablePing {
		// first ping all the plugin remotes immediately, synchronously.
		rcs.pingAllNow(model.RemoteClusterQueryFilter{OnlyPlugins: true})

		// start the async ping loop
		rcs.pingLoop(rcs.pingDone)
	}

	rcs.server.Log().Debug("Remote Cluster Service ping active")
}

func (rcs *Service) pingStop() {
	rcs.mux.Lock()
	defer rcs.mux.Unlock()

	if rcs.pingDone == nil {
		return
	}
	close(rcs.pingDone)
	rcs.pingDone = nil

	rcs.server.Log().Debug("Remote Cluster Service ping inactive")
}