Add sunrise and anchor tenant discounts to invoicing pipeline

This officially adds a 15% discount to sunrise creates and makes anchor tenant
creates free for the first 2 years.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=204805141

java/google/registry/beam/BUILD

@@ -10,6 +10,7 @@ java_library(
     resources = glob(["sql/*"]),
     deps = [
         "//java/google/registry/config",
+        "//java/google/registry/model",
         "//java/google/registry/reporting/billing",
         "//java/google/registry/util",
         "@com_google_apis_google_api_services_bigquery",

java/google/registry/beam/BillingEvent.java

@@ -19,11 +19,13 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.flogger.FluentLogger;
+import google.registry.model.billing.BillingEvent.Flag;
 import google.registry.reporting.billing.BillingModule;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.Serializable;
+import java.text.DecimalFormat;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
@@ -52,6 +54,9 @@ public abstract class BillingEvent implements Serializable {
       DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss zzz");
 
 
+  /** The amount we multiply the price for sunrise creates. This is currently a 15% discount. */
+  private static final double SUNRISE_DISCOUNT_PRICE_MODIFIER = 0.85;
+
   private static final ImmutableList<String> FIELD_NAMES =
       ImmutableList.of(
           "id",
@@ -105,6 +110,8 @@ public abstract class BillingEvent implements Serializable {
   static BillingEvent parseFromRecord(SchemaAndRecord schemaAndRecord) {
     checkFieldsNotNull(schemaAndRecord);
     GenericRecord record = schemaAndRecord.getRecord();
+    String flags = extractField(record, "flags");
+    double amount = getDiscountedAmount(Double.parseDouble(extractField(record, "amount")), flags);
     return create(
         // We need to chain parsers off extractField because GenericRecord only returns
         // Objects, which contain a string representation of their underlying types.
@@ -122,8 +129,30 @@ public abstract class BillingEvent implements Serializable {
         extractField(record, "repositoryId"),
         Integer.parseInt(extractField(record, "years")),
         extractField(record, "currency"),
-        Double.parseDouble(extractField(record, "amount")),
-        extractField(record, "flags"));
+        amount,
+        flags);
+  }
+
+  /**
+   * Applies a discount to sunrise creates and anchor tenant creates if applicable.
+   *
+   * Currently sunrise creates are discounted 15% and anchor tenant creates are free for 2 years.
+   * All anchor tenant creates are enforced to be 2 years in
+   * {@link google.registry.flows.domain.DomainCreateFlow#verifyAnchorTenantValidPeriod}.
+   */
+  private static double getDiscountedAmount(double amount, String flags) {
+    // Apply a configurable discount to sunrise creates.
+    if (flags.contains(Flag.SUNRISE.name())) {
+      amount =
+          Double.parseDouble(
+              new DecimalFormat("#.##").format(amount * SUNRISE_DISCOUNT_PRICE_MODIFIER));
+    }
+    // Anchor tenant creates are free for the initial create. This is enforced to be a 2 year period
+    // upon domain create.
+    if (flags.contains(Flag.ANCHOR_TENANT.name())) {
+      amount = 0;
+    }
+    return amount;
   }
 
   /**

javatests/google/registry/beam/BillingEventTest.java

@@ -99,6 +99,22 @@ public class BillingEventTest {
     assertThat(event.flags()).isEqualTo("AUTO_RENEW SYNTHETIC");
   }
 
+  @Test
+  public void testParseBillingEventFromRecord_sunriseCreate_reducedPrice_success() {
+    schemaAndRecord.getRecord().put("flags", "SUNRISE");
+    BillingEvent event = BillingEvent.parseFromRecord(schemaAndRecord);
+    assertThat(event.amount()).isEqualTo(17.43);
+    assertThat(event.flags()).isEqualTo("SUNRISE");
+  }
+
+  @Test
+  public void testParseBillingEventFromRecord_anchorTenant_zeroPrice_success() {
+    schemaAndRecord.getRecord().put("flags", "SUNRISE ANCHOR_TENANT");
+    BillingEvent event = BillingEvent.parseFromRecord(schemaAndRecord);
+    assertThat(event.amount()).isZero();
+    assertThat(event.flags()).isEqualTo("SUNRISE ANCHOR_TENANT");
+  }
+
   @Test
   public void testParseBillingEventFromRecord_nullValue_throwsException() {
     schemaAndRecord.getRecord().put("tld", null);

javatests/google/registry/beam/InvoicingPipelineTest.java

@@ -125,7 +125,21 @@ public class InvoicingPipelineTest {
             1,
             "USD",
             20.5,
-            ""));
+            ""),
+        BillingEvent.create(
+            1,
+            ZonedDateTime.of(2017, 10, 4, 0, 0, 0, 0, ZoneId.of("UTC")),
+            ZonedDateTime.of(2017, 10, 4, 0, 0, 0, 0, ZoneId.of("UTC")),
+            "anotherRegistrar",
+            "789",
+            "test",
+            "CREATE",
+            "mydomain5.test",
+            "REPO-ID",
+            1,
+            "USD",
+            0,
+            "SUNRISE ANCHOR_TENANT"));
   }
 
   /** Returns a map from filename to expected contents for detail reports. */
@@ -144,7 +158,11 @@ public class InvoicingPipelineTest {
         "invoice_details_2017-10_googledomains_test.csv",
         ImmutableList.of(
             "1,2017-10-04 00:00:00 UTC,2017-10-04 00:00:00 UTC,googledomains,456,"
-                + "test,RENEW,mydomain4.test,REPO-ID,1,USD,20.50,"));
+                + "test,RENEW,mydomain4.test,REPO-ID,1,USD,20.50,"),
+        "invoice_details_2017-10_anotherRegistrar_test.csv",
+        ImmutableList.of(
+            "1,2017-10-04 00:00:00 UTC,2017-10-04 00:00:00 UTC,anotherRegistrar,789,"
+                + "test,CREATE,mydomain5.test,REPO-ID,1,USD,0.00,SUNRISE ANCHOR_TENANT"));
   }
 
   private ImmutableList<String> getExpectedInvoiceOutput() {
@@ -154,7 +172,9 @@ public class InvoicingPipelineTest {
         "2017-10-01,2022-09-30,234,70.75,JPY,10125,1,PURCHASE,theRegistrar - hello,1,"
             + "CREATE | TLD: hello | TERM: 5-year,70.75,JPY,",
         "2017-10-01,2018-09-30,456,20.50,USD,10125,1,PURCHASE,googledomains - test,1,"
-            + "RENEW | TLD: test | TERM: 1-year,20.50,USD,");
+            + "RENEW | TLD: test | TERM: 1-year,20.50,USD,",
+        "2017-10-01,2018-09-30,789,0.00,USD,10125,1,PURCHASE,anotherRegistrar - test,1,"
+            + "CREATE | TLD: test | TERM: 1-year,0.00,USD,");
   }
 
   @Test