seanime-docker/seanime-2.9.10/internal/nakama/nakama.go

package nakama

import (
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"seanime/internal/database/models"
	debrid_client "seanime/internal/debrid/client"
	"seanime/internal/events"
	"seanime/internal/library/playbackmanager"
	"seanime/internal/platforms/platform"
	"seanime/internal/torrentstream"
	"seanime/internal/util"
	"seanime/internal/util/result"
	"strings"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	"github.com/imroc/req/v3"
	"github.com/rs/zerolog"
)

type Manager struct {
	serverHost              string
	serverPort              int
	username                string
	logger                  *zerolog.Logger
	settings                *models.NakamaSettings
	wsEventManager          events.WSEventManagerInterface
	platform                platform.Platform
	playbackManager         *playbackmanager.PlaybackManager
	torrentstreamRepository *torrentstream.Repository
	debridClientRepository  *debrid_client.Repository
	peerId                  string

	// Host connections (when acting as host)
	peerConnections *result.Map[string, *PeerConnection]

	// Host connection (when connecting to a host)
	hostConnection       *HostConnection
	hostConnectionCtx    context.Context
	hostConnectionCancel context.CancelFunc
	hostMu               sync.RWMutex
	reconnecting         bool // Flag to prevent multiple concurrent reconnection attempts

	// Connection management
	cancel context.CancelFunc
	ctx    context.Context

	// Message handlers
	messageHandlers map[MessageType]func(*Message, string) error
	handlerMu       sync.RWMutex

	// Cleanup functions
	cleanups []func()

	reqClient         *req.Client
	watchPartyManager *WatchPartyManager

	previousPath string // latest file streamed by the peer - real path on the host
}

type NewManagerOptions struct {
	Logger                  *zerolog.Logger
	WSEventManager          events.WSEventManagerInterface
	PlaybackManager         *playbackmanager.PlaybackManager
	TorrentstreamRepository *torrentstream.Repository
	DebridClientRepository  *debrid_client.Repository
	Platform                platform.Platform
	ServerHost              string
	ServerPort              int
}

type ConnectionType string

const (
	ConnectionTypeHost ConnectionType = "host"
	ConnectionTypePeer ConnectionType = "peer"
)

// MessageType represents the type of message being sent
type MessageType string

const (
	MessageTypeAuth      MessageType = "auth"
	MessageTypeAuthReply MessageType = "auth_reply"
	MessageTypePing      MessageType = "ping"
	MessageTypePong      MessageType = "pong"
	MessageTypeError     MessageType = "error"

	MessageTypeCustom MessageType = "custom"
)

// Message represents a message sent between Nakama instances
type Message struct {
	Type      MessageType `json:"type"`
	Payload   interface{} `json:"payload"`
	RequestID string      `json:"requestId,omitempty"`
	Timestamp time.Time   `json:"timestamp"`
}

// PeerConnection represents a connection from a peer to this host
type PeerConnection struct {
	ID             string // Internal connection ID (websocket)
	PeerId         string // UUID generated by the peer (primary identifier)
	Username       string // Display name (kept for UI purposes)
	Conn           *websocket.Conn
	ConnectionType ConnectionType
	Authenticated  bool
	LastPing       time.Time
	mu             sync.RWMutex
}

// HostConnection represents this instance's connection to a host
type HostConnection struct {
	URL            string
	PeerId         string // UUID generated by this peer instance
	Username       string
	Conn           *websocket.Conn
	Authenticated  bool
	LastPing       time.Time
	reconnectTimer *time.Timer
	mu             sync.RWMutex
}

// NakamaEvent represents events sent to the client
type NakamaEvent struct {
	Type    string      `json:"type"`
	Payload interface{} `json:"payload"`
}

// AuthPayload represents authentication data
type AuthPayload struct {
	Password string `json:"password"`
	PeerId   string `json:"peerId"` // UUID generated by the peer
}

// AuthReplyPayload represents authentication response
type AuthReplyPayload struct {
	Success  bool   `json:"success"`
	Message  string `json:"message"`
	Username string `json:"username"`
	PeerId   string `json:"peerId"` // Echo back the peer's UUID
}

