registry/internal/service/registry_service.go

package service

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/modelcontextprotocol/registry/internal/config"
	"github.com/modelcontextprotocol/registry/internal/database"
	"github.com/modelcontextprotocol/registry/internal/validators"
	apiv0 "github.com/modelcontextprotocol/registry/pkg/api/v0"
)

const maxServerVersionsPerServer = 10000

// registryServiceImpl implements the RegistryService interface using our Database
type registryServiceImpl struct {
	db  database.Database
	cfg *config.Config
}

// NewRegistryService creates a new registry service with the provided database
func NewRegistryService(db database.Database, cfg *config.Config) RegistryService {
	return &registryServiceImpl{
		db:  db,
		cfg: cfg,
	}
}

// List returns registry entries with cursor-based pagination and optional filtering
func (s *registryServiceImpl) List(filter *database.ServerFilter, cursor string, limit int) ([]apiv0.ServerJSON, string, error) {
	// Create a timeout context for the database operation
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// If limit is not set or negative, use a default limit
	if limit <= 0 {
		limit = 30
	}

	// Use the database's ListServers method with pagination and filtering
	serverRecords, nextCursor, err := s.db.List(ctx, nil, filter, cursor, limit)
	if err != nil {
		return nil, "", err
	}

	// Return ServerJSONs directly
	result := make([]apiv0.ServerJSON, len(serverRecords))
	for i, record := range serverRecords {
		result[i] = *record
	}

	return result, nextCursor, nil
}

// GetByVersionID retrieves a specific server by its registry metadata version ID
func (s *registryServiceImpl) GetByVersionID(versionID string) (*apiv0.ServerJSON, error) {
	// Create a timeout context for the database operation
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	serverRecord, err := s.db.GetByVersionID(ctx, nil, versionID)
	if err != nil {
		return nil, err
	}

	// Return the server record directly
	return serverRecord, nil
}

// GetByServerID retrieves the latest version of a server by its server ID
func (s *registryServiceImpl) GetByServerID(serverID string) (*apiv0.ServerJSON, error) {
	// Create a timeout context for the database operation
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	serverRecord, err := s.db.GetByServerID(ctx, nil, serverID)
	if err != nil {
		return nil, err
	}

	// Return the server record directly
	return serverRecord, nil
}

// GetByServerIDAndVersion retrieves a specific version of a server by server ID and version
func (s *registryServiceImpl) GetByServerIDAndVersion(serverID string, version string) (*apiv0.ServerJSON, error) {
	// Create a timeout context for the database operation
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	serverRecord, err := s.db.GetByServerIDAndVersion(ctx, nil, serverID, version)
	if err != nil {
		return nil, err
	}

	// Return the server record directly
	return serverRecord, nil
}

// GetAllVersionsByServerID retrieves all versions of a server by server ID
func (s *registryServiceImpl) GetAllVersionsByServerID(serverID string) ([]apiv0.ServerJSON, error) {
	// Create a timeout context for the database operation
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	serverRecords, err := s.db.GetAllVersionsByServerID(ctx, nil, serverID)
	if err != nil {
		return nil, err
	}

	// Return ServerJSONs directly
	result := make([]apiv0.ServerJSON, len(serverRecords))
	for i, record := range serverRecords {
		result[i] = *record
	}

	return result, nil
}

// Publish publishes a server with flattened _meta extensions
func (s *registryServiceImpl) Publish(req apiv0.ServerJSON) (*apiv0.ServerJSON, error) {
	// Create a timeout context for the database operation
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Validate the request
	if err := validators.ValidatePublishRequest(req, s.cfg); err != nil {
		return nil, err
	}

	result, err := database.InTransactionT(ctx, s.db, func(txCtx context.Context, tx pgx.Tx) (*apiv0.ServerJSON, error) {
		publishTime := time.Now()
		serverJSON := req

		// Acquire advisory lock to prevent concurrent publishes of the same server
		if err := s.db.AcquirePublishLock(txCtx, tx, serverJSON.Name); err != nil {
			return nil, err
		}

		// Check for duplicate remote URLs
		if err := s.validateNoDuplicateRemoteURLs(txCtx, tx, serverJSON); err != nil {
			return nil, err
		}

		filter := &database.ServerFilter{Name: &serverJSON.Name}
		existingServerVersions, _, err := s.db.List(txCtx, tx, filter, "", maxServerVersionsPerServer)
		if err != nil && !errors.Is(err, database.ErrNotFound) {
			return nil, err
		}

		// Check we haven't exceeded the maximum versions allowed for a server
		if len(existingServerVersions) >= maxServerVersionsPerServer {
			return nil, database.ErrMaxServersReached
		}

		// Check this isn't a duplicate version
		for _, server := range existingServerVersions {
			existingVersion := server.Version
			if existingVersion == serverJSON.Version {
				return nil, database.ErrInvalidVersion
			}
		}

		// Determine if this version should be marked as latest
		existingLatest := s.getCurrentLatestVersion(existingServerVersions)
		isNewLatest := true
		if existingLatest != nil {
			var existingPublishedAt time.Time
			if existingLatest.Meta != nil && existingLatest.Meta.Official != nil {
				existingPublishedAt = existingLatest.Meta.Official.PublishedAt
			}
			isNewLatest = CompareVersions(
				serverJSON.Version,
				existingLatest.Version,
				publishTime,
				existingPublishedAt,
			) > 0
		}

		// Unmark old latest version if needed
		if isNewLatest && existingLatest != nil {
			if existingLatest.Meta != nil && existingLatest.Meta.Official != nil {
				existingLatest.Meta.Official.IsLatest = false
				if _, err := s.db.UpdateServer(txCtx, tx, existingLatest.Meta.Official.VersionID, existingLatest); err != nil {
					return nil, err
				}
			}
		}

		// Create complete server with metadata
		server := s.createServerWithMetadata(serverJSON, existingServerVersions, publishTime, isNewLatest)

		// Insert new server version
		return s.db.CreateServer(txCtx, tx, &server)
	})

	if err != nil {
		return nil, err
	}

	return result, nil
}

