feat: payment settlement support refactor

Author: turip

.agents/skills/billing/SKILL.md

@@ -0,0 +1,132 @@
+# Subscription → Billing Sync Algorithm
+
+Primary location: `openmeter/billing/worker/subscriptionsync/`
+
+## Entry Points
+
+The sync is triggered by the billing worker (`worker/worker.go`) on these events:
+
+| Event | Handler |
+|---|---|
+| `subscription.Created/Updated/Continued` | `SynchronizeSubscriptionAndInvoiceCustomer` |
+| `subscription.Cancelled` | `HandleCancelledEvent` |
+| `subscription.SubscriptionSyncEvent` | `HandleSubscriptionSyncEvent` (self-loop) |
+| `billing.StandardInvoiceCreatedEvent` | `HandleInvoiceCreation` → re-syncs all subscriptions in the new invoice's lines |
+
+The self-loop mechanism (`SubscriptionSyncEvent`) ensures the gathering invoice is refilled with the next period's lines immediately after an invoice is issued, without waiting for the cron.
+
+## Main Algorithm (`service/sync.go`)
+
+```
+SynchronizeSubscriptionAndInvoiceCustomer(ctx, SubscriptionView, asOf)
+  → SynchronizeSubscription
+    → billingService.WithLock(customerID)  ← advisory lock, serializes per customer
+      → persistedstate.Loader.LoadForSubscription   ← fetch current DB state
+      → targetstate.Builder.Build                   ← compute desired state
+      → reconciler.Plan                             ← diff: new / delete / upsert
+      → reconciler.Apply                            ← write patches to DB
+      → updateSyncState                             ← upsert SubscriptionBillingSyncState row
+  → invoicePendingLines  ← InvoicePendingLines(ProgressiveBillingOverride=false)
+```
+
+## Target State Builder (`service/targetstate/`)
+
+`Builder.Build` iterates each phase and calls `PhaseIterator.Generate(ctx, generationLimit)`.
+
+**Generation limit**: the end of the current aligned billing period (`spec.GetAlignedBillingPeriodAt(asOf)`). Lines are never generated past the current period boundary, except for in-advance items whose `InvoiceAt` equals the period start — these are pre-generated so the gathering invoice is immediately ready.
+
+**Phase iterator loop** (`phaseiterator.go`): for each item in the phase, loops by `BillingCadence` until `InvoiceAt > iterationEnd`. Each iteration produces a `SubscriptionItemWithPeriods` with:
+- `ServicePeriod`: the actual usage window for this item cadence period
+- `BillingPeriod`: the subscription-aligned outer billing period
+
+### `GetInvoiceAt()` — When Does a Line Appear on an Invoice?
+
+```go
+// Flat-fee in-advance: bill at the start of the billing period
+if price.Type() == FlatPrice && paymentTerm == InAdvance {
+    return BillingPeriod.Start
+}
+// Everything else (in-arrears, usage-based): bill at end of service or billing period
+return max(ServicePeriod.End, BillingPeriod.End)
+```
+
+This is defined in `phaseiterator.go` and is the canonical source of billing timing truth.
+
+## Reconciler (`service/reconciler/`)
+
+### Plan (`reconciler.go`)
+
+Computes symmetric difference on `ChildUniqueReferenceID` keys:
+- **Only in persisted**: `LinesToDelete`
+- **Only in target**: `NewSubscriptionItems`
+- **In both**: `LinesToUpsert`
+
+### ChildUniqueReferenceID Format
+
+```
+{subscriptionID}/{phaseKey}/{itemKey}/v[{version}]/period[{periodIndex}]
+```
+
+One-time (non-recurring) items omit `/period[N]`. Version increments when the rate card changes within a phase.
+
+### Apply (`apply.go` + `invoiceupdate.go`)
+
+Translates the plan into `Patch` objects and calls `InvoiceUpdater.ApplyPatches`. The updater handles three invoice states:
+
+| Invoice state | Action |
+|---|---|
+| Gathering invoice | Direct edit via `UpdateGatheringInvoice` |
+| Mutable standard invoice | `UpdateStandardInvoice` + `SnapshotLineQuantity` |
+| Immutable standard invoice | Emit `ValidationIssue` (no actual mutation — detects drift) |
+
+## Persisted State (`service/persistedstate/`)
+
+`Loader.LoadForSubscription` calls `billingService.GetLinesForSubscription`, which queries `BillingInvoiceLine` rows where:
+- `subscription_id = ?`
+- `parent_line_id IS NULL` (top-level lines only; split children handled by mapper)
+- Either `deleted_at IS NULL` OR (`deleted_at IS NOT NULL` AND `managed_by = manually_managed`)
+
+Returns `[]billing.LineOrHierarchy` — gathering lines and standard lines/split hierarchies mixed.
+
+## SubscriptionBillingSyncState Table
+
+Managed in `adapter/syncstate.go`. Upserted on `(subscription_id, namespace)` after each sync:
+
+```go
+.OnConflictColumns(FieldSubscriptionID, FieldNamespace).
+    UpdateHasBillables().UpdateSyncedAt().UpdateNextSyncAfter().Exec(ctx)
+```
+
+`NextSyncAfter` is set to `SubscriptionMaxGenerationTimeLimit` (the end of the last generated billing period). The worker uses this to know when to next sync without processing every event repeatedly.
+
+## Billing Cadence and Alignment
+
+`SubscriptionSpec.BillingCadence` (ISO duration, e.g. `P1M`) is the master cadence.
+`BillingAnchor` is the fixed time anchor for period alignment.
+
+`GetAlignedBillingPeriodAt(at)` in `subscription/subscriptionspec.go`:
+1. Finds the active phase at `at`
+2. Builds `timeutil.Recurrence` from `(BillingCadence, BillingAnchor)`
+3. Calls `recurrence.GetPeriodAt(at)` to find the enclosing period
+4. Clips boundaries to phase start/end
+
+## Subscription-Level Timing Resolution
+
+`Timing.ResolveForSpec` in `subscription/timing.go` resolves `next_billing_cycle` to `GetAlignedBillingPeriodAt(now).To`. Subscription edits with `next_billing_cycle` timing are deferred to the next billing period boundary.
+
+## Subscription Validator (`validators/subscription/validator.go`)
+
+Implements `subscription.SubscriptionCommandHook`. On `AfterCreate` and `AfterUpdate`, validates:
+- Customer has a `CustomerOverride` pointing to a billing profile
+- Profile apps have the required capabilities: `CapabilityTypeCalculateTax`, `CapabilityTypeInvoiceCustomers`, `CapabilityTypeCollectPayments`
+
+If the subscription has no priced rate cards, validation is skipped entirely.
+
+## Annotations
+
+On `GatheringLine` and `StandardLine`:
+
+- `AnnotationSubscriptionSyncIgnore` — sync algorithm skips this line (managed externally)
+- `AnnotationSubscriptionSyncForceContinuousLines` — forces progressive billing to treat lines as continuous even if the pricing type would normally disallow it (use with care)
+
+Both are defined in `billing/annotations.go`.

