From 47708fc87a725378fefeb88436ae4e564d0edb25 Mon Sep 17 00:00:00 2001 From: Hein Date: Fri, 24 Jul 2026 12:40:27 +0200 Subject: [PATCH] fix(security): run RLS-scoping hooks in the same transaction as their queries BeforeRead/BeforeCreate hooks (used to set session-scoped RLS GUCs) were firing against the pooled db handle while the actual queries ran as separate calls to the same pool. Under connection pooling these could land on different physical connections, silently bypassing row-level security on creates and reads. handleUpdate already did this correctly; handleRead/handleCreate in both resolvespec and restheadspec now wrap hook execution and queries in a single RunInTransaction call. --- pkg/resolvespec/handler.go | 745 ++++++++++++++++++------------ pkg/restheadspec/handler.go | 897 +++++++++++++++++++----------------- 2 files changed, 910 insertions(+), 732 deletions(-) diff --git a/pkg/resolvespec/handler.go b/pkg/resolvespec/handler.go index bb35db1..6cbf696 100644 --- a/pkg/resolvespec/handler.go +++ b/pkg/resolvespec/handler.go @@ -259,278 +259,315 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st sliceType := reflect.SliceOf(reflect.PointerTo(modelType)) modelPtr := reflect.New(sliceType).Interface() - // Start with Model() using the slice pointer to avoid "Model(nil)" errors in Count() - // Bun's Model() accepts both single pointers and slice pointers - query := h.db.NewSelect().Model(modelPtr) + // Everything below runs inside a single transaction so that the BeforeRead + // hook (which sets session-scoped RLS GUCs via SET LOCAL) executes on the + // same physical connection as the queries it is meant to protect. Under + // connection pooling, firing the hook against h.db and then querying + // against h.db again may hand out two different connections, silently + // bypassing RLS. + var ( + result interface{} + total int + rowNumber *int64 + limit int + offset int + statusCode int + errCode string + errMsg string + ) - // Only set Table() if the model doesn't provide a table name via the underlying type - // Create a temporary instance to check for TableNameProvider - tempInstance := reflect.New(modelType).Interface() - if provider, ok := tempInstance.(common.TableNameProvider); !ok || provider.TableName() == "" { - query = query.Table(tableName) - } - - if len(options.Columns) == 0 && (len(options.ComputedColumns) > 0) { - logger.Debug("Populating options.Columns with all model columns since computed columns are additions") - options.Columns = reflection.GetSQLModelColumns(model) - } - - // Apply column selection - if len(options.Columns) > 0 { - logger.Debug("Selecting columns: %v", options.Columns) - for _, col := range options.Columns { - query = query.Column(reflection.ExtractSourceColumn(col)) + txErr := h.db.RunInTransaction(ctx, func(tx common.Database) error { + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + Model: model, + Options: options, + ID: id, + Writer: w, + Tx: tx, } - } - - if len(options.ComputedColumns) > 0 { - for _, cu := range options.ComputedColumns { - logger.Debug("Applying computed column: %s", cu.Name) - query = query.ColumnExpr(fmt.Sprintf("(%s) AS %s", cu.Expression, cu.Name)) + if err := h.hooks.Execute(BeforeRead, hookCtx); err != nil { + statusCode, errCode, errMsg = http.StatusInternalServerError, "hook_error", "BeforeRead hook failed" + return err } - } + options = hookCtx.Options - // Apply preloading - if len(options.Preload) > 0 { - var err error - query, err = h.applyPreloads(model, query, options.Preload) - if err != nil { - logger.Error("Failed to apply preloads: %v", err) - h.sendError(w, http.StatusBadRequest, "invalid_preload", "Failed to apply preloads", err) - return - } - } + // Start with Model() using the slice pointer to avoid "Model(nil)" errors in Count() + // Bun's Model() accepts both single pointers and slice pointers + query := tx.NewSelect().Model(modelPtr) - // Apply filters with proper grouping for OR logic - query = h.applyFilters(query, options.Filters) - - // Apply custom operators - for _, customOp := range options.CustomOperators { - logger.Debug("Applying custom operator: %s - %s", customOp.Name, customOp.SQL) - query = query.Where(customOp.SQL) - } - - // Apply sorting - for _, sort := range options.Sort { - direction := "ASC" - if strings.EqualFold(sort.Direction, "desc") { - direction = "DESC" - } - logger.Debug("Applying sort: %s %s", sort.Column, direction) - query = query.Order(fmt.Sprintf("%s %s", sort.Column, direction)) - } - - // Apply cursor-based pagination - if len(options.CursorForward) > 0 || len(options.CursorBackward) > 0 { - logger.Debug("Applying cursor pagination") - - // Get primary key name - pkName := reflection.GetPrimaryKeyName(model) - - // Extract model columns for validation - modelColumns := reflection.GetModelColumns(model) - - // Default sort to primary key when none provided - if len(options.Sort) == 0 { - options.Sort = []common.SortOption{{Column: pkName, Direction: "ASC"}} + // Only set Table() if the model doesn't provide a table name via the underlying type + // Create a temporary instance to check for TableNameProvider + tempInstance := reflect.New(modelType).Interface() + if provider, ok := tempInstance.(common.TableNameProvider); !ok || provider.TableName() == "" { + query = query.Table(tableName) } - // Get cursor filter SQL (expandJoins is empty for resolvespec — no custom SQL join support yet) - cursorFilter, err := GetCursorFilter(tableName, pkName, modelColumns, options, nil) - if err != nil { - logger.Error("Error building cursor filter: %v", err) - h.sendError(w, http.StatusBadRequest, "cursor_error", "Invalid cursor pagination", err) - return + if len(options.Columns) == 0 && (len(options.ComputedColumns) > 0) { + logger.Debug("Populating options.Columns with all model columns since computed columns are additions") + options.Columns = reflection.GetSQLModelColumns(model) } - // Apply cursor filter to query - if cursorFilter != "" { - logger.Debug("Applying cursor filter: %s", cursorFilter) - sanitizedCursor := common.SanitizeWhereClause(cursorFilter, reflection.ExtractTableNameOnly(tableName), &options) - // Ensure outer parentheses to prevent OR logic from escaping - sanitizedCursor = common.EnsureOuterParentheses(sanitizedCursor) - if sanitizedCursor != "" { - query = query.Where(sanitizedCursor) + // Apply column selection + if len(options.Columns) > 0 { + logger.Debug("Selecting columns: %v", options.Columns) + for _, col := range options.Columns { + query = query.Column(reflection.ExtractSourceColumn(col)) } } - } - // Get total count before pagination - var total int - - // Try to get from cache first - // Use extended cache key if cursors are present - var cacheKeyHash string - if len(options.CursorForward) > 0 || len(options.CursorBackward) > 0 { - cacheKeyHash = buildExtendedQueryCacheKey( - tableName, - options.Filters, - options.Sort, - "", // No custom SQL WHERE in resolvespec - "", // No custom SQL OR in resolvespec - options.CursorForward, - options.CursorBackward, - ) - } else { - cacheKeyHash = buildQueryCacheKey( - tableName, - options.Filters, - options.Sort, - "", // No custom SQL WHERE in resolvespec - "", // No custom SQL OR in resolvespec - ) - } - cacheKey := getQueryTotalCacheKey(cacheKeyHash) - - // Try to retrieve from cache - var cachedTotal cachedTotal - err := cache.GetDefaultCache().Get(ctx, cacheKey, &cachedTotal) - if err == nil { - total = cachedTotal.Total - logger.Debug("Total records (from cache): %d", total) - } else { - // Cache miss - execute count query - logger.Debug("Cache miss for query total") - count, err := query.Count(ctx) - if err != nil { - logger.Error("Error counting records: %v", err) - h.sendError(w, http.StatusInternalServerError, "query_error", "Error counting records", err) - return - } - total = count - logger.Debug("Total records (from query): %d", total) - - // Store in cache with schema and table tags - cacheTTL := time.Minute * 2 // Default 2 minutes TTL - if err := setQueryTotalCache(ctx, cacheKey, total, schema, tableName, cacheTTL); err != nil { - logger.Warn("Failed to cache query total: %v", err) - // Don't fail the request if caching fails - } else { - logger.Debug("Cached query total with key: %s", cacheKey) - } - } - - // Handle FetchRowNumber if requested - var rowNumber *int64 - if options.FetchRowNumber != nil && *options.FetchRowNumber != "" { - logger.Debug("Fetching row number for ID: %s", *options.FetchRowNumber) - pkName := reflection.GetPrimaryKeyName(model) - - // Build ROW_NUMBER window function SQL - rowNumberSQL := "ROW_NUMBER() OVER (" - if len(options.Sort) > 0 { - rowNumberSQL += "ORDER BY " - for i, sort := range options.Sort { - if i > 0 { - rowNumberSQL += ", " - } - direction := "ASC" - if strings.EqualFold(sort.Direction, "desc") { - direction = "DESC" - } - rowNumberSQL += fmt.Sprintf("%s %s", sort.Column, direction) + if len(options.ComputedColumns) > 0 { + for _, cu := range options.ComputedColumns { + logger.Debug("Applying computed column: %s", cu.Name) + query = query.ColumnExpr(fmt.Sprintf("(%s) AS %s", cu.Expression, cu.Name)) } } - rowNumberSQL += ")" - // Create a query to fetch the row number using a subquery approach - // We'll select the PK and row_number, then filter by the target ID - type RowNumResult struct { - RowNum int64 `bun:"row_num"` + // Apply preloading + if len(options.Preload) > 0 { + var err error + query, err = h.applyPreloads(model, query, options.Preload) + if err != nil { + logger.Error("Failed to apply preloads: %v", err) + statusCode, errCode, errMsg = http.StatusBadRequest, "invalid_preload", "Failed to apply preloads" + return err + } } - rowNumQuery := h.db.NewSelect().Table(tableName). - ColumnExpr(fmt.Sprintf("%s AS row_num", rowNumberSQL)). - Column(pkName) - - // Apply the same filters as the main query - for _, filter := range options.Filters { - rowNumQuery = h.applyFilter(rowNumQuery, filter) - } + // Apply filters with proper grouping for OR logic + query = h.applyFilters(query, options.Filters) // Apply custom operators for _, customOp := range options.CustomOperators { - rowNumQuery = rowNumQuery.Where(customOp.SQL) + logger.Debug("Applying custom operator: %s - %s", customOp.Name, customOp.SQL) + query = query.Where(customOp.SQL) } - // Filter for the specific ID we want the row number for - rowNumQuery = rowNumQuery.Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), *options.FetchRowNumber) - - // Execute query to get row number - var result RowNumResult - if err := rowNumQuery.Scan(ctx, &result); err != nil { - if err == sql.ErrNoRows { - // Build filter description for error message - filterInfo := fmt.Sprintf("filters: %d", len(options.Filters)) - if len(options.CustomOperators) > 0 { - customOps := make([]string, 0, len(options.CustomOperators)) - for _, op := range options.CustomOperators { - customOps = append(customOps, op.SQL) - } - filterInfo += fmt.Sprintf(", custom operators: [%s]", strings.Join(customOps, "; ")) - } - logger.Warn("No row found for primary key %s=%s with %s", pkName, *options.FetchRowNumber, filterInfo) - } else { - logger.Warn("Error fetching row number: %v", err) + // Apply sorting + for _, sort := range options.Sort { + direction := "ASC" + if strings.EqualFold(sort.Direction, "desc") { + direction = "DESC" } + logger.Debug("Applying sort: %s %s", sort.Column, direction) + query = query.Order(fmt.Sprintf("%s %s", sort.Column, direction)) + } + + // Apply cursor-based pagination + if len(options.CursorForward) > 0 || len(options.CursorBackward) > 0 { + logger.Debug("Applying cursor pagination") + + // Get primary key name + pkName := reflection.GetPrimaryKeyName(model) + + // Extract model columns for validation + modelColumns := reflection.GetModelColumns(model) + + // Default sort to primary key when none provided + if len(options.Sort) == 0 { + options.Sort = []common.SortOption{{Column: pkName, Direction: "ASC"}} + } + + // Get cursor filter SQL (expandJoins is empty for resolvespec — no custom SQL join support yet) + cursorFilter, err := GetCursorFilter(tableName, pkName, modelColumns, options, nil) + if err != nil { + logger.Error("Error building cursor filter: %v", err) + statusCode, errCode, errMsg = http.StatusBadRequest, "cursor_error", "Invalid cursor pagination" + return err + } + + // Apply cursor filter to query + if cursorFilter != "" { + logger.Debug("Applying cursor filter: %s", cursorFilter) + sanitizedCursor := common.SanitizeWhereClause(cursorFilter, reflection.ExtractTableNameOnly(tableName), &options) + // Ensure outer parentheses to prevent OR logic from escaping + sanitizedCursor = common.EnsureOuterParentheses(sanitizedCursor) + if sanitizedCursor != "" { + query = query.Where(sanitizedCursor) + } + } + } + + // Try to get from cache first + // Use extended cache key if cursors are present + var cacheKeyHash string + if len(options.CursorForward) > 0 || len(options.CursorBackward) > 0 { + cacheKeyHash = buildExtendedQueryCacheKey( + tableName, + options.Filters, + options.Sort, + "", // No custom SQL WHERE in resolvespec + "", // No custom SQL OR in resolvespec + options.CursorForward, + options.CursorBackward, + ) } else { - rowNumber = &result.RowNum - logger.Debug("Found row number: %d", *rowNumber) + cacheKeyHash = buildQueryCacheKey( + tableName, + options.Filters, + options.Sort, + "", // No custom SQL WHERE in resolvespec + "", // No custom SQL OR in resolvespec + ) } - } + cacheKey := getQueryTotalCacheKey(cacheKeyHash) - // Apply pagination (skip if FetchRowNumber is set - we want only that record) - if options.FetchRowNumber == nil || *options.FetchRowNumber == "" { - if options.Limit != nil && *options.Limit > 0 { - logger.Debug("Applying limit: %d", *options.Limit) - query = query.Limit(*options.Limit) - } - if options.Offset != nil && *options.Offset > 0 { - logger.Debug("Applying offset: %d", *options.Offset) - query = query.Offset(*options.Offset) - } - } - - // Execute query - var result interface{} - if id != "" || (options.FetchRowNumber != nil && *options.FetchRowNumber != "") { - // Single record query - either by URL ID or FetchRowNumber - var targetID string - if id != "" { - targetID = id - logger.Debug("Querying single record with URL ID: %s", id) + // Try to retrieve from cache + var cachedTotal cachedTotal + if err := cache.GetDefaultCache().Get(ctx, cacheKey, &cachedTotal); err == nil { + total = cachedTotal.Total + logger.Debug("Total records (from cache): %d", total) } else { - targetID = *options.FetchRowNumber - logger.Debug("Querying single record with FetchRowNumber ID: %s", targetID) + // Cache miss - execute count query + logger.Debug("Cache miss for query total") + count, err := query.Count(ctx) + if err != nil { + logger.Error("Error counting records: %v", err) + statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error counting records" + return err + } + total = count + logger.Debug("Total records (from query): %d", total) + + // Store in cache with schema and table tags + cacheTTL := time.Minute * 2 // Default 2 minutes TTL + if err := setQueryTotalCache(ctx, cacheKey, total, schema, tableName, cacheTTL); err != nil { + logger.Warn("Failed to cache query total: %v", err) + // Don't fail the request if caching fails + } else { + logger.Debug("Cached query total with key: %s", cacheKey) + } } - // For single record, create a new pointer to the struct type - singleResult := reflect.New(modelType).Interface() - pkName := reflection.GetPrimaryKeyName(singleResult) + // Handle FetchRowNumber if requested + if options.FetchRowNumber != nil && *options.FetchRowNumber != "" { + logger.Debug("Fetching row number for ID: %s", *options.FetchRowNumber) + pkName := reflection.GetPrimaryKeyName(model) - query = query.Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), targetID) - if err := query.Scan(ctx, singleResult); err != nil { - logger.Error("Error querying record: %v", err) - h.sendError(w, http.StatusInternalServerError, "query_error", "Error executing query", err) - return + // Build ROW_NUMBER window function SQL + rowNumberSQL := "ROW_NUMBER() OVER (" + if len(options.Sort) > 0 { + rowNumberSQL += "ORDER BY " + for i, sort := range options.Sort { + if i > 0 { + rowNumberSQL += ", " + } + direction := "ASC" + if strings.EqualFold(sort.Direction, "desc") { + direction = "DESC" + } + rowNumberSQL += fmt.Sprintf("%s %s", sort.Column, direction) + } + } + rowNumberSQL += ")" + + // Create a query to fetch the row number using a subquery approach + // We'll select the PK and row_number, then filter by the target ID + type RowNumResult struct { + RowNum int64 `bun:"row_num"` + } + + rowNumQuery := tx.NewSelect().Table(tableName). + ColumnExpr(fmt.Sprintf("%s AS row_num", rowNumberSQL)). + Column(pkName) + + // Apply the same filters as the main query + for _, filter := range options.Filters { + rowNumQuery = h.applyFilter(rowNumQuery, filter) + } + + // Apply custom operators + for _, customOp := range options.CustomOperators { + rowNumQuery = rowNumQuery.Where(customOp.SQL) + } + + // Filter for the specific ID we want the row number for + rowNumQuery = rowNumQuery.Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), *options.FetchRowNumber) + + // Execute query to get row number + var rnResult RowNumResult + if err := rowNumQuery.Scan(ctx, &rnResult); err != nil { + if err == sql.ErrNoRows { + // Build filter description for error message + filterInfo := fmt.Sprintf("filters: %d", len(options.Filters)) + if len(options.CustomOperators) > 0 { + customOps := make([]string, 0, len(options.CustomOperators)) + for _, op := range options.CustomOperators { + customOps = append(customOps, op.SQL) + } + filterInfo += fmt.Sprintf(", custom operators: [%s]", strings.Join(customOps, "; ")) + } + logger.Warn("No row found for primary key %s=%s with %s", pkName, *options.FetchRowNumber, filterInfo) + } else { + logger.Warn("Error fetching row number: %v", err) + } + } else { + rowNumber = &rnResult.RowNum + logger.Debug("Found row number: %d", *rowNumber) + } } - result = singleResult - } else { - logger.Debug("Querying multiple records") - // Use the modelPtr already created and set on the query - if err := query.Scan(ctx, modelPtr); err != nil { - logger.Error("Error querying records: %v", err) - h.sendError(w, http.StatusInternalServerError, "query_error", "Error executing query", err) - return + + // Apply pagination (skip if FetchRowNumber is set - we want only that record) + if options.FetchRowNumber == nil || *options.FetchRowNumber == "" { + if options.Limit != nil && *options.Limit > 0 { + logger.Debug("Applying limit: %d", *options.Limit) + query = query.Limit(*options.Limit) + } + if options.Offset != nil && *options.Offset > 0 { + logger.Debug("Applying offset: %d", *options.Offset) + query = query.Offset(*options.Offset) + } } - result = reflect.ValueOf(modelPtr).Elem().Interface() + + // Execute query + if id != "" || (options.FetchRowNumber != nil && *options.FetchRowNumber != "") { + // Single record query - either by URL ID or FetchRowNumber + var targetID string + if id != "" { + targetID = id + logger.Debug("Querying single record with URL ID: %s", id) + } else { + targetID = *options.FetchRowNumber + logger.Debug("Querying single record with FetchRowNumber ID: %s", targetID) + } + + // For single record, create a new pointer to the struct type + singleResult := reflect.New(modelType).Interface() + pkName := reflection.GetPrimaryKeyName(singleResult) + + query = query.Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), targetID) + if err := query.Scan(ctx, singleResult); err != nil { + logger.Error("Error querying record: %v", err) + statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error executing query" + return err + } + result = singleResult + } else { + logger.Debug("Querying multiple records") + // Use the modelPtr already created and set on the query + if err := query.Scan(ctx, modelPtr); err != nil { + logger.Error("Error querying records: %v", err) + statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error executing query" + return err + } + result = reflect.ValueOf(modelPtr).Elem().Interface() + } + + logger.Info("Successfully retrieved records") + return nil + }) + + if txErr != nil { + if statusCode == 0 { + statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error executing query" + } + h.sendError(w, statusCode, errCode, errMsg, txErr) + return } - logger.Info("Successfully retrieved records") - // Build metadata - limit := 0 - offset := 0 count := int64(total) // When FetchRowNumber is used, we only return 1 record @@ -585,54 +622,107 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat // Check if we should use nested processing if h.shouldUseNestedProcessor(v, model) { logger.Info("Using nested CUD processor for create operation") - result, err := h.nestedProcessor.ProcessNestedCUD(ctx, "insert", v, model, make(map[string]interface{}), tableName) + var nestedResult *common.ProcessResult + err := h.db.RunInTransaction(ctx, func(tx common.Database) error { + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + Model: model, + Options: options, + Data: v, + Writer: w, + Tx: tx, + } + if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { + return fmt.Errorf("BeforeCreate hook failed: %w", err) + } + if modifiedData, ok := hookCtx.Data.(map[string]interface{}); ok { + v = modifiedData + } + + originalProcessor := h.nestedProcessor + h.nestedProcessor = common.NewNestedCUDProcessor(tx, h.registry, h) + defer func() { + h.nestedProcessor = originalProcessor + }() + + var procErr error + nestedResult, procErr = h.nestedProcessor.ProcessNestedCUD(ctx, "insert", v, model, make(map[string]interface{}), tableName) + return procErr + }) if err != nil { logger.Error("Error in nested create: %v", err) h.sendError(w, http.StatusInternalServerError, "create_error", "Error creating record with nested data", err) return } - logger.Info("Successfully created record with nested data, ID: %v", result.ID) + logger.Info("Successfully created record with nested data, ID: %v", nestedResult.ID) // Invalidate cache for this table cacheTags := buildCacheTags(schema, tableName) if err := invalidateCacheForTags(ctx, cacheTags); err != nil { logger.Warn("Failed to invalidate cache for table %s: %v", tableName, err) } - h.sendResponse(w, result.Data, nil) + h.sendResponse(w, nestedResult.Data, nil) return } // Standard processing without nested relations pkName := reflection.GetPrimaryKeyName(model) - query := h.db.NewInsert().Table(tableName) - for key, value := range v { - query = query.Value(key, common.ConvertSliceForBun(value)) - } var responseData interface{} = v - if pkName == "" { - // No PK on model — insert and return input as-is. - result, err := query.Exec(ctx) - if err != nil { - logger.Error("Error creating record: %v", err) - h.sendError(w, http.StatusInternalServerError, "create_error", "Error creating record", err) - return + err := h.db.RunInTransaction(ctx, func(tx common.Database) error { + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + Model: model, + Options: options, + Data: v, + Writer: w, + Tx: tx, } - logger.Info("Successfully created record, rows affected: %d", result.RowsAffected()) - } else { + if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { + return fmt.Errorf("BeforeCreate hook failed: %w", err) + } + if modifiedData, ok := hookCtx.Data.(map[string]interface{}); ok { + v = modifiedData + } + responseData = v + + query := tx.NewInsert().Table(tableName) + for key, value := range v { + query = query.Value(key, common.ConvertSliceForBun(value)) + } + if pkName == "" { + // No PK on model — insert and return input as-is. + result, err := query.Exec(ctx) + if err != nil { + return err + } + logger.Info("Successfully created record, rows affected: %d", result.RowsAffected()) + return nil + } + var insertedID interface{} if err := query.Returning(pkName).Scan(ctx, &insertedID); err != nil { - logger.Error("Error creating record: %v", err) - h.sendError(w, http.StatusInternalServerError, "create_error", "Error creating record", err) - return + return err } logger.Info("Successfully created record with %s: %v", pkName, insertedID) fetchedRecord := reflect.New(reflection.GetPointerElement(reflect.TypeOf(model))).Interface() - if fetchErr := h.db.NewSelect().Model(fetchedRecord). + if fetchErr := tx.NewSelect().Model(fetchedRecord). Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), insertedID). ScanModel(ctx); fetchErr == nil { responseData = mergeWithInput(fetchedRecord, v) } else { logger.Warn("Failed to re-fetch created record with %s=%v: %v", pkName, insertedID, fetchErr) } + return nil + }) + if err != nil { + logger.Error("Error creating record: %v", err) + h.sendError(w, http.StatusInternalServerError, "create_error", "Error creating record", err) + return } // Invalidate cache for this table cacheTags := buildCacheTags(schema, tableName) @@ -663,6 +753,24 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat }() for _, item := range v { + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + Model: model, + Options: options, + Data: item, + Writer: w, + Tx: tx, + } + if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { + return fmt.Errorf("BeforeCreate hook failed: %w", err) + } + if modifiedData, ok := hookCtx.Data.(map[string]interface{}); ok { + item = modifiedData + } + result, err := h.nestedProcessor.ProcessNestedCUD(ctx, "insert", item, model, make(map[string]interface{}), tableName) if err != nil { return fmt.Errorf("failed to process item: %w", err) @@ -689,10 +797,27 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat // Standard batch insert without nested relations pkName := reflection.GetPrimaryKeyName(model) modelElemType := reflection.GetPointerElement(reflect.TypeOf(model)) - originals := make([]map[string]interface{}, 0, len(v)) - insertedIDs := make([]interface{}, 0, len(v)) + responseItems := make([]interface{}, 0, len(v)) err := h.db.RunInTransaction(ctx, func(tx common.Database) error { for _, item := range v { + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + Model: model, + Options: options, + Data: item, + Writer: w, + Tx: tx, + } + if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { + return fmt.Errorf("BeforeCreate hook failed: %w", err) + } + if modifiedData, ok := hookCtx.Data.(map[string]interface{}); ok { + item = modifiedData + } + txQuery := tx.NewInsert().Table(tableName) for key, value := range item { txQuery = txQuery.Value(key, common.ConvertSliceForBun(value)) @@ -701,16 +826,22 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat if _, err := txQuery.Exec(ctx); err != nil { return err } - originals = append(originals, item) - insertedIDs = append(insertedIDs, nil) + responseItems = append(responseItems, item) continue } var returnedID interface{} if err := txQuery.Returning(pkName).Scan(ctx, &returnedID); err != nil { return err } - originals = append(originals, item) - insertedIDs = append(insertedIDs, returnedID) + fetchedRecord := reflect.New(modelElemType).Interface() + if fetchErr := tx.NewSelect().Model(fetchedRecord). + Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), returnedID). + ScanModel(ctx); fetchErr == nil { + responseItems = append(responseItems, mergeWithInput(fetchedRecord, item)) + } else { + logger.Warn("Failed to re-fetch created record with %s=%v: %v", pkName, returnedID, fetchErr) + responseItems = append(responseItems, item) + } } return nil }) @@ -725,23 +856,6 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat if err := invalidateCacheForTags(ctx, cacheTags); err != nil { logger.Warn("Failed to invalidate cache for table %s: %v", tableName, err) } - // Re-fetch each record after transaction commits; fall back to input on failure. - responseItems := make([]interface{}, 0, len(insertedIDs)) - for i, pkVal := range insertedIDs { - if pkVal == nil { - responseItems = append(responseItems, originals[i]) - continue - } - fetchedRecord := reflect.New(modelElemType).Interface() - if fetchErr := h.db.NewSelect().Model(fetchedRecord). - Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), pkVal). - ScanModel(ctx); fetchErr == nil { - responseItems = append(responseItems, mergeWithInput(fetchedRecord, originals[i])) - } else { - logger.Warn("Failed to re-fetch created record with %s=%v: %v", pkName, pkVal, fetchErr) - responseItems = append(responseItems, originals[i]) - } - } h.sendResponse(w, responseItems, nil) case []interface{}: @@ -770,6 +884,24 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat for _, item := range v { if itemMap, ok := item.(map[string]interface{}); ok { + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + Model: model, + Options: options, + Data: itemMap, + Writer: w, + Tx: tx, + } + if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { + return fmt.Errorf("BeforeCreate hook failed: %w", err) + } + if modifiedData, ok := hookCtx.Data.(map[string]interface{}); ok { + itemMap = modifiedData + } + result, err := h.nestedProcessor.ProcessNestedCUD(ctx, "insert", itemMap, model, make(map[string]interface{}), tableName) if err != nil { return fmt.Errorf("failed to process item: %w", err) @@ -797,14 +929,32 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat // Standard batch insert without nested relations pkName := reflection.GetPrimaryKeyName(model) modelElemType := reflection.GetPointerElement(reflect.TypeOf(model)) - originals := make([]map[string]interface{}, 0, len(v)) - insertedIDs := make([]interface{}, 0, len(v)) + responseItems := make([]interface{}, 0, len(v)) err := h.db.RunInTransaction(ctx, func(tx common.Database) error { for _, item := range v { itemMap, ok := item.(map[string]interface{}) if !ok { continue } + + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + Model: model, + Options: options, + Data: itemMap, + Writer: w, + Tx: tx, + } + if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { + return fmt.Errorf("BeforeCreate hook failed: %w", err) + } + if modifiedData, ok := hookCtx.Data.(map[string]interface{}); ok { + itemMap = modifiedData + } + txQuery := tx.NewInsert().Table(tableName) for key, value := range itemMap { txQuery = txQuery.Value(key, common.ConvertSliceForBun(value)) @@ -813,16 +963,22 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat if _, err := txQuery.Exec(ctx); err != nil { return err } - originals = append(originals, itemMap) - insertedIDs = append(insertedIDs, nil) + responseItems = append(responseItems, itemMap) continue } var returnedID interface{} if err := txQuery.Returning(pkName).Scan(ctx, &returnedID); err != nil { return err } - originals = append(originals, itemMap) - insertedIDs = append(insertedIDs, returnedID) + fetchedRecord := reflect.New(modelElemType).Interface() + if fetchErr := tx.NewSelect().Model(fetchedRecord). + Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), returnedID). + ScanModel(ctx); fetchErr == nil { + responseItems = append(responseItems, mergeWithInput(fetchedRecord, itemMap)) + } else { + logger.Warn("Failed to re-fetch created record with %s=%v: %v", pkName, returnedID, fetchErr) + responseItems = append(responseItems, itemMap) + } } return nil }) @@ -837,23 +993,6 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat if err := invalidateCacheForTags(ctx, cacheTags); err != nil { logger.Warn("Failed to invalidate cache for table %s: %v", tableName, err) } - // Re-fetch each record after transaction commits; fall back to input on failure. - responseItems := make([]interface{}, 0, len(insertedIDs)) - for i, pkVal := range insertedIDs { - if pkVal == nil { - responseItems = append(responseItems, originals[i]) - continue - } - fetchedRecord := reflect.New(modelElemType).Interface() - if fetchErr := h.db.NewSelect().Model(fetchedRecord). - Where(fmt.Sprintf("%s = ?", common.QuoteIdent(pkName)), pkVal). - ScanModel(ctx); fetchErr == nil { - responseItems = append(responseItems, mergeWithInput(fetchedRecord, originals[i])) - } else { - logger.Warn("Failed to re-fetch created record with %s=%v: %v", pkName, pkVal, fetchErr) - responseItems = append(responseItems, originals[i]) - } - } h.sendResponse(w, responseItems, nil) default: diff --git a/pkg/restheadspec/handler.go b/pkg/restheadspec/handler.go index 8d87fa6..8b6c6e3 100644 --- a/pkg/restheadspec/handler.go +++ b/pkg/restheadspec/handler.go @@ -327,26 +327,6 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st options.SingleRecordAsObject = false } - // Execute BeforeRead hooks - hookCtx := &HookContext{ - Context: ctx, - Handler: h, - Schema: schema, - Entity: entity, - TableName: tableName, - Model: model, - Options: options, - ID: id, - Writer: w, - Tx: h.db, - } - - if err := h.hooks.Execute(BeforeRead, hookCtx); err != nil { - logger.Error("BeforeRead hook failed: %v", err) - h.sendError(w, http.StatusBadRequest, "hook_error", "Hook execution failed", err) - return - } - // Validate and unwrap model type to get base struct modelType := reflect.TypeOf(model) for modelType != nil && (modelType.Kind() == reflect.Pointer || modelType.Kind() == reflect.Slice || modelType.Kind() == reflect.Array) { @@ -364,437 +344,480 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st logger.Info("Reading records from %s.%s", schema, entity) - // Start with Model() using the slice pointer to avoid "Model(nil)" errors in Count() - // Bun's Model() accepts both single pointers and slice pointers - query := h.db.NewSelect().Model(modelPtr) - - // Only set Table() if the model doesn't provide a table name via the underlying type - // Create a temporary instance to check for TableNameProvider - tempInstance := reflect.New(modelType).Interface() - if provider, ok := tempInstance.(common.TableNameProvider); !ok || provider.TableName() == "" { - query = query.Table(tableName) + hookCtx := &HookContext{ + Context: ctx, + Handler: h, + Schema: schema, + Entity: entity, + TableName: tableName, + Model: model, + Options: options, + ID: id, + Writer: w, } - // If we have computed columns/expressions but options.Columns is empty, - // populate it with all model columns first since computed columns are additions - if len(options.Columns) == 0 && (len(options.ComputedQL) > 0 || len(options.ComputedColumns) > 0) { - logger.Debug("Populating options.Columns with all model columns since computed columns are additions") - options.Columns = reflection.GetSQLModelColumns(model) - } + // Everything below runs inside a single transaction so that the BeforeRead/BeforeScan + // hooks (which may set session-scoped RLS GUCs via SET LOCAL) execute on the same + // physical connection as the queries they are meant to protect. Under connection + // pooling, firing a hook against h.db and then querying against h.db again may hand + // out two different connections, silently bypassing RLS. + var ( + total int + fetchedRowNumber *int64 + statusCode int + errCode string + errMsg string + ) - // Apply ComputedQL fields if any - if len(options.ComputedQL) > 0 { - for colName, colExpr := range options.ComputedQL { - logger.Debug("Applying computed column: %s", colName) - if strings.Contains(colName, "cql") { - query = query.ColumnExpr(fmt.Sprintf("(%s)::text AS %s", colExpr, colName)) - } else { - query = query.ColumnExpr(fmt.Sprintf("(%s)AS %s", colExpr, colName)) - } + txErr := h.db.RunInTransaction(ctx, func(tx common.Database) error { + hookCtx.Tx = tx - for colIndex := range options.Columns { - if options.Columns[colIndex] == colName { - // Remove the computed column from the selected columns to avoid duplication - options.Columns = append(options.Columns[:colIndex], options.Columns[colIndex+1:]...) - break - } - } + if err := h.hooks.Execute(BeforeRead, hookCtx); err != nil { + statusCode, errCode, errMsg = http.StatusBadRequest, "hook_error", "Hook execution failed" + return err } - } + options = hookCtx.Options - if len(options.ComputedColumns) > 0 { - for _, cu := range options.ComputedColumns { - logger.Debug("Applying computed column: %s", cu.Name) - if strings.Contains(cu.Name, "cql") { - query = query.ColumnExpr(fmt.Sprintf("(%s)::text AS %s", cu.Expression, cu.Name)) - } else { - query = query.ColumnExpr(fmt.Sprintf("(%s) AS %s", cu.Expression, cu.Name)) - } + // Start with Model() using the slice pointer to avoid "Model(nil)" errors in Count() + // Bun's Model() accepts both single pointers and slice pointers + query := tx.NewSelect().Model(modelPtr) - for colIndex := range options.Columns { - if options.Columns[colIndex] == cu.Name { - // Remove the computed column from the selected columns to avoid duplication - options.Columns = append(options.Columns[:colIndex], options.Columns[colIndex+1:]...) - break - } - } - } - } - - // Apply column selection - if len(options.Columns) > 0 { - logger.Debug("Selecting columns: %v", options.Columns) - for _, col := range options.Columns { - query = query.Column(reflection.ExtractSourceColumn(col)) + // Only set Table() if the model doesn't provide a table name via the underlying type + // Create a temporary instance to check for TableNameProvider + tempInstance := reflect.New(modelType).Interface() + if provider, ok := tempInstance.(common.TableNameProvider); !ok || provider.TableName() == "" { + query = query.Table(tableName) } - } - - // Apply expand (Just expand to Preload for now) - for _, expand := range options.Expand { - logger.Debug("Applying expand: %s", expand.Relation) - sorts := make([]common.SortOption, 0) - for _, s := range strings.Split(expand.Sort, ",") { - if s == "" { - continue - } - dir := "ASC" - if strings.HasPrefix(s, "-") || strings.HasSuffix(strings.ToUpper(s), " DESC") { - dir = "DESC" - s = strings.TrimPrefix(s, "-") - s = strings.TrimSuffix(strings.ToLower(s), " desc") - } - sorts = append(sorts, common.SortOption{ - Column: s, Direction: dir, - }) - } - // Note: Expand would require JOIN implementation - // For now, we'll use Preload as a fallback - // query = query.Preload(expand.Relation) - if options.Preload == nil { - options.Preload = make([]common.PreloadOption, 0) - } - skip := false - for idx := range options.Preload { - if options.Preload[idx].Relation == expand.Relation { - skip = true - continue - } - } - if !skip { - options.Preload = append(options.Preload, common.PreloadOption{ - Relation: expand.Relation, - Columns: expand.Columns, - Sort: sorts, - Where: expand.Where, - }) - } - } - - // Apply preloading - logger.Debug("Total preloads to apply: %d", len(options.Preload)) - for idx := range options.Preload { - preload := options.Preload[idx] - logger.Debug("Applying preload [%d]: Relation=%s, Recursive=%v, RelatedKey=%s, Where=%s", - idx, preload.Relation, preload.Recursive, preload.RelatedKey, preload.Where) - - // Validate and fix WHERE clause to ensure it contains the relation prefix - if len(preload.Where) > 0 { - fixedWhere, err := common.ValidateAndFixPreloadWhere(preload.Where, preload.Relation) - if err != nil { - logger.Error("Invalid preload WHERE clause for relation '%s': %v", preload.Relation, err) - h.sendError(w, http.StatusBadRequest, "invalid_preload_where", - fmt.Sprintf("Invalid preload WHERE clause for relation '%s'", preload.Relation), err) - return - } - preload.Where = fixedWhere + // If we have computed columns/expressions but options.Columns is empty, + // populate it with all model columns first since computed columns are additions + if len(options.Columns) == 0 && (len(options.ComputedQL) > 0 || len(options.ComputedColumns) > 0) { + logger.Debug("Populating options.Columns with all model columns since computed columns are additions") + options.Columns = reflection.GetSQLModelColumns(model) } - // Apply the preload with recursive support - query = h.applyPreloadWithRecursion(query, preload, options.Preload, model, 0) - } - - // Apply DISTINCT if requested - if options.Distinct { - logger.Debug("Applying DISTINCT") - // Note: DISTINCT implementation depends on ORM support - // This may need to be handled differently per database adapter - } - - // Apply filters - validate and adjust for column types first - // Group consecutive OR filters together to prevent OR logic from escaping - for i := 0; i < len(options.Filters); { - filter := &options.Filters[i] - - // Validate and adjust filter based on column type - castInfo := h.ValidateAndAdjustFilterForColumnType(filter, model) - - // Default to AND if LogicOperator is not set - logicOp := filter.LogicOperator - if logicOp == "" { - logicOp = "AND" - } - - // Check if this is the start of an OR group - if logicOp == "OR" { - // Collect all consecutive OR filters - orFilters := []*common.FilterOption{filter} - orCastInfo := []ColumnCastInfo{castInfo} - - j := i + 1 - for j < len(options.Filters) { - nextFilter := &options.Filters[j] - nextLogicOp := nextFilter.LogicOperator - if nextLogicOp == "" { - nextLogicOp = "AND" - } - if nextLogicOp == "OR" { - nextCastInfo := h.ValidateAndAdjustFilterForColumnType(nextFilter, model) - orFilters = append(orFilters, nextFilter) - orCastInfo = append(orCastInfo, nextCastInfo) - j++ + // Apply ComputedQL fields if any + if len(options.ComputedQL) > 0 { + for colName, colExpr := range options.ComputedQL { + logger.Debug("Applying computed column: %s", colName) + if strings.Contains(colName, "cql") { + query = query.ColumnExpr(fmt.Sprintf("(%s)::text AS %s", colExpr, colName)) } else { - break + query = query.ColumnExpr(fmt.Sprintf("(%s)AS %s", colExpr, colName)) + } + + for colIndex := range options.Columns { + if options.Columns[colIndex] == colName { + // Remove the computed column from the selected columns to avoid duplication + options.Columns = append(options.Columns[:colIndex], options.Columns[colIndex+1:]...) + break + } } } - - // Apply the OR group as a single grouped condition - logger.Debug("Applying OR filter group with %d conditions", len(orFilters)) - query = h.applyOrFilterGroup(query, orFilters, orCastInfo, tableName) - i = j - } else { - // Single AND filter - apply normally - logger.Debug("Applying filter: %s %s %v (needsCast=%v, logic=%s)", filter.Column, filter.Operator, filter.Value, castInfo.NeedsCast, logicOp) - query = h.applyFilter(query, *filter, tableName, castInfo.NeedsCast, logicOp) - i++ } - } - // Apply custom SQL WHERE clause (AND condition) - if options.CustomSQLWhere != "" { - logger.Debug("Applying custom SQL WHERE: %s", options.CustomSQLWhere) - // First add table prefixes to unqualified columns (but skip columns inside function calls) - prefixedWhere := common.AddTablePrefixToColumns(options.CustomSQLWhere, reflection.ExtractTableNameOnly(tableName)) - // Then sanitize and allow preload table prefixes since custom SQL may reference multiple tables - sanitizedWhere := common.SanitizeWhereClause(prefixedWhere, reflection.ExtractTableNameOnly(tableName), &options.RequestOptions) - // Ensure outer parentheses to prevent OR logic from escaping - sanitizedWhere = common.EnsureOuterParentheses(sanitizedWhere) - if sanitizedWhere != "" { - query = query.Where(sanitizedWhere) - } - } + if len(options.ComputedColumns) > 0 { + for _, cu := range options.ComputedColumns { + logger.Debug("Applying computed column: %s", cu.Name) + if strings.Contains(cu.Name, "cql") { + query = query.ColumnExpr(fmt.Sprintf("(%s)::text AS %s", cu.Expression, cu.Name)) + } else { + query = query.ColumnExpr(fmt.Sprintf("(%s) AS %s", cu.Expression, cu.Name)) + } - // Apply custom SQL WHERE clause (OR condition) - if options.CustomSQLOr != "" { - logger.Debug("Applying custom SQL OR: %s", options.CustomSQLOr) - customOr := common.AddTablePrefixToColumns(options.CustomSQLOr, reflection.ExtractTableNameOnly(tableName)) - // Sanitize and allow preload table prefixes since custom SQL may reference multiple tables - sanitizedOr := common.SanitizeWhereClause(customOr, reflection.ExtractTableNameOnly(tableName), &options.RequestOptions) - // Ensure outer parentheses to prevent OR logic from escaping - sanitizedOr = common.EnsureOuterParentheses(sanitizedOr) - if sanitizedOr != "" { - query = query.WhereOr(sanitizedOr) - } - } - - // Apply custom SQL JOIN clauses, skipping any whose alias is already provided by a - // preload LEFT JOIN (to prevent "table name specified more than once" errors). - if len(options.CustomSQLJoin) > 0 { - preloadAliasSet := make(map[string]bool, len(options.Preload)) - for i := range options.Preload { - if alias := common.RelationPathToBunAlias(options.Preload[i].Relation); alias != "" { - preloadAliasSet[alias] = true + for colIndex := range options.Columns { + if options.Columns[colIndex] == cu.Name { + // Remove the computed column from the selected columns to avoid duplication + options.Columns = append(options.Columns[:colIndex], options.Columns[colIndex+1:]...) + break + } + } } } - for i, joinClause := range options.CustomSQLJoin { - if i < len(options.JoinAliases) && options.JoinAliases[i] != "" { - alias := strings.ToLower(options.JoinAliases[i]) - if preloadAliasSet[alias] { - logger.Debug("Skipping custom SQL JOIN (alias '%s' already joined by preload): %s", alias, joinClause) + // Apply column selection + if len(options.Columns) > 0 { + logger.Debug("Selecting columns: %v", options.Columns) + for _, col := range options.Columns { + query = query.Column(reflection.ExtractSourceColumn(col)) + } + + } + + // Apply expand (Just expand to Preload for now) + for _, expand := range options.Expand { + logger.Debug("Applying expand: %s", expand.Relation) + sorts := make([]common.SortOption, 0) + for _, s := range strings.Split(expand.Sort, ",") { + if s == "" { + continue + } + dir := "ASC" + if strings.HasPrefix(s, "-") || strings.HasSuffix(strings.ToUpper(s), " DESC") { + dir = "DESC" + s = strings.TrimPrefix(s, "-") + s = strings.TrimSuffix(strings.ToLower(s), " desc") + } + sorts = append(sorts, common.SortOption{ + Column: s, Direction: dir, + }) + } + // Note: Expand would require JOIN implementation + // For now, we'll use Preload as a fallback + // query = query.Preload(expand.Relation) + if options.Preload == nil { + options.Preload = make([]common.PreloadOption, 0) + } + skip := false + for idx := range options.Preload { + if options.Preload[idx].Relation == expand.Relation { + skip = true continue } } - logger.Debug("Applying custom SQL JOIN: %s", joinClause) - query = query.Join(joinClause) - } - } - - // Handle FetchRowNumber before applying ID filter - // This must happen before the query to get the row position, then filter by PK - var fetchedRowNumber *int64 - var fetchRowNumberPKValue string - if options.FetchRowNumber != nil && *options.FetchRowNumber != "" { - pkName := reflection.GetPrimaryKeyName(model) - fetchRowNumberPKValue = *options.FetchRowNumber - - logger.Debug("FetchRowNumber: Fetching row number for PK %s = %s", pkName, fetchRowNumberPKValue) - - rowNum, err := h.FetchRowNumber(ctx, tableName, pkName, fetchRowNumberPKValue, options, model) - if err != nil { - logger.Error("Failed to fetch row number: %v", err) - h.sendError(w, http.StatusBadRequest, "fetch_rownumber_error", "Failed to fetch row number", err) - return + if !skip { + options.Preload = append(options.Preload, common.PreloadOption{ + Relation: expand.Relation, + Columns: expand.Columns, + Sort: sorts, + Where: expand.Where, + }) + } } - fetchedRowNumber = &rowNum - logger.Debug("FetchRowNumber: Row number %d for PK %s = %s", rowNum, pkName, fetchRowNumberPKValue) + // Apply preloading + logger.Debug("Total preloads to apply: %d", len(options.Preload)) + for idx := range options.Preload { + preload := options.Preload[idx] + logger.Debug("Applying preload [%d]: Relation=%s, Recursive=%v, RelatedKey=%s, Where=%s", + idx, preload.Relation, preload.Recursive, preload.RelatedKey, preload.Where) - // Now filter the main query to this specific primary key - tableAlias := reflection.ExtractTableNameOnly(tableName) - query = query.Where(fmt.Sprintf("%s.%s = ?", common.QuoteIdent(tableAlias), common.QuoteIdent(pkName)), fetchRowNumberPKValue) - } else if id != "" { - // If ID is provided (and not FetchRowNumber), filter by ID - pkName := reflection.GetPrimaryKeyName(model) - logger.Debug("Filtering by ID=%s: %s", pkName, id) - - tableAlias := reflection.ExtractTableNameOnly(tableName) - query = query.Where(fmt.Sprintf("%s.%s = ?", common.QuoteIdent(tableAlias), common.QuoteIdent(pkName)), id) - } - - // Apply sorting - tableAlias := reflection.ExtractTableNameOnly(tableName) - for _, sort := range options.Sort { - direction := "ASC" - if strings.EqualFold(sort.Direction, "desc") { - direction = "DESC" - } - logger.Debug("Applying sort: %s %s", sort.Column, direction) - - // Check if it's an expression (enclosed in brackets) - use directly without quoting - if strings.HasPrefix(sort.Column, "(") && strings.HasSuffix(sort.Column, ")") { - // For expressions, pass as raw SQL to prevent auto-quoting - query = query.OrderExpr(fmt.Sprintf("%s %s", sort.Column, direction)) - } else if strings.Contains(sort.Column, ".") { - // Already qualified (e.g. alias.column) - pass as raw expression to preserve the dot - query = query.OrderExpr(fmt.Sprintf("%s %s", sort.Column, direction)) - } else { - // Unqualified column - prefix with main table alias to avoid ambiguity on JOINs - query = query.OrderExpr(fmt.Sprintf("%s.%s %s", common.QuoteIdent(tableAlias), common.QuoteIdent(sort.Column), direction)) - } - } - - // Get total count before pagination (unless skip count is requested) - var total int - if !options.SkipCount { - // Try to get from cache first (unless SkipCache is true) - var cachedTotalData *cachedTotal - var cacheKey string - - if !options.SkipCache { - // Build cache key from query parameters - // Convert expand options to interface slice for the cache key builder - expandOpts := make([]interface{}, len(options.Expand)) - for i, exp := range options.Expand { - expandOpts[i] = map[string]interface{}{ - "relation": exp.Relation, - "where": exp.Where, + // Validate and fix WHERE clause to ensure it contains the relation prefix + if len(preload.Where) > 0 { + fixedWhere, err := common.ValidateAndFixPreloadWhere(preload.Where, preload.Relation) + if err != nil { + logger.Error("Invalid preload WHERE clause for relation '%s': %v", preload.Relation, err) + statusCode, errCode, errMsg = http.StatusBadRequest, "invalid_preload_where", + fmt.Sprintf("Invalid preload WHERE clause for relation '%s'", preload.Relation) + return err } + preload.Where = fixedWhere } - cacheKeyHash := buildExtendedQueryCacheKey( - tableName, - options.Filters, - options.Sort, - options.CustomSQLWhere, - options.CustomSQLOr, - options.CustomSQLJoin, - expandOpts, - options.Distinct, - options.CursorForward, - options.CursorBackward, - ) - cacheKey = getQueryTotalCacheKey(cacheKeyHash) + // Apply the preload with recursive support + query = h.applyPreloadWithRecursion(query, preload, options.Preload, model, 0) + } - // Try to retrieve from cache - cachedTotalData = &cachedTotal{} - err := cache.GetDefaultCache().Get(ctx, cacheKey, cachedTotalData) - if err == nil { - total = cachedTotalData.Total - logger.Debug("Total records (from cache): %d", total) + // Apply DISTINCT if requested + if options.Distinct { + logger.Debug("Applying DISTINCT") + // Note: DISTINCT implementation depends on ORM support + // This may need to be handled differently per database adapter + } + + // Apply filters - validate and adjust for column types first + // Group consecutive OR filters together to prevent OR logic from escaping + for i := 0; i < len(options.Filters); { + filter := &options.Filters[i] + + // Validate and adjust filter based on column type + castInfo := h.ValidateAndAdjustFilterForColumnType(filter, model) + + // Default to AND if LogicOperator is not set + logicOp := filter.LogicOperator + if logicOp == "" { + logicOp = "AND" + } + + // Check if this is the start of an OR group + if logicOp == "OR" { + // Collect all consecutive OR filters + orFilters := []*common.FilterOption{filter} + orCastInfo := []ColumnCastInfo{castInfo} + + j := i + 1 + for j < len(options.Filters) { + nextFilter := &options.Filters[j] + nextLogicOp := nextFilter.LogicOperator + if nextLogicOp == "" { + nextLogicOp = "AND" + } + if nextLogicOp == "OR" { + nextCastInfo := h.ValidateAndAdjustFilterForColumnType(nextFilter, model) + orFilters = append(orFilters, nextFilter) + orCastInfo = append(orCastInfo, nextCastInfo) + j++ + } else { + break + } + } + + // Apply the OR group as a single grouped condition + logger.Debug("Applying OR filter group with %d conditions", len(orFilters)) + query = h.applyOrFilterGroup(query, orFilters, orCastInfo, tableName) + i = j } else { - logger.Debug("Cache miss for query total") - cachedTotalData = nil + // Single AND filter - apply normally + logger.Debug("Applying filter: %s %s %v (needsCast=%v, logic=%s)", filter.Column, filter.Operator, filter.Value, castInfo.NeedsCast, logicOp) + query = h.applyFilter(query, *filter, tableName, castInfo.NeedsCast, logicOp) + i++ } } - // If not in cache or cache skip, execute count query - if cachedTotalData == nil { - count, err := query.Count(ctx) - if err != nil { - logger.Error("Error counting records: %v", err) - h.sendError(w, http.StatusInternalServerError, "query_error", "Error counting records", err) - return + // Apply custom SQL WHERE clause (AND condition) + if options.CustomSQLWhere != "" { + logger.Debug("Applying custom SQL WHERE: %s", options.CustomSQLWhere) + // First add table prefixes to unqualified columns (but skip columns inside function calls) + prefixedWhere := common.AddTablePrefixToColumns(options.CustomSQLWhere, reflection.ExtractTableNameOnly(tableName)) + // Then sanitize and allow preload table prefixes since custom SQL may reference multiple tables + sanitizedWhere := common.SanitizeWhereClause(prefixedWhere, reflection.ExtractTableNameOnly(tableName), &options.RequestOptions) + // Ensure outer parentheses to prevent OR logic from escaping + sanitizedWhere = common.EnsureOuterParentheses(sanitizedWhere) + if sanitizedWhere != "" { + query = query.Where(sanitizedWhere) } - total = count - logger.Debug("Total records (from query): %d", total) + } - // Store in cache with schema and table tags (if caching is enabled) - if !options.SkipCache && cacheKey != "" { - cacheTTL := time.Minute * 2 // Default 2 minutes TTL - if err := setQueryTotalCache(ctx, cacheKey, total, schema, tableName, cacheTTL); err != nil { - logger.Warn("Failed to cache query total: %v", err) - // Don't fail the request if caching fails + // Apply custom SQL WHERE clause (OR condition) + if options.CustomSQLOr != "" { + logger.Debug("Applying custom SQL OR: %s", options.CustomSQLOr) + customOr := common.AddTablePrefixToColumns(options.CustomSQLOr, reflection.ExtractTableNameOnly(tableName)) + // Sanitize and allow preload table prefixes since custom SQL may reference multiple tables + sanitizedOr := common.SanitizeWhereClause(customOr, reflection.ExtractTableNameOnly(tableName), &options.RequestOptions) + // Ensure outer parentheses to prevent OR logic from escaping + sanitizedOr = common.EnsureOuterParentheses(sanitizedOr) + if sanitizedOr != "" { + query = query.WhereOr(sanitizedOr) + } + } + + // Apply custom SQL JOIN clauses, skipping any whose alias is already provided by a + // preload LEFT JOIN (to prevent "table name specified more than once" errors). + if len(options.CustomSQLJoin) > 0 { + preloadAliasSet := make(map[string]bool, len(options.Preload)) + for i := range options.Preload { + if alias := common.RelationPathToBunAlias(options.Preload[i].Relation); alias != "" { + preloadAliasSet[alias] = true + } + } + + for i, joinClause := range options.CustomSQLJoin { + if i < len(options.JoinAliases) && options.JoinAliases[i] != "" { + alias := strings.ToLower(options.JoinAliases[i]) + if preloadAliasSet[alias] { + logger.Debug("Skipping custom SQL JOIN (alias '%s' already joined by preload): %s", alias, joinClause) + continue + } + } + logger.Debug("Applying custom SQL JOIN: %s", joinClause) + query = query.Join(joinClause) + } + } + + // Handle FetchRowNumber before applying ID filter + // This must happen before the query to get the row position, then filter by PK + var fetchRowNumberPKValue string + if options.FetchRowNumber != nil && *options.FetchRowNumber != "" { + pkName := reflection.GetPrimaryKeyName(model) + fetchRowNumberPKValue = *options.FetchRowNumber + + logger.Debug("FetchRowNumber: Fetching row number for PK %s = %s", pkName, fetchRowNumberPKValue) + + rowNum, err := h.FetchRowNumber(ctx, tx, tableName, pkName, fetchRowNumberPKValue, options, model) + if err != nil { + logger.Error("Failed to fetch row number: %v", err) + statusCode, errCode, errMsg = http.StatusBadRequest, "fetch_rownumber_error", "Failed to fetch row number" + return err + } + + fetchedRowNumber = &rowNum + logger.Debug("FetchRowNumber: Row number %d for PK %s = %s", rowNum, pkName, fetchRowNumberPKValue) + + // Now filter the main query to this specific primary key + tableAlias := reflection.ExtractTableNameOnly(tableName) + query = query.Where(fmt.Sprintf("%s.%s = ?", common.QuoteIdent(tableAlias), common.QuoteIdent(pkName)), fetchRowNumberPKValue) + } else if id != "" { + // If ID is provided (and not FetchRowNumber), filter by ID + pkName := reflection.GetPrimaryKeyName(model) + logger.Debug("Filtering by ID=%s: %s", pkName, id) + + tableAlias := reflection.ExtractTableNameOnly(tableName) + query = query.Where(fmt.Sprintf("%s.%s = ?", common.QuoteIdent(tableAlias), common.QuoteIdent(pkName)), id) + } + + // Apply sorting + tableAlias := reflection.ExtractTableNameOnly(tableName) + for _, sort := range options.Sort { + direction := "ASC" + if strings.EqualFold(sort.Direction, "desc") { + direction = "DESC" + } + logger.Debug("Applying sort: %s %s", sort.Column, direction) + + // Check if it's an expression (enclosed in brackets) - use directly without quoting + if strings.HasPrefix(sort.Column, "(") && strings.HasSuffix(sort.Column, ")") { + // For expressions, pass as raw SQL to prevent auto-quoting + query = query.OrderExpr(fmt.Sprintf("%s %s", sort.Column, direction)) + } else if strings.Contains(sort.Column, ".") { + // Already qualified (e.g. alias.column) - pass as raw expression to preserve the dot + query = query.OrderExpr(fmt.Sprintf("%s %s", sort.Column, direction)) + } else { + // Unqualified column - prefix with main table alias to avoid ambiguity on JOINs + query = query.OrderExpr(fmt.Sprintf("%s.%s %s", common.QuoteIdent(tableAlias), common.QuoteIdent(sort.Column), direction)) + } + } + + // Get total count before pagination (unless skip count is requested) + if !options.SkipCount { + // Try to get from cache first (unless SkipCache is true) + var cachedTotalData *cachedTotal + var cacheKey string + + if !options.SkipCache { + // Build cache key from query parameters + // Convert expand options to interface slice for the cache key builder + expandOpts := make([]interface{}, len(options.Expand)) + for i, exp := range options.Expand { + expandOpts[i] = map[string]interface{}{ + "relation": exp.Relation, + "where": exp.Where, + } + } + + cacheKeyHash := buildExtendedQueryCacheKey( + tableName, + options.Filters, + options.Sort, + options.CustomSQLWhere, + options.CustomSQLOr, + options.CustomSQLJoin, + expandOpts, + options.Distinct, + options.CursorForward, + options.CursorBackward, + ) + cacheKey = getQueryTotalCacheKey(cacheKeyHash) + + // Try to retrieve from cache + cachedTotalData = &cachedTotal{} + err := cache.GetDefaultCache().Get(ctx, cacheKey, cachedTotalData) + if err == nil { + total = cachedTotalData.Total + logger.Debug("Total records (from cache): %d", total) } else { - logger.Debug("Cached query total with key: %s", cacheKey) + logger.Debug("Cache miss for query total") + cachedTotalData = nil + } + } + + // If not in cache or cache skip, execute count query + if cachedTotalData == nil { + count, err := query.Count(ctx) + if err != nil { + logger.Error("Error counting records: %v", err) + statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error counting records" + return err + } + total = count + logger.Debug("Total records (from query): %d", total) + + // Store in cache with schema and table tags (if caching is enabled) + if !options.SkipCache && cacheKey != "" { + cacheTTL := time.Minute * 2 // Default 2 minutes TTL + if err := setQueryTotalCache(ctx, cacheKey, total, schema, tableName, cacheTTL); err != nil { + logger.Warn("Failed to cache query total: %v", err) + // Don't fail the request if caching fails + } else { + logger.Debug("Cached query total with key: %s", cacheKey) + } + } + } + } else { + logger.Debug("Skipping count as requested") + total = -1 // Indicate count was skipped + } + + // Apply pagination + if options.Limit != nil && *options.Limit > 0 { + logger.Debug("Applying limit: %d", *options.Limit) + query = query.Limit(*options.Limit) + } + if options.Offset != nil && *options.Offset > 0 { + logger.Debug("Applying offset: %d", *options.Offset) + query = query.Offset(*options.Offset) + } + + // Apply cursor-based pagination + if len(options.CursorForward) > 0 || len(options.CursorBackward) > 0 { + logger.Debug("Applying cursor pagination") + + // Get primary key name + pkName := reflection.GetPrimaryKeyName(model) + + // Extract model columns for validation using the generic database function + modelColumns := reflection.GetModelColumns(model) + + // Build expand joins map: custom SQL joins are available in cursor subquery + expandJoins := make(map[string]string) + for _, joinClause := range options.CustomSQLJoin { + alias := extractJoinAlias(joinClause) + if alias != "" { + expandJoins[alias] = joinClause + } + } + // TODO: also add Expand relation JOINs when those are built as SQL rather than Preload + + // Default sort to primary key when none provided + if len(options.Sort) == 0 { + options.Sort = []common.SortOption{{Column: pkName, Direction: "ASC"}} + } + + // Get cursor filter SQL + cursorFilter, err := options.GetCursorFilter(tableName, pkName, modelColumns, expandJoins) + if err != nil { + logger.Error("Error building cursor filter: %v", err) + statusCode, errCode, errMsg = http.StatusBadRequest, "cursor_error", "Invalid cursor pagination" + return err + } + + // Apply cursor filter to query + if cursorFilter != "" { + logger.Debug("Applying cursor filter: %s", cursorFilter) + sanitizedCursor := common.SanitizeWhereClause(cursorFilter, reflection.ExtractTableNameOnly(tableName), &options.RequestOptions) + if sanitizedCursor != "" { + query = query.Where(sanitizedCursor) } } } - } else { - logger.Debug("Skipping count as requested") - total = -1 // Indicate count was skipped - } - // Apply pagination - if options.Limit != nil && *options.Limit > 0 { - logger.Debug("Applying limit: %d", *options.Limit) - query = query.Limit(*options.Limit) - } - if options.Offset != nil && *options.Offset > 0 { - logger.Debug("Applying offset: %d", *options.Offset) - query = query.Offset(*options.Offset) - } - - // Apply cursor-based pagination - if len(options.CursorForward) > 0 || len(options.CursorBackward) > 0 { - logger.Debug("Applying cursor pagination") - - // Get primary key name - pkName := reflection.GetPrimaryKeyName(model) - - // Extract model columns for validation using the generic database function - modelColumns := reflection.GetModelColumns(model) - - // Build expand joins map: custom SQL joins are available in cursor subquery - expandJoins := make(map[string]string) - for _, joinClause := range options.CustomSQLJoin { - alias := extractJoinAlias(joinClause) - if alias != "" { - expandJoins[alias] = joinClause - } - } - // TODO: also add Expand relation JOINs when those are built as SQL rather than Preload - - // Default sort to primary key when none provided - if len(options.Sort) == 0 { - options.Sort = []common.SortOption{{Column: pkName, Direction: "ASC"}} + // Execute BeforeScan hooks - pass query chain so hooks can modify it + hookCtx.Query = query + if err := h.hooks.Execute(BeforeScan, hookCtx); err != nil { + logger.Error("BeforeScan hook failed: %v", err) + statusCode, errCode, errMsg = http.StatusBadRequest, "hook_error", "Hook execution failed" + return err } - // Get cursor filter SQL - cursorFilter, err := options.GetCursorFilter(tableName, pkName, modelColumns, expandJoins) - if err != nil { - logger.Error("Error building cursor filter: %v", err) - h.sendError(w, http.StatusBadRequest, "cursor_error", "Invalid cursor pagination", err) - return + // Use potentially modified query from hook context + if modifiedQuery, ok := hookCtx.Query.(common.SelectQuery); ok { + query = modifiedQuery } - // Apply cursor filter to query - if cursorFilter != "" { - logger.Debug("Applying cursor filter: %s", cursorFilter) - sanitizedCursor := common.SanitizeWhereClause(cursorFilter, reflection.ExtractTableNameOnly(tableName), &options.RequestOptions) - if sanitizedCursor != "" { - query = query.Where(sanitizedCursor) - } + // Execute query - modelPtr was already created earlier + if err := query.ScanModel(ctx); err != nil { + logger.Error("Error executing query: %v", err) + statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error executing query" + return err } - } - // Execute BeforeScan hooks - pass query chain so hooks can modify it - hookCtx.Query = query - if err := h.hooks.Execute(BeforeScan, hookCtx); err != nil { - logger.Error("BeforeScan hook failed: %v", err) - h.sendError(w, http.StatusBadRequest, "hook_error", "Hook execution failed", err) - return - } + return nil + }) - // Use potentially modified query from hook context - if modifiedQuery, ok := hookCtx.Query.(common.SelectQuery); ok { - query = modifiedQuery - } - - // Execute query - modelPtr was already created earlier - if err := query.ScanModel(ctx); err != nil { - logger.Error("Error executing query: %v", err) - h.sendError(w, http.StatusInternalServerError, "query_error", "Error executing query", err) + if txErr != nil { + if statusCode == 0 { + statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error executing query" + } + h.sendError(w, statusCode, errCode, errMsg, txErr) return } @@ -839,7 +862,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st logger.Debug("FetchRowNumber: Row number %d set in metadata", *fetchedRowNumber) } - // Execute AfterRead hooks + // Execute AfterRead hooks (runs after the transaction commits, against the pooled db) + hookCtx.Tx = h.db hookCtx.Result = modelPtr hookCtx.Error = nil @@ -1133,7 +1157,6 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat logger.Info("Creating record in %s.%s", schema, entity) - // Execute BeforeCreate hooks hookCtx := &HookContext{ Context: ctx, Handler: h, @@ -1144,28 +1167,38 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat Options: options, Data: data, Writer: w, - Tx: h.db, } - if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { - logger.Error("BeforeCreate hook failed: %v", err) - h.sendError(w, http.StatusBadRequest, "hook_error", "Hook execution failed", err) - return - } - - // Use potentially modified data from hook context - data = hookCtx.Data - - // Normalize data to slice for unified processing - dataSlice := h.normalizeToSlice(data) - logger.Debug("Processing %d item(s) for creation", len(dataSlice)) - - // Store original data maps for merging later - originalDataMaps := make([]map[string]interface{}, 0, len(dataSlice)) + // Everything below (including the BeforeCreate hook) runs inside a single + // transaction so that session-scoped RLS GUCs set by the hook execute on the + // same physical connection as the inserts they are meant to protect. + var ( + dataSlice []interface{} + originalDataMaps []map[string]interface{} + statusCode int + errCode string + errMsg string + ) // Process all items in a transaction - results := make([]interface{}, 0, len(dataSlice)) - err := h.db.RunInTransaction(ctx, func(tx common.Database) error { + results := make([]interface{}, 0) + txErr := h.db.RunInTransaction(ctx, func(tx common.Database) error { + hookCtx.Tx = tx + if err := h.hooks.Execute(BeforeCreate, hookCtx); err != nil { + statusCode, errCode, errMsg = http.StatusBadRequest, "hook_error", "Hook execution failed" + return err + } + + // Use potentially modified data from hook context + data = hookCtx.Data + + // Normalize data to slice for unified processing + dataSlice = h.normalizeToSlice(data) + logger.Debug("Processing %d item(s) for creation", len(dataSlice)) + + // Store original data maps for merging later + originalDataMaps = make([]map[string]interface{}, 0, len(dataSlice)) + // Create temporary nested processor with transaction txNestedProcessor := common.NewNestedCUDProcessor(tx, h.registry, h) @@ -1266,9 +1299,12 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat return nil }) - if err != nil { - logger.Error("Error creating records: %v", err) - h.sendError(w, http.StatusInternalServerError, "create_error", "Error creating records", err) + if txErr != nil { + if statusCode == 0 { + statusCode, errCode, errMsg = http.StatusInternalServerError, "create_error", "Error creating records" + } + logger.Error("Error creating records: %v", txErr) + h.sendError(w, statusCode, errCode, errMsg, txErr) return } @@ -1284,7 +1320,10 @@ func (h *Handler) handleCreate(ctx context.Context, w common.ResponseWriter, dat } } - // Execute AfterCreate hooks + // Execute AfterCreate hooks (runs after the transaction commits, against the + // pooled db — hookCtx.Tx was pointed at the now-closed transaction inside the + // RunInTransaction closure above and must not be reused here). + hookCtx.Tx = h.db var responseData interface{} if len(mergedResults) == 1 { responseData = mergedResults[0] @@ -2828,7 +2867,7 @@ func (h *Handler) sendError(w common.ResponseWriter, statusCode int, code, messa // FetchRowNumber calculates the row number of a specific record based on sorting and filtering // Returns the 1-based row number of the record with the given primary key value -func (h *Handler) FetchRowNumber(ctx context.Context, tableName string, pkName string, pkValue string, options ExtendedRequestOptions, model any) (int64, error) { +func (h *Handler) FetchRowNumber(ctx context.Context, db common.Database, tableName string, pkName string, pkValue string, options ExtendedRequestOptions, model any) (int64, error) { defer func() { if r := recover(); r != nil { logger.Error("Panic during FetchRowNumber: %v", r) @@ -2912,7 +2951,7 @@ func (h *Handler) FetchRowNumber(ctx context.Context, tableName string, pkName s RN int64 `bun:"rn"` } logger.Debug("[FetchRowNumber] BEFORE Query call - about to execute raw query") - err := h.db.Query(ctx, &result, queryStr, pkValue) + err := db.Query(ctx, &result, queryStr, pkValue) logger.Debug("[FetchRowNumber] AFTER Query call - query completed with %d results, err: %v", len(result), err) if err != nil { return 0, fmt.Errorf("failed to fetch row number: %w", err)