// ErrorPayload represents error messages
type ErrorPayload struct {
	Message string `json:"message"`
	Code    string `json:"code,omitempty"`
}

// HostConnectionStatus represents the status of the host connection
type HostConnectionStatus struct {
	Connected     bool      `json:"connected"`
	Authenticated bool      `json:"authenticated"`
	URL           string    `json:"url"`
	LastPing      time.Time `json:"lastPing"`
	PeerId        string    `json:"peerId"`
	Username      string    `json:"username"`
}

// NakamaStatus represents the overall status of Nakama connections
type NakamaStatus struct {
	IsHost                   bool                  `json:"isHost"`
	ConnectedPeers           []string              `json:"connectedPeers"`
	IsConnectedToHost        bool                  `json:"isConnectedToHost"`
	HostConnectionStatus     *HostConnectionStatus `json:"hostConnectionStatus"`
	CurrentWatchPartySession *WatchPartySession    `json:"currentWatchPartySession"`
}

// MessageResponse represents a response to message sending requests
type MessageResponse struct {
	Success bool   `json:"success"`
	Message string `json:"message"`
}

type ClientEvent struct {
	Type    string      `json:"type"`
	Payload interface{} `json:"payload"`
}

func NewManager(opts *NewManagerOptions) *Manager {
	ctx, cancel := context.WithCancel(context.Background())

	m := &Manager{
		username:                "",
		logger:                  opts.Logger,
		wsEventManager:          opts.WSEventManager,
		playbackManager:         opts.PlaybackManager,
		peerConnections:         result.NewResultMap[string, *PeerConnection](),
		platform:                opts.Platform,
		ctx:                     ctx,
		cancel:                  cancel,
		messageHandlers:         make(map[MessageType]func(*Message, string) error),
		cleanups:                make([]func(), 0),
		reqClient:               req.C(),
		serverHost:              opts.ServerHost,
		serverPort:              opts.ServerPort,
		settings:                &models.NakamaSettings{},
		torrentstreamRepository: opts.TorrentstreamRepository,
		debridClientRepository:  opts.DebridClientRepository,
		previousPath:            "",
	}

	m.watchPartyManager = NewWatchPartyManager(m)

	// Register default message handlers
	m.registerDefaultHandlers()

	eventListener := m.wsEventManager.SubscribeToClientEvents("nakama")
	go func() {
		for event := range eventListener.Channel {
			if event.Type == events.NakamaStatusRequested {
				currSession, _ := m.GetWatchPartyManager().GetCurrentSession()
				status := &NakamaStatus{
					IsHost:                   m.IsHost(),
					ConnectedPeers:           m.GetConnectedPeers(),
					IsConnectedToHost:        m.IsConnectedToHost(),
					HostConnectionStatus:     m.GetHostConnectionStatus(),
					CurrentWatchPartySession: currSession,
				}
				m.wsEventManager.SendEvent(events.NakamaStatus, status)
			}

			if event.Type == events.NakamaWatchPartyEnableRelayMode {
				var payload WatchPartyEnableRelayModePayload
				marshaledPayload, err := json.Marshal(event.Payload)
				if err != nil {
					m.logger.Error().Err(err).Msg("nakama: Failed to marshal watch party enable relay mode payload")
					continue
				}
				err = json.Unmarshal(marshaledPayload, &payload)
				if err != nil {
					m.logger.Error().Err(err).Msg("nakama: Failed to unmarshal watch party enable relay mode payload")
					continue
				}
				m.GetWatchPartyManager().EnableRelayMode(payload.PeerId)
			}
		}
	}()

	return m
}