.agents/skills/billing/references/subscription-sync.md

@@ -0,0 +1,211 @@
+# Billing Test Patterns
+
+## Suite Hierarchy
+
+```
+billingtest.BaseSuite              (test/billing/suite.go)
+    ↳ billingtest.SubscriptionMixin  (test/billing/subscription_suite.go)
+        ↳ subscriptionsync.SuiteBase (worker/subscriptionsync/service/suitebase_test.go)
+```
+
+Embed the lowest suite that gives you what you need. Most billing service tests only need `BaseSuite`. Tests involving subscription creation need `SubscriptionMixin`. Tests for the sync algorithm need `SuiteBase`.
+
+## BaseSuite (`test/billing/suite.go`)
+
+Sets up a full in-process stack with a real PostgreSQL database:
+
+1. `testutils.InitPostgresDB(t)` — real Postgres test DB (requires `POSTGRES_HOST=127.0.0.1`)
+2. Atlas migrations unless `TEST_DISABLE_ATLAS` is set (falls back to Ent auto-create)
+3. Full billing service + adapter chain
+4. `ForegroundAdvancementStrategy` — state machine runs synchronously in tests (no async events)
+5. `MockStreamingConnector` for meter queries
+6. `invoicecalc.NewMockableCalculator` for overriding pricing calculations
+7. Sandbox app factory + customer/subject sync hooks
+
+**Key exposed fields:**
+```go
+BillingService  billing.Service
+BillingAdapter  billing.Adapter
+MockStreamingConnector *streamingtestutils.MockStreamingConnector
+MeterAdapter    *metermock.MockRepository
+CustomerService customer.Service
+FeatureService  feature.FeatureConnector
+```
+
+**Namespace isolation:**
+```go
+ns := s.GetUniqueNamespace("my-test")  // returns "my-test-{ulid}"
+```
+Always use unique namespaces to isolate test data between test cases.
+
+## SubscriptionMixin (`test/billing/subscription_suite.go`)
+
+Call `mixin.SetupSuite(t, deps)` from your suite's `SetupSuite`. Adds:
+- `PlanService`, `SubscriptionService`, `SubscriptionAddonService`, `SubscriptionWorkflowService`
+- `EntitlementConnector` (full entitlement/credit/grant stack for metered entitlements)
+
+**Access via:**
+```go
+s.PlanService
+s.SubscriptionService
+s.SubscriptionWorkflowService
+```
+
+## SuiteBase for Sync Tests (`worker/subscriptionsync/service/suitebase_test.go`)
+
+Embeds both `BaseSuite` and `SubscriptionMixin`. Also provides:
+- `SubscriptionSyncService subscriptionsync.Service`
+- `SubscriptionSyncAdapter subscriptionsync.Adapter`
+
+**Per-test setup** (`BeforeTest`):
+```go
+// Creates fresh per-test state:
+ns := getUniqueTestNamespace(suiteName, testName)
+s.InstallSandboxApp(t, ns)
+s.ProvisionBillingProfile(t, ns)
+// Creates test meter + feature
+// Creates test customer
+```
+
+**Per-test teardown** (`AfterTest`):
+```go
+clock.UnFreeze()
+s.MockStreamingConnector.Reset()
+// resets feature flags on the service
+```
+
+## Provisioning Helpers
+
+### `InstallSandboxApp(t, ns)`
+Required before any invoice operations. Installs the sandbox invoicing app in the namespace.
+
+### `ProvisionBillingProfile(t, ns, opts...)`
+Creates a billing profile with option functions:
+
+```go
+s.ProvisionBillingProfile(t, ns,
+    billingtest.WithProgressiveBilling(),
+    billingtest.WithCollectionInterval(isodate.MustParse("P1D")),
+    billingtest.WithManualApproval(),
+    billingtest.WithBillingProfileEditFn(func(p *billing.CreateProfileInput) {
+        p.WorkflowConfig.Tax.Enabled = true
+    }),
+)
+```
+
+Default: auto-advance, monthly collection, immediate approval.
+
+### Subscription Creation Helpers (on `SuiteBase`)
+
+```go
+// Create from explicit phase definitions
+sub, err := s.createSubscriptionFromPlanPhases([]subscriptiontestutils.CreatePhasesInput{...})
+
+// Create from a full plan input
+sub, err := s.createSubscriptionFromPlan(plan.CreatePlanInput{...})
+```
+
+## Gathering Invoice Helpers (on `SuiteBase`)
+
+```go
+// Assert exactly 1 gathering invoice exists and return it with lines expanded
+gi := s.gatheringInvoice(ctx, ns, customerID)
+
+// Assert no gathering invoice exists
+s.expectNoGatheringInvoice(ctx, ns, customerID)
+
+// Verify lines on an invoice
+s.expectLines(invoice, subscriptionID, []expectedLine{
+    {PhaseKey: "default", ItemKey: "api-calls", ...},
+})
+```
+
+### `recurringLineMatcher{PhaseKey, ItemKey, Version, PeriodMin, PeriodMax}`
+Generates expected `ChildUniqueReferenceID` strings for a range of billing periods:
+```go
+matcher := recurringLineMatcher{
+    PhaseKey:  "default",
+    ItemKey:   "api-calls",
+    Version:   0,
+    PeriodMin: 0,
+    PeriodMax: 2,
+}
+// Generates: {subID}/default/api-calls/v[0]/period[0], /period[1], /period[2]
+```
+
+## MockStreamingConnector
+
+```go
+// Set meter values returned by queries
+s.MockStreamingConnector.AddSimpleEvent(meterSlug, value, at)
+
+// Or set a fixed return value for all queries
+s.MockStreamingConnector.SetDefaultMeter(meterSlug, value)
+
+// Reset all values (called in AfterTest)
+s.MockStreamingConnector.Reset()
+```
+
+## MockableInvoiceCalculator
+
+Override the invoice calculator for a single test:
+```go
+s.BillingService.GetInvoiceCalculator().(*invoicecalc.MockableCalculator).
+    SetupMock(func(inv *billing.StandardInvoice) {
+        // modify the invoice directly before totals are calculated
+    })
+defer s.BillingService.GetInvoiceCalculator().(*invoicecalc.MockableCalculator).Reset()
+```
+
+## Clock Control
+
+```go
+clock.SetTime(t1)           // advance clock without freezing
+clock.FreezeTime(t)         // freeze at specific time (t is *testing.T for cleanup)
+clock.UnFreeze()            // manual unfreeze (also called in AfterTest)
+clock.ResetTime()           // reset to wall clock time
+```
+
+Always call `clock.UnFreeze()` in `AfterTest` or use `clock.FreezeTime(t)` which registers cleanup automatically.
+
+## Progressive Billing Test Helpers
+
+```go
+// Enable progressive billing on the billing profile
+s.enableProgressiveBilling()
+
+// Set feature flags on the service directly (bypasses profile)
+s.enableProrating()
+```
+
+## `RemoveMetaForCompare()`
+
+Use before `require.Equal` comparisons to strip DB-only fields:
+```go
+expected.RemoveMetaForCompare()
+actual.RemoveMetaForCompare()
+s.Equal(expected, actual)
+```
+
+Available on both `StandardInvoice` and `StandardLine`. Strips: `DBState`, `DetailedLines`, IDs, timestamps.
+
+## Running Billing Tests
+
+```bash
+# All billing tests (requires postgres)
+POSTGRES_HOST=127.0.0.1 go test -tags=dynamic -v ./openmeter/billing/...
+
+# Just sync algorithm tests
+POSTGRES_HOST=127.0.0.1 go test -tags=dynamic -v ./openmeter/billing/worker/subscriptionsync/service/...
+
+# Just charges tests
+POSTGRES_HOST=127.0.0.1 go test -tags=dynamic -v ./openmeter/billing/charges/...
+
+# Full billing integration tests (test/ package)
+POSTGRES_HOST=127.0.0.1 go test -tags=dynamic -v ./test/billing/...
+```
+
+Skip atlas migrations (faster, uses Ent auto-create):
+```bash
+TEST_DISABLE_ATLAS=true POSTGRES_HOST=127.0.0.1 go test -tags=dynamic -v ./openmeter/billing/...
+```

