feat: subscription sync allow disabling continuous line logic

openmeter/billing/annotations.go

@@ -1,5 +1,14 @@
 package billing
 
 const (
+	// AnnotationSubscriptionSyncIgnore is used to mark a line or hierarchy as ignored in subscription syncing.
+	// Should be used in case there is a breaking change in the subscription synchronization process, preventing billing
+	// from issuing credit notes for the past periods.
 	AnnotationSubscriptionSyncIgnore = "billing.subscription.sync.ignore"
+
+	// AnnotationSubscriptionSyncForceContinuousLines is used to force the creation of continuous subscription item lines.
+	// If the sync process finds a previously existing line with this annotation, and the next line generated will not start at the end of the previously
+	// found line, the sync process will adjust the start of the next line to the end of the previously found line, so that we don't have gaps in the
+	// invoices.
+	AnnotationSubscriptionSyncForceContinuousLines = "billing.subscription.sync.force-continuous-lines"
 )

openmeter/billing/worker/subscription/patch.go

@@ -148,7 +148,7 @@ func (h *Handler) getDeletePatchesForLine(lineOrHierarchy billing.LineOrHierarch
 		}
 
 		// Ignored lines do not take part in syncing so we skip them
-		if ignore, ok := line.Annotations[billing.AnnotationSubscriptionSyncIgnore]; ok && ignore == true {
+		if line.Annotations.GetBool(billing.AnnotationSubscriptionSyncIgnore) {
 			return nil, nil
 		}
 

openmeter/billing/worker/subscription/sync.go

@@ -351,12 +351,12 @@ func (h *Handler) correctPeriodStartForUpcomingLines(ctx context.Context, subscr
 
 		// We are not correcting periods for lines that are already ignored
 		if existingCurrentLine, ok := existingLinesByUniqueID[line.UniqueID]; ok {
-			hasAnnotation, err := h.lineOrHierarchyHasSubscriptionSyncIgnoreAnnotation(existingCurrentLine)
+			syncIgnore, err := h.lineOrHierarchyHasAnnotation(existingCurrentLine, billing.AnnotationSubscriptionSyncIgnore)
 			if err != nil {
 				return nil, fmt.Errorf("checking if line has subscription sync ignore annotation: %w", err)
 			}
 
-			if hasAnnotation {
+			if syncIgnore {
 				continue
 			}
 		}
@@ -375,12 +375,22 @@ func (h *Handler) correctPeriodStartForUpcomingLines(ctx context.Context, subscr
 			continue
 		}
 
-		existingPreviousLineHasAnnotation, err := h.lineOrHierarchyHasSubscriptionSyncIgnoreAnnotation(existingPreviousLine)
+		existingPreviousLineSyncIgnoreAnnotation, err := h.lineOrHierarchyHasAnnotation(existingPreviousLine, billing.AnnotationSubscriptionSyncIgnore)
 		if err != nil {
 			return nil, fmt.Errorf("checking if previous line has subscription sync ignore annotation: %w", err)
 		}
 
-		if !existingPreviousLineHasAnnotation {
+		if !existingPreviousLineSyncIgnoreAnnotation {
+			continue
+		}
+
+		// If the previous line does not have the AnnotationSubscriptionSyncForceContinuousLines annotations, we don't need to perform the period correction
+		existingPreviousLineSyncForceContinuousLinesAnnotation, err := h.lineOrHierarchyHasAnnotation(existingPreviousLine, billing.AnnotationSubscriptionSyncForceContinuousLines)
+		if err != nil {
+			return nil, fmt.Errorf("checking if previous line has subscription sync force continuous lines annotation: %w", err)
+		}
+
+		if !existingPreviousLineSyncForceContinuousLinesAnnotation {
 			continue
 		}
 
@@ -409,60 +419,45 @@ func (h *Handler) correctPeriodStartForUpcomingLines(ctx context.Context, subscr
 	return inScopeLines, nil
 }
 
-func (h *Handler) lineOrHierarchyHasSubscriptionSyncIgnoreAnnotation(lineOrHierarchy billing.LineOrHierarchy) (bool, error) {
+func (h *Handler) lineOrHierarchyHasAnnotation(lineOrHierarchy billing.LineOrHierarchy, annotation string) (bool, error) {
 	switch lineOrHierarchy.Type() {
 	case billing.LineOrHierarchyTypeLine:
 		previousLine, err := lineOrHierarchy.AsLine()
 		if err != nil {
 			return false, fmt.Errorf("getting previous line: %w", err)
 		}
 
-		return h.lineHasSubscriptionSyncIgnoreAnnotation(previousLine), nil
+		return h.lineHasAnnotation(previousLine, annotation), nil
 	case billing.LineOrHierarchyTypeHierarchy:
 		hierarchy, err := lineOrHierarchy.AsHierarchy()
 		if err != nil {
 			return false, fmt.Errorf("getting previous hierarchy: %w", err)
 		}
 
-		return h.hierarchyHasSubscriptionSyncIgnoreAnnotation(hierarchy), nil
+		return h.hierarchyHasAnnotation(hierarchy, annotation), nil
 	default:
 		return false, nil
 	}
 }
 
-func (h *Handler) lineHasSubscriptionSyncIgnoreAnnotation(line *billing.Line) bool {
+func (h *Handler) lineHasAnnotation(line *billing.Line, annotation string) bool {
 	if line.ManagedBy != billing.SubscriptionManagedLine {
 		// We only correct the period start for subscription managed lines, for manual edits
 		// we should not apply this logic, as the user might have created a setup where the period start
 		// is no longer valid.
 		return false
 	}
 
-	if line.Annotations == nil {
-		// If the previous line is not annotated to be frozen we should not correct the period start
-		return false
-	}
-
-	val, ok := line.Annotations[billing.AnnotationSubscriptionSyncIgnore]
-	if !ok {
-		return false
-	}
-
-	boolVal, ok := val.(bool)
-	if !ok {
-		return false
-	}
-
-	return boolVal
+	return line.Annotations.GetBool(annotation)
 }
 
-func (h *Handler) hierarchyHasSubscriptionSyncIgnoreAnnotation(hierarchy *billing.SplitLineHierarchy) bool {
+func (h *Handler) hierarchyHasAnnotation(hierarchy *billing.SplitLineHierarchy, annotation string) bool {
 	servicePeriod := hierarchy.Group.ServicePeriod
 
 	// The correction can only happen if the last line the progressively billed group is in scope for the period correction
 	for _, line := range hierarchy.Lines {
 		if line.Line.Period.End.Equal(servicePeriod.End) {
-			return h.lineHasSubscriptionSyncIgnoreAnnotation(line.Line)
+			return h.lineHasAnnotation(line.Line, annotation)
 		}
 	}
 
@@ -760,11 +755,11 @@ 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 {
+	if expectedLine.Annotations.GetBool(billing.AnnotationSubscriptionSyncIgnore) {
 		return nil, nil
 	}
 
-	if ignore, ok := existingLine.Annotations[billing.AnnotationSubscriptionSyncIgnore]; ok && ignore == true {
+	if existingLine.Annotations.GetBool(billing.AnnotationSubscriptionSyncIgnore) {
 		return nil, nil
 	}
 

openmeter/billing/worker/subscription/sync_test.go

@@ -3499,7 +3499,8 @@ func (s *SubscriptionHandlerTestSuite) TestManualIgnoringOfSyncedLines() {
 			line := s.getLineByChildID(*invoice, draftLineReferenceID)
 
 			line.Annotations = models.Annotations{
-				billing.AnnotationSubscriptionSyncIgnore: true,
+				billing.AnnotationSubscriptionSyncIgnore:               true,
+				billing.AnnotationSubscriptionSyncForceContinuousLines: true,
 			}
 
 			return nil
@@ -3515,7 +3516,8 @@ func (s *SubscriptionHandlerTestSuite) TestManualIgnoringOfSyncedLines() {
 			line := s.getLineByChildID(*invoice, gatheringLineReferenceID)
 
 			line.Annotations = models.Annotations{
-				billing.AnnotationSubscriptionSyncIgnore: true,
+				billing.AnnotationSubscriptionSyncIgnore:               true,
+				billing.AnnotationSubscriptionSyncForceContinuousLines: true,
 			}
 
 			gatheringInvoiceIgnoredLine = line.Clone()
@@ -3558,7 +3560,8 @@ func (s *SubscriptionHandlerTestSuite) TestManualIgnoringOfSyncedLines() {
 	expectedInvoice.Lines = expectedInvoice.Lines.Map(func(line *billing.Line) *billing.Line {
 		if line.ChildUniqueReferenceID != nil && *line.ChildUniqueReferenceID == draftLineReferenceID {
 			line.Annotations = models.Annotations{
-				billing.AnnotationSubscriptionSyncIgnore: true,
+				billing.AnnotationSubscriptionSyncIgnore:               true,
+				billing.AnnotationSubscriptionSyncForceContinuousLines: true,
 			}
 		}
 
@@ -3673,7 +3676,8 @@ func (s *SubscriptionHandlerTestSuite) TestManualIgnoringOfSyncedLinesWhenPeriod
 			line := s.getLineByChildID(*invoice, markedLineReferenceID)
 
 			line.Annotations = models.Annotations{
-				billing.AnnotationSubscriptionSyncIgnore: true,
+				billing.AnnotationSubscriptionSyncIgnore:               true,
+				billing.AnnotationSubscriptionSyncForceContinuousLines: true,
 			}
 
 			return nil
@@ -4514,7 +4518,8 @@ func (s *SubscriptionHandlerTestSuite) TestSyncronizeSubscriptionPeriodAlgorithm
 			line.Period.Start = s.mustParseTime("2025-01-31T00:00:00Z")
 			line.Period.End = s.mustParseTime("2025-03-02T00:00:00Z")
 			line.Annotations = models.Annotations{
-				billing.AnnotationSubscriptionSyncIgnore: true,
+				billing.AnnotationSubscriptionSyncIgnore:               true,
+				billing.AnnotationSubscriptionSyncForceContinuousLines: true,
 			}
 
 			invoice.Lines = billing.NewLineChildren([]*billing.Line{

openmeter/customer/adapter/entitymapping.go

@@ -29,7 +29,7 @@ func CustomerFromDBEntity(e db.Customer) (*customer.Customer, error) {
 	var annotations *models.Annotations
 
 	if len(e.Annotations) > 0 {
-		annotations = lo.ToPtr(e.Annotations)
+		annotations = &e.Annotations
 	}
 
 	result := &customer.Customer{