func (m *Manager) SetSettings(settings *models.NakamaSettings) {
	var previousSettings *models.NakamaSettings
	if m.settings != nil {
		previousSettings = &[]models.NakamaSettings{*m.settings}[0]
	}

	// If the host password has changed, stop host service
	// This will cause a restart of the host service
	disconnectAsHost := false
	if m.settings != nil && m.settings.HostPassword != settings.HostPassword {
		disconnectAsHost = true
		m.stopHostServices()
	}

	m.settings = settings
	m.username = cmp.Or(settings.Username, "Peer_"+util.RandomStringWithAlphabet(8, "bcdefhijklmnopqrstuvwxyz0123456789"))
	m.logger.Debug().Bool("isHost", settings.IsHost).Str("username", m.username).Str("remoteURL", settings.RemoteServerURL).Msg("nakama: Settings updated")

	if previousSettings == nil || previousSettings.IsHost != settings.IsHost || previousSettings.Enabled != settings.Enabled || disconnectAsHost {
		// Determine if we should stop host services
		shouldStopHost := m.IsHost() && (!settings.Enabled || // Nakama disabled
			!settings.IsHost || // Switching to peer mode
			disconnectAsHost) // Password changed (requires restart)

		// Determine if we should start host services
		shouldStartHost := settings.IsHost && settings.Enabled

		// Always stop first if needed, then start
		if shouldStopHost {
			m.stopHostServices()
		}
		if shouldStartHost {
			m.startHostServices()
		}
	}

	if previousSettings == nil || previousSettings.RemoteServerURL != settings.RemoteServerURL || previousSettings.RemoteServerPassword != settings.RemoteServerPassword || previousSettings.Enabled != settings.Enabled {
		// Determine if we should disconnect from current host
		shouldDisconnect := m.IsConnectedToHost() && (!settings.Enabled || // Nakama disabled
			settings.IsHost || // Switching to host mode
			settings.RemoteServerURL == "" || // No remote URL
			settings.RemoteServerPassword == "" || // No password
			(previousSettings != nil && previousSettings.RemoteServerURL != settings.RemoteServerURL) || // URL changed
			(previousSettings != nil && previousSettings.RemoteServerPassword != settings.RemoteServerPassword)) // Password changed

		// Determine if we should connect to a host
		shouldConnect := !settings.IsHost &&
			settings.Enabled &&
			settings.RemoteServerURL != "" &&
			settings.RemoteServerPassword != ""

		// Always disconnect first if needed, then connect
		if shouldDisconnect {
			m.disconnectFromHost()
		}
		if shouldConnect {
			m.connectToHost()
		}
	}

	// if previousSettings == nil || previousSettings.Username != settings.Username {
	// 	m.SendMessage(MessageTypeCustom, map[string]interface{}{
	// 		"type": "nakama_username_changed",
	// 		"username": settings.Username,
	// 	})
	// }
}

func (m *Manager) GetHostBaseServerURL() string {
	url := m.settings.RemoteServerURL
	if strings.HasSuffix(url, "/") {
		url = strings.TrimSuffix(url, "/")
	}
	return url
}

func (m *Manager) IsHost() bool {
	return m.settings.IsHost
}

func (m *Manager) GetHostConnection() (*HostConnection, bool) {
	m.hostMu.RLock()
	defer m.hostMu.RUnlock()
	return m.hostConnection, m.hostConnection != nil
}

// GetWatchPartyManager returns the watch party manager
func (m *Manager) GetWatchPartyManager() *WatchPartyManager {
	return m.watchPartyManager
}

// Cleanup stops all connections and services
func (m *Manager) Cleanup() {
	m.logger.Debug().Msg("nakama: Cleaning up")

	if m.cancel != nil {
		m.cancel()
	}

	// Cancel any ongoing host connection attempts
	m.hostMu.Lock()
	if m.hostConnectionCancel != nil {
		m.hostConnectionCancel()
		m.hostConnectionCancel = nil
	}
	m.hostMu.Unlock()

	// Cleanup host connections
	m.peerConnections.Range(func(id string, conn *PeerConnection) bool {
		conn.Close()
		return true
	})
	m.peerConnections.Clear()

	// Cleanup client connection
	m.hostMu.Lock()
	if m.hostConnection != nil {
		m.hostConnection.Close()
		m.hostConnection = nil
	}
	m.hostMu.Unlock()

	// Run cleanup functions
	for _, cleanup := range m.cleanups {
		cleanup()
	}
}

