server.go 6.4 KB

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