feat(09-05): make bulk delete atomic, ordered, and retry-safe
- Reject an empty selection and dedupe ids before locking rows in pk order - Return deleted 0 when every requested row is already gone, without hooks - Roll back mixed, hook, and cancelled batches so no partial delete commits
This commit is contained in:
146
cabana/crud.go
146
cabana/crud.go
@@ -7,6 +7,7 @@ import (
|
|||||||
"math"
|
"math"
|
||||||
"net/http"
|
"net/http"
|
||||||
"reflect"
|
"reflect"
|
||||||
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
@@ -169,9 +170,58 @@ func (s CRUDService) Delete(ctx context.Context, cc *CompiledController, id any)
|
|||||||
}
|
}
|
||||||
|
|
||||||
// BulkDelete removes a normalized id set in one transaction.
|
// BulkDelete removes a normalized id set in one transaction.
|
||||||
// The deterministic lifecycle is added with the bulk route.
|
// An empty selection is validation_failed. A wholly absent selection is a
|
||||||
|
// successful no-op. A mixed present/absent selection conflicts and rolls back.
|
||||||
func (s CRUDService) BulkDelete(ctx context.Context, cc *CompiledController, in BulkDeleteInput) (BulkResult, error) {
|
func (s CRUDService) BulkDelete(ctx context.Context, cc *CompiledController, in BulkDeleteInput) (BulkResult, error) {
|
||||||
return BulkResult{}, errors.New("cabana: bulk delete is not implemented")
|
if s.DB == nil {
|
||||||
|
return BulkResult{}, errors.New("cabana: database is not configured")
|
||||||
|
}
|
||||||
|
if ctx == nil {
|
||||||
|
ctx = context.Background()
|
||||||
|
}
|
||||||
|
ids, err := normalizeIDs(in.IDs)
|
||||||
|
if err != nil {
|
||||||
|
return BulkResult{}, err
|
||||||
|
}
|
||||||
|
if _, err := newWritableModel(cc); err != nil {
|
||||||
|
return BulkResult{}, err
|
||||||
|
}
|
||||||
|
var result BulkResult
|
||||||
|
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||||
|
tx = tx.WithContext(ctx)
|
||||||
|
if err := ctx.Err(); err != nil {
|
||||||
|
return lifecycleFailure(cc, err)
|
||||||
|
}
|
||||||
|
proto, err := newWritableModel(cc)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
rows, err := lockScoped(ctx, tx, cc, proto, ids)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if len(rows) == 0 {
|
||||||
|
result.Deleted = 0
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if len(rows) != len(ids) {
|
||||||
|
return partialSelection{}
|
||||||
|
}
|
||||||
|
for _, row := range rows {
|
||||||
|
if err := ctx.Err(); err != nil {
|
||||||
|
return lifecycleFailure(cc, err)
|
||||||
|
}
|
||||||
|
if err := deleteRecord(ctx, tx, cc, row); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
result.Deleted = len(rows)
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return BulkResult{}, err
|
||||||
|
}
|
||||||
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Show loads one scoped record. Missing and out-of-scope ids are identical.
|
// Show loads one scoped record. Missing and out-of-scope ids are identical.
|
||||||
@@ -292,6 +342,11 @@ func writeCRUDError(w http.ResponseWriter, err error) {
|
|||||||
WriteError(w, http.StatusNotFound, "not_found", msgNotFound)
|
WriteError(w, http.StatusNotFound, "not_found", msgNotFound)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
var partial partialSelection
|
||||||
|
if errors.As(err, &partial) {
|
||||||
|
WriteError(w, http.StatusConflict, "conflict", "Conflict")
|
||||||
|
return
|
||||||
|
}
|
||||||
WriteError(w, http.StatusInternalServerError, "error", msgServerError)
|
WriteError(w, http.StatusInternalServerError, "error", msgServerError)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -373,9 +428,96 @@ func lifecycleFailure(cc *CompiledController, err error) error {
|
|||||||
if errors.As(err, &closed) {
|
if errors.As(err, &closed) {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
var partial partialSelection
|
||||||
|
if errors.As(err, &partial) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
var life *lifecycleError
|
||||||
|
if errors.As(err, &life) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
return &lifecycleError{ControllerID: controllerID(cc)}
|
return &lifecycleError{ControllerID: controllerID(cc)}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type partialSelection struct{}
|
||||||
|
|
||||||
|
func (partialSelection) Error() string { return "cabana: partial selection" }
|
||||||
|
|
||||||
|
func normalizeIDs(raw []any) ([]uint, error) {
|
||||||
|
if len(raw) == 0 {
|
||||||
|
return nil, &ValidationError{Details: map[string]any{"ids": []string{"The ids field is required."}}}
|
||||||
|
}
|
||||||
|
seen := map[uint]struct{}{}
|
||||||
|
ids := make([]uint, 0, len(raw))
|
||||||
|
for _, item := range raw {
|
||||||
|
n, err := asUint(item)
|
||||||
|
if err != nil {
|
||||||
|
return nil, &ValidationError{Details: map[string]any{"ids": []string{"The ids field must be a list of integers."}}}
|
||||||
|
}
|
||||||
|
if _, ok := seen[n]; ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seen[n] = struct{}{}
|
||||||
|
ids = append(ids, n)
|
||||||
|
}
|
||||||
|
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
||||||
|
return ids, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func lockScoped(ctx context.Context, tx *gorm.DB, cc *CompiledController, proto any, ids []uint) ([]any, error) {
|
||||||
|
mt := reflect.TypeOf(proto)
|
||||||
|
if mt == nil || mt.Kind() != reflect.Pointer || mt.Elem().Kind() != reflect.Struct {
|
||||||
|
return nil, &CapabilityError{ControllerID: controllerID(cc)}
|
||||||
|
}
|
||||||
|
holder := reflect.New(reflect.SliceOf(mt.Elem()))
|
||||||
|
q := tx.WithContext(ctx).Model(proto)
|
||||||
|
if cc != nil {
|
||||||
|
if ext, ok := cc.Controller.(pact.ListExtendQuery); ok && ext != nil {
|
||||||
|
if next := ext.ListExtendQuery(ctx, q); next != nil {
|
||||||
|
q = next
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
col := primaryColumn(proto)
|
||||||
|
vals := make([]any, len(ids))
|
||||||
|
for i, id := range ids {
|
||||||
|
vals[i] = castPK(proto, id)
|
||||||
|
}
|
||||||
|
err := q.Clauses(clause.Locking{Strength: "UPDATE"}).
|
||||||
|
Where(clause.IN{Column: clause.Column{Name: col}, Values: vals}).
|
||||||
|
Order(clause.OrderByColumn{Column: clause.Column{Name: col}, Desc: false}).
|
||||||
|
Find(holder.Interface()).Error
|
||||||
|
if err != nil {
|
||||||
|
return nil, lifecycleFailure(cc, err)
|
||||||
|
}
|
||||||
|
values := holder.Elem()
|
||||||
|
out := make([]any, values.Len())
|
||||||
|
for i := 0; i < values.Len(); i++ {
|
||||||
|
out[i] = values.Index(i).Addr().Interface()
|
||||||
|
}
|
||||||
|
sort.SliceStable(out, func(i, j int) bool { return pkUint(out[i]) < pkUint(out[j]) })
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func pkUint(model any) uint {
|
||||||
|
v := reflect.ValueOf(model)
|
||||||
|
for v.Kind() == reflect.Pointer {
|
||||||
|
if v.IsNil() {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
v = v.Elem()
|
||||||
|
}
|
||||||
|
field := fieldByColumn(v, primaryColumn(model))
|
||||||
|
if !field.IsValid() || !field.CanInterface() {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
n, err := asUint(field.Interface())
|
||||||
|
if err != nil {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
|
||||||
func formBeforeCreate(ctx context.Context, cc *CompiledController, model any) error {
|
func formBeforeCreate(ctx context.Context, cc *CompiledController, model any) error {
|
||||||
hook, ok := cc.Controller.(pact.FormBeforeCreate)
|
hook, ok := cc.Controller.(pact.FormBeforeCreate)
|
||||||
if !ok || hook == nil {
|
if !ok || hook == nil {
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package cabana
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -128,6 +129,8 @@ func (s *service) mount(r pact.Router) {
|
|||||||
constrainController(g)
|
constrainController(g)
|
||||||
g.Post("/{vendor}/{plugin}/{controller}", s.create)
|
g.Post("/{vendor}/{plugin}/{controller}", s.create)
|
||||||
constrainController(g)
|
constrainController(g)
|
||||||
|
g.Post("/{vendor}/{plugin}/{controller}/bulk-delete", s.bulkDelete)
|
||||||
|
constrainController(g)
|
||||||
g.Get("/{vendor}/{plugin}/{controller}/{id}", s.show)
|
g.Get("/{vendor}/{plugin}/{controller}/{id}", s.show)
|
||||||
constrainController(g)
|
constrainController(g)
|
||||||
g.Put("/{vendor}/{plugin}/{controller}/{id}", s.update)
|
g.Put("/{vendor}/{plugin}/{controller}/{id}", s.update)
|
||||||
@@ -242,7 +245,34 @@ func (s *service) update(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *service) bulkDelete(w http.ResponseWriter, r *http.Request) {
|
func (s *service) bulkDelete(w http.ResponseWriter, r *http.Request) {
|
||||||
WriteError(w, http.StatusInternalServerError, "error", msgServerError)
|
s.protect(w, r, func(cc *CompiledController) {
|
||||||
|
in, err := decodeBulk(r)
|
||||||
|
if err != nil {
|
||||||
|
writeCRUDError(w, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
svc, err := s.crud()
|
||||||
|
if err != nil {
|
||||||
|
WriteError(w, http.StatusInternalServerError, "error", msgServerError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
result, err := svc.BulkDelete(r.Context(), cc, in)
|
||||||
|
if err != nil {
|
||||||
|
writeCRUDError(w, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
WriteData(w, http.StatusOK, result, nil)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func decodeBulk(r *http.Request) (BulkDeleteInput, error) {
|
||||||
|
dec := json.NewDecoder(r.Body)
|
||||||
|
dec.UseNumber()
|
||||||
|
var in BulkDeleteInput
|
||||||
|
if err := dec.Decode(&in); err != nil {
|
||||||
|
return BulkDeleteInput{}, &ValidationError{Details: map[string]any{"body": []string{"The request body is invalid."}}}
|
||||||
|
}
|
||||||
|
return in, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *service) deleteRecord(w http.ResponseWriter, r *http.Request) {
|
func (s *service) deleteRecord(w http.ResponseWriter, r *http.Request) {
|
||||||
|
|||||||
Reference in New Issue
Block a user