// RegisterMessageHandler registers a custom message handler
func (m *Manager) RegisterMessageHandler(msgType MessageType, handler func(*Message, string) error) {
	m.handlerMu.Lock()
	defer m.handlerMu.Unlock()
	m.messageHandlers[msgType] = handler
}

// SendMessage sends a message to all connected peers (when acting as host)
func (m *Manager) SendMessage(msgType MessageType, payload interface{}) error {
	if !m.settings.IsHost {
		return errors.New("not acting as host")
	}

	message := &Message{
		Type:      msgType,
		Payload:   payload,
		Timestamp: time.Now(),
	}

	var lastError error
	m.peerConnections.Range(func(id string, conn *PeerConnection) bool {
		if err := conn.SendMessage(message); err != nil {
			m.logger.Error().Err(err).Str("peerId", conn.PeerId).Msg("nakama: Failed to send message to peer")
			lastError = err
		}
		return true
	})

	return lastError
}

// SendMessageToPeer sends a message to a specific peer by their PeerID
func (m *Manager) SendMessageToPeer(peerID string, msgType MessageType, payload interface{}) error {
	if !m.settings.IsHost {
		return errors.New("only hosts can send messages to peers")
	}

	// Find peer by PeerID
	var targetConn *PeerConnection
	m.peerConnections.Range(func(id string, conn *PeerConnection) bool {
		if conn.PeerId == peerID {
			targetConn = conn
			return false // Stop iteration
		}
		return true
	})

	if targetConn == nil {
		return errors.New("peer not found: " + peerID)
	}

	message := &Message{
		Type:      msgType,
		Payload:   payload,
		Timestamp: time.Now(),
	}

	return targetConn.SendMessage(message)
}

// SendMessageToHost sends a message to the host (when acting as peer)
func (m *Manager) SendMessageToHost(msgType MessageType, payload interface{}) error {
	m.hostMu.RLock()
	defer m.hostMu.RUnlock()

	if m.hostConnection == nil || !m.hostConnection.Authenticated {
		return errors.New("not connected to host")
	}

	message := &Message{
		Type:      msgType,
		Payload:   payload,
		Timestamp: time.Now(),
	}

	return m.hostConnection.SendMessage(message)
}

// GetConnectedPeers returns a list of connected peer IDs
func (m *Manager) GetConnectedPeers() []string {
	if !m.settings.IsHost {
		return []string{}
	}

	peers := make([]string, 0)

	m.peerConnections.Range(func(id string, conn *PeerConnection) bool {
		if conn.Authenticated {
			// Use PeerID as the primary identifier
			peerDisplayName := conn.Username
			if peerDisplayName == "" {
				peerDisplayName = "Unknown"
			}
			// Format: "Username (PeerID_short)"
			// peers = append(peers, fmt.Sprintf("%s (%s)", peerDisplayName, conn.PeerId[:8]))
			peers = append(peers, fmt.Sprintf("%s", peerDisplayName))
		}
		return true
	})

	return peers
}

// IsConnectedToHost returns whether this instance is connected to a host
func (m *Manager) IsConnectedToHost() bool {
	m.hostMu.RLock()
	defer m.hostMu.RUnlock()
	return m.hostConnection != nil && m.hostConnection.Authenticated
}

// GetHostConnectionStatus returns the status of the host connection
func (m *Manager) GetHostConnectionStatus() *HostConnectionStatus {
	m.hostMu.RLock()
	defer m.hostMu.RUnlock()

	if m.hostConnection == nil {
		return nil
	}

	return &HostConnectionStatus{
		Connected:     m.hostConnection != nil,
		Authenticated: m.hostConnection != nil && m.hostConnection.Authenticated,
		URL:           m.hostConnection.URL,
		LastPing:      m.hostConnection.LastPing,
		PeerId:        m.hostConnection.PeerId,
		Username:      m.hostConnection.Username,
	}
}

func (pc *PeerConnection) SendMessage(message *Message) error {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	return pc.Conn.WriteJSON(message)
}