.agents/skills/billing/references/testing.md

@@ -850,8 +850,15 @@ func (a *adapter) IsAppUsed(ctx context.Context, appID app.AppID) error {
 				billing.StandardInvoiceStatusIssuingChargeBookingFailed,
 				billing.StandardInvoiceStatusIssued,
 				billing.StandardInvoiceStatusPaymentProcessingPending,
+				billing.StandardInvoiceStatusPaymentProcessingBookingAuthorized,
+				billing.StandardInvoiceStatusPaymentProcessingBookingAuthorizedFailed,
+				billing.StandardInvoiceStatusPaymentProcessingBookingAuthorizedAndSettled,
+				billing.StandardInvoiceStatusPaymentProcessingBookingAuthorizedAndSettledFailed,
+				billing.StandardInvoiceStatusPaymentProcessingAuthorized,
 				billing.StandardInvoiceStatusPaymentProcessingFailed,
 				billing.StandardInvoiceStatusPaymentProcessingActionRequired,
+				billing.StandardInvoiceStatusPaymentProcessingBookingSettled,
+				billing.StandardInvoiceStatusPaymentProcessingBookingSettledFailed,
 				billing.StandardInvoiceStatusOverdue,
 			),
 			billinginvoice.DeletedAtIsNil(),

openmeter/billing/adapter/invoice.go

