modelcontextprotocol/registry
Watch
1
0
Fork
You've already forked registry
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Changes on the server.json format and the API (#562)

Browse source
This commit is contained in:
Radoslav Dimitrov 2025-09-29 23:30:35 +03:00 • committed by GitHub
commit b21c6f564d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 4056 additions and 5608 deletions
Download patch file Download diff file Expand all files Collapse all files

34
internal/database/database.go
View file

@ -31,20 +31,28 @@ type ServerFilter struct {
// Database defines the interface for database operations
type Database interface {
// Retrieve server entries with optional filtering
List(ctx context.Context, tx pgx.Tx, filter *ServerFilter, cursor string, limit int) ([]*apiv0.ServerJSON, string, error)
// Retrieve a single server by its version ID
GetByVersionID(ctx context.Context, tx pgx.Tx, versionID string) (*apiv0.ServerJSON, error)
// Retrieve latest version of a server by server ID
GetByServerID(ctx context.Context, tx pgx.Tx, serverID string) (*apiv0.ServerJSON, error)
// Retrieve specific version of a server by server ID and version
GetByServerIDAndVersion(ctx context.Context, tx pgx.Tx, serverID string, version string) (*apiv0.ServerJSON, error)
// Retrieve all versions of a server by server ID
GetAllVersionsByServerID(ctx context.Context, tx pgx.Tx, serverID string) ([]*apiv0.ServerJSON, error)
// CreateServer inserts a new server version
CreateServer(ctx context.Context, tx pgx.Tx, newServer *apiv0.ServerJSON) (*apiv0.ServerJSON, error)
// CreateServer inserts a new server version with official metadata
CreateServer(ctx context.Context, tx pgx.Tx, serverJSON *apiv0.ServerJSON, officialMeta *apiv0.RegistryExtensions) (*apiv0.ServerResponse, error)
// UpdateServer updates an existing server record
UpdateServer(ctx context.Context, tx pgx.Tx, id string, server *apiv0.ServerJSON) (*apiv0.ServerJSON, error)
UpdateServer(ctx context.Context, tx pgx.Tx, serverName, version string, serverJSON *apiv0.ServerJSON) (*apiv0.ServerResponse, error)
// SetServerStatus updates the status of a specific server version
SetServerStatus(ctx context.Context, tx pgx.Tx, serverName, version string, status string) (*apiv0.ServerResponse, error)
// ListServers retrieve server entries with optional filtering
ListServers(ctx context.Context, tx pgx.Tx, filter *ServerFilter, cursor string, limit int) ([]*apiv0.ServerResponse, string, error)
// GetServerByName retrieve a single server by its name
GetServerByName(ctx context.Context, tx pgx.Tx, serverName string) (*apiv0.ServerResponse, error)
// GetServerByNameAndVersion retrieve specific version of a server by server name and version
GetServerByNameAndVersion(ctx context.Context, tx pgx.Tx, serverName string, version string) (*apiv0.ServerResponse, error)
// GetAllVersionsByServerName retrieve all versions of a server by server name
GetAllVersionsByServerName(ctx context.Context, tx pgx.Tx, serverName string) ([]*apiv0.ServerResponse, error)
// GetCurrentLatestVersion retrieve the current latest version of a server by server name
GetCurrentLatestVersion(ctx context.Context, tx pgx.Tx, serverName string) (*apiv0.ServerResponse, error)
// CountServerVersions count the number of versions for a server
CountServerVersions(ctx context.Context, tx pgx.Tx, serverName string) (int, error)
// CheckVersionExists check if a specific version exists for a server
CheckVersionExists(ctx context.Context, tx pgx.Tx, serverName, version string) (bool, error)
// UnmarkAsLatest marks the current latest version of a server as no longer latest
UnmarkAsLatest(ctx context.Context, tx pgx.Tx, serverName string) error
// AcquirePublishLock acquires an exclusive advisory lock for publishing a server
// This prevents race conditions when multiple versions are published concurrently
AcquirePublishLock(ctx context.Context, tx pgx.Tx, serverName string) error

143
internal/database/migrations/008_separate_official_metadata.sql Normal file
View file

@ -0,0 +1,143 @@
-- Separate official metadata from JSON to dedicated columns for better performance and consistency
-- This migration transforms the database from JSON-heavy to columnar structure for registry-managed data
BEGIN;
-- Add new columns for official metadata and core identifiers
ALTER TABLE servers ADD COLUMN server_name VARCHAR(255);
ALTER TABLE servers ADD COLUMN version VARCHAR(255);
ALTER TABLE servers ADD COLUMN status VARCHAR(50);
ALTER TABLE servers ADD COLUMN published_at TIMESTAMP WITH TIME ZONE;
ALTER TABLE servers ADD COLUMN updated_at TIMESTAMP WITH TIME ZONE;
ALTER TABLE servers ADD COLUMN is_latest BOOLEAN;
-- Create a function to migrate existing data from JSON to columns
CREATE OR REPLACE FUNCTION migrate_official_metadata()
RETURNS VOID AS $$
DECLARE
rec RECORD;
official_meta JSONB;
updated_value JSONB;
server_meta JSONB;
publisher_provided JSONB;
BEGIN
-- Iterate through all existing records
FOR rec IN SELECT version_id, value FROM servers ORDER BY version_id LOOP
-- Extract core identifiers
UPDATE servers
SET
server_name = rec.value->>'name',
version = rec.value->>'version'
WHERE version_id = rec.version_id;
-- Extract official metadata from nested JSON structure
official_meta := rec.value->'_meta'->'io.modelcontextprotocol.registry/official';
IF official_meta IS NOT NULL THEN
-- Update columns with extracted metadata
UPDATE servers
SET
status = (official_meta->>'status')::VARCHAR(50),
published_at = (official_meta->>'publishedAt')::TIMESTAMP WITH TIME ZONE,
updated_at = (official_meta->>'updatedAt')::TIMESTAMP WITH TIME ZONE,
is_latest = (official_meta->>'isLatest')::BOOLEAN
WHERE version_id = rec.version_id;
ELSE
-- Handle records without official metadata (set defaults)
UPDATE servers
SET
status = COALESCE((rec.value->>'status'), 'active'),
published_at = NOW(),
updated_at = NOW(),
is_latest = true
WHERE version_id = rec.version_id;
END IF;
-- Clean up the JSON: remove status field and official metadata
updated_value := rec.value - 'status';
-- Reconstruct _meta with only publisher-provided data
publisher_provided := rec.value->'_meta'->'io.modelcontextprotocol.registry/publisher-provided';
IF publisher_provided IS NOT NULL THEN
server_meta := jsonb_build_object('io.modelcontextprotocol.registry/publisher-provided', publisher_provided);
updated_value := jsonb_set(updated_value, '{_meta}', server_meta);
ELSE
-- Remove _meta entirely if no publisher-provided data
updated_value := updated_value - '_meta';
END IF;
-- Update the JSON with cleaned structure (keeping name and version for immutable server.json)
UPDATE servers
SET value = updated_value
WHERE version_id = rec.version_id;
END LOOP;
END;
$$ LANGUAGE plpgsql;
-- Execute the migration
SELECT migrate_official_metadata();
-- Drop the migration function
DROP FUNCTION migrate_official_metadata();
-- Make the new columns NOT NULL now that all records have values
ALTER TABLE servers ALTER COLUMN server_name SET NOT NULL;
ALTER TABLE servers ALTER COLUMN version SET NOT NULL;
ALTER TABLE servers ALTER COLUMN status SET NOT NULL;
ALTER TABLE servers ALTER COLUMN published_at SET NOT NULL;
ALTER TABLE servers ALTER COLUMN is_latest SET NOT NULL;
-- Drop the old primary key constraint
ALTER TABLE servers DROP CONSTRAINT servers_pkey;
-- Create new composite primary key using name + version (natural key)
ALTER TABLE servers ADD CONSTRAINT servers_pkey PRIMARY KEY (server_name, version);
-- Drop the old version_id column since we're using name-based keys now
ALTER TABLE servers DROP COLUMN version_id;
-- Drop old indexes that used JSON paths and UUIDs
DROP INDEX IF EXISTS idx_servers_version_id;
DROP INDEX IF EXISTS idx_servers_name_latest;
DROP INDEX IF EXISTS idx_servers_updated_at;
DROP INDEX IF EXISTS idx_servers_remotes;
DROP INDEX IF EXISTS idx_unique_server_version;
DROP INDEX IF EXISTS idx_unique_latest_version;
-- Create new efficient indexes on the dedicated columns
CREATE INDEX idx_servers_name ON servers (server_name);
CREATE INDEX idx_servers_name_version ON servers (server_name, version);
CREATE INDEX idx_servers_name_latest ON servers (server_name, is_latest) WHERE is_latest = true;
CREATE INDEX idx_servers_status ON servers (status);
CREATE INDEX idx_servers_published_at ON servers (published_at DESC);
CREATE INDEX idx_servers_updated_at ON servers (updated_at DESC) WHERE updated_at IS NOT NULL;
-- Ensure only one version per server can be marked as latest
CREATE UNIQUE INDEX idx_unique_latest_per_server
ON servers (server_name)
WHERE is_latest = true;
-- Create GIN index for remaining JSON queries (remotes, packages, etc.)
CREATE INDEX idx_servers_json_remotes ON servers USING GIN((value->'remotes'));
CREATE INDEX idx_servers_json_packages ON servers USING GIN((value->'packages'));
-- Update $schema field to latest version for all entries
UPDATE servers
SET value = jsonb_set(value, '{$schema}', '"https://static.modelcontextprotocol.io/schemas/2025-09-29/server.schema.json"')
WHERE value ? '$schema' AND value IS NOT NULL;
-- Add check constraints for data integrity
ALTER TABLE servers ADD CONSTRAINT check_status_valid
CHECK (status IN ('active', 'deprecated', 'deleted'));
ALTER TABLE servers ADD CONSTRAINT check_server_name_format
CHECK (server_name ~ '^[a-zA-Z0-9][a-zA-Z0-9.-]*[a-zA-Z0-9]/[a-zA-Z0-9][a-zA-Z0-9._-]*[a-zA-Z0-9]$');
ALTER TABLE servers ADD CONSTRAINT check_version_not_empty
CHECK (length(trim(version)) > 0);
ALTER TABLE servers ADD CONSTRAINT check_published_at_reasonable
CHECK (published_at >= '2020-01-01'::timestamp AND published_at <= NOW() + interval '1 day');
COMMIT;

461
internal/database/postgres.go
View file

@ -9,12 +9,12 @@ import (
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool"
apiv0 "github.com/modelcontextprotocol/registry/pkg/api/v0"
"github.com/modelcontextprotocol/registry/pkg/model"
)
// PostgreSQL is an implementation of the Database interface using PostgreSQL
@ -79,14 +79,13 @@ func NewPostgreSQL(ctx context.Context, connectionURI string) (*PostgreSQL, erro
}, nil
}
//nolint:cyclop // Database filtering logic is inherently complex but clear
func (db *PostgreSQL) List(
func (db *PostgreSQL) ListServers(
ctx context.Context,
tx pgx.Tx,
filter *ServerFilter,
cursor string,
limit int,
) ([]*apiv0.ServerJSON, string, error) {
) ([]*apiv0.ServerResponse, string, error) {
if limit <= 0 {
limit = 10
}
@ -95,15 +94,15 @@ func (db *PostgreSQL) List(
return nil, "", ctx.Err()
}
// Build WHERE clause for filtering
// Build WHERE clause for filtering using dedicated columns
var whereConditions []string
args := []any{}
argIndex := 1
// Add filters using JSON operators
// Add filters using dedicated columns for better performance
if filter != nil {
if filter.Name != nil {
whereConditions = append(whereConditions, fmt.Sprintf("value->>'name' = $%d", argIndex))
whereConditions = append(whereConditions, fmt.Sprintf("server_name = $%d", argIndex))
args = append(args, *filter.Name)
argIndex++
}
@ -113,35 +112,45 @@ func (db *PostgreSQL) List(
argIndex++
}
if filter.UpdatedSince != nil {
whereConditions = append(whereConditions, fmt.Sprintf("(value->'_meta'->'io.modelcontextprotocol.registry/official'->>'updatedAt')::timestamp > $%d", argIndex))
whereConditions = append(whereConditions, fmt.Sprintf("updated_at > $%d", argIndex))
args = append(args, *filter.UpdatedSince)
argIndex++
}
if filter.SubstringName != nil {
whereConditions = append(whereConditions, fmt.Sprintf("value->>'name' ILIKE $%d", argIndex))
whereConditions = append(whereConditions, fmt.Sprintf("server_name ILIKE $%d", argIndex))
args = append(args, "%"+*filter.SubstringName+"%")
argIndex++
}
if filter.Version != nil {
whereConditions = append(whereConditions, fmt.Sprintf("(value->'version_detail'->>'version') = $%d", argIndex))
whereConditions = append(whereConditions, fmt.Sprintf("version = $%d", argIndex))
args = append(args, *filter.Version)
argIndex++
}
if filter.IsLatest != nil {
whereConditions = append(whereConditions, fmt.Sprintf("(value->'_meta'->'io.modelcontextprotocol.registry/official'->>'isLatest')::boolean = $%d", argIndex))
whereConditions = append(whereConditions, fmt.Sprintf("is_latest = $%d", argIndex))
args = append(args, *filter.IsLatest)
argIndex++
}
}
// Add cursor pagination using primary key version_id
// Add cursor pagination using compound serverName:version cursor
if cursor != "" {
if _, err := uuid.Parse(cursor); err != nil {
return nil, "", fmt.Errorf("invalid cursor format: %w", err)
// Parse cursor format: "serverName:version"
parts := strings.SplitN(cursor, ":", 2)
if len(parts) == 2 {
cursorServerName := parts[0]
cursorVersion := parts[1]
// Use compound condition: (server_name > cursor_name) OR (server_name = cursor_name AND version > cursor_version)
whereConditions = append(whereConditions, fmt.Sprintf("(server_name > $%d OR (server_name = $%d AND version > $%d))", argIndex, argIndex+1, argIndex+2))
args = append(args, cursorServerName, cursorServerName, cursorVersion)
argIndex += 3
} else {
// Fallback for malformed cursor - treat as server name only for backwards compatibility
whereConditions = append(whereConditions, fmt.Sprintf("server_name > $%d", argIndex))
args = append(args, cursor)
argIndex++
}
whereConditions = append(whereConditions, fmt.Sprintf("version_id > $%d", argIndex))
args = append(args, cursor)
argIndex++
}
// Build the WHERE clause
@ -150,12 +159,12 @@ func (db *PostgreSQL) List(
whereClause = "WHERE " + strings.Join(whereConditions, " AND ")
}
// Simple query on servers table
// Query servers table with hybrid column/JSON data
query := fmt.Sprintf(`
SELECT value
SELECT server_name, version, status, published_at, updated_at, is_latest, value
FROM servers
%s
ORDER BY version_id
ORDER BY server_name, version
LIMIT $%d
`, whereClause, argIndex)
args = append(args, limit)
@ -166,170 +175,202 @@ func (db *PostgreSQL) List(
}
defer rows.Close()
var results []*apiv0.ServerJSON
var results []*apiv0.ServerResponse
for rows.Next() {
var serverName, version, status string
var publishedAt, updatedAt time.Time
var isLatest bool
var valueJSON []byte
err := rows.Scan(&valueJSON)
err := rows.Scan(&serverName, &version, &status, &publishedAt, &updatedAt, &isLatest, &valueJSON)
if err != nil {
return nil, "", fmt.Errorf("failed to scan server row: %w", err)
}
// Parse the complete ServerJSON from JSONB
// Parse the ServerJSON from JSONB
var serverJSON apiv0.ServerJSON
if err := json.Unmarshal(valueJSON, &serverJSON); err != nil {
return nil, "", fmt.Errorf("failed to unmarshal server JSON: %w", err)
}
results = append(results, &serverJSON)
// Build ServerResponse with separated metadata
serverResponse := &apiv0.ServerResponse{
Server: serverJSON,
Meta: apiv0.ResponseMeta{
Official: &apiv0.RegistryExtensions{
Status: model.Status(status),
PublishedAt: publishedAt,
UpdatedAt: updatedAt,
IsLatest: isLatest,
},
},
}
results = append(results, serverResponse)
}
if err := rows.Err(); err != nil {
return nil, "", fmt.Errorf("error iterating rows: %w", err)
}
// Determine next cursor using registry metadata VersionID
// Determine next cursor using compound serverName:version format
nextCursor := ""
if len(results) > 0 && len(results) >= limit {
lastResult := results[len(results)-1]
if lastResult.Meta != nil && lastResult.Meta.Official != nil {
nextCursor = lastResult.Meta.Official.VersionID
}
nextCursor = lastResult.Server.Name + ":" + lastResult.Server.Version
}
return results, nextCursor, nil
}
func (db *PostgreSQL) GetByVersionID(ctx context.Context, tx pgx.Tx, versionID string) (*apiv0.ServerJSON, error) {
// GetServerByName retrieves the latest version of a server by server name
func (db *PostgreSQL) GetServerByName(ctx context.Context, tx pgx.Tx, serverName string) (*apiv0.ServerResponse, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
query := `
SELECT value
SELECT server_name, version, status, published_at, updated_at, is_latest, value
FROM servers
WHERE version_id = $1
`
var valueJSON []byte
err := db.getExecutor(tx).QueryRow(ctx, query, versionID).Scan(&valueJSON)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNotFound
}
return nil, fmt.Errorf("failed to get server by ID: %w", err)
}
// Parse the complete ServerJSON from JSONB
var serverJSON apiv0.ServerJSON
if err := json.Unmarshal(valueJSON, &serverJSON); err != nil {
return nil, fmt.Errorf("failed to unmarshal server JSON: %w", err)
}
return &serverJSON, nil
}
// GetByServerID retrieves the latest version of a server by server ID
func (db *PostgreSQL) GetByServerID(ctx context.Context, tx pgx.Tx, serverID string) (*apiv0.ServerJSON, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
query := `
SELECT value
FROM servers
WHERE (value->'_meta'->'io.modelcontextprotocol.registry/official'->>'serverId') = $1 AND (value->'_meta'->'io.modelcontextprotocol.registry/official'->>'isLatest')::boolean = true
ORDER BY (value->'_meta'->'io.modelcontextprotocol.registry/official'->>'publishedAt')::timestamp DESC
WHERE server_name = $1 AND is_latest = true
ORDER BY published_at DESC
LIMIT 1
`
var name, version, status string
var publishedAt, updatedAt time.Time
var isLatest bool
var valueJSON []byte
err := db.getExecutor(tx).QueryRow(ctx, query, serverID).Scan(&valueJSON)
err := db.getExecutor(tx).QueryRow(ctx, query, serverName).Scan(&name, &version, &status, &publishedAt, &updatedAt, &isLatest, &valueJSON)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNotFound
}
return nil, fmt.Errorf("failed to get server by server ID: %w", err)
return nil, fmt.Errorf("failed to get server by name: %w", err)
}
// Parse the complete ServerJSON from JSONB
// Parse the ServerJSON from JSONB
var serverJSON apiv0.ServerJSON
if err := json.Unmarshal(valueJSON, &serverJSON); err != nil {
return nil, fmt.Errorf("failed to unmarshal server JSON: %w", err)
}
return &serverJSON, nil
// Build ServerResponse with separated metadata
serverResponse := &apiv0.ServerResponse{
Server: serverJSON,
Meta: apiv0.ResponseMeta{
Official: &apiv0.RegistryExtensions{
Status: model.Status(status),
PublishedAt: publishedAt,
UpdatedAt: updatedAt,
IsLatest: isLatest,
},
},
}
return serverResponse, nil
}
// GetByServerIDAndVersion retrieves a specific version of a server by server ID and version
func (db *PostgreSQL) GetByServerIDAndVersion(ctx context.Context, tx pgx.Tx, serverID string, version string) (*apiv0.ServerJSON, error) {
// GetServerByNameAndVersion retrieves a specific version of a server by server name and version
func (db *PostgreSQL) GetServerByNameAndVersion(ctx context.Context, tx pgx.Tx, serverName string, version string) (*apiv0.ServerResponse, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
query := `
SELECT value
SELECT server_name, version, status, published_at, updated_at, is_latest, value
FROM servers
WHERE (value->'_meta'->'io.modelcontextprotocol.registry/official'->>'serverId') = $1 AND value->>'version' = $2
WHERE server_name = $1 AND version = $2
LIMIT 1
`
var name, vers, status string
var publishedAt, updatedAt time.Time
var isLatest bool
var valueJSON []byte
err := db.getExecutor(tx).QueryRow(ctx, query, serverID, version).Scan(&valueJSON)
err := db.getExecutor(tx).QueryRow(ctx, query, serverName, version).Scan(&name, &vers, &status, &publishedAt, &updatedAt, &isLatest, &valueJSON)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNotFound
}
return nil, fmt.Errorf("failed to get server by server ID and version: %w", err)
return nil, fmt.Errorf("failed to get server by name and version: %w", err)
}
// Parse the complete ServerJSON from JSONB
// Parse the ServerJSON from JSONB
var serverJSON apiv0.ServerJSON
if err := json.Unmarshal(valueJSON, &serverJSON); err != nil {
return nil, fmt.Errorf("failed to unmarshal server JSON: %w", err)
}
return &serverJSON, nil
// Build ServerResponse with separated metadata
serverResponse := &apiv0.ServerResponse{
Server: serverJSON,
Meta: apiv0.ResponseMeta{
Official: &apiv0.RegistryExtensions{
Status: model.Status(status),
PublishedAt: publishedAt,
UpdatedAt: updatedAt,
IsLatest: isLatest,
},
},
}
return serverResponse, nil
}
// GetAllVersionsByServerID retrieves all versions of a server by server ID
func (db *PostgreSQL) GetAllVersionsByServerID(ctx context.Context, tx pgx.Tx, serverID string) ([]*apiv0.ServerJSON, error) {
// GetAllVersionsByServerName retrieves all versions of a server by server name
func (db *PostgreSQL) GetAllVersionsByServerName(ctx context.Context, tx pgx.Tx, serverName string) ([]*apiv0.ServerResponse, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
query := `
SELECT value
SELECT server_name, version, status, published_at, updated_at, is_latest, value
FROM servers
WHERE (value->'_meta'->'io.modelcontextprotocol.registry/official'->>'serverId') = $1
ORDER BY (value->'_meta'->'io.modelcontextprotocol.registry/official'->>'publishedAt')::timestamp DESC
WHERE server_name = $1
ORDER BY published_at DESC
`
rows, err := db.getExecutor(tx).Query(ctx, query, serverID)
rows, err := db.getExecutor(tx).Query(ctx, query, serverName)
if err != nil {
return nil, fmt.Errorf("failed to query server versions: %w", err)
}
defer rows.Close()
var results []*apiv0.ServerJSON
var results []*apiv0.ServerResponse
for rows.Next() {
var name, version, status string
var publishedAt, updatedAt time.Time
var isLatest bool
var valueJSON []byte
err := rows.Scan(&valueJSON)
err := rows.Scan(&name, &version, &status, &publishedAt, &updatedAt, &isLatest, &valueJSON)
if err != nil {
return nil, fmt.Errorf("failed to scan server row: %w", err)
}
// Parse the complete ServerJSON from JSONB
// Parse the ServerJSON from JSONB
var serverJSON apiv0.ServerJSON
if err := json.Unmarshal(valueJSON, &serverJSON); err != nil {
return nil, fmt.Errorf("failed to unmarshal server JSON: %w", err)
}
results = append(results, &serverJSON)
// Build ServerResponse with separated metadata
serverResponse := &apiv0.ServerResponse{
Server: serverJSON,
Meta: apiv0.ResponseMeta{
Official: &apiv0.RegistryExtensions{
Status: model.Status(status),
PublishedAt: publishedAt,
UpdatedAt: updatedAt,
IsLatest: isLatest,
},
},
}
results = append(results, serverResponse)
}
if err := rows.Err(); err != nil {
@ -343,77 +384,163 @@ func (db *PostgreSQL) GetAllVersionsByServerID(ctx context.Context, tx pgx.Tx, s
return results, nil
}
// CreateServer inserts a new server version
func (db *PostgreSQL) CreateServer(ctx context.Context, tx pgx.Tx, server *apiv0.ServerJSON) (*apiv0.ServerJSON, error) {
// CreateServer inserts a new server version with official metadata
func (db *PostgreSQL) CreateServer(ctx context.Context, tx pgx.Tx, serverJSON *apiv0.ServerJSON, officialMeta *apiv0.RegistryExtensions) (*apiv0.ServerResponse, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
// Get the IDs from the registry metadata
if server.Meta == nil || server.Meta.Official == nil {
return nil, fmt.Errorf("server must have registry metadata with ServerID and VersionID")
// Validate inputs
if serverJSON == nil || officialMeta == nil {
return nil, fmt.Errorf("serverJSON and officialMeta are required")
}
versionID := server.Meta.Official.VersionID
if versionID == "" {
return nil, fmt.Errorf("server must have VersionID in registry metadata")
if serverJSON.Name == "" || serverJSON.Version == "" {
return nil, fmt.Errorf("server name and version are required")
}
// Marshal the complete server to JSONB
valueJSON, err := json.Marshal(server)
// Marshal the ServerJSON to JSONB
valueJSON, err := json.Marshal(serverJSON)
if err != nil {
return nil, fmt.Errorf("failed to marshal server JSON: %w", err)
}
// Insert the new version
// Insert the new server version using composite primary key
insertQuery := `
INSERT INTO servers (version_id, value)
VALUES ($1, $2)
INSERT INTO servers (server_name, version, status, published_at, updated_at, is_latest, value)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`
_, err = db.getExecutor(tx).Exec(ctx, insertQuery, versionID, valueJSON)
_, err = db.getExecutor(tx).Exec(ctx, insertQuery,
serverJSON.Name,
serverJSON.Version,
string(officialMeta.Status),
officialMeta.PublishedAt,
officialMeta.UpdatedAt,
officialMeta.IsLatest,
valueJSON,
)
if err != nil {
return nil, fmt.Errorf("failed to insert server: %w", err)
}
return server, nil
// Return the complete ServerResponse
serverResponse := &apiv0.ServerResponse{
Server: *serverJSON,
Meta: apiv0.ResponseMeta{
Official: officialMeta,
},
}
return serverResponse, nil
}
// UpdateServer updates an existing server record with new server details
func (db *PostgreSQL) UpdateServer(ctx context.Context, tx pgx.Tx, id string, server *apiv0.ServerJSON) (*apiv0.ServerJSON, error) {
func (db *PostgreSQL) UpdateServer(ctx context.Context, tx pgx.Tx, serverName, version string, serverJSON *apiv0.ServerJSON) (*apiv0.ServerResponse, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
// Validate that meta structure exists and VersionID matches path
if server.Meta == nil || server.Meta.Official == nil || server.Meta.Official.VersionID != id {
return nil, fmt.Errorf("%w: io.modelcontextprotocol.registry/official.version_id must match path id (%s)", ErrInvalidInput, id)
// Validate inputs
if serverJSON == nil {
return nil, fmt.Errorf("serverJSON is required")
}
// Marshal updated server
valueJSON, err := json.Marshal(server)
// Ensure the serverJSON matches the provided serverName and version
if serverJSON.Name != serverName || serverJSON.Version != version {
return nil, fmt.Errorf("%w: server name and version in JSON must match parameters", ErrInvalidInput)
}
// Marshal updated ServerJSON
valueJSON, err := json.Marshal(serverJSON)
if err != nil {
return nil, fmt.Errorf("failed to marshal updated server: %w", err)
}
// Update the complete server record using version_id
// Update only the JSON data (keep existing metadata columns)
query := `
UPDATE servers
SET value = $1
WHERE version_id = $2
UPDATE servers
SET value = $1, updated_at = NOW()
WHERE server_name = $2 AND version = $3
RETURNING server_name, version, status, published_at, updated_at, is_latest
`
result, err := db.getExecutor(tx).Exec(ctx, query, valueJSON, id)
var name, vers, status string
var publishedAt, updatedAt time.Time
var isLatest bool
err = db.getExecutor(tx).QueryRow(ctx, query, valueJSON, serverName, version).Scan(&name, &vers, &status, &publishedAt, &updatedAt, &isLatest)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNotFound
}
return nil, fmt.Errorf("failed to update server: %w", err)
}
if result.RowsAffected() == 0 {
return nil, ErrNotFound
// Return the updated ServerResponse
serverResponse := &apiv0.ServerResponse{
Server: *serverJSON,
Meta: apiv0.ResponseMeta{
Official: &apiv0.RegistryExtensions{
Status: model.Status(status),
PublishedAt: publishedAt,
UpdatedAt: updatedAt,
IsLatest: isLatest,
},
},
}
return server, nil
return serverResponse, nil
}
// SetServerStatus updates the status of a specific server version
func (db *PostgreSQL) SetServerStatus(ctx context.Context, tx pgx.Tx, serverName, version string, status string) (*apiv0.ServerResponse, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
// Update the status column
query := `
UPDATE servers
SET status = $1, updated_at = NOW()
WHERE server_name = $2 AND version = $3
RETURNING server_name, version, status, value, published_at, updated_at, is_latest
`
var name, vers, currentStatus string
var publishedAt, updatedAt time.Time
var isLatest bool
var valueJSON []byte
err := db.getExecutor(tx).QueryRow(ctx, query, status, serverName, version).Scan(&name, &vers, &currentStatus, &valueJSON, &publishedAt, &updatedAt, &isLatest)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNotFound
}
return nil, fmt.Errorf("failed to update server status: %w", err)
}
// Unmarshal the JSON data
var serverJSON apiv0.ServerJSON
if err := json.Unmarshal(valueJSON, &serverJSON); err != nil {
return nil, fmt.Errorf("failed to unmarshal server JSON: %w", err)
}
// Return the updated ServerResponse
serverResponse := &apiv0.ServerResponse{
Server: serverJSON,
Meta: apiv0.ResponseMeta{
Official: &apiv0.RegistryExtensions{
Status: model.Status(currentStatus),
PublishedAt: publishedAt,
UpdatedAt: updatedAt,
IsLatest: isLatest,
},
},
}
return serverResponse, nil
}
// InTransaction executes a function within a database transaction
@ -479,6 +606,112 @@ func hashServerName(name string) int64 {
return int64(hash & 0x7FFFFFFFFFFFFFFF)
}
// GetCurrentLatestVersion retrieves the current latest version of a server by server name
func (db *PostgreSQL) GetCurrentLatestVersion(ctx context.Context, tx pgx.Tx, serverName string) (*apiv0.ServerResponse, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
executor := db.getExecutor(tx)
query := `
SELECT server_name, version, status, value, published_at, updated_at, is_latest
FROM servers
WHERE server_name = $1 AND is_latest = true
`
row := executor.QueryRow(ctx, query, serverName)
var name, version, status string
var publishedAt, updatedAt time.Time
var isLatest bool
var jsonValue []byte
err := row.Scan(&name, &version, &status, &jsonValue, &publishedAt, &updatedAt, &isLatest)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrNotFound
}
return nil, fmt.Errorf("failed to scan server row: %w", err)
}
// Parse the JSON value to get the server details
var serverJSON apiv0.ServerJSON
if err := json.Unmarshal(jsonValue, &serverJSON); err != nil {
return nil, fmt.Errorf("failed to unmarshal server JSON: %w", err)
}
// Build ServerResponse with separated metadata
serverResponse := &apiv0.ServerResponse{
Server: serverJSON,
Meta: apiv0.ResponseMeta{
Official: &apiv0.RegistryExtensions{
PublishedAt: publishedAt,
UpdatedAt: updatedAt,
IsLatest: isLatest,
},
},
}
return serverResponse, nil
}
// CountServerVersions counts the number of versions for a server
func (db *PostgreSQL) CountServerVersions(ctx context.Context, tx pgx.Tx, serverName string) (int, error) {
if ctx.Err() != nil {
return 0, ctx.Err()
}
executor := db.getExecutor(tx)
query := `SELECT COUNT(*) FROM servers WHERE server_name = $1`
var count int
err := executor.QueryRow(ctx, query, serverName).Scan(&count)
if err != nil {
return 0, fmt.Errorf("failed to count server versions: %w", err)
}
return count, nil
}
// CheckVersionExists checks if a specific version exists for a server
func (db *PostgreSQL) CheckVersionExists(ctx context.Context, tx pgx.Tx, serverName, version string) (bool, error) {
if ctx.Err() != nil {
return false, ctx.Err()
}
executor := db.getExecutor(tx)
query := `SELECT EXISTS(SELECT 1 FROM servers WHERE server_name = $1 AND version = $2)`
var exists bool
err := executor.QueryRow(ctx, query, serverName, version).Scan(&exists)
if err != nil {
return false, fmt.Errorf("failed to check version existence: %w", err)
}
return exists, nil
}
// UnmarkAsLatest marks the current latest version of a server as no longer latest
func (db *PostgreSQL) UnmarkAsLatest(ctx context.Context, tx pgx.Tx, serverName string) error {
if ctx.Err() != nil {
return ctx.Err()
}
executor := db.getExecutor(tx)
query := `UPDATE servers SET is_latest = false WHERE server_name = $1 AND is_latest = true`
_, err := executor.Exec(ctx, query, serverName)
if err != nil {
return fmt.Errorf("failed to unmark latest version: %w", err)
}
return nil
}
// Close closes the database connection
func (db *PostgreSQL) Close() error {
db.pool.Close()

969
internal/database/postgres_test.go Normal file
View file

@ -0,0 +1,969 @@
package database_test
import (
"context"
"fmt"
"testing"
"time"
"github.com/jackc/pgx/v5"
"github.com/modelcontextprotocol/registry/internal/database"
apiv0 "github.com/modelcontextprotocol/registry/pkg/api/v0"
"github.com/modelcontextprotocol/registry/pkg/model"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestPostgreSQL_CreateServer(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
tests := []struct {
name string
serverJSON *apiv0.ServerJSON
officialMeta *apiv0.RegistryExtensions
expectError bool
errorType error
}{
{
name: "successful server creation",
serverJSON: &apiv0.ServerJSON{
Name: "com.example/test-server",
Description: "A test server",
Version: "1.0.0",
Remotes: []model.Transport{
{Type: "http", URL: "https://api.example.com/mcp"},
},
},
officialMeta: &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
},
expectError: false,
},
{
name: "duplicate server version should fail",
serverJSON: &apiv0.ServerJSON{
Name: "com.example/duplicate-server",
Description: "A duplicate test server",
Version: "1.0.0",
},
officialMeta: &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
},
expectError: true,
// Note: Expecting generic database error for constraint violation
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Create the first server to test duplicates
if tt.name == "duplicate server version should fail" {
_, err := db.CreateServer(ctx, nil, tt.serverJSON, tt.officialMeta)
require.NoError(t, err, "First creation should succeed")
}
result, err := db.CreateServer(ctx, nil, tt.serverJSON, tt.officialMeta)
if tt.expectError {
assert.Error(t, err)
if tt.errorType != nil {
assert.ErrorIs(t, err, tt.errorType)
}
assert.Nil(t, result)
} else {
assert.NoError(t, err)
assert.NotNil(t, result)
assert.Equal(t, tt.serverJSON.Name, result.Server.Name)
assert.Equal(t, tt.serverJSON.Version, result.Server.Version)
assert.Equal(t, tt.serverJSON.Description, result.Server.Description)
assert.NotNil(t, result.Meta.Official)
assert.Equal(t, tt.officialMeta.Status, result.Meta.Official.Status)
assert.Equal(t, tt.officialMeta.IsLatest, result.Meta.Official.IsLatest)
}
})
}
}
func TestPostgreSQL_GetServerByName(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
// Setup test data
serverJSON := &apiv0.ServerJSON{
Name: "com.example/get-test-server",
Description: "A server for get testing",
Version: "1.0.0",
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
}
// Create the server
_, err := db.CreateServer(ctx, nil, serverJSON, officialMeta)
require.NoError(t, err)
tests := []struct {
name string
serverName string
expectError bool
errorType error
}{
{
name: "get existing server",
serverName: "com.example/get-test-server",
},
{
name: "get non-existent server",
serverName: "com.example/non-existent",
expectError: true,
errorType: database.ErrNotFound,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result, err := db.GetServerByName(ctx, nil, tt.serverName)
if tt.expectError {
assert.Error(t, err)
if tt.errorType != nil {
assert.ErrorIs(t, err, tt.errorType)
}
assert.Nil(t, result)
} else {
assert.NoError(t, err)
assert.NotNil(t, result)
assert.Equal(t, tt.serverName, result.Server.Name)
assert.NotNil(t, result.Meta.Official)
}
})
}
}
func TestPostgreSQL_GetServerByNameAndVersion(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
// Setup test data with multiple versions
serverName := "com.example/version-test-server"
versions := []string{"1.0.0", "1.1.0", "2.0.0"}
for i, version := range versions {
serverJSON := &apiv0.ServerJSON{
Name: serverName,
Description: "A server for version testing",
Version: version,
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: i == len(versions)-1, // Only last version is latest
}
_, err := db.CreateServer(ctx, nil, serverJSON, officialMeta)
require.NoError(t, err)
}
tests := []struct {
name string
serverName string
version string
expectError bool
errorType error
}{
{
name: "get existing server version",
serverName: serverName,
version: "1.1.0",
},
{
name: "get non-existent version",
serverName: serverName,
version: "3.0.0",
expectError: true,
errorType: database.ErrNotFound,
},
{
name: "get non-existent server",
serverName: "com.example/non-existent",
version: "1.0.0",
expectError: true,
errorType: database.ErrNotFound,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result, err := db.GetServerByNameAndVersion(ctx, nil, tt.serverName, tt.version)
if tt.expectError {
assert.Error(t, err)
if tt.errorType != nil {
assert.ErrorIs(t, err, tt.errorType)
}
assert.Nil(t, result)
} else {
assert.NoError(t, err)
assert.NotNil(t, result)
assert.Equal(t, tt.serverName, result.Server.Name)
assert.Equal(t, tt.version, result.Server.Version)
assert.NotNil(t, result.Meta.Official)
}
})
}
}
func TestPostgreSQL_ListServers(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
// Setup test data
testServers := []struct {
name string
version string
status model.Status
remoteURL string
isLatest bool
publishedAt time.Time
}{
{
name: "com.example/server-a",
version: "1.0.0",
status: model.StatusActive,
remoteURL: "https://api-a.example.com/mcp",
isLatest: true,
publishedAt: time.Now().Add(-2 * time.Hour),
},
{
name: "com.example/server-b",
version: "2.0.0",
status: model.StatusActive,
remoteURL: "https://api-b.example.com/mcp",
isLatest: true,
publishedAt: time.Now().Add(-1 * time.Hour),
},
{
name: "com.example/server-c",
version: "1.0.0",
status: model.StatusDeprecated,
remoteURL: "https://api-c.example.com/mcp",
isLatest: true,
publishedAt: time.Now().Add(-30 * time.Minute),
},
}
// Create test servers
for _, server := range testServers {
serverJSON := &apiv0.ServerJSON{
Name: server.name,
Description: "Test server for listing",
Version: server.version,
Remotes: []model.Transport{
{Type: "http", URL: server.remoteURL},
},
}
officialMeta := &apiv0.RegistryExtensions{
Status: server.status,
PublishedAt: server.publishedAt,
UpdatedAt: server.publishedAt,
IsLatest: server.isLatest,
}
_, err := db.CreateServer(ctx, nil, serverJSON, officialMeta)
require.NoError(t, err)
}
tests := []struct {
name string
filter *database.ServerFilter
cursor string
limit int
expectedCount int
expectedNames []string
expectError bool
}{
{
name: "list all servers",
filter: nil,
limit: 10,
expectedCount: 3,
expectedNames: []string{"com.example/server-a", "com.example/server-b", "com.example/server-c"},
},
{
name: "filter by name",
filter: &database.ServerFilter{
Name: stringPtr("com.example/server-a"),
},
limit: 10,
expectedCount: 1,
expectedNames: []string{"com.example/server-a"},
},
{
name: "filter by remote URL",
filter: &database.ServerFilter{
RemoteURL: stringPtr("https://api-b.example.com/mcp"),
},
limit: 10,
expectedCount: 1,
expectedNames: []string{"com.example/server-b"},
},
{
name: "filter by substring name",
filter: &database.ServerFilter{
SubstringName: stringPtr("server-"),
},
limit: 10,
expectedCount: 3,
},
{
name: "filter by version",
filter: &database.ServerFilter{
Version: stringPtr("1.0.0"),
},
limit: 10,
expectedCount: 2,
},
{
name: "filter by isLatest",
filter: &database.ServerFilter{
IsLatest: boolPtr(true),
},
limit: 10,
expectedCount: 3,
},
{
name: "filter by updatedSince",
filter: &database.ServerFilter{
UpdatedSince: timePtr(time.Now().Add(-45 * time.Minute)),
},
limit: 10,
expectedCount: 1, // Only server-c was updated in the last 45 minutes
},
{
name: "test pagination with limit",
filter: nil,
limit: 2,
expectedCount: 2,
},
{
name: "test cursor pagination",
filter: nil,
cursor: "com.example/server-a",
limit: 10,
// Should return servers after 'server-a' alphabetically
expectedCount: 2,
expectedNames: []string{"com.example/server-b", "com.example/server-c"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
results, nextCursor, err := db.ListServers(ctx, nil, tt.filter, tt.cursor, tt.limit)
if tt.expectError {
assert.Error(t, err)
return
}
assert.NoError(t, err)
assert.Len(t, results, tt.expectedCount)
if len(tt.expectedNames) > 0 {
actualNames := make([]string, len(results))
for i, result := range results {
actualNames[i] = result.Server.Name
}
assert.Subset(t, tt.expectedNames, actualNames)
}
// Test cursor behavior
if tt.limit < len(testServers) && len(results) == tt.limit {
assert.NotEmpty(t, nextCursor, "Should return next cursor when results are limited")
}
})
}
}
func TestPostgreSQL_UpdateServer(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
// Setup test data
serverName := "com.example/update-test-server"
version := "1.0.0"
serverJSON := &apiv0.ServerJSON{
Name: serverName,
Description: "Original description",
Version: version,
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
}
_, err := db.CreateServer(ctx, nil, serverJSON, officialMeta)
require.NoError(t, err)
tests := []struct {
name string
serverName string
version string
updatedServer *apiv0.ServerJSON
expectError bool
errorType error
}{
{
name: "successful server update",
serverName: serverName,
version: version,
updatedServer: &apiv0.ServerJSON{
Name: serverName,
Description: "Updated description",
Version: version,
Remotes: []model.Transport{
{Type: "http", URL: "https://updated.example.com/mcp"},
},
},
},
{
name: "update non-existent server",
serverName: "com.example/non-existent",
version: "1.0.0",
updatedServer: &apiv0.ServerJSON{
Name: "com.example/non-existent",
Description: "Should fail",
Version: "1.0.0",
},
expectError: true,
errorType: database.ErrNotFound,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result, err := db.UpdateServer(ctx, nil, tt.serverName, tt.version, tt.updatedServer)
if tt.expectError {
assert.Error(t, err)
if tt.errorType != nil {
assert.ErrorIs(t, err, tt.errorType)
}
assert.Nil(t, result)
} else {
assert.NoError(t, err)
assert.NotNil(t, result)
assert.Equal(t, tt.updatedServer.Description, result.Server.Description)
assert.NotNil(t, result.Meta.Official)
assert.NotZero(t, result.Meta.Official.UpdatedAt)
}
})
}
}
func TestPostgreSQL_SetServerStatus(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
// Setup test data
serverName := "com.example/status-test-server"
version := "1.0.0"
serverJSON := &apiv0.ServerJSON{
Name: serverName,
Description: "A server for status testing",
Version: version,
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
}
_, err := db.CreateServer(ctx, nil, serverJSON, officialMeta)
require.NoError(t, err)
tests := []struct {
name string
serverName string
version string
newStatus string
expectError bool
errorType error
}{
{
name: "active to deprecated",
serverName: serverName,
version: version,
newStatus: string(model.StatusDeprecated),
},
{
name: "invalid status",
serverName: serverName,
version: version,
newStatus: "invalid_status",
expectError: true,
},
{
name: "non-existent server",
serverName: "com.example/non-existent",
version: "1.0.1",
newStatus: string(model.StatusDeprecated),
expectError: true,
errorType: database.ErrNotFound,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result, err := db.SetServerStatus(ctx, nil, tt.serverName, tt.version, tt.newStatus)
if tt.expectError {
assert.Error(t, err)
if tt.errorType != nil {
assert.ErrorIs(t, err, tt.errorType)
}
assert.Nil(t, result)
} else {
assert.NoError(t, err)
assert.NotNil(t, result)
assert.Equal(t, model.Status(tt.newStatus), result.Meta.Official.Status)
assert.NotZero(t, result.Meta.Official.UpdatedAt)
}
})
}
}
func TestPostgreSQL_TransactionHandling(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
t.Run("successful transaction", func(t *testing.T) {
err := db.InTransaction(ctx, func(ctx context.Context, tx pgx.Tx) error {
serverJSON := &apiv0.ServerJSON{
Name: "com.example/transaction-success",
Description: "Transaction test server",
Version: "1.0.0",
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
}
_, err := db.CreateServer(ctx, tx, serverJSON, officialMeta)
return err
})
assert.NoError(t, err)
// Verify server was created
result, err := db.GetServerByName(ctx, nil, "com.example/transaction-success")
assert.NoError(t, err)
assert.NotNil(t, result)
})
t.Run("failed transaction rollback", func(t *testing.T) {
err := db.InTransaction(ctx, func(ctx context.Context, tx pgx.Tx) error {
serverJSON := &apiv0.ServerJSON{
Name: "com.example/transaction-rollback",
Description: "Transaction rollback test server",
Version: "1.0.0",
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
}
_, err := db.CreateServer(ctx, tx, serverJSON, officialMeta)
if err != nil {
return err
}
// Force an error to trigger rollback
return assert.AnError
})
assert.Error(t, err)
assert.Equal(t, assert.AnError, err)
// Verify server was NOT created due to rollback
result, err := db.GetServerByName(ctx, nil, "com.example/transaction-rollback")
assert.Error(t, err)
assert.ErrorIs(t, err, database.ErrNotFound)
assert.Nil(t, result)
})
}
func TestPostgreSQL_ConcurrencyAndLocking(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
serverName := "com.example/concurrent-server"
// Test advisory locking prevents concurrent publishes
t.Run("advisory locking prevents race conditions", func(t *testing.T) {
published := make(chan bool, 2)
errors := make(chan error, 2)
// Launch two concurrent publish operations
for i := 0; i < 2; i++ {
go func(version string) {
err := db.InTransaction(ctx, func(ctx context.Context, tx pgx.Tx) error {
// Acquire lock
if err := db.AcquirePublishLock(ctx, tx, serverName); err != nil {
return err
}
// Simulate some processing time
time.Sleep(100 * time.Millisecond)
serverJSON := &apiv0.ServerJSON{
Name: serverName,
Description: "Concurrent test server",
Version: version,
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
}
_, err := db.CreateServer(ctx, tx, serverJSON, officialMeta)
if err != nil {
return err
}
published <- true
return nil
})
errors <- err
}(time.Now().Format("20060102150405.000000"))
}
// Wait for both goroutines to complete
err1 := <-errors
err2 := <-errors
// One should succeed, one should wait (or fail if timeout)
successCount := 0
if err1 == nil {
successCount++
}
if err2 == nil {
successCount++
}
// At least one should succeed (both can succeed if advisory lock works properly)
assert.GreaterOrEqual(t, successCount, 1, "At least one concurrent operation should succeed")
close(published)
close(errors)
})
}
func TestPostgreSQL_HelperMethods(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
serverName := "com.example/helper-test-server"
// Setup test data with multiple versions
versions := []string{"1.0.0", "1.1.0", "2.0.0"}
for _, version := range versions {
serverJSON := &apiv0.ServerJSON{
Name: serverName,
Description: "Helper methods test server",
Version: version,
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: version == "2.0.0",
}
_, err := db.CreateServer(ctx, nil, serverJSON, officialMeta)
require.NoError(t, err)
}
t.Run("CountServerVersions", func(t *testing.T) {
count, err := db.CountServerVersions(ctx, nil, serverName)
assert.NoError(t, err)
assert.Equal(t, 3, count)
// Test non-existent server
count, err = db.CountServerVersions(ctx, nil, "com.example/non-existent")
assert.NoError(t, err)
assert.Equal(t, 0, count)
})
t.Run("CheckVersionExists", func(t *testing.T) {
exists, err := db.CheckVersionExists(ctx, nil, serverName, "1.1.0")
assert.NoError(t, err)
assert.True(t, exists)
exists, err = db.CheckVersionExists(ctx, nil, serverName, "3.0.0")
assert.NoError(t, err)
assert.False(t, exists)
})
t.Run("GetCurrentLatestVersion", func(t *testing.T) {
latest, err := db.GetCurrentLatestVersion(ctx, nil, serverName)
assert.NoError(t, err)
assert.NotNil(t, latest)
assert.Equal(t, "2.0.0", latest.Server.Version)
assert.True(t, latest.Meta.Official.IsLatest)
})
t.Run("GetAllVersionsByServerName", func(t *testing.T) {
allVersions, err := db.GetAllVersionsByServerName(ctx, nil, serverName)
assert.NoError(t, err)
assert.Len(t, allVersions, 3)
// Check versions are present
versionSet := make(map[string]bool)
for _, server := range allVersions {
versionSet[server.Server.Version] = true
}
for _, expectedVersion := range versions {
assert.True(t, versionSet[expectedVersion], "Version %s should be present", expectedVersion)
}
})
t.Run("UnmarkAsLatest", func(t *testing.T) {
err := db.UnmarkAsLatest(ctx, nil, serverName)
assert.NoError(t, err)
// Verify no version is marked as latest
latest, err := db.GetCurrentLatestVersion(ctx, nil, serverName)
assert.Error(t, err)
assert.ErrorIs(t, err, database.ErrNotFound)
assert.Nil(t, latest)
})
}
func TestPostgreSQL_EdgeCases(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
t.Run("input validation", func(t *testing.T) {
// Test nil inputs
_, err := db.CreateServer(ctx, nil, nil, nil)
assert.Error(t, err)
assert.Contains(t, err.Error(), "serverJSON and officialMeta are required")
// Test empty required fields
_, err = db.CreateServer(ctx, nil, &apiv0.ServerJSON{}, &apiv0.RegistryExtensions{})
assert.Error(t, err)
assert.Contains(t, err.Error(), "server name and version are required")
})
t.Run("database constraints", func(t *testing.T) {
// Test server name format constraint (should be caught by database constraint)
invalidServer := &apiv0.ServerJSON{
Name: "invalid-name-format", // Missing namespace/name format
Description: "Invalid server",
Version: "1.0.0",
}
officialMeta := &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
}
_, err := db.CreateServer(ctx, nil, invalidServer, officialMeta)
assert.Error(t, err, "Should fail due to server name format constraint")
})
t.Run("pagination edge cases", func(t *testing.T) {
// Test pagination with no results
results, cursor, err := db.ListServers(ctx, nil, &database.ServerFilter{
Name: stringPtr("com.example/non-existent-server"),
}, "", 10)
assert.NoError(t, err)
assert.Empty(t, results)
assert.Empty(t, cursor)
// Test pagination with limit 0 (should use default)
_, _, err = db.ListServers(ctx, nil, nil, "", 0)
assert.NoError(t, err)
// Should still work with default limit
})
t.Run("complex filtering", func(t *testing.T) {
// Setup test data
serverName := "com.example/complex-filter-server"
testTime := time.Now().Add(-1 * time.Hour)
_, err := db.CreateServer(ctx, nil, &apiv0.ServerJSON{
Name: serverName,
Description: "Complex filter test server",
Version: "1.0.0",
Remotes: []model.Transport{
{Type: "streamable-http", URL: "https://complex.example.com/mcp"},
},
}, &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: testTime,
UpdatedAt: testTime,
IsLatest: true,
})
require.NoError(t, err)
// Test multiple filters combined
filter := &database.ServerFilter{
SubstringName: stringPtr("complex"),
UpdatedSince: timePtr(testTime.Add(-30 * time.Minute)),
IsLatest: boolPtr(true),
Version: stringPtr("1.0.0"),
}
results, _, err := db.ListServers(ctx, nil, filter, "", 10)
assert.NoError(t, err)
assert.Len(t, results, 1)
assert.Equal(t, serverName, results[0].Server.Name)
})
t.Run("status transitions", func(t *testing.T) {
serverName := "com.example/status-transition-server"
version := "1.0.2"
// Create server
_, err := db.CreateServer(ctx, nil, &apiv0.ServerJSON{
Name: serverName,
Description: "Status transition test",
Version: version,
}, &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
})
require.NoError(t, err)
// Test all valid status transitions
statuses := []string{
string(model.StatusDeprecated),
string(model.StatusDeleted),
string(model.StatusActive), // Can transition back
}
for _, status := range statuses {
result, err := db.SetServerStatus(ctx, nil, serverName, version, status)
assert.NoError(t, err, "Should allow transition to %s", status)
assert.Equal(t, model.Status(status), result.Meta.Official.Status)
}
})
}
func TestPostgreSQL_PerformanceScenarios(t *testing.T) {
db := database.NewTestDB(t)
ctx := context.Background()
t.Run("many versions management", func(t *testing.T) {
serverName := "com.example/many-versions-server"
// Create many versions (but stay under the limit)
versionCount := 50
for i := 0; i < versionCount; i++ {
_, err := db.CreateServer(ctx, nil, &apiv0.ServerJSON{
Name: serverName,
Description: fmt.Sprintf("Version %d", i),
Version: fmt.Sprintf("1.0.%d", i),
}, &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: i == versionCount-1, // Only last one is latest
})
require.NoError(t, err)
}
// Test counting versions
count, err := db.CountServerVersions(ctx, nil, serverName)
assert.NoError(t, err)
assert.Equal(t, versionCount, count)
// Test getting all versions
allVersions, err := db.GetAllVersionsByServerName(ctx, nil, serverName)
assert.NoError(t, err)
assert.Len(t, allVersions, versionCount)
// Test only one is marked as latest
latestCount := 0
for _, version := range allVersions {
if version.Meta.Official.IsLatest {
latestCount++
}
}
assert.Equal(t, 1, latestCount)
})
t.Run("large result pagination", func(t *testing.T) {
// Create multiple servers for pagination testing
serverCount := 25
for i := 0; i < serverCount; i++ {
_, err := db.CreateServer(ctx, nil, &apiv0.ServerJSON{
Name: fmt.Sprintf("com.example/pagination-server-%02d", i),
Description: "Pagination test server",
Version: "1.0.0",
}, &apiv0.RegistryExtensions{
Status: model.StatusActive,
PublishedAt: time.Now(),
UpdatedAt: time.Now(),
IsLatest: true,
})
require.NoError(t, err)
}
// Test paginated retrieval
allResults := []*apiv0.ServerResponse{}
cursor := ""
pageSize := 10
for {
results, nextCursor, err := db.ListServers(ctx, nil, nil, cursor, pageSize)
assert.NoError(t, err)
allResults = append(allResults, results...)
if nextCursor == "" || len(results) < pageSize {
break
}
cursor = nextCursor
}
// Should have retrieved all servers including the ones we just created
assert.GreaterOrEqual(t, len(allResults), serverCount)
})
}
// Helper functions for creating pointers to basic types
func stringPtr(s string) *string {
return &s
}
func boolPtr(b bool) *bool {
return &b
}
func timePtr(t time.Time) *time.Time {
return &t
}