fix(items): centralize creation collision retries (WI-955)

Author: stefan-ernst

internal/repository/fracindex.go

@@ -79,8 +79,7 @@ func isFracIndexRetryableTransactionError(err error) bool {
 
 // IsSerializationAbort reports PostgreSQL serialization failures and
 // deadlocks — the cases where retrying the whole transaction on a fresh
-// snapshot is the documented recovery. Exported for service-level clone
-// transactions that need the same retry classification.
+// snapshot is the documented recovery.
 func IsSerializationAbort(err error) bool {
 	return isFracIndexRetryableTransactionError(err)
 }

internal/repository/item_create_transaction.go

@@ -0,0 +1,104 @@
+package repository
+
+import (
+	"context"
+	"fmt"
+
+	"windshift/internal/database"
+	"windshift/internal/models"
+)
+
+// IsItemCreateRetryable reports whether item creation should restart on a
+// fresh transaction.
+func IsItemCreateRetryable(err error) bool {
+	return IsFracIndexUniqueViolation(err) ||
+		IsWorkspaceItemNumberUniqueViolation(err) ||
+		IsSerializationAbort(err)
+}
+
+// WithItemCreateTransaction runs one complete item creation transaction and
+// retries conflicts that require a fresh snapshot. The callback must allocate
+// ranks and workspace item numbers inside the supplied transaction.
+func WithItemCreateTransaction(
+	ctx context.Context,
+	db database.Database,
+	create func(tx database.Tx) (int, error),
+) (int, error) {
+	if ctx == nil {
+		return 0, fmt.Errorf("item creation requires a context")
+	}
+	if db == nil {
+		return 0, fmt.Errorf("item creation requires a database")
+	}
+	if create == nil {
+		return 0, fmt.Errorf("item creation requires a transaction callback")
+	}
+
+	var lastErr error
+	for range FracIndexMaxRetries {
+		if err := ctx.Err(); err != nil {
+			return 0, err
+		}
+
+		tx, err := db.BeginTx(ctx, nil)
+		if err != nil {
+			return 0, fmt.Errorf("begin item creation transaction: %w", err)
+		}
+
+		itemID, err := create(tx)
+		if err == nil {
+			if commitErr := tx.Commit(); commitErr != nil {
+				err = fmt.Errorf("commit item creation transaction: %w", commitErr)
+			}
+		}
+		if err == nil {
+			return itemID, nil
+		}
+		_ = tx.Rollback()
+
+		if !IsItemCreateRetryable(err) {
+			return 0, err
+		}
+		lastErr = err
+	}
+
+	return 0, fmt.Errorf("item creation failed after %d attempts: %w", FracIndexMaxRetries, lastErr)
+}
+
+// CreateWithRetry appends an item and any source-specific records in one
+// retryable transaction.
+func (r *ItemRepository) CreateWithRetry(
+	ctx context.Context,
+	item *models.Item,
+	afterCreate func(tx database.Tx, itemID int) error,
+) (int, error) {
+	if item == nil {
+		return 0, fmt.Errorf("item creation requires an item")
+	}
+
+	itemID, err := WithItemCreateTransaction(ctx, r.db, func(tx database.Tx) (int, error) {
+		attempt := *item
+		nextNumber, err := r.GetNextWorkspaceItemNumber(tx, attempt.WorkspaceID)
+		if err != nil {
+			return 0, fmt.Errorf("allocate workspace item number: %w", err)
+		}
+		attempt.WorkspaceItemNumber = nextNumber
+		attempt.FracIndex = nil
+
+		itemID, err := r.Create(tx, &attempt)
+		if err != nil {
+			return 0, err
+		}
+		if afterCreate != nil {
+			if err := afterCreate(tx, itemID); err != nil {
+				return 0, err
+			}
+		}
+		return itemID, nil
+	})
+	if err != nil {
+		return 0, err
+	}
+	InvalidateItemListCountCache(r.db)
+	return itemID, nil
+}

internal/repository/item_repository.go

@@ -755,6 +755,9 @@ func (r *ItemRepository) GetNextWorkspaceItemNumber(tx database.Tx, workspaceID
 	return int(maxNumber + 1), nil
 }
 
+// Create inserts an item in a caller-owned transaction. Production creation
+// flows should use CreateWithRetry so rank and item-number conflicts restart
+// the full transaction.
 func (r *ItemRepository) Create(tx database.Tx, item *models.Item) (int, error) {
 	if err := acquireGlobalRankMutationLock(tx, r.db.GetDriverName()); err != nil {
 		return 0, err

internal/repository/workspace_template_repository.go

@@ -317,7 +317,7 @@ func (r *WorkspaceTemplateRepository) InsertClonedItemTx(ctx context.Context, tx
 		insert.CreatorID, insert.CreatedAt, insert.CreatedAt, insert.CreatedAt).Scan(&id)
 	if err != nil {
 		if database.IsUniqueConstraintError(err) {
-			return 0, ErrDuplicateEntry
+			return 0, fmt.Errorf("%w: %w", ErrDuplicateEntry, err)
 		}
 		return 0, fmt.Errorf("insert cloned item: %w", err)
 	}

internal/scheduler/recurrence_scheduler.go

@@ -1,6 +1,7 @@
 package scheduler
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -265,24 +266,6 @@ func (rs *RecurrenceScheduler) generateInstancesForRule(rule *models.RecurrenceR
 
 // createInstance creates a single recurring task instance
 func (rs *RecurrenceScheduler) createInstance(rule *models.RecurrenceRule, template *models.Item, scheduledDate time.Time) error {
-	tx, err := rs.db.Begin()
-	if err != nil {
-		return fmt.Errorf("failed to begin transaction: %w", err)
-	}
-	defer func() { _ = tx.Rollback() }()
-
-	// Get next workspace item number
-	nextNum, err := rs.itemRepo.GetNextWorkspaceItemNumber(tx, template.WorkspaceID)
-	if err != nil {
-		return fmt.Errorf("failed to get next item number: %w", err)
-	}
-
-	// Get next sequence number for this rule
-	seqNum, err := rs.recurrenceRepo.GetNextSequenceNumber(tx, rule.ID)
-	if err != nil {
-		return fmt.Errorf("failed to get sequence number: %w", err)
-	}
-
 	// Determine status. The user can pin one explicitly; otherwise resolve via the
 	// workspace's configuration set → workflow → initial transition (the same path
 	// used by handlers/portal.go and handlers/forms.go). The previous hard-coded
@@ -306,14 +289,13 @@ func (rs *RecurrenceScheduler) createInstance(rule *models.RecurrenceRule, templ
 	// goes through ItemRepository so the items-table write lives in the
 	// repository layer (same path scm/issue_sync.go uses), not in the scheduler.
 	item := &models.Item{
-		WorkspaceID:         template.WorkspaceID,
-		WorkspaceItemNumber: nextNum,
-		ItemTypeID:          template.ItemTypeID,
-		Title:               template.Title,
-		StatusID:            &statusID,
-		DueDate:             &scheduledDate,
-		IsTask:              template.IsTask,
-		ParentID:            template.ParentID,
+		WorkspaceID: template.WorkspaceID,
+		ItemTypeID:  template.ItemTypeID,
+		Title:       template.Title,
+		StatusID:    &statusID,
+		DueDate:     &scheduledDate,
+		IsTask:      template.IsTask,
+		ParentID:    template.ParentID,
 	}
 	if rule.CopyDescription {
 		item.Description = template.Description
@@ -328,24 +310,23 @@ func (rs *RecurrenceScheduler) createInstance(rule *models.RecurrenceRule, templ
 		item.CustomFieldValues = template.CustomFieldValues
 	}
 
-	itemID, err := rs.itemRepo.Create(tx, item)
-	if err != nil {
-		return fmt.Errorf("failed to create item: %w", err)
-	}
-
-	// Create the instance record
-	err = rs.recurrenceRepo.CreateInstance(tx, &models.RecurrenceInstance{
-		RecurrenceRuleID: rule.ID,
-		InstanceItemID:   itemID,
-		ScheduledDate:    scheduledDate,
-		SequenceNumber:   seqNum,
+	itemID, err := rs.itemRepo.CreateWithRetry(context.Background(), item, func(tx database.Tx, itemID int) error {
+		seqNum, err := rs.recurrenceRepo.GetNextSequenceNumber(tx, rule.ID)
+		if err != nil {
+			return fmt.Errorf("get next recurrence sequence number: %w", err)
+		}
+		if err := rs.recurrenceRepo.CreateInstance(tx, &models.RecurrenceInstance{
+			RecurrenceRuleID: rule.ID,
+			InstanceItemID:   itemID,
+			ScheduledDate:    scheduledDate,
+			SequenceNumber:   seqNum,
+		}); err != nil {
+			return fmt.Errorf("create recurrence instance: %w", err)
+		}
+		return nil
 	})
 	if err != nil {
-		return fmt.Errorf("failed to create instance record: %w", err)
-	}
-
-	if err := tx.Commit(); err != nil {
-		return err
+		return fmt.Errorf("create recurring item: %w", err)
 	}
 
 	// Live-update publish (WI-483): the recurrence instance committed. Announce

internal/scm/issue_sync.go

@@ -283,61 +283,46 @@ func (s *IssueSyncService) createItemFromIssue(ctx context.Context, config *mode
 
 	milestoneID := s.resolveMilestoneID(config, issue)
 
-	tx, err := s.db.BeginTx(ctx, nil)
-	if err != nil {
-		return fmt.Errorf("begin tx: %w", err)
-	}
-	defer func() { _ = tx.Rollback() }()
-
-	nextNum, err := s.itemRepo.GetNextWorkspaceItemNumber(tx, config.WorkspaceID)
-	if err != nil {
-		return fmt.Errorf("get next item number: %w", err)
-	}
-
 	newItem := &models.Item{
-		WorkspaceID:         config.WorkspaceID,
-		WorkspaceItemNumber: nextNum,
-		ItemTypeID:          config.DefaultItemTypeID,
-		Title:               issue.Title,
-		Description:         issue.Body,
-		StatusID:            statusID,
-		PriorityID:          config.DefaultPriorityID,
-		AssigneeID:          assigneeID,
-	}
-	itemID, err := s.itemRepo.Create(tx, newItem)
-	if err != nil {
-		return fmt.Errorf("create item: %w", err)
+		WorkspaceID: config.WorkspaceID,
+		ItemTypeID:  config.DefaultItemTypeID,
+		Title:       issue.Title,
+		Description: issue.Body,
+		StatusID:    statusID,
+		PriorityID:  config.DefaultPriorityID,
+		AssigneeID:  assigneeID,
 	}
 
 	milestoneIDs := []int{}
 	if milestoneID != nil {
 		milestoneIDs = append(milestoneIDs, *milestoneID)
 	}
-	if err := repository.NewMilestoneAttachRepository(s.db).ReplaceItemMilestonesTx(ctx, tx, itemID, milestoneIDs); err != nil {
-		return err
-	}
-
-	now := time.Now()
-	_, err = tx.Exec(`
-		INSERT INTO issue_sync_items (
-			issue_sync_config_id, item_id, github_issue_number, github_issue_id,
-			github_issue_url, last_synced_at, last_github_updated_at, created_at, updated_at
-		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
-	`,
-		config.ID, itemID, issue.Number, issue.ID,
-		issue.URL, now, issue.UpdatedAt, now, now,
-	)
-	if err != nil {
-		return fmt.Errorf("insert sync item: %w", err)
-	}
+	itemID, err := s.itemRepo.CreateWithRetry(ctx, newItem, func(tx database.Tx, itemID int) error {
+		if err := repository.NewMilestoneAttachRepository(s.db).ReplaceItemMilestonesTx(ctx, tx, itemID, milestoneIDs); err != nil {
+			return err
+		}
 
-	// Keep item, labels, and sync metadata atomic.
-	if err := s.syncLabels(ctx, tx, config, issue, itemID); err != nil {
-		return fmt.Errorf("sync labels: %w", err)
-	}
+		now := time.Now()
+		if _, err := tx.Exec(`
+			INSERT INTO issue_sync_items (
+				issue_sync_config_id, item_id, github_issue_number, github_issue_id,
+				github_issue_url, last_synced_at, last_github_updated_at, created_at, updated_at
+			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+		`,
+			config.ID, itemID, issue.Number, issue.ID,
+			issue.URL, now, issue.UpdatedAt, now, now,
+		); err != nil {
+			return fmt.Errorf("insert sync item: %w", err)
+		}
 
-	if err := tx.Commit(); err != nil {
-		return fmt.Errorf("commit: %w", err)
+		// Keep item, labels, and sync metadata atomic.
+		if err := s.syncLabels(ctx, tx, config, issue, itemID); err != nil {
+			return fmt.Errorf("sync labels: %w", err)
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("create item from issue: %w", err)
 	}
 
 	slog.Info("created item from GitHub issue",