Skip to content

Commit 840a665

Browse files
turipturip
committed
feat: add bulk invoice schema migration function
1 parent f1eac14 commit 840a665

5 files changed

Lines changed: 185 additions & 2 deletions

File tree

‎openmeter/billing/adapter/schemamigration.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ func (a *adapter) migrateCustomerInvoices(ctx context.Context, customerID custom
6969
func (a *adapter) migrateSchemaLevel1(ctx context.Context, customerID customer.CustomerID) error {
7070
return entutils.TransactingRepoWithNoValue(ctx, a, func(ctx context.Context, tx *adapter) error {
7171
// Schema level 1 -> 2 migration is implemented as a DB function (see migrations).
72-
rows, err := tx.db.QueryContext(ctx, `SELECT om_func_migrate_customer_invoices_to_schema_level_2($1)`, customerID.ID)
72+
rows, err := tx.db.QueryContext(ctx, `SELECT om_func_migrate_customer_invoices_to_schema_level_2_bulk(ARRAY[$1]::TEXT[])`, customerID.ID)
7373
if err != nil {
7474
return err
7575
}

‎test/billing/schemamigration_test.go‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ func (s *SchemaMigrationTestSuite) TestSchemaLevel1Migration() {
5252
var (
5353
customerEntity *customer.Customer
5454
invoiceID billing.InvoiceID
55+
gatheringID billing.InvoiceID
5556

5657
deletedAtSet time.Time
5758

@@ -210,13 +211,51 @@ func (s *SchemaMigrationTestSuite) TestSchemaLevel1Migration() {
210211
s.GreaterOrEqual(len(lineActive.DetailedLines[0].AmountDiscounts), 1)
211212
})
212213

214+
s.Run("Given a schema level 1 gathering invoice exists", func() {
215+
periodStart := time.Now().Add(-time.Hour)
216+
periodEnd := time.Now().Add(time.Hour)
217+
218+
result, err := s.BillingService.CreatePendingInvoiceLines(ctx, billing.CreatePendingInvoiceLinesInput{
219+
Customer: customerEntity.GetID(),
220+
Currency: currencyx.Code(currency.USD),
221+
Lines: []billing.GatheringLine{
222+
{
223+
GatheringLineBase: billing.GatheringLineBase{
224+
ManagedResource: models.NewManagedResource(models.ManagedResourceInput{
225+
Namespace: namespace,
226+
Name: "Gathering item",
227+
}),
228+
ServicePeriod: timeutil.ClosedPeriod{From: periodStart, To: periodEnd},
229+
InvoiceAt: periodEnd,
230+
ManagedBy: billing.ManuallyManagedLine,
231+
Currency: currencyx.Code(currency.USD),
232+
FeatureKey: featureFlatPerUnit.Key,
233+
Price: lo.FromPtr(productcatalog.NewPriceFrom(productcatalog.FlatPrice{
234+
Amount: alpacadecimal.NewFromFloat(100),
235+
})),
236+
},
237+
},
238+
},
239+
})
240+
s.Require().NoError(err)
241+
242+
gatheringID = result.Invoice.GetInvoiceID()
243+
gatheringInvoice, err := s.DBClient.BillingInvoice.Get(ctx, gatheringID.ID)
244+
s.Require().NoError(err)
245+
s.Require().Equal(1, gatheringInvoice.SchemaLevel)
246+
})
247+
213248
s.Run("When the write schema level is set to 2 and a lock is obtained on the customer", func() {
214249
s.NoError(s.BillingAdapter.SetInvoiceDefaultSchemaLevel(ctx, 2))
215250
// Side-effect: migration happens due to the previous line.
216251
s.NoError(s.BillingAdapter.LockCustomerForUpdate(ctx, customerEntity.GetID()))
217252
})
218253

219254
s.Run("Then the invoice is migrated and lines (incl detailed lines) match exactly", func() {
255+
gatheringInvoice, err := s.DBClient.BillingInvoice.Get(ctx, gatheringID.ID)
256+
s.Require().NoError(err)
257+
s.Require().Equal(2, gatheringInvoice.SchemaLevel)
258+
220259
invoiceAfter, err := s.BillingAdapter.GetStandardInvoiceById(ctx, billing.GetStandardInvoiceByIdInput{
221260
Invoice: invoiceID,
222261
Expand: billing.StandardInvoiceExpands{

‎tools/migrate/migrations/20260717060208_add_bulk_invoice_schema_level_2_migration_function.down.sql‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
DROP FUNCTION IF EXISTS om_func_migrate_customer_invoices_to_schema_level_2_bulk(TEXT[]);

‎tools/migrate/migrations/20260717060208_add_bulk_invoice_schema_level_2_migration_function.up.sql‎

Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
-- Migrates schema-level-1 invoices for an arbitrary set of customers to schema level 2.
2+
--
3+
-- Gathering invoice details are intentionally excluded: they are invalid legacy data and are
4+
-- removed by the preceding gathering-detail cleanup migration.
5+
CREATE OR REPLACE FUNCTION om_func_migrate_customer_invoices_to_schema_level_2_bulk(p_customer_ids TEXT[])
6+
RETURNS BIGINT
7+
AS $$
8+
DECLARE
9+
v_updated_invoices BIGINT;
10+
BEGIN
11+
INSERT INTO billing_standard_invoice_detailed_lines (
12+
id,
13+
namespace,
14+
created_at,
15+
updated_at,
16+
deleted_at,
17+
name,
18+
description,
19+
currency,
20+
amount,
21+
taxes_total,
22+
taxes_inclusive_total,
23+
taxes_exclusive_total,
24+
charges_total,
25+
discounts_total,
26+
total,
27+
service_period_start,
28+
service_period_end,
29+
quantity,
30+
invoicing_app_external_id,
31+
child_unique_reference_id,
32+
per_unit_amount,
33+
category,
34+
payment_term,
35+
index,
36+
invoice_id,
37+
parent_line_id,
38+
credits_total,
39+
credits_applied
40+
)
41+
SELECT
42+
l.id,
43+
l.namespace,
44+
l.created_at,
45+
l.updated_at,
46+
l.deleted_at,
47+
l.name,
48+
l.description,
49+
l.currency,
50+
l.amount,
51+
l.taxes_total,
52+
l.taxes_inclusive_total,
53+
l.taxes_exclusive_total,
54+
l.charges_total,
55+
l.discounts_total,
56+
l.total,
57+
l.period_start AS service_period_start,
58+
l.period_end AS service_period_end,
59+
l.quantity,
60+
l.invoicing_app_external_id,
61+
l.child_unique_reference_id,
62+
c.per_unit_amount,
63+
c.category,
64+
c.payment_term,
65+
c.index,
66+
l.invoice_id,
67+
l.parent_line_id,
68+
l.credits_total,
69+
l.credits_applied
70+
FROM billing_invoices i
71+
JOIN billing_invoice_lines l
72+
ON l.invoice_id = i.id
73+
AND l.namespace = i.namespace
74+
JOIN billing_invoice_flat_fee_line_configs c
75+
ON c.id = l.fee_line_config_id
76+
AND c.namespace = l.namespace
77+
WHERE
78+
i.customer_id = ANY(p_customer_ids)
79+
AND i.schema_level = 1
80+
AND i.status <> 'gathering'
81+
AND l.status = 'detailed'
82+
AND l.type = 'flat_fee'
83+
ON CONFLICT (id) DO NOTHING;
84+
85+
INSERT INTO billing_standard_invoice_detailed_line_amount_discounts (
86+
id,
87+
namespace,
88+
created_at,
89+
updated_at,
90+
deleted_at,
91+
child_unique_reference_id,
92+
description,
93+
reason,
94+
invoicing_app_external_id,
95+
amount,
96+
rounding_amount,
97+
source_discount,
98+
line_id
99+
)
100+
SELECT
101+
d.id,
102+
d.namespace,
103+
d.created_at,
104+
d.updated_at,
105+
d.deleted_at,
106+
d.child_unique_reference_id,
107+
d.description,
108+
d.reason,
109+
d.invoicing_app_external_id,
110+
d.amount,
111+
d.rounding_amount,
112+
d.source_discount,
113+
d.line_id
114+
FROM billing_invoices i
115+
JOIN billing_invoice_lines l
116+
ON l.invoice_id = i.id
117+
AND l.namespace = i.namespace
118+
JOIN billing_standard_invoice_detailed_lines migrated_l
119+
ON migrated_l.id = l.id
120+
AND migrated_l.namespace = l.namespace
121+
JOIN billing_invoice_line_discounts d
122+
ON d.line_id = l.id
123+
AND d.namespace = l.namespace
124+
WHERE
125+
i.customer_id = ANY(p_customer_ids)
126+
AND i.schema_level = 1
127+
AND i.status <> 'gathering'
128+
AND l.status = 'detailed'
129+
AND l.type = 'flat_fee'
130+
ON CONFLICT (id) DO NOTHING;
131+
132+
UPDATE billing_invoices
133+
SET schema_level = 2
134+
WHERE
135+
customer_id = ANY(p_customer_ids)
136+
AND schema_level = 1;
137+
138+
GET DIAGNOSTICS v_updated_invoices = ROW_COUNT;
139+
140+
RETURN v_updated_invoices;
141+
END;
142+
$$ LANGUAGE plpgsql VOLATILE;

‎tools/migrate/migrations/atlas.sum‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
h1:sHp6EpaiKRdL9B9trZ/xwvONr0fdPUZGrOmNafRNcYQ=
1+
h1:ACEAs9ku7RQcRBVm3VP14IhdgxZ8ByODYwzo0ebdwrc=
22
20240826120919_init.up.sql h1:tc1V91/smlmaeJGQ8h+MzTEeFjjnrrFDbDAjOYJK91o=
33
20240903155435_entitlement-expired-index.up.sql h1:Hp8u5uckmLXc1cRvWU0AtVnnK8ShlpzZNp8pbiJLhac=
44
20240917172257_billing-entities.up.sql h1:Q1dAMo0Vjiit76OybClNfYPGC5nmvov2/M2W1ioi4Kw=
@@ -237,3 +237,4 @@ h1:sHp6EpaiKRdL9B9trZ/xwvONr0fdPUZGrOmNafRNcYQ=
237237
20260716150455_cleanup_gathering_invoice_details.up.sql h1:W4u33tHXJy+GfryC8j46bHi2SiQwDk6brBOex/fQ4rw=
238238
20260716150456_drop_gathering_detail_cleanup_indexes.up.sql h1:IJ6P54dqEy/LFG9hID2T9yBoN29FRTH/iLTSSTkdpDQ=
239239
20260716152332_rescope-credit-grant-key-to-customer.up.sql h1:gbeATYlchOi0pwaTUFX+nrbLoS0jZbVDMHn8uMTck0c=
240+
20260717060208_add_bulk_invoice_schema_level_2_migration_function.up.sql h1:G5G6jYndCh8yx903RzDzVzrStmPDpdNhLiziNBT5ul0=

0 commit comments

Comments
 (0)