fix: sync deleted subscriptions at least once (#4382)

Author: turip

.agents/skills/subscriptionsync/SKILL.md

@@ -11,7 +11,7 @@ Guidance for working with `openmeter/billing/worker/subscriptionsync/`.
 
 `subscriptionsync` is the bridge between the subscription domain and billing.
 
-- Input: `subscription.SubscriptionView`
+- Input: either a subscription ID (`models.NamespacedID`) or an already-expanded `subscription.SubscriptionView`
 - Output: reconciled billing state
   - invoice lines / split-line groups
   - charges
@@ -25,7 +25,8 @@ It does not own subscription editing rules and it does not own billing primitive
 openmeter/billing/worker/subscriptionsync/
 ├── service/                         # orchestration entrypoint used by the worker
 │   ├── service.go                   # Service struct, Config, FeatureFlags, constructor
-│   ├── sync.go                      # SynchronizeSubscription + SynchronizeSubscriptionAndInvoiceCustomer
+│   ├── sync.go                      # internal sync orchestration for SyncByID/SyncByView entrypoints
+│   ├── ref.go                       # normalizes subscription ID vs view references
 │   ├── reconcile.go                 # build persisted snapshot + target state + plan
 │   ├── handlers.go                  # event handlers (HandleCancelledEvent, HandleInvoiceCreation)
 │   ├── base_test.go                 # shared SuiteBase + SyncSuiteBase test harness
@@ -54,20 +55,29 @@ openmeter/billing/worker/subscriptionsync/
 
 ## Core Flow
 
-`Service.SynchronizeSubscription(...)` in `service/sync.go` is the main entrypoint.
+The public sync entrypoints are defined by `subscriptionsync.SyncService`:
+
+- `SyncByID(...)` and `SyncByIDAndInvoiceCustomer(...)` fetch subscription state internally.
+- `SyncByView(...)` and `SyncByViewAndInvoiceCustomer(...)` use an already-expanded subscription view.
+- The `AndInvoiceCustomer` variants run the regular sync and then invoice pending lines for the subscription customer.
+
+The internal orchestration lives in `service/sync.go` (`synchronizeSubscription`, `synchronizeSubscriptionAndInvoiceCustomer`) and normalizes the public input through `subscriptionReferenceOrView` from `service/ref.go`.
 
 High-level flow:
-1. Load persisted sync state and persisted billing artifacts.
-2. Build target state from the subscription view.
-3. Reconcile target vs persisted.
-4. Apply billing patches.
-5. Persist sync state.
+1. Load the subscription entity, including deleted subscriptions when the ID path is used.
+2. Load persisted sync state and persisted billing artifacts.
+3. Build target state from the subscription view, or an empty target state when the subscription is deleted.
+4. Reconcile target vs persisted.
+5. Apply billing patches.
+6. Persist sync state.
 
 The important bridge boundaries are:
 - `persistedstate`: current billing-side reality relevant to this subscription
 - `targetstate`: expected billing-side reality derived from the subscription view
 - `reconciler`: diff between the two, expressed as backend-specific patches
 
+Deleted-subscription cleanup uses the ID path. `subscription.Service.GetView` follows the normal non-deleted read path, so code that needs to reconcile deleted subscriptions must use `List(... IncludeDeleted: true)` or the sync service's ID entrypoints rather than calling `GetView` first.
+
 ## Persisted State
 
 `service/persistedstate` owns the billing-side read model used by sync.
@@ -100,6 +110,7 @@ Invoice loading notes:
 `service/targetstate` converts a subscription view into expected billing/charge items.
 
 Useful points:
+- `BuildInput.SubscriptionView` is a pointer. A nil view represents a deleted subscription and builds an empty target state so persisted billing artifacts are removed.
 - `StateItem.IsBillable()` is the first gate
 - `StateItem.GetServicePeriod()` is the diff-level period source
 - `StateItem.GetExpectedLine()` is invoice-specific rendering; keep direct billing assumptions isolated to places that really need invoice lines
@@ -191,6 +202,8 @@ Test suite hierarchy:
 
 Use `setupChargesService(config)` on `SuiteBase` to rebuild the sync service with a charge-capable stack (replaces the default no-charges service).
 
+When a billing sync test needs to exercise deleted-subscription cleanup, prefer the public ID entrypoint (`SyncByID` / `SyncByIDAndInvoiceCustomer`) so the service owns the `IncludeDeleted` lookup. Subscription-domain coverage for deleted scheduled replacements belongs under `openmeter/subscription/service/sync_test.go`; keep billing-package tests focused on billing artifacts.
+
 For charge-backed sync tests:
 - prefer `openmeter/billing/charges/testutils.NewMockHandlers()`
 - these mocks are intentionally minimal but valid enough for charge creation/advancement
@@ -216,7 +229,7 @@ Pattern for credit-only tests:
 
 Key methods:
 - `ListSubscriptions(...)` — pages through active subscriptions with their sync states
-- `ReconcileSubscription(...)` — fetches subscription view and calls `SynchronizeSubscription`
+- `ReconcileSubscription(...)` — calls `SyncByID` for the subscription under reconciliation
 - `All(...)` — reconciles all eligible subscriptions, skipping those with no billables or whose `NextSyncAfter` is in the future (unless `Force` is set)
 
 ## Common Refactor Rules

migrations/atlas.sum

@@ -47,8 +47,9 @@ func (v *Validator) ValidateDeleteCustomer(ctx context.Context, input customer.D
 
 	// Let's sync any subscriptions pending for this customer
 	subs, err := v.subscriptionService.List(ctx, subscription.ListSubscriptionsInput{
-		Namespaces: []string{input.Namespace},
-		CustomerID: &filter.FilterULID{FilterString: filter.FilterString{Eq: &input.ID}},
+		Namespaces:     []string{input.Namespace},
+		CustomerID:     &filter.FilterULID{FilterString: filter.FilterString{Eq: &input.ID}},
+		IncludeDeleted: true,
 	})
 	if err != nil {
 		return err
@@ -57,13 +58,8 @@ func (v *Validator) ValidateDeleteCustomer(ctx context.Context, input customer.D
 	watermark := time.Now().Add(-24 * time.Hour)
 
 	for _, sub := range subs.Items {
-		if sub.ActiveTo == nil || watermark.Before(*sub.ActiveTo) {
-			view, err := v.subscriptionService.GetView(ctx, sub.NamespacedID)
-			if err != nil {
-				return err
-			}
-
-			if err := v.syncService.SynchronizeSubscription(ctx, view, time.Now()); err != nil {
+		if sub.ActiveTo == nil || sub.DeletedAt != nil || watermark.Before(*sub.ActiveTo) {
+			if err := v.syncService.SyncByID(ctx, sub.NamespacedID, time.Now()); err != nil {
 				return err
 			}
 		}

openmeter/billing/validators/customer/customer.go

@@ -127,67 +127,89 @@ func (r *Reconciler) ListSubscriptions(ctx context.Context, in ReconcilerListSub
 			break
 		}
 
-		syncStates, err := r.subscriptionSync.GetSyncStates(ctx, lo.Map(subscriptions.Items, func(item subscription.Subscription, _ int) models.NamespacedID {
-			return models.NamespacedID{
-				ID:        item.ID,
-				Namespace: item.Namespace,
-			}
-		}))
+		mapped, err := r.mapToSubscriptionWithSyncState(ctx, subscriptions.Items)
 		if err != nil {
-			return nil, fmt.Errorf("failed to get sync states: %w", err)
+			return nil, fmt.Errorf("failed to map subscriptions to subscription with sync state: %w", err)
 		}
 
-		syncStatesBySubscriptionID := lo.SliceToMap(syncStates, func(syncState subscriptionsync.SyncState) (models.NamespacedID, subscriptionsync.SyncState) {
-			return models.NamespacedID{
-				ID:        syncState.SubscriptionID.ID,
-				Namespace: syncState.SubscriptionID.Namespace,
-			}, syncState
+		out = append(out, mapped...)
+
+		pageIndex++
+	}
+
+	pageIndex = 1
+
+	for {
+		subscriptions, err := r.subscriptionService.List(ctx, subscription.ListSubscriptionsInput{
+			Namespaces: in.Namespaces,
+			CustomerID: customerID,
+			// TODO: Later we might want to have a lookback for deleted subscriptions as well, but as of 2026-05-19
+			// we had a delete bug, so we need to reconcile all deleted subscriptions.
+			DeletedAt: &filter.FilterTime{
+				Lte: lo.ToPtr(clock.Now()),
+			},
+			IncludeDeleted: true,
+			Page: pagination.Page{
+				PageNumber: pageIndex,
+				PageSize:   defaultWindowSize,
+			},
 		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to list subscriptions: %w", err)
+		}
 
-		out = append(out, lo.Map(subscriptions.Items, func(item subscription.Subscription, _ int) SubscriptionWithSyncState {
-			existingSyncState, ok := syncStatesBySubscriptionID[item.NamespacedID]
+		if len(subscriptions.Items) == 0 {
+			break
+		}
 
-			var syncState *subscriptionsync.SyncState
-			if ok {
-				syncState = lo.ToPtr(existingSyncState)
-			}
+		mapped, err := r.mapToSubscriptionWithSyncState(ctx, subscriptions.Items)
+		if err != nil {
+			return nil, fmt.Errorf("failed to map subscriptions to subscription with sync state: %w", err)
+		}
 
-			return SubscriptionWithSyncState{
-				Subscription: item,
-				SyncState:    syncState,
-			}
-		})...)
+		out = append(out, mapped...)
 
 		pageIndex++
 	}
 
 	return out, nil
 }
 
-func (r *Reconciler) ReconcileSubscription(ctx context.Context, subsID models.NamespacedID) error {
-	subsView, err := r.subscriptionService.GetView(ctx, subsID)
+func (r *Reconciler) mapToSubscriptionWithSyncState(ctx context.Context, subs []subscription.Subscription) ([]SubscriptionWithSyncState, error) {
+	syncStates, err := r.subscriptionSync.GetSyncStates(ctx, lo.Map(subs, func(item subscription.Subscription, _ int) models.NamespacedID {
+		return models.NamespacedID{
+			ID:        item.ID,
+			Namespace: item.Namespace,
+		}
+	}))
 	if err != nil {
-		return fmt.Errorf("failed to get subscription: %w", err)
+		return nil, fmt.Errorf("failed to get sync states: %w", err)
 	}
 
-	cus, err := r.customerService.GetCustomer(ctx, customer.GetCustomerInput{
-		CustomerID: &customer.CustomerID{
-			ID:        subsView.Customer.ID,
-			Namespace: subsID.Namespace,
-		},
+	syncStatesBySubscriptionID := lo.SliceToMap(syncStates, func(syncState subscriptionsync.SyncState) (models.NamespacedID, subscriptionsync.SyncState) {
+		return models.NamespacedID{
+			ID:        syncState.SubscriptionID.ID,
+			Namespace: syncState.SubscriptionID.Namespace,
+		}, syncState
 	})
-	if err != nil {
-		return fmt.Errorf("failed to get customer: %w", err)
-	}
 
-	// TODO: remove this check to make sure that deleted customers are fully invoiced
-	if cus != nil && cus.IsDeleted() {
-		r.logger.WarnContext(ctx, "customer is deleted, skipping reconciliation", "customer_id", subsView.Customer.ID)
+	return lo.Map(subs, func(item subscription.Subscription, _ int) SubscriptionWithSyncState {
+		existingSyncState, ok := syncStatesBySubscriptionID[item.NamespacedID]
 
-		return nil
-	}
+		var syncState *subscriptionsync.SyncState
+		if ok {
+			syncState = lo.ToPtr(existingSyncState)
+		}
 
-	return r.subscriptionSync.SynchronizeSubscription(ctx, subsView, time.Now())
+		return SubscriptionWithSyncState{
+			Subscription: item,
+			SyncState:    syncState,
+		}
+	}), nil
+}
+
+func (r *Reconciler) ReconcileSubscription(ctx context.Context, subsID models.NamespacedID) error {
+	return r.subscriptionSync.SyncByID(ctx, subsID, time.Now())
 }
 
 type ReconcilerAllInput struct {

openmeter/billing/worker/subscriptionsync/reconciler/reconciler.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/openmeterio/openmeter/openmeter/billing"
 	"github.com/openmeterio/openmeter/openmeter/subscription"
+	"github.com/openmeterio/openmeter/pkg/models"
 )
 
 type Service interface {
@@ -15,12 +16,15 @@ type Service interface {
 }
 
 type SyncService interface {
-	SynchronizeSubscriptionAndInvoiceCustomer(ctx context.Context, subs subscription.SubscriptionView, asOf time.Time) error
-	SynchronizeSubscription(ctx context.Context, subs subscription.SubscriptionView, asOf time.Time, opts ...SynchronizeSubscriptionOption) error
+	SyncByViewAndInvoiceCustomer(ctx context.Context, view subscription.SubscriptionView, asOf time.Time) error
+	SyncByIDAndInvoiceCustomer(ctx context.Context, subscriptionID models.NamespacedID, asOf time.Time) error
+	SyncByView(ctx context.Context, view subscription.SubscriptionView, asOf time.Time, opts ...SynchronizeSubscriptionOption) error
+	SyncByID(ctx context.Context, subscriptionID models.NamespacedID, asOf time.Time, opts ...SynchronizeSubscriptionOption) error
 }
 
 type EventHandler interface {
 	HandleCancelledEvent(ctx context.Context, event *subscription.CancelledEvent) error
+	HandleDeletedEvent(ctx context.Context, event *subscription.DeletedEvent) error
 	HandleSubscriptionSyncEvent(ctx context.Context, event *subscription.SubscriptionSyncEvent) error
 	HandleInvoiceCreation(ctx context.Context, event *billing.StandardInvoiceCreatedEvent) error
 }