server.go 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236
  1. package httpserver
  2. import (
  3. "context"
  4. "fmt"
  5. "log"
  6. "net/http"
  7. "sync"
  8. "time"
  9. "github.com/danfragoso/pizzasql-next/pkg/executor"
  10. "github.com/danfragoso/pizzasql-next/pkg/storage"
  11. )
  12. // Config holds HTTP server configuration.
  13. type Config struct {
  14. Host string
  15. Port int
  16. ReadTimeout time.Duration
  17. WriteTimeout time.Duration
  18. MaxConnections int
  19. EnableCORS bool
  20. EnableAuth bool
  21. EnableCompression bool
  22. APIKeys []string
  23. TLSCertFile string
  24. TLSKeyFile string
  25. }
  26. // DefaultConfig returns default server configuration.
  27. func DefaultConfig() *Config {
  28. return &Config{
  29. Host: "localhost",
  30. Port: 8080,
  31. ReadTimeout: 30 * time.Second,
  32. WriteTimeout: 30 * time.Second,
  33. MaxConnections: 1000,
  34. EnableCORS: true,
  35. EnableAuth: false,
  36. EnableCompression: true,
  37. APIKeys: []string{},
  38. }
  39. }
  40. // Server represents the HTTP API server.
  41. type Server struct {
  42. config *Config
  43. executor *executor.Executor // Default executor (for backward compatibility)
  44. schema *storage.SchemaManager
  45. dbManager *storage.DatabaseManager // Multi-database support
  46. server *http.Server
  47. stats *Stats
  48. // Per-server executor cache for multi-database support
  49. executorCache map[string]*executor.Executor
  50. executorCacheMu sync.RWMutex
  51. }
  52. // Stats tracks server statistics.
  53. type Stats struct {
  54. QueriesExecuted int64
  55. QueriesSuccess int64
  56. QueriesError int64
  57. StartTime time.Time
  58. }
  59. // New creates a new HTTP server.
  60. // Deprecated: Use NewWithDatabaseManager for multi-database support.
  61. func New(config *Config, exec *executor.Executor, schema *storage.SchemaManager) *Server {
  62. if config == nil {
  63. config = DefaultConfig()
  64. }
  65. s := &Server{
  66. config: config,
  67. executor: exec,
  68. schema: schema,
  69. executorCache: make(map[string]*executor.Executor),
  70. stats: &Stats{
  71. StartTime: time.Now(),
  72. },
  73. }
  74. return s.init()
  75. }
  76. // NewWithDatabaseManager creates a new HTTP server with multi-database support.
  77. func NewWithDatabaseManager(config *Config, dbManager *storage.DatabaseManager) *Server {
  78. if config == nil {
  79. config = DefaultConfig()
  80. }
  81. // Initialize executor cache
  82. execCache := make(map[string]*executor.Executor)
  83. // Get the default database for backward compatibility
  84. defaultDB, _ := dbManager.GetDatabase("")
  85. var defaultExec *executor.Executor
  86. var defaultSchema *storage.SchemaManager
  87. if defaultDB != nil {
  88. defaultExec = executor.New(defaultDB.Schema, defaultDB.Table)
  89. defaultExec.SyncCatalog()
  90. defaultSchema = defaultDB.Schema
  91. // Pre-populate cache with default executor
  92. execCache[defaultDB.Name] = defaultExec
  93. }
  94. s := &Server{
  95. config: config,
  96. executor: defaultExec,
  97. schema: defaultSchema,
  98. dbManager: dbManager,
  99. executorCache: execCache,
  100. stats: &Stats{
  101. StartTime: time.Now(),
  102. },
  103. }
  104. return s.init()
  105. }
  106. // init initializes the server routes and middleware.
  107. func (s *Server) init() *Server {
  108. mux := http.NewServeMux()
  109. // Apply middleware (order matters: logging -> auth -> cors -> compression -> handler)
  110. var handler http.Handler = mux
  111. if s.config.EnableCompression {
  112. handler = s.compressionMiddleware(handler)
  113. }
  114. if s.config.EnableCORS {
  115. handler = s.corsMiddleware(handler)
  116. }
  117. if s.config.EnableAuth {
  118. handler = s.authMiddleware(handler)
  119. }
  120. handler = s.loggingMiddleware(handler)
  121. // Register routes
  122. mux.HandleFunc("/query", s.handleQuery)
  123. mux.HandleFunc("/execute", s.handleExecute)
  124. mux.HandleFunc("/schema/tables", s.handleSchemaTables)
  125. mux.HandleFunc("/schema/tables/", s.handleSchemaTable)
  126. mux.HandleFunc("/health", s.handleHealth)
  127. mux.HandleFunc("/stats", s.handleStats)
  128. mux.HandleFunc("/metrics", s.handleMetrics)
  129. mux.HandleFunc("/transaction/begin", s.handleTransactionBegin)
  130. mux.HandleFunc("/transaction/commit", s.handleTransactionCommit)
  131. mux.HandleFunc("/transaction/rollback", s.handleTransactionRollback)
  132. mux.HandleFunc("/export", s.handleExport)
  133. mux.HandleFunc("/import", s.handleImport)
  134. s.server = &http.Server{
  135. Addr: fmt.Sprintf("%s:%d", s.config.Host, s.config.Port),
  136. Handler: handler,
  137. ReadTimeout: s.config.ReadTimeout,
  138. WriteTimeout: s.config.WriteTimeout,
  139. }
  140. return s
  141. }
  142. // Start starts the HTTP server.
  143. func (s *Server) Start() error {
  144. addr := s.server.Addr
  145. log.Printf("Starting HTTP server on http://%s", addr)
  146. if s.config.TLSCertFile != "" && s.config.TLSKeyFile != "" {
  147. return s.server.ListenAndServeTLS(s.config.TLSCertFile, s.config.TLSKeyFile)
  148. }
  149. return s.server.ListenAndServe()
  150. }
  151. // Shutdown gracefully shuts down the server.
  152. func (s *Server) Shutdown(ctx context.Context) error {
  153. log.Println("Shutting down HTTP server...")
  154. return s.server.Shutdown(ctx)
  155. }
  156. // Addr returns the server address.
  157. func (s *Server) Addr() string {
  158. return s.server.Addr
  159. }
  160. // getExecutorForDatabase returns an executor for the specified database.
  161. // If dbName is empty, returns the default executor.
  162. // If multi-database support is not enabled, always returns the default executor.
  163. func (s *Server) getExecutorForDatabase(dbName string) (*executor.Executor, *storage.SchemaManager, error) {
  164. // If no database manager, use the default executor
  165. if s.dbManager == nil {
  166. return s.executor, s.schema, nil
  167. }
  168. // Get the database instance - this ensures we get the correct SchemaManager
  169. dbInstance, err := s.dbManager.GetDatabase(dbName)
  170. if err != nil {
  171. return nil, nil, err
  172. }
  173. // IMPORTANT: Always use dbInstance.Schema for isolation
  174. // The SchemaManager contains the database name and ensures queries
  175. // are scoped to the correct database namespace
  176. // Check per-server executor cache
  177. s.executorCacheMu.RLock()
  178. exec, exists := s.executorCache[dbInstance.Name]
  179. s.executorCacheMu.RUnlock()
  180. if exists {
  181. // Return cached executor with the correct schema from dbInstance
  182. return exec, dbInstance.Schema, nil
  183. }
  184. // Create new executor and cache it
  185. s.executorCacheMu.Lock()
  186. defer s.executorCacheMu.Unlock()
  187. // Double-check after acquiring write lock
  188. if exec, exists := s.executorCache[dbInstance.Name]; exists {
  189. return exec, dbInstance.Schema, nil
  190. }
  191. // Create executor with the database-specific schema and table managers
  192. exec = executor.New(dbInstance.Schema, dbInstance.Table)
  193. exec.SyncCatalog()
  194. s.executorCache[dbInstance.Name] = exec
  195. log.Printf("Created executor for database: %s", dbInstance.Name)
  196. return exec, dbInstance.Schema, nil
  197. }