pizzasql/pkg/httpserver/server.go

package httpserver

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/danfragoso/pizzasql-next/pkg/executor"
	"github.com/danfragoso/pizzasql-next/pkg/lexer"
	"github.com/danfragoso/pizzasql-next/pkg/parser"
	"github.com/danfragoso/pizzasql-next/pkg/storage"
)

const httpTransactionTTL = 30 * time.Minute

// Config holds HTTP server configuration.
type Config struct {
	Host              string
	Port              int
	ReadTimeout       time.Duration
	WriteTimeout      time.Duration
	MaxConnections    int
	EnableCORS        bool
	EnableAuth        bool
	EnableCompression bool
	EnableLogging     bool
	APIKeys           []string
	TLSCertFile       string
	TLSKeyFile        string
}

// DefaultConfig returns default server configuration.
func DefaultConfig() *Config {
	return &Config{
		Host:              "localhost",
		Port:              8080,
		ReadTimeout:       30 * time.Second,
		WriteTimeout:      30 * time.Second,
		MaxConnections:    1000,
		EnableCORS:        true,
		EnableAuth:        false,
		EnableCompression: true,
		EnableLogging:     true,
		APIKeys:           []string{},
	}
}

// Server represents the HTTP API server.
type Server struct {
	config    *Config
	executor  *executor.Executor // Default executor (for backward compatibility)
	schema    *storage.SchemaManager
	dbManager *storage.DatabaseManager // Multi-database support
	server    *http.Server
	stats     *Stats

	// transactionExecutors holds a per-session executor for the HTTP
	// transaction endpoints (BEGIN/COMMIT/ROLLBACK), keyed by transaction ID.
	// This prevents transaction state and caches from being shared across
	// concurrent HTTP requests.
	transactionExecutorsMu sync.RWMutex
	transactionExecutors   map[string]*transactionExecutor
}

type transactionExecutor struct {
	mu        sync.Mutex
	exec      *executor.Executor
	database  string
	expiresAt time.Time
}

// Stats tracks server statistics.
type Stats struct {
	QueriesExecuted int64
	QueriesSuccess  int64
	QueriesError    int64
	StartTime       time.Time
}

// New creates a new HTTP server.
// Deprecated: Use NewWithDatabaseManager for multi-database support.
func New(config *Config, exec *executor.Executor, schema *storage.SchemaManager) *Server {
	if config == nil {
		config = DefaultConfig()
	}

	s := &Server{
		config:               config,
		executor:             exec,
		schema:               schema,
		transactionExecutors: make(map[string]*transactionExecutor),
		stats: &Stats{
			StartTime: time.Now(),
		},
	}

	return s.init()
}

// NewWithDatabaseManager creates a new HTTP server with multi-database support.
func NewWithDatabaseManager(config *Config, dbManager *storage.DatabaseManager) *Server {
	if config == nil {
		config = DefaultConfig()
	}

	defaultDB, _ := dbManager.GetDatabase("")
	var defaultExec *executor.Executor
	var defaultSchema *storage.SchemaManager
	if defaultDB != nil {
		defaultExec = executor.New(defaultDB.Schema, defaultDB.Table)
		defaultExec.SetNotificationHub(defaultDB.Hub)
		defaultExec.SyncCatalog()
		defaultSchema = defaultDB.Schema
	}

	s := &Server{
		config:               config,
		executor:             defaultExec,
		schema:               defaultSchema,
		dbManager:            dbManager,
		transactionExecutors: make(map[string]*transactionExecutor),
		stats: &Stats{
			StartTime: time.Now(),
		},
	}

	return s.init()
}

// init initializes the server routes and middleware.
func (s *Server) init() *Server {
	mux := http.NewServeMux()

	// Apply middleware (order matters: logging -> auth -> cors -> compression -> handler)
	var handler http.Handler = mux

	if s.config.EnableCompression {
		handler = s.compressionMiddleware(handler)
	}

	if s.config.EnableCORS {
		handler = s.corsMiddleware(handler)
	}

	if s.config.EnableAuth {
		handler = s.authMiddleware(handler)
	}

	if s.config.EnableLogging {
		handler = s.loggingMiddleware(handler)
	}

	// Register routes
	mux.HandleFunc("/query", s.handleQuery)
	mux.HandleFunc("/execute", s.handleExecute)
	mux.HandleFunc("/schema/tables", s.handleSchemaTables)
	mux.HandleFunc("/schema/tables/", s.handleSchemaTable)
	mux.HandleFunc("/health", s.handleHealth)
	mux.HandleFunc("/version", s.handleVersion)
	mux.HandleFunc("/capabilities", s.handleCapabilities)
	mux.HandleFunc("/stats", s.handleStats)
	mux.HandleFunc("/metrics", s.handleMetrics)
	mux.HandleFunc("/transaction/begin", s.handleTransactionBegin)
	mux.HandleFunc("/transaction/commit", s.handleTransactionCommit)
	mux.HandleFunc("/transaction/rollback", s.handleTransactionRollback)
	mux.HandleFunc("/export", s.handleExport)
	mux.HandleFunc("/import", s.handleImport)

	s.server = &http.Server{
		Addr:         fmt.Sprintf("%s:%d", s.config.Host, s.config.Port),
		Handler:      handler,
		ReadTimeout:  s.config.ReadTimeout,
		WriteTimeout: s.config.WriteTimeout,
	}

	return s
}