func (pc *PeerConnection) Close() {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	_ = pc.Conn.Close()
}

func (hc *HostConnection) SendMessage(message *Message) error {
	hc.mu.Lock()
	defer hc.mu.Unlock()
	return hc.Conn.WriteJSON(message)
}

func (hc *HostConnection) Close() {
	hc.mu.Lock()
	defer hc.mu.Unlock()
	if hc.reconnectTimer != nil {
		hc.reconnectTimer.Stop()
	}
	_ = hc.Conn.Close()
}

// Helper function to generate connection IDs
func generateConnectionID() string {
	return fmt.Sprintf("conn_%d", time.Now().UnixNano())
}

// ReconnectToHost attempts to reconnect to the host
func (m *Manager) ReconnectToHost() error {
	if m.settings == nil || m.settings.RemoteServerURL == "" || m.settings.RemoteServerPassword == "" {
		return errors.New("no host connection configured")
	}

	// Check if already reconnecting
	m.hostMu.Lock()
	if m.reconnecting {
		m.hostMu.Unlock()
		return errors.New("reconnection already in progress")
	}
	m.hostMu.Unlock()

	m.logger.Info().Msg("nakama: Manual reconnection to host requested")

	// Disconnect current connection if exists
	m.disconnectFromHost()

	// Wait a moment before reconnecting
	time.Sleep(1 * time.Second)

	// Reconnect
	m.connectToHost()
	return nil
}

// RemoveStaleConnections removes connections that haven't responded to ping in a while
func (m *Manager) RemoveStaleConnections() {
	if !m.settings.IsHost {
		return
	}

	staleThreshold := 90 * time.Second // Consider connections stale after 90 seconds of no ping
	now := time.Now()

	var staleConnections []string

	m.peerConnections.Range(func(id string, conn *PeerConnection) bool {
		conn.mu.RLock()
		lastPing := conn.LastPing
		authenticated := conn.Authenticated
		conn.mu.RUnlock()

		// Only check authenticated connections
		if !authenticated {
			return true
		}

		// If LastPing is zero, use connection time as reference
		if lastPing.IsZero() {
			lastPing = now.Add(-staleThreshold - time.Minute)
		}

		if now.Sub(lastPing) > staleThreshold {
			staleConnections = append(staleConnections, id)
		}
		return true
	})

	// Remove stale connections
	for _, id := range staleConnections {
		if conn, exists := m.peerConnections.Get(id); exists {
			// Double-check to avoid race conditions
			conn.mu.RLock()
			lastPing := conn.LastPing
			if lastPing.IsZero() {
				lastPing = now.Add(-staleThreshold - time.Minute)
			}
			isStale := now.Sub(lastPing) > staleThreshold
			conn.mu.RUnlock()

			if isStale {
				m.logger.Info().Str("peerId", conn.PeerId).Str("internalConnID", id).Msg("nakama: Removing stale peer connection")

				// Remove from map first to prevent re-addition
				m.peerConnections.Delete(id)

				// Remove peer from watch party if they were participating
				m.watchPartyManager.HandlePeerDisconnected(conn.PeerId)

				// Then close the connection (this will trigger the defer cleanup in handlePeerConnection)
				conn.Close()

				// Send event about peer disconnection
				m.wsEventManager.SendEvent(events.NakamaPeerDisconnected, map[string]interface{}{
					"peerId": conn.PeerId,
					"reason": "stale_connection",
				})
			}
		}
	}

	if len(staleConnections) > 0 {
		m.logger.Info().Int("count", len(staleConnections)).Msg("nakama: Removed stale peer connections")
	}
}

// FindPeerByPeerID finds a peer connection by their PeerID
func (m *Manager) FindPeerByPeerID(peerID string) (*PeerConnection, bool) {
	var found *PeerConnection
	m.peerConnections.Range(func(id string, conn *PeerConnection) bool {
		if conn.PeerId == peerID {
			found = conn
			return false // Stop iteration
		}
		return true
	})
	return found, found != nil
}