// createServerWithMetadata creates a server with proper metadata including server_id and version_id
func (s *registryServiceImpl) createServerWithMetadata(
	serverJSON apiv0.ServerJSON,
	existingServerVersions []*apiv0.ServerJSON,
	publishTime time.Time,
	isNewLatest bool,
) apiv0.ServerJSON {
	server := serverJSON // Copy the input

	// Initialize meta if not present
	if server.Meta == nil {
		server.Meta = &apiv0.ServerMeta{}
	}

	// Determine server_id - either from existing versions or generate new one
	var serverID string
	if len(existingServerVersions) > 0 {
		// Use existing server_id from any existing version
		firstExisting := existingServerVersions[0]
		if firstExisting.Meta != nil && firstExisting.Meta.Official != nil {
			serverID = firstExisting.Meta.Official.ServerID
		}
	}
	if serverID == "" {
		// This is the first version of a new server
		serverID = uuid.New().String()
	}

	// Set registry metadata
	server.Meta.Official = &apiv0.RegistryExtensions{
		ServerID:    serverID,
		VersionID:   uuid.New().String(),
		PublishedAt: publishTime,
		UpdatedAt:   publishTime,
		IsLatest:    isNewLatest,
	}

	return server
}

// validateNoDuplicateRemoteURLs checks that no other server is using the same remote URLs
func (s *registryServiceImpl) validateNoDuplicateRemoteURLs(ctx context.Context, tx pgx.Tx, serverDetail apiv0.ServerJSON) error {
	// Check each remote URL in the new server for conflicts
	for _, remote := range serverDetail.Remotes {
		// Use filter to find servers with this remote URL
		filter := &database.ServerFilter{RemoteURL: &remote.URL}

		conflictingServers, _, err := s.db.List(ctx, tx, filter, "", 1000)
		if err != nil {
			return fmt.Errorf("failed to check remote URL conflict: %w", err)
		}

		// Check if any conflicting server has a different name
		for _, conflictingServer := range conflictingServers {
			if conflictingServer.Name != serverDetail.Name {
				return fmt.Errorf("remote URL %s is already used by server %s", remote.URL, conflictingServer.Name)
			}
		}
	}

	return nil
}

// getCurrentLatestVersion finds the current latest version from existing server versions
func (s *registryServiceImpl) getCurrentLatestVersion(existingServerVersions []*apiv0.ServerJSON) *apiv0.ServerJSON {
	for _, server := range existingServerVersions {
		if server.Meta != nil && server.Meta.Official != nil &&
			server.Meta.Official.IsLatest {
			return server
		}
	}
	return nil
}

// EditServer updates an existing server with new details (admin operation)
func (s *registryServiceImpl) EditServer(versionID string, req apiv0.ServerJSON) (*apiv0.ServerJSON, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Validate the request
	if err := validators.ValidatePublishRequest(req, s.cfg); err != nil {
		return nil, err
	}

	return database.InTransactionT(ctx, s.db, func(txCtx context.Context, tx pgx.Tx) (*apiv0.ServerJSON, error) {
		// Get the current server to preserve metadata
		currentServer, err := s.db.GetByVersionID(txCtx, tx, versionID)
		if err != nil {
			return nil, err
		}

		// Acquire advisory lock to prevent concurrent edits of servers with same name
		if err := s.db.AcquirePublishLock(txCtx, tx, currentServer.Name); err != nil {
			return nil, err
		}

		// Merge the request with the current server, preserving metadata
		updatedServer := *currentServer

		// Update only the user-modifiable fields from the request
		updatedServer.Name = req.Name
		updatedServer.Description = req.Description
		updatedServer.Version = req.Version
		updatedServer.Status = req.Status
		updatedServer.Repository = req.Repository
		updatedServer.Remotes = req.Remotes
		updatedServer.Packages = req.Packages

		// Update the UpdatedAt timestamp in metadata
		if updatedServer.Meta != nil && updatedServer.Meta.Official != nil {
			updatedServer.Meta.Official.UpdatedAt = time.Now()
		}

		// Check for duplicate remote URLs using the updated server
		if err := s.validateNoDuplicateRemoteURLs(txCtx, tx, updatedServer); err != nil {
			return nil, err
		}

		// Update server in database
		return s.db.UpdateServer(txCtx, tx, versionID, &updatedServer)
	})
}