fix: address review comments

- refactor diff calculation
- use jsonb for usage attribution
- fix README.md typos
- document period

openmeter/billing/README.md

@@ -37,31 +37,31 @@ TODO: document when implemented
 
 ### Invoices
 
-The invoices are goverened by the [invoice state machine](./service/invoicestate.go).
+The invoices are governed by the [invoice state machine](./service/invoicestate.go).
 
 Invoices are composed of [lines](./entity/invoiceline.go). Each invoice can only have lines from the same currency.
 
 The lines can be of different types:
 - ManualFee: one time manually added charge
-- ManualUsageBased: manually added usage-based charge (can be used to charge addition usage-based prices without the product catalog features)
+- ManualUsageBased: manually added usage-based charge (can be used to charge additional usage-based prices without the product catalog features)
 
 Each line has a `period` (`start`, `end`) and an `invoiceAt` property. The period specifies which period of time the line is referring to (in case of usage-based pricing, the underlying meter will be queried for this time-period). `invoiceAt` specifies the time when it is expected to create an invoice that contains this line. The invoice's collection settings can defer this.
 
-Invoices are always created by collecting one or more line from the `gathering` invoices. The `/v1/api/billing/invoices/lines` endpoint can be used to create new future line items. A new invoice can be created any time. In such case the `gathering` items that are to be invoiced (`invoiceAt`) already are added to the invoice. Any usage-based line, that we can bill early is also added to the invoice for the period between the `period.start` of the line and the time of invoice creation.
+Invoices are always created by collecting one or more line from the `gathering` invoices. The `/v1/api/billing/invoices/lines` endpoint can be used to create new future line items. A new invoice can be created any time. In such case, the `gathering` items to be invoiced (`invoiceAt`) are already added to the invoice. Any usage-based line, that we can bill early is also added to the invoice for the period between the `period.start` of the line and the time of invoice creation.
 
 ### Line splitting
 
 To achieve the behavior described above, we are using line splitting. By default we would have one line per billing period that would eventually be part of an invoice:
 
 ```
-                period.start                                              period.end
+ period.start                                              period.end
 Line1 [status=valid] |--------------------------------------------------------|
 ```
 
 When the usage-based line can be billed mid-period, we `split` the line into two:
 
 ```
-                     period.start              asOf                              period.end
+ period.start              asOf                              period.end
 Line1 [status=split]         |--------------------------------------------------------|
 SplitLine1 [status=valid]    |------------------|
 SplitLine2 [status=valid]                       |-------------------------------------|
@@ -75,18 +75,18 @@ As visible:
 When creating a new invoice between `asof` and `period.end` the same logic continues, but without marking SplitLine2 `split`, instead the new line is added to the original line's parent line:
 
 ```