@@ -87,6 +87,14 @@ func (e *Engine) OnInvoiceIssued(_ context.Context, _ billing.OnInvoiceIssuedInp
 	return nil
 }
 
+func (e *Engine) OnPaymentAuthorized(_ context.Context, _ billing.OnPaymentAuthorizedInput) error {
+	return nil
+}
+
+func (e *Engine) OnPaymentSettled(_ context.Context, _ billing.OnPaymentSettledInput) error {
+	return nil
+}
+
 func (e *Engine) CalculateLines(input billing.CalculateLinesInput) (billing.StandardLines, error) {
 	if input.Invoice.ID == "" {
 		return nil, fmt.Errorf("invoice id is required")

openmeter/billing/charges/creditpurchase/lineengine/engine.go

@@ -130,6 +130,14 @@ func (e *Engine) OnInvoiceIssued(_ context.Context, _ billing.OnInvoiceIssuedInp
 	return nil
 }
 
+func (e *Engine) OnPaymentAuthorized(_ context.Context, _ billing.OnPaymentAuthorizedInput) error {
+	return nil
+}
+
+func (e *Engine) OnPaymentSettled(_ context.Context, _ billing.OnPaymentSettledInput) error {
+	return nil
+}
+
 func (e *Engine) CalculateLines(input billing.CalculateLinesInput) (billing.StandardLines, error) {
 	if input.Invoice.ID == "" {
 		return nil, fmt.Errorf("invoice id is required")