Change use of BillingIdentifier to BillingAccountMap in invoicing pipeline (#1577)

* Change billingIdentifier to BillingAccountMap in invoicing pipeline

* Add a default for billing account map

* Throw error on missing PAK

* Add unit test

core/src/main/java/google/registry/beam/invoicing/InvoicingPipeline.java

@@ -14,6 +14,7 @@
 
 package google.registry.beam.invoicing;
 
+import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static google.registry.beam.BeamUtils.getQueryFromFile;
 import static org.apache.beam.sdk.values.TypeDescriptors.strings;
@@ -53,6 +54,7 @@ import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.TypeDescriptor;
+import org.joda.money.CurrencyUnit;
 
 /**
  * Definition of a Dataflow Flex pipeline template, which generates a given month's invoices.
@@ -122,12 +124,19 @@ public class InvoicingPipeline implements Serializable {
     google.registry.model.billing.BillingEvent.OneTime oneTime =
         (google.registry.model.billing.BillingEvent.OneTime) row[0];
     Registrar registrar = (Registrar) row[1];
+    CurrencyUnit currency = oneTime.getCost().getCurrencyUnit();
+    checkState(
+        registrar.getBillingAccountMap().containsKey(currency),
+        "Registrar %s does not have a product account key for the currency unit: %s",
+        registrar.getRegistrarId(),
+        currency);
+
     return BillingEvent.create(
         oneTime.getId(),
         DateTimeUtils.toZonedDateTime(oneTime.getBillingTime(), ZoneId.of("UTC")),
         DateTimeUtils.toZonedDateTime(oneTime.getEventTime(), ZoneId.of("UTC")),
         registrar.getRegistrarId(),
-        registrar.getBillingIdentifier().toString(),
+        registrar.getBillingAccountMap().get(currency),
         registrar.getPoNumber().orElse(""),
         DomainNameUtils.getTldFromDomainName(oneTime.getTargetId()),
         oneTime.getReason().toString(),

core/src/main/resources/google/registry/beam/invoicing/sql/cloud_sql_billing_events.sql

@@ -31,7 +31,7 @@ JOIN Domain d ON b.domainRepoId = d.repoId
 JOIN Tld t ON t.tldStrId = d.tld
 LEFT JOIN BillingCancellation c ON b.id = c.refOneTime.billingId
 LEFT JOIN BillingCancellation cr ON b.cancellationMatchingBillingEvent = cr.refRecurring.billingId
-WHERE r.billingIdentifier IS NOT NULL
+WHERE r.billingAccountMap IS NOT NULL
 AND r.type = 'REAL'
 AND t.invoicingEnabled IS TRUE
 AND b.billingTime BETWEEN CAST('%FIRST_TIMESTAMP_OF_MONTH%' AS timestamp) AND CAST('%LAST_TIMESTAMP_OF_MONTH%' AS timestamp)

core/src/test/java/google/registry/beam/invoicing/InvoicingPipelineTest.java

@@ -24,6 +24,10 @@ import static google.registry.testing.DatabaseHelper.persistNewRegistrar;
 import static google.registry.testing.DatabaseHelper.persistResource;
 import static google.registry.util.DateTimeUtils.END_OF_TIME;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
+import static org.joda.money.CurrencyUnit.CAD;
+import static org.joda.money.CurrencyUnit.JPY;
+import static org.joda.money.CurrencyUnit.USD;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -55,6 +59,7 @@ import java.time.ZonedDateTime;
 import java.util.Arrays;
 import java.util.Map.Entry;
 import java.util.Optional;
+import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
 import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.testing.PAssert;
@@ -294,6 +299,37 @@ class InvoicingPipelineTest {
     pipeline.run().waitUntilFinish();
   }
 
+  @Test
+  void testFailure_readFromCloudSqlMissingPAK() throws Exception {
+    Registrar registrar = persistNewRegistrar("TheRegistrar");
+    registrar =
+        registrar
+            .asBuilder()
+            .setBillingAccountMap(ImmutableMap.of(USD, "789"))
+            .setPoNumber(Optional.of("22446688"))
+            .build();
+    persistResource(registrar);
+    Registry test =
+        newRegistry("test", "_TEST", ImmutableSortedMap.of(START_OF_TIME, GENERAL_AVAILABILITY))
+            .asBuilder()
+            .setInvoicingEnabled(true)
+            .build();
+    persistResource(test);
+    DomainBase domain = persistActiveDomain("mycanadiandomain.test");
+
+    persistOneTimeBillingEvent(1, domain, registrar, Reason.RENEW, 3, Money.of(CAD, 20.5));
+    PCollection<BillingEvent> billingEvents = InvoicingPipeline.readFromCloudSql(options, pipeline);
+    billingEvents = billingEvents.apply(new ChangeDomainRepo());
+    PAssert.that(billingEvents).empty();
+    PipelineExecutionException thrown =
+        assertThrows(PipelineExecutionException.class, () -> pipeline.run().waitUntilFinish());
+    assertThat(thrown)
+        .hasMessageThat()
+        .contains(
+            "Registrar TheRegistrar does not have a product account key for the currency unit:"
+                + " CAD");
+  }
+
   @Test
   void testSuccess_saveInvoiceCsv() throws Exception {
     InvoicingPipeline.saveInvoiceCsv(billingEvents, options);
@@ -338,7 +374,7 @@ class InvoicingPipelineTest {
                 + "LEFT JOIN BillingCancellation c ON b.id = c.refOneTime.billingId\n"
                 + "LEFT JOIN BillingCancellation cr ON b.cancellationMatchingBillingEvent ="
                 + " cr.refRecurring.billingId\n"
-                + "WHERE r.billingIdentifier IS NOT NULL\n"
+                + "WHERE r.billingAccountMap IS NOT NULL\n"
                 + "AND r.type = 'REAL'\n"
                 + "AND t.invoicingEnabled IS TRUE\n"
                 + "AND b.billingTime BETWEEN CAST('2017-10-01' AS timestamp) AND CAST('2017-11-01'"
@@ -362,18 +398,22 @@ class InvoicingPipelineTest {
     persistNewRegistrar("NewRegistrar");
     persistNewRegistrar("TheRegistrar");
     Registrar registrar1 = persistNewRegistrar("theRegistrar");
-    registrar1 = registrar1.asBuilder().setBillingIdentifier(234L).build();
+    registrar1 =
+        registrar1
+            .asBuilder()
+            .setBillingAccountMap(ImmutableMap.of(JPY, "234", USD, "234"))
+            .build();
     persistResource(registrar1);
     Registrar registrar2 = persistNewRegistrar("bestdomains");
     registrar2 =
         registrar2
             .asBuilder()
-            .setBillingIdentifier(456L)
+            .setBillingAccountMap(ImmutableMap.of(USD, "456"))
             .setPoNumber(Optional.of("116688"))
             .build();
     persistResource(registrar2);
     Registrar registrar3 = persistNewRegistrar("anotherRegistrar");
-    registrar3 = registrar3.asBuilder().setBillingIdentifier(789L).build();
+    registrar3 = registrar3.asBuilder().setBillingAccountMap(ImmutableMap.of(USD, "789")).build();
     persistResource(registrar3);
 
     Registry test =
@@ -397,10 +437,8 @@ class InvoicingPipelineTest {
     DomainBase domain6 = persistActiveDomain("locked.test");
     DomainBase domain7 = persistActiveDomain("update-prohibited.test");
 
-    persistOneTimeBillingEvent(
+    persistOneTimeBillingEvent(1, domain1, registrar1, Reason.RENEW, 3, Money.of(USD, 20.5));
-        1, domain1, registrar1, Reason.RENEW, 3, Money.of(CurrencyUnit.USD, 20.5));
+    persistOneTimeBillingEvent(2, domain2, registrar1, Reason.RENEW, 3, Money.of(USD, 20.5));
-    persistOneTimeBillingEvent(
-        2, domain2, registrar1, Reason.RENEW, 3, Money.of(CurrencyUnit.USD, 20.5));
     persistOneTimeBillingEvent(
         3,
         domain3,
@@ -410,31 +448,27 @@ class InvoicingPipelineTest {
         Money.ofMajor(CurrencyUnit.JPY, 70),
         DateTime.parse("2017-09-29T00:00:00.0Z"),
         DateTime.parse("2017-10-02T00:00:00.0Z"));
-    persistOneTimeBillingEvent(
+    persistOneTimeBillingEvent(4, domain4, registrar2, Reason.RENEW, 1, Money.of(USD, 20.5));
-        4, domain4, registrar2, Reason.RENEW, 1, Money.of(CurrencyUnit.USD, 20.5));
     persistOneTimeBillingEvent(
         5,
         domain5,
         registrar3,
         Reason.CREATE,
         1,
-        Money.of(CurrencyUnit.USD, 0),
+        Money.of(USD, 0),
         DateTime.parse("2017-10-04T00:00:00.0Z"),
         DateTime.parse("2017-10-04T00:00:00.0Z"),
         Flag.SUNRISE,
         Flag.ANCHOR_TENANT);
-    persistOneTimeBillingEvent(
+    persistOneTimeBillingEvent(6, domain6, registrar1, Reason.SERVER_STATUS, 0, Money.of(USD, 0));
-        6, domain6, registrar1, Reason.SERVER_STATUS, 0, Money.of(CurrencyUnit.USD, 0));
+    persistOneTimeBillingEvent(7, domain7, registrar1, Reason.SERVER_STATUS, 0, Money.of(USD, 20));
-    persistOneTimeBillingEvent(
-        7, domain7, registrar1, Reason.SERVER_STATUS, 0, Money.of(CurrencyUnit.USD, 20));
 
     // Add billing event for a non-billable registrar
     Registrar registrar4 = persistNewRegistrar("noBillRegistrar");
-    registrar4 = registrar4.asBuilder().setBillingIdentifier(null).build();
+    registrar4 = registrar4.asBuilder().setBillingAccountMap(null).build();
     persistResource(registrar4);
     DomainBase domain8 = persistActiveDomain("non-billable.test");
-    persistOneTimeBillingEvent(
+    persistOneTimeBillingEvent(8, domain8, registrar4, Reason.RENEW, 3, Money.of(USD, 20.5));
-        8, domain8, registrar4, Reason.RENEW, 3, Money.of(CurrencyUnit.USD, 20.5));
 
     // Add billing event for a non-real registrar
     Registrar registrar5 = persistNewRegistrar("notRealRegistrar");
@@ -442,19 +476,17 @@ class InvoicingPipelineTest {
         registrar5
             .asBuilder()
             .setIanaIdentifier(null)
-            .setBillingIdentifier(456L)
+            .setBillingAccountMap(ImmutableMap.of(USD, "456"))
             .setType(Registrar.Type.OTE)
             .build();
     persistResource(registrar5);
     DomainBase domain9 = persistActiveDomain("not-real.test");
-    persistOneTimeBillingEvent(
+    persistOneTimeBillingEvent(9, domain9, registrar5, Reason.RENEW, 3, Money.of(USD, 20.5));
-        9, domain9, registrar5, Reason.RENEW, 3, Money.of(CurrencyUnit.USD, 20.5));
 
     // Add billing event for a non-invoicing TLD
     createTld("nobill");
     DomainBase domain10 = persistActiveDomain("test.nobill");
-    persistOneTimeBillingEvent(
+    persistOneTimeBillingEvent(10, domain10, registrar1, Reason.RENEW, 3, Money.of(USD, 20.5));
-        10, domain10, registrar1, Reason.RENEW, 3, Money.of(CurrencyUnit.USD, 20.5));
 
     // Add billing event before October 2017
     DomainBase domain11 = persistActiveDomain("july.test");
@@ -471,8 +503,7 @@ class InvoicingPipelineTest {
     // Add a billing event with a corresponding cancellation
     DomainBase domain12 = persistActiveDomain("cancel.test");
     OneTime oneTime =
-        persistOneTimeBillingEvent(
+        persistOneTimeBillingEvent(12, domain12, registrar1, Reason.RENEW, 3, Money.of(USD, 20.5));
-            12, domain12, registrar1, Reason.RENEW, 3, Money.of(CurrencyUnit.USD, 20.5));
     DomainHistory domainHistory = persistDomainHistory(domain12, registrar1);
 
     Cancellation cancellation =
@@ -507,8 +538,7 @@ class InvoicingPipelineTest {
             .build();
     persistResource(recurring);
     OneTime oneTimeRecurring =
-        persistOneTimeBillingEvent(
+        persistOneTimeBillingEvent(13, domain13, registrar1, Reason.RENEW, 3, Money.of(USD, 20.5));
-            13, domain13, registrar1, Reason.RENEW, 3, Money.of(CurrencyUnit.USD, 20.5));
     oneTimeRecurring =
         oneTimeRecurring
             .asBuilder()