fix(purchase): claim the exact managed schedule root

internal/config/store_postgres_managed_root_integration_test.go

@@ -0,0 +1,29 @@
+//go:build integration
+
+package config
+
+import (
+	"context"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/require"
+)
+
+func TestManagedRootCapabilityRefusesUnmanaged(t *testing.T) {
+	ctx := t.Context()
+	f := newRampFixture(ctx, t, 0, 0, 3)
+	capability, ok := any(f.store).(interface {
+		ResolveManagedPlanRootTx(context.Context, pgx.Tx, *PurchasePlan) (*PurchaseExecution, error)
+	})
+	if !ok {
+		t.Skip("parent lacks the capability; behavioral parent proof uses the actual Manager")
+	}
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		p, err := f.store.LockPurchasePlanTx(ctx, tx, f.planID)
+		require.NoError(t, err)
+		_, err = capability.ResolveManagedPlanRootTx(ctx, tx, p)
+		require.Error(t, err)
+		return nil
+	}))
+}

internal/config/store_postgres_plan_schedule_clock.go

@@ -47,6 +47,33 @@ func (s *PostgresStore) planScheduleNextDateTx(ctx context.Context, tx pgx.Tx, p
 	return &root.ScheduledDate, nil
 }
 
+// ResolveManagedPlanRootTx requires the plan row lock before execution locks.
+func (s *PostgresStore) ResolveManagedPlanRootTx(ctx context.Context, tx pgx.Tx, plan *PurchasePlan) (*PurchaseExecution, error) {
+	if plan == nil || !plan.RampSchedule.DatesManaged {
+		return nil, fmt.Errorf("managed root resolution requires a managed plan")
+	}
+	if plan.RampSchedule.IsComplete() {
+		return nil, nil
+	}
+	rows, err := tx.Query(ctx, `SELECT execution_id FROM purchase_executions
+		WHERE plan_id=$1 AND step_number=$2 ORDER BY execution_id FOR UPDATE`, plan.ID, plan.RampSchedule.CurrentStep+1)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	rows.Close()
+	return s.managedPlanNextRootTx(ctx, tx, plan)
+}
+
 func (s *PostgresStore) managedPlanNextRootTx(ctx context.Context, tx pgx.Tx, plan *PurchasePlan) (*PurchaseExecution, error) {
 	rows, err := s.ListExecutionsForPlanStepTx(ctx, tx, plan.ID, plan.RampSchedule.CurrentStep+1)
 	if err != nil {

internal/purchase/notifications.go

@@ -130,9 +130,12 @@ func (m *Manager) dueManualRootTx(ctx context.Context, tx pgx.Tx, planID string)
 	if err != nil {
 		return nil, err
 	}
-	if locked == nil || !locked.Enabled || locked.AutoPurchase {
+	if !enabledManualPlan(locked) {
 		return nil, nil
 	}
+	if locked.RampSchedule.DatesManaged {
+		return m.dueManagedPlanRootTx(ctx, tx, locked)
+	}
 	rows, err := m.config.ListExecutionsForPlanStepTx(ctx, tx, planID, locked.RampSchedule.CurrentStep+1)
 	if err != nil || len(rows) == 0 {
 		return nil, err

internal/purchase/plan_step_attach.go

@@ -66,6 +66,12 @@ func (m *Manager) claimPlanStep(ctx context.Context, tx pgx.Tx, plan *config.Pur
 	if locked == nil {
 		return stepClaim{}, fmt.Errorf("plan not found: %s", plan.ID)
 	}
+	if locked.RampSchedule.DatesManaged != plan.RampSchedule.DatesManaged {
+		return stepClaim{}, errPlanStepSkipped
+	}
+	if locked.RampSchedule.DatesManaged {
+		return m.claimManagedPlanStep(ctx, tx, plan, locked, providers)
+	}
 	if !planUnchanged(plan, locked, providers) {
 		return stepClaim{}, fmt.Errorf("%w: plan %s changed since it was listed", errPlanStepSkipped, plan.ID)
 	}

internal/purchase/plan_step_managed.go

@@ -0,0 +1,58 @@
+package purchase
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/jackc/pgx/v5"
+)
+
+type managedPlanRootStore interface {
+	ResolveManagedPlanRootTx(context.Context, pgx.Tx, *config.PurchasePlan) (*config.PurchaseExecution, error)
+}
+
+func (m *Manager) managedPlanRootTx(ctx context.Context, tx pgx.Tx, plan *config.PurchasePlan) (*config.PurchaseExecution, error) {
+	store, ok := m.config.(managedPlanRootStore)
+	if !ok {
+		return nil, fmt.Errorf("%w: managed root resolution unavailable", errPlanStepSkipped)
+	}
+	return store.ResolveManagedPlanRootTx(ctx, tx, plan)
+}
+
+func (m *Manager) claimManagedPlanStep(ctx context.Context, tx pgx.Tx, listed, locked *config.PurchasePlan, providers []string) (stepClaim, error) {
+	if !managedPlanUnchanged(listed, locked, providers) {
+		return stepClaim{}, errPlanStepSkipped
+	}
+	root, err := m.managedPlanRootTx(ctx, tx, locked)
+	if err != nil {
+		return stepClaim{}, err
+	}
+	if root == nil || root.Status == "scheduled" {
+		return stepClaim{}, errPlanStepSkipped
+	}
+	locked.NextExecutionDate = &root.ScheduledDate
+	return m.claimFromRows(ctx, tx, locked, stepClaim{number: locked.RampSchedule.CurrentStep + 1}, []config.PurchaseExecution{*root})
+}
+
+func managedPlanUnchanged(listed, locked *config.PurchasePlan, providers []string) bool {
+	return locked.Enabled && locked.AutoPurchase == listed.AutoPurchase && locked.UpdatedAt.Equal(listed.UpdatedAt) &&
+		locked.RampSchedule.CurrentStep == listed.RampSchedule.CurrentStep && strings.Join(planProviders(locked), ",") == strings.Join(providers, ",")
+}
+
+func enabledManualPlan(plan *config.PurchasePlan) bool {
+	return plan != nil && plan.Enabled && !plan.AutoPurchase
+}
+
+func (m *Manager) dueManagedPlanRootTx(ctx context.Context, tx pgx.Tx, plan *config.PurchasePlan) (*config.PurchaseExecution, error) {
+	root, err := m.managedPlanRootTx(ctx, tx, plan)
+	if err != nil || root == nil {
+		return nil, err
+	}
+	if root.Status != "scheduled" && len(root.Recommendations) == 0 && !root.ScheduledDate.After(time.Now()) {
+		return root, nil
+	}
+	return nil, nil
+}