// Start starts the HTTP server.
func (s *Server) Start() error {
	addr := s.server.Addr
	log.Printf("Starting HTTP server on http://%s", addr)

	if s.config.TLSCertFile != "" && s.config.TLSKeyFile != "" {
		return s.server.ListenAndServeTLS(s.config.TLSCertFile, s.config.TLSKeyFile)
	}

	return s.server.ListenAndServe()
}

// Shutdown gracefully shuts down the server.
func (s *Server) Shutdown(ctx context.Context) error {
	log.Println("Shutting down HTTP server...")
	return s.server.Shutdown(ctx)
}

// Addr returns the server address.
func (s *Server) Addr() string {
	return s.server.Addr
}

// getExecutorForDatabase returns a fresh executor for the specified database.
// A new executor is created per call so transaction state, subquery caches, and
// other mutable per-executor fields are never shared across concurrent HTTP
// requests. If dbName is empty, the default database is used.
func (s *Server) getExecutorForDatabase(dbName string) (*executor.Executor, *storage.SchemaManager, error) {
	// If no database manager, create a fresh executor from the default managers.
	if s.dbManager == nil {
		return s.executor.NewSessionExecutor(), s.schema, nil
	}

	dbInstance, err := s.dbManager.GetDatabase(dbName)
	if err != nil {
		return nil, nil, err
	}

	exec := executor.New(dbInstance.Schema, dbInstance.Table)
	exec.SetNotificationHub(dbInstance.Hub)
	exec.SyncCatalog()
	return exec, dbInstance.Schema, nil
}

// beginTransaction starts a transaction bound to a new session executor and
// registers it under the returned transaction ID.
func (s *Server) beginTransaction(dbName string) (string, *executor.Executor, error) {
	dbName = strings.TrimSpace(dbName)
	exec, _, err := s.getExecutorForDatabase(dbName)
	if err != nil {
		return "", nil, err
	}

	l := lexer.New("BEGIN")
	p := parser.New(l)
	stmt, _ := p.Parse()
	if _, err := exec.Execute(stmt); err != nil {
		return "", nil, err
	}

	idBytes := make([]byte, 16)
	if _, err := rand.Read(idBytes); err != nil {
		return "", nil, fmt.Errorf("generate transaction ID: %w", err)
	}
	txID := "tx-" + hex.EncodeToString(idBytes)
	now := time.Now()
	s.transactionExecutorsMu.Lock()
	for id, tx := range s.transactionExecutors {
		if !tx.expiresAt.After(now) {
			delete(s.transactionExecutors, id)
		}
	}
	s.transactionExecutors[txID] = &transactionExecutor{
		exec:      exec,
		database:  dbName,
		expiresAt: now.Add(httpTransactionTTL),
	}
	s.transactionExecutorsMu.Unlock()
	return txID, exec, nil
}

func (s *Server) getTransactionExecutor(txID, dbName string) (*transactionExecutor, bool) {
	dbName = strings.TrimSpace(dbName)
	s.transactionExecutorsMu.Lock()
	tx, ok := s.transactionExecutors[txID]
	if ok && !tx.expiresAt.After(time.Now()) {
		delete(s.transactionExecutors, txID)
		ok = false
	}
	if ok && tx.database == dbName {
		tx.mu.Lock()
	} else {
		ok = false
	}
	s.transactionExecutorsMu.Unlock()
	return tx, ok
}

// takeTransactionExecutor removes a session before COMMIT or ROLLBACK so it
// cannot receive another request while its terminal command is running.
func (s *Server) takeTransactionExecutor(txID, dbName string) (*transactionExecutor, bool) {
	dbName = strings.TrimSpace(dbName)
	s.transactionExecutorsMu.Lock()
	tx, ok := s.transactionExecutors[txID]
	if ok && !tx.expiresAt.After(time.Now()) {
		delete(s.transactionExecutors, txID)
		ok = false
	}
	if ok && tx.database == dbName {
		delete(s.transactionExecutors, txID)
	} else {
		ok = false
	}
	s.transactionExecutorsMu.Unlock()
	if ok {
		tx.mu.Lock()
	}
	return tx, ok
}