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

@@ -259,9 +259,44 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 	sliceType := reflect.SliceOf(reflect.PointerTo(modelType))
 	modelPtr := reflect.New(sliceType).Interface()
 
+	// 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
+	)
+
+	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 err := h.hooks.Execute(BeforeRead, hookCtx); err != nil {
+			statusCode, errCode, errMsg = http.StatusInternalServerError, "hook_error", "BeforeRead hook failed"
+			return err
+		}
+		options = hookCtx.Options
+
 		// 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)
+		query := tx.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
@@ -296,8 +331,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			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
+				statusCode, errCode, errMsg = http.StatusBadRequest, "invalid_preload", "Failed to apply preloads"
+				return err
 			}
 		}
 
@@ -339,8 +374,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			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
+				statusCode, errCode, errMsg = http.StatusBadRequest, "cursor_error", "Invalid cursor pagination"
+				return err
 			}
 
 			// Apply cursor filter to query
@@ -355,9 +390,6 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			}
 		}
 
-	// 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
@@ -384,8 +416,7 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 
 		// Try to retrieve from cache
 		var cachedTotal cachedTotal
-	err := cache.GetDefaultCache().Get(ctx, cacheKey, &cachedTotal)
-	if err == nil {
+		if err := cache.GetDefaultCache().Get(ctx, cacheKey, &cachedTotal); err == nil {
 			total = cachedTotal.Total
 			logger.Debug("Total records (from cache): %d", total)
 		} else {
@@ -394,8 +425,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			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
+				statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error counting records"
+				return err
 			}
 			total = count
 			logger.Debug("Total records (from query): %d", total)
@@ -411,7 +442,6 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 		}
 
 		// 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)
@@ -439,7 +469,7 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 				RowNum int64 `bun:"row_num"`
 			}
 
-		rowNumQuery := h.db.NewSelect().Table(tableName).
+			rowNumQuery := tx.NewSelect().Table(tableName).
 				ColumnExpr(fmt.Sprintf("%s AS row_num", rowNumberSQL)).
 				Column(pkName)
 
@@ -457,8 +487,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			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 {
+			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))
@@ -474,7 +504,7 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 					logger.Warn("Error fetching row number: %v", err)
 				}
 			} else {
-			rowNumber = &result.RowNum
+				rowNumber = &rnResult.RowNum
 				logger.Debug("Found row number: %d", *rowNumber)
 			}
 		}
@@ -492,7 +522,6 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 		}
 
 		// Execute query
-	var result interface{}
 		if id != "" || (options.FetchRowNumber != nil && *options.FetchRowNumber != "") {
 			// Single record query - either by URL ID or FetchRowNumber
 			var targetID string
@@ -511,8 +540,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			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
+				statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error executing query"
+				return err
 			}
 			result = singleResult
 		} else {
@@ -520,17 +549,25 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			// 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
+				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
+	}
 
 	// 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)
+		var responseData interface{} = v
+		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
+			}
+			responseData = v
+
+			query := tx.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
+					return err
 				}
 				logger.Info("Successfully created record, rows affected: %d", result.RowsAffected())
-		} else {
+				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:

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,9 +344,43 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 
 	logger.Info("Reading records from %s.%s", schema, entity)
 
+	hookCtx := &HookContext{
+		Context:   ctx,
+		Handler:   h,
+		Schema:    schema,
+		Entity:    entity,
+		TableName: tableName,
+		Model:     model,
+		Options:   options,
+		ID:        id,
+		Writer:    w,
+	}
+
+	// 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
+	)
+
+	txErr := h.db.RunInTransaction(ctx, func(tx common.Database) error {
+		hookCtx.Tx = tx
+
+		if err := h.hooks.Execute(BeforeRead, hookCtx); err != nil {
+			statusCode, errCode, errMsg = http.StatusBadRequest, "hook_error", "Hook execution failed"
+			return err
+		}
+		options = hookCtx.Options
+
 		// 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)
+		query := tx.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
@@ -483,9 +497,9 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 				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
+					statusCode, errCode, errMsg = http.StatusBadRequest, "invalid_preload_where",
+						fmt.Sprintf("Invalid preload WHERE clause for relation '%s'", preload.Relation)
+					return err
 				}
 				preload.Where = fixedWhere
 			}
@@ -602,7 +616,6 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 
 		// 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)
@@ -610,11 +623,11 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 
 			logger.Debug("FetchRowNumber: Fetching row number for PK %s = %s", pkName, fetchRowNumberPKValue)
 
-		rowNum, err := h.FetchRowNumber(ctx, tableName, pkName, fetchRowNumberPKValue, options, model)
+			rowNum, err := h.FetchRowNumber(ctx, tx, 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
+				statusCode, errCode, errMsg = http.StatusBadRequest, "fetch_rownumber_error", "Failed to fetch row number"
+				return err
 			}
 
 			fetchedRowNumber = &rowNum
@@ -655,7 +668,6 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 		}
 
 		// 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
@@ -703,8 +715,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 				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
+					statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error counting records"
+					return err
 				}
 				total = count
 				logger.Debug("Total records (from query): %d", total)
@@ -764,8 +776,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 			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
+				statusCode, errCode, errMsg = http.StatusBadRequest, "cursor_error", "Invalid cursor pagination"
+				return err
 			}
 
 			// Apply cursor filter to query
@@ -782,8 +794,8 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 		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
+			statusCode, errCode, errMsg = http.StatusBadRequest, "hook_error", "Hook execution failed"
+			return err
 		}
 
 		// Use potentially modified query from hook context
@@ -794,7 +806,18 @@ func (h *Handler) handleRead(ctx context.Context, w common.ResponseWriter, id st
 		// 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)
+			statusCode, errCode, errMsg = http.StatusInternalServerError, "query_error", "Error executing query"
+			return err
+		}
+
+		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
 	}
 
@@ -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,
 	}
 
+	// 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)
+	txErr := h.db.RunInTransaction(ctx, func(tx common.Database) error {
+		hookCtx.Tx = tx
 		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
+			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)
+		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))
+		originalDataMaps = make([]map[string]interface{}, 0, len(dataSlice))
 
-	// Process all items in a transaction
-	results := make([]interface{}, 0, len(dataSlice))
-	err := h.db.RunInTransaction(ctx, func(tx common.Database) error {
 		// 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)