-                     period.start              asOf1          asof2                period.end
+ period.start              asOf1          asof2                period.end
 Line1 [status=split]         |--------------------------------------------------------|
 SplitLine1 [status=valid]    |------------------|
 SplitLine2 [status=valid]                       |---------------|
 SplitLine3 [status=valid]                                       |---------------------|
 ```
 
-This flattening approach allows us to not to have to recusively traverse lines in the database.
+This flattening approach allows us not to have to recursively traverse lines in the database.
 
 ### Usage-based quantity
 
-When a line is created for an invoice, the quantity of the underlying merter is captured into the line's qty field. This information is never updated, so late events will have to create new invoice lines when needed.
+When a line is created for an invoice, the quantity of the underlying meter is captured into the line's qty field. This information is never updated, so late events will have to create new invoice lines when needed.
 
 ### Detailed Lines
 
@@ -116,16 +116,14 @@ Apps can choose to syncronize the original line (if the upstream system understa
 
 TODO: this is TBD as not implemented.
 
-When we are dealing with a split line, the calculation is by taking the meter's quantity for the whole line period ([`parent.period.start`, `splitline.period.end`]) and the splitline's period (`splitline.period.start`, `splitline.period.end`).
+When we are dealing with a split line, the calculation of the quantity is by taking the meter's quantity for the whole line period ([`parent.period.start`, `splitline.period.end`]) and the amount before the period (`splitline.period.start`, `splitline.period.start`).
 
-When substracting the two we get two values:
-- line qty: splitline's period
-- before line usage: whole line period - splitline's period
+When substracting the two we get the delta for the period (this gets the delta for all supported meter types except of Min and Avg).
 
 We execute the pricing logic (e.g. tiered pricing) for the line qty, while considering the before usage, as it reflects the already billed for items.
 
 Corner cases:
-- Graduating tiered prices cannot be billed mid billing period (always arrears)
-- Min, Avg meters are always billed arrears as they are not composable
+- Graduating tiered prices cannot be billed mid-billing period (always arrears, as the calculation cannot be split into multiple items)
+- Min, Avg meters are always billed arrears as we cannot calculate the delta.
 
 

openmeter/billing/adapter/invoice.go

@@ -246,7 +246,10 @@ func (r *adapter) CreateInvoice(ctx context.Context, input billing.CreateInvoice
 		SetNillableDueAt(input.DueAt).
 		SetNillableCustomerTimezone(customer.Timezone).
 		SetNillableIssuedAt(lo.EmptyableToPtr(input.IssuedAt)).
-		SetCustomerSubjectKeys(input.Customer.UsageAttribution.SubjectKeys).
+		SetCustomerUsageAttribution(&billingentity.VersionedCustomerUsageAttribution{
+			Type:                     billingentity.CustomerUsageAttributionTypeVersion,
+			CustomerUsageAttribution: input.Customer.UsageAttribution,
+		}).
 		// Workflow (cloned)
 		SetBillingWorkflowConfigID(clonedWorkflowConfig.ID).
 		// TODO[later]: By cloning the AppIDs here we could support changing the apps in the billing profile if needed
@@ -490,8 +493,8 @@ func mapInvoiceFromDB(invoice db.BillingInvoice, expand billingentity.InvoiceExp
 				Line2:       invoice.CustomerAddressLine2,
 				PhoneNumber: invoice.CustomerAddressPhoneNumber,
 			},
-			Timezone: invoice.CustomerTimezone,
-			Subjects: invoice.CustomerSubjectKeys,
+			Timezone:         invoice.CustomerTimezone,
+			UsageAttribution: invoice.CustomerUsageAttribution.CustomerUsageAttribution,
 		},
 		Period:    mapPeriodFromDB(invoice.PeriodStart, invoice.PeriodEnd),
 		IssuedAt:  invoice.IssuedAt,

openmeter/billing/entity/invoice.go

@@ -9,6 +9,7 @@ import (
 	"github.com/invopop/gobl/bill"
 	"github.com/samber/lo"
 
+	customerentity "github.com/openmeterio/openmeter/openmeter/customer/entity"
 	"github.com/openmeterio/openmeter/pkg/currencyx"
 	"github.com/openmeterio/openmeter/pkg/models"
 	"github.com/openmeterio/openmeter/pkg/timezone"
@@ -215,13 +216,25 @@ type InvoiceStatusDetails struct {
 	AvailableActions []InvoiceAction `json:"availableActions"`
 }
 
+const (
+	CustomerUsageAttributionTypeVersion = "customer_usage_attribution.v1"
+)
+
+type (
+	CustomerUsageAttribution          = customerentity.CustomerUsageAttribution
+	VersionedCustomerUsageAttribution struct {
+		CustomerUsageAttribution `json:",inline"`
+		Type                     string `json:"type"`
+	}
+)
+
 type InvoiceCustomer struct {
 	CustomerID string `json:"customerId,omitempty"`
 
-	Name           string             `json:"name"`
-	BillingAddress *models.Address    `json:"billingAddress,omitempty"`
-	Timezone       *timezone.Timezone `json:"timezone,omitempty"`
-	Subjects       []string           `json:"subjects,omitempty"`
+	Name             string                   `json:"name"`
+	BillingAddress   *models.Address          `json:"billingAddress,omitempty"`
+	Timezone         *timezone.Timezone       `json:"timezone,omitempty"`
+	UsageAttribution CustomerUsageAttribution `json:"usageAttribution"`
 }
 
 func (i *InvoiceCustomer) Validate() error {

openmeter/billing/entity/invoiceline.go

@@ -49,6 +49,8 @@ func (InvoiceLineStatus) Values() []string {
 	}
 }
 
+// Period represents a time period, in billing the time period is always interpreted as
+// [start, end) (i.e. start is inclusive, end is exclusive).
 type Period struct {
 	Start time.Time `json:"start"`
 	End   time.Time `json:"end"`

openmeter/billing/service/lineservice/manualusagebasedline.go

@@ -32,7 +32,7 @@ func (l manualUsageBasedLine) Validate(ctx context.Context, targetInvoice *billi
 		return err
 	}
 
-	if len(targetInvoice.Customer.Subjects) == 0 {
+	if len(targetInvoice.Customer.UsageAttribution.SubjectKeys) == 0 {
 		return billingentity.ValidationError{
 			Err: billingentity.ErrInvoiceCreateUBPLineCustomerHasNoSubjects,
 		}
@@ -114,7 +114,7 @@ func (l manualUsageBasedLine) SnapshotQuantity(ctx context.Context, invoice *bil
 			ParentLine: l.line.ParentLine,
 			Feature:    featureMeter.feature,
 			Meter:      featureMeter.meter,
-			Subjects:   invoice.Customer.Subjects,
+			Subjects:   invoice.Customer.UsageAttribution.SubjectKeys,
 		},
 	)
 	if err != nil {

openmeter/billing/service/lineservice/meters.go

@@ -83,42 +83,61 @@ func (s *Service) getFeatureUsage(ctx context.Context, in getFeatureUsageInput)
 		}
 	}
 
-	meterValues, err := s.StreamingConnector.QueryMeter(
+	// If we are the first line in the split, we don't need to calculate the pre period
+	if in.ParentLine == nil || in.ParentLine.Period.Start.Equal(in.Line.Period.Start) {
+		meterValues, err := s.StreamingConnector.QueryMeter(
+			ctx,
+			in.Line.Namespace,
+			in.Meter.Slug,
+			meterQueryParams,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("querying line[%s] meter[%s]: %w", in.Line.ID, in.Meter.Slug, err)
+		}
+
+		return &featureUsageResponse{
+			LinePeriodQty: summarizeMeterQueryRow(meterValues),
+		}, nil
+	}
+
+	// Let's calculate [parent.start ... line.start] values
+	preLineQuery := meterQueryParams
+	preLineQuery.From = &in.ParentLine.Period.Start
+	preLineQuery.To = &in.Line.Period.Start
+
+	preLineResult, err := s.StreamingConnector.QueryMeter(
 		ctx,
 		in.Line.Namespace,
 		in.Meter.Slug,
 		meterQueryParams,
 	)
 	if err != nil {
-		return nil, fmt.Errorf("querying line[%s] meter[%s]: %w", in.Line.ID, in.Meter.Slug, err)
+		return nil, fmt.Errorf("querying pre line[%s] period meter[%s]: %w", in.ParentLine.ID, in.Meter.Slug, err)
 	}
 
-	res := &featureUsageResponse{
-		LinePeriodQty: summarizeMeterQueryRow(meterValues),
-	}
-
-	// If we are the first line in the split, we don't need to calculate the pre period
-	if in.ParentLine == nil || in.ParentLine.Period.Start.Equal(in.Line.Period.Start) {
-		return res, nil
-	}
+	preLineQty := summarizeMeterQueryRow(preLineResult)
 
-	// Let's get the usage for the parent line to calculate the pre period
-	meterQueryParams.From = &in.ParentLine.Period.Start
+	// Let's calculate [parent.start ... line.end] values
+	upToLineEnd := meterQueryParams
+	upToLineEnd.From = &in.Line.Period.Start
+	upToLineEnd.To = &in.Line.Period.End
 
-	meterValues, err = s.StreamingConnector.QueryMeter(
+	upToLineEndResult, err := s.StreamingConnector.QueryMeter(
 		ctx,
 		in.Line.Namespace,
 		in.Meter.Slug,
-		meterQueryParams,
+		upToLineEnd,
 	)
 	if err != nil {
-		return nil, fmt.Errorf("querying parent line[%s] meter[%s]: %w", in.ParentLine.ID, in.Meter.Slug, err)
+		return nil, fmt.Errorf("querying up to line[%s] end meter[%s]: %w", in.ParentLine.ID, in.Meter.Slug, err)
 	}
 
-	fullPeriodQty := summarizeMeterQueryRow(meterValues)
-	res.PreLinePeriodQty = fullPeriodQty.Sub(res.LinePeriodQty)
+	upToLineQty := summarizeMeterQueryRow(upToLineEndResult)
 
-	return res, nil
+	return &featureUsageResponse{
+		LinePeriodQty:    upToLineQty.Sub(preLineQty),
+		PreLinePeriodQty: preLineQty,
+	}, nil
 }
 
 func summarizeMeterQueryRow(in []models.MeterQueryRow) alpacadecimal.Decimal {