feat(billing): ignore sync annotation

openmeter/billing/adapter/invoicelinemapper.go

@@ -137,10 +137,11 @@ func (a *adapter) mapInvoiceLineWithoutReferences(dbLine *db.BillingInvoiceLine)
 			UpdatedAt: dbLine.UpdatedAt.In(time.UTC),
 			DeletedAt: convert.TimePtrIn(dbLine.DeletedAt, time.UTC),
 
-			Metadata:  dbLine.Metadata,
-			InvoiceID: dbLine.InvoiceID,
-			Status:    dbLine.Status,
-			ManagedBy: dbLine.ManagedBy,
+			Metadata:    dbLine.Metadata,
+			Annotations: dbLine.Annotations,
+			InvoiceID:   dbLine.InvoiceID,
+			Status:      dbLine.Status,
+			ManagedBy:   dbLine.ManagedBy,
 
 			Period: billing.Period{
 				Start: dbLine.PeriodStart.In(time.UTC),

openmeter/billing/adapter/invoicelines.go

@@ -87,6 +87,7 @@ func (a *adapter) UpsertInvoiceLines(ctx context.Context, inputIn billing.Upsert
 					SetNillableDescription(line.Description).
 					SetCurrency(line.Currency).
 					SetMetadata(line.Metadata).
+					SetAnnotations(line.Annotations).
 					SetNillableChildUniqueReferenceID(line.ChildUniqueReferenceID).
 					// Totals
 					SetAmount(line.Totals.Amount).

openmeter/billing/annotations.go

@@ -0,0 +1,5 @@
+package billing
+
+const (
+	AnnotationSubscriptionSyncIgnore = "billing.subscription.sync.ignore"
+)

openmeter/billing/invoiceline.go

@@ -132,6 +132,7 @@ type LineBase struct {
 	DeletedAt *time.Time `json:"deletedAt,omitempty"`
 
 	Metadata    map[string]string    `json:"metadata"`
+	Annotations models.Annotations   `json:"annotations"`
 	Name        string               `json:"name"`
 	Type        InvoiceLineType      `json:"type"`
 	ManagedBy   InvoiceLineManagedBy `json:"managedBy"`
@@ -213,6 +214,13 @@ func (i LineBase) Clone() LineBase {
 		}
 	}
 
+	if i.Annotations != nil {
+		out.Annotations = make(models.Annotations, len(i.Annotations))
+		for k, v := range i.Annotations {
+			out.Annotations[k] = v
+		}
+	}
+
 	if i.TaxConfig != nil {
 		tc := *i.TaxConfig
 		out.TaxConfig = &tc
@@ -591,6 +599,7 @@ type NewFlatFeeLineInput struct {
 
 	Name        string
 	Metadata    map[string]string
+	Annotations models.Annotations
 	Description *string
 
 	Currency currencyx.Code
@@ -617,6 +626,7 @@ func NewFlatFeeLine(input NewFlatFeeLineInput) *Line {
 
 			Name:        input.Name,
 			Metadata:    input.Metadata,
+			Annotations: input.Annotations,
 			Description: input.Description,
 
 			Status: InvoiceLineStatusValid,
@@ -673,6 +683,7 @@ func NewUsageBasedFlatFeeLine(input NewFlatFeeLineInput, opts ...usageBasedLineO
 
 			Name:        input.Name,
 			Metadata:    input.Metadata,
+			Annotations: input.Annotations,
 			Description: input.Description,
 
 			Status: InvoiceLineStatusValid,
@@ -1102,11 +1113,12 @@ func (u UpdateInvoiceLineInput) Apply(l *Line) (*Line, error) {
 type UpdateInvoiceLineBaseInput struct {
 	InvoiceAt mo.Option[time.Time]
 
-	Metadata  mo.Option[map[string]string]
-	Name      mo.Option[string]
-	ManagedBy mo.Option[InvoiceLineManagedBy]
-	Period    mo.Option[Period]
-	TaxConfig mo.Option[*productcatalog.TaxConfig]
+	Metadata    mo.Option[map[string]string]
+	Annotations mo.Option[models.Annotations]
+	Name        mo.Option[string]
+	ManagedBy   mo.Option[InvoiceLineManagedBy]
+	Period      mo.Option[Period]
+	TaxConfig   mo.Option[*productcatalog.TaxConfig]
 }
 
 func (u UpdateInvoiceLineBaseInput) Validate() error {
@@ -1154,6 +1166,10 @@ func (u UpdateInvoiceLineBaseInput) Apply(l *Line) error {
 		l.Metadata = u.Metadata.OrEmpty()
 	}
 
+	if u.Annotations.IsPresent() {
+		l.Annotations = u.Annotations.OrEmpty()
+	}
+
 	if u.Name.IsPresent() {
 		l.Name = u.Name.OrEmpty()
 	}

openmeter/billing/worker/subscription/patch.go

@@ -147,6 +147,11 @@ func (h *Handler) getDeletePatchesForLine(lineOrHierarchy billing.LineOrHierarch
 			return nil, fmt.Errorf("getting line: %w", err)
 		}
 
+		// Ignored lines do not take part in syncing so we skip them
+		if ignore, ok := line.Annotations[billing.AnnotationSubscriptionSyncIgnore]; ok && ignore == true {
+			return nil, nil
+		}
+
 		return []linePatch{
 			newDeleteLinePatch(line.LineID(), line.InvoiceID),
 		}, nil

openmeter/billing/worker/subscription/phaseiterator.go

@@ -349,7 +349,15 @@ func (it *PhaseIterator) generateForAlignedItemVersionPeriod(ctx context.Context
 			return empty, err
 		}
 
-		servicePeriod, err := fullServicePeriod.Open().Intersection(item.SubscriptionItem.CadencedModel.AsPeriod()).Closed()
+		inter := fullServicePeriod.Open().Intersection(item.SubscriptionItem.CadencedModel.AsPeriod())
+
+		// .Intersection() treats zero length periods as non-intersecting (to be consistent with .Contains() calls)
+		// We need to handle this case separately
+		if cl, err := item.SubscriptionItem.CadencedModel.AsPeriod().Closed(); err == nil && cl.From.Equal(cl.To) {
+			inter = lo.ToPtr(cl.Open())
+		}
+
+		servicePeriod, err := inter.Closed()
 		if err != nil {
 			logger.ErrorContext(ctx, "failed to get service period", slog.Any("error", err))
 			return empty, err

openmeter/billing/worker/subscription/sync.go

@@ -629,6 +629,15 @@ func (h *Handler) getPatchesForExistingLineOrHierarchy(existingLine billing.Line
 }
 
 func (h *Handler) getPatchesForExistingLine(existingLine *billing.Line, expectedLine *billing.Line, invoiceByID InvoiceByID) ([]linePatch, error) {
+	// Lines can be manually marked as ignored in syncing, which is used for cases where we're doing backwards incompatible changes
+	if ignore, ok := expectedLine.Annotations[billing.AnnotationSubscriptionSyncIgnore]; ok && ignore == true {
+		return nil, nil
+	}
+
+	if ignore, ok := existingLine.Annotations[billing.AnnotationSubscriptionSyncIgnore]; ok && ignore == true {
+		return nil, nil
+	}
+
 	// Manual edits prevent resyncronization so that we preserve the user intent
 	if existingLine.ManagedBy != billing.SubscriptionManagedLine {
 		return nil, nil