package httpserver import ( "encoding/json" "fmt" "net/http" "strings" "sync/atomic" "time" "github.com/danfragoso/pizzasql-next/pkg/lexer" "github.com/danfragoso/pizzasql-next/pkg/parser" ) // QueryRequest represents a single query request. type QueryRequest struct { SQL string `json:"sql"` Params []interface{} `json:"params"` } // ExecuteRequest represents a batch execution request. type ExecuteRequest struct { Statements []QueryRequest `json:"statements"` Transaction bool `json:"transaction"` } // TransactionRequest represents a transaction management request. type TransactionRequest struct { TransactionID string `json:"transactionId"` } // handleQuery handles POST /query func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only POST method is allowed", nil) return } var req QueryRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "INVALID_JSON", "Invalid JSON in request body", nil) return } if req.SQL == "" { writeError(w, http.StatusBadRequest, "MISSING_SQL", "SQL query is required", nil) return } // Check for pretty print pretty := r.URL.Query().Get("pretty") == "true" explain := r.URL.Query().Get("explain") == "true" readonly := r.URL.Query().Get("readonly") == "true" // Parse timeout timeout := 5 * time.Minute if t := r.URL.Query().Get("timeout"); t != "" { if d, err := time.ParseDuration(t); err == nil { timeout = d } } // Execute with timeout resultChan := make(chan *QueryResponse, 1) errorChan := make(chan error, 1) go func() { start := time.Now() // Check readonly mode if readonly { upper := strings.ToUpper(strings.TrimSpace(req.SQL)) if strings.HasPrefix(upper, "INSERT") || strings.HasPrefix(upper, "UPDATE") || strings.HasPrefix(upper, "DELETE") || strings.HasPrefix(upper, "CREATE") || strings.HasPrefix(upper, "DROP") || strings.HasPrefix(upper, "ALTER") { errorChan <- &HTTPError{ Code: "READ_ONLY_MODE", Message: "Write operations not allowed in read-only mode", Status: http.StatusForbidden, } return } } // Substitute parameters sql := substituteParams(req.SQL, req.Params) // Parse SQL l := lexer.New(sql) p := parser.New(l) stmt, err := p.Parse() if err != nil { errorChan <- &HTTPError{ Code: "SYNTAX_ERROR", Message: err.Error(), Status: http.StatusBadRequest, } return } // Execute result, err := s.executor.Execute(stmt) if err != nil { errorChan <- &HTTPError{ Code: "EXECUTION_ERROR", Message: err.Error(), Status: http.StatusInternalServerError, } return } duration := time.Since(start) // Build response resp := &QueryResponse{ Columns: make([]ColumnInfo, len(result.Columns)), Rows: result.Rows, RowsAffected: result.RowsAffected, LastInsertID: result.LastInsertID, ExecutionTime: duration.String(), } for i, col := range result.Columns { colType := result.GetColumnType(i) // If no type info, infer from first row values if colType == "ANY" && len(result.Rows) > 0 && i < len(result.Rows[0]) { colType = inferType(result.Rows[0][i]) } resp.Columns[i] = ColumnInfo{ Name: col, Type: colType, } } if explain { resp.QueryPlan = []string{"Full table scan"} // TODO: Real query plan } resultChan <- resp }() select { case resp := <-resultChan: atomic.AddInt64(&s.stats.QueriesExecuted, 1) atomic.AddInt64(&s.stats.QueriesSuccess, 1) writeJSON(w, http.StatusOK, resp, pretty) case err := <-errorChan: atomic.AddInt64(&s.stats.QueriesExecuted, 1) atomic.AddInt64(&s.stats.QueriesError, 1) if httpErr, ok := err.(*HTTPError); ok { writeError(w, httpErr.Status, httpErr.Code, httpErr.Message, httpErr.Details) } else { writeError(w, http.StatusInternalServerError, "INTERNAL_ERROR", err.Error(), nil) } case <-time.After(timeout): atomic.AddInt64(&s.stats.QueriesExecuted, 1) atomic.AddInt64(&s.stats.QueriesError, 1) writeError(w, http.StatusRequestTimeout, "TIMEOUT", "Query execution timeout", nil) } } // handleExecute handles POST /execute for batch operations func (s *Server) handleExecute(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only POST method is allowed", nil) return } var req ExecuteRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "INVALID_JSON", "Invalid JSON in request body", nil) return } if len(req.Statements) == 0 { writeError(w, http.StatusBadRequest, "MISSING_STATEMENTS", "At least one statement is required", nil) return } pretty := r.URL.Query().Get("pretty") == "true" start := time.Now() results := make([]ExecuteResult, 0, len(req.Statements)) // Start transaction if requested if req.Transaction { l := lexer.New("BEGIN") p := parser.New(l) stmt, _ := p.Parse() s.executor.Execute(stmt) } var executeErr error for _, stmt := range req.Statements { // Substitute parameters sql := substituteParams(stmt.SQL, stmt.Params) l := lexer.New(sql) p := parser.New(l) parsed, err := p.Parse() if err != nil { executeErr = err break } result, err := s.executor.Execute(parsed) if err != nil { executeErr = err break } results = append(results, ExecuteResult{ RowsAffected: result.RowsAffected, LastInsertID: result.LastInsertID, }) } // Handle transaction if req.Transaction { if executeErr != nil { // Rollback on error l := lexer.New("ROLLBACK") p := parser.New(l) stmt, _ := p.Parse() s.executor.Execute(stmt) writeError(w, http.StatusBadRequest, "TRANSACTION_ERROR", executeErr.Error(), nil) return } else { // Commit on success l := lexer.New("COMMIT") p := parser.New(l) stmt, _ := p.Parse() s.executor.Execute(stmt) } } else if executeErr != nil { writeError(w, http.StatusBadRequest, "EXECUTION_ERROR", executeErr.Error(), nil) return } resp := &ExecuteResponse{ Results: results, ExecutionTime: time.Since(start).String(), } writeJSON(w, http.StatusOK, resp, pretty) } // handleSchemaTables handles GET /schema/tables func (s *Server) handleSchemaTables(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only GET method is allowed", nil) return } tables, err := s.schema.ListTables() if err != nil { writeError(w, http.StatusInternalServerError, "SCHEMA_ERROR", err.Error(), nil) return } resp := map[string]interface{}{ "tables": tables, } pretty := r.URL.Query().Get("pretty") == "true" writeJSON(w, http.StatusOK, resp, pretty) } // handleSchemaTable handles GET /schema/tables/{table} func (s *Server) handleSchemaTable(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only GET method is allowed", nil) return } // Extract table name from path path := strings.TrimPrefix(r.URL.Path, "/schema/tables/") tableName := strings.TrimSpace(path) if tableName == "" { writeError(w, http.StatusBadRequest, "MISSING_TABLE_NAME", "Table name is required", nil) return } schema, err := s.schema.GetSchema(tableName) if err != nil { writeError(w, http.StatusNotFound, "TABLE_NOT_FOUND", fmt.Sprintf("Table '%s' not found", tableName), nil) return } columns := make([]map[string]interface{}, len(schema.Columns)) for i, col := range schema.Columns { columns[i] = map[string]interface{}{ "name": col.Name, "type": col.Type, "nullable": col.Nullable, "primaryKey": col.PrimaryKey, "default": col.Default, } } resp := map[string]interface{}{ "name": schema.Name, "columns": columns, } pretty := r.URL.Query().Get("pretty") == "true" writeJSON(w, http.StatusOK, resp, pretty) } // handleHealth handles GET /health func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { resp := map[string]interface{}{ "status": "ok", "version": "0.1.0", "uptime": time.Since(s.stats.StartTime).String(), } pretty := r.URL.Query().Get("pretty") == "true" writeJSON(w, http.StatusOK, resp, pretty) } // handleStats handles GET /stats func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) { tables, _ := s.schema.ListTables() var avgQueryTime string if s.stats.QueriesExecuted > 0 { avgQueryTime = "N/A" // TODO: Track actual query times } else { avgQueryTime = "0ms" } resp := map[string]interface{}{ "queriesExecuted": atomic.LoadInt64(&s.stats.QueriesExecuted), "queriesSuccess": atomic.LoadInt64(&s.stats.QueriesSuccess), "queriesError": atomic.LoadInt64(&s.stats.QueriesError), "tablesCount": len(tables), "avgQueryTime": avgQueryTime, "uptime": time.Since(s.stats.StartTime).String(), } pretty := r.URL.Query().Get("pretty") == "true" writeJSON(w, http.StatusOK, resp, pretty) } // handleTransactionBegin handles POST /transaction/begin func (s *Server) handleTransactionBegin(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only POST method is allowed", nil) return } l := lexer.New("BEGIN") p := parser.New(l) stmt, _ := p.Parse() _, err := s.executor.Execute(stmt) if err != nil { writeError(w, http.StatusInternalServerError, "TRANSACTION_ERROR", err.Error(), nil) return } // Generate transaction ID (simple implementation) txID := fmt.Sprintf("tx-%d", time.Now().UnixNano()) resp := map[string]interface{}{ "transactionId": txID, } pretty := r.URL.Query().Get("pretty") == "true" writeJSON(w, http.StatusOK, resp, pretty) } // handleTransactionCommit handles POST /transaction/commit func (s *Server) handleTransactionCommit(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only POST method is allowed", nil) return } var req TransactionRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { // Allow commit without transaction ID for simplicity } l := lexer.New("COMMIT") p := parser.New(l) stmt, _ := p.Parse() _, err := s.executor.Execute(stmt) if err != nil { writeError(w, http.StatusInternalServerError, "TRANSACTION_ERROR", err.Error(), nil) return } resp := map[string]interface{}{ "status": "committed", } pretty := r.URL.Query().Get("pretty") == "true" writeJSON(w, http.StatusOK, resp, pretty) } // substituteParams replaces ? placeholders with actual parameter values. // This is a simple implementation that handles basic SQL escaping. func substituteParams(sql string, params []interface{}) string { if len(params) == 0 { return sql } result := sql for _, param := range params { idx := strings.Index(result, "?") if idx == -1 { break } var replacement string switch v := param.(type) { case nil: replacement = "NULL" case string: // Escape single quotes in strings escaped := strings.ReplaceAll(v, "'", "''") replacement = "'" + escaped + "'" case int, int64, int32, int16, int8: replacement = fmt.Sprintf("%d", v) case float64, float32: replacement = fmt.Sprintf("%g", v) case bool: if v { replacement = "1" } else { replacement = "0" } default: // For other types, convert to string escaped := strings.ReplaceAll(fmt.Sprintf("%v", v), "'", "''") replacement = "'" + escaped + "'" } result = result[:idx] + replacement + result[idx+1:] } return result } // inferType infers SQL type from a Go value. func inferType(v interface{}) string { switch v.(type) { case nil: return "NULL" case int, int64, int32, int16, int8: return "INTEGER" case float64, float32: return "REAL" case string: return "TEXT" case []byte: return "BLOB" case bool: return "INTEGER" default: return "ANY" } } // handleTransactionRollback handles POST /transaction/rollback func (s *Server) handleTransactionRollback(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only POST method is allowed", nil) return } var req TransactionRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { // Allow rollback without transaction ID for simplicity } l := lexer.New("ROLLBACK") p := parser.New(l) stmt, _ := p.Parse() _, err := s.executor.Execute(stmt) if err != nil { writeError(w, http.StatusInternalServerError, "TRANSACTION_ERROR", err.Error(), nil) return } resp := map[string]interface{}{ "status": "rolled back", } pretty := r.URL.Query().Get("pretty") == "true" writeJSON(w, http.StatusOK, resp, pretty) } // handleMetrics handles GET /metrics in Prometheus format func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Only GET method is allowed", nil) return } tables, _ := s.schema.ListTables() uptime := time.Since(s.stats.StartTime).Seconds() queriesTotal := atomic.LoadInt64(&s.stats.QueriesExecuted) queriesSuccess := atomic.LoadInt64(&s.stats.QueriesSuccess) queriesError := atomic.LoadInt64(&s.stats.QueriesError) w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") // Write Prometheus format metrics fmt.Fprintf(w, "# HELP pizzasql_queries_total Total number of queries executed\n") fmt.Fprintf(w, "# TYPE pizzasql_queries_total counter\n") fmt.Fprintf(w, "pizzasql_queries_total{status=\"success\"} %d\n", queriesSuccess) fmt.Fprintf(w, "pizzasql_queries_total{status=\"error\"} %d\n", queriesError) fmt.Fprintf(w, "\n") fmt.Fprintf(w, "# HELP pizzasql_queries_executed_total Total queries executed (all statuses)\n") fmt.Fprintf(w, "# TYPE pizzasql_queries_executed_total counter\n") fmt.Fprintf(w, "pizzasql_queries_executed_total %d\n", queriesTotal) fmt.Fprintf(w, "\n") fmt.Fprintf(w, "# HELP pizzasql_tables_count Number of tables in the database\n") fmt.Fprintf(w, "# TYPE pizzasql_tables_count gauge\n") fmt.Fprintf(w, "pizzasql_tables_count %d\n", len(tables)) fmt.Fprintf(w, "\n") fmt.Fprintf(w, "# HELP pizzasql_uptime_seconds Server uptime in seconds\n") fmt.Fprintf(w, "# TYPE pizzasql_uptime_seconds gauge\n") fmt.Fprintf(w, "pizzasql_uptime_seconds %.2f\n", uptime) fmt.Fprintf(w, "\n") fmt.Fprintf(w, "# HELP pizzasql_info PizzaSQL server information\n") fmt.Fprintf(w, "# TYPE pizzasql_info gauge\n") fmt.Fprintf(w, "pizzasql_info{version=\"0.1.0\"} 1\n") }