Verify billing recurrence expansion is caught up before invoicing (#3121)

core/src/main/java/google/registry/reporting/billing/GenerateInvoicesAction.java

@@ -14,10 +14,14 @@
 
 package google.registry.reporting.billing;
 
+import static com.google.common.base.Preconditions.checkState;
 import static google.registry.beam.BeamUtils.createJobName;
+import static google.registry.model.common.Cursor.CursorType.RECURRING_BILLING;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.POST;
 import static jakarta.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static jakarta.servlet.http.HttpServletResponse.SC_OK;
+import static java.time.ZoneOffset.UTC;
 
 import com.google.api.services.dataflow.Dataflow;
 import com.google.api.services.dataflow.model.LaunchFlexTemplateParameter;
@@ -29,6 +33,7 @@ import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
 import google.registry.batch.CloudTasksUtils;
 import google.registry.config.RegistryConfig.Config;
+import google.registry.model.common.Cursor;
 import google.registry.persistence.PersistenceModule;
 import google.registry.reporting.ReportingModule;
 import google.registry.request.Action;
@@ -40,7 +45,9 @@ import google.registry.util.RegistryEnvironment;
 import jakarta.inject.Inject;
 import java.io.IOException;
 import java.time.Duration;
+import java.time.Instant;
 import java.time.YearMonth;
+import java.util.Optional;
 
 /**
  * Invokes the {@code InvoicingPipeline} beam template via the REST api, and enqueues the {@link
@@ -107,6 +114,7 @@ public class GenerateInvoicesAction implements Runnable {
     response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
     logger.atInfo().log("Launching invoicing pipeline for %s.", yearMonth);
     try {
+      checkBillingRecurrenceCursor();
       LaunchFlexTemplateParameter parameter =
           new LaunchFlexTemplateParameter()
               .setJobName(createJobName("invoicing", clock))
@@ -156,4 +164,20 @@ public class GenerateInvoicesAction implements Runnable {
       response.setPayload(String.format("Pipeline launch failed: %s", e.getMessage()));
     }
   }
+
+  private void checkBillingRecurrenceCursor() {
+    Optional<Cursor> previousCursor =
+        tm().transact(() -> tm().loadByKeyIfPresent(Cursor.createGlobalVKey(RECURRING_BILLING)));
+    checkState(
+        previousCursor.isPresent(),
+        "BillingRecurrence expansion cursor is not present. Run ExpandBillingRecurrencesAction.");
+    Instant startOfNextMonth = yearMonth.plusMonths(1).atDay(1).atStartOfDay(UTC).toInstant();
+    Instant previousCursorTime = previousCursor.get().getCursorTime();
+    checkState(
+        !previousCursorTime.isBefore(startOfNextMonth),
+        "BillingRecurrence expansion cursor (%s) is before the start of the next month (%s). "
+            + "Run ExpandBillingRecurrencesAction.",
+        previousCursorTime,
+        startOfNextMonth);
+  }
 }

core/src/test/java/google/registry/reporting/billing/GenerateInvoicesActionTest.java

@@ -15,8 +15,11 @@
 package google.registry.reporting.billing;
 
 import static com.google.common.truth.Truth.assertThat;
+import static google.registry.model.common.Cursor.CursorType.RECURRING_BILLING;
+import static google.registry.testing.DatabaseHelper.persistResource;
 import static jakarta.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static jakarta.servlet.http.HttpServletResponse.SC_OK;
+import static org.mockito.ArgumentMatchers.startsWith;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -25,6 +28,7 @@ import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.net.MediaType;
 import google.registry.batch.CloudTasksUtils;
 import google.registry.beam.BeamActionTestBase;
+import google.registry.model.common.Cursor;
 import google.registry.persistence.transaction.JpaTestExtensions;
 import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationTestExtension;
 import google.registry.reporting.ReportingModule;
@@ -33,6 +37,7 @@ import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.FakeClock;
 import java.io.IOException;
 import java.time.Duration;
+import java.time.Instant;
 import java.time.YearMonth;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -50,8 +55,13 @@ class GenerateInvoicesActionTest extends BeamActionTestBase {
   private CloudTasksUtils cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
   private GenerateInvoicesAction action;
 
+  private void setCursor(Instant cursorTime) {
+    persistResource(Cursor.createGlobal(RECURRING_BILLING, cursorTime));
+  }
+
   @Test
   void testLaunchTemplateJob_withPublish() throws Exception {
+    setCursor(Instant.parse("2017-11-01T00:00:00Z"));
     action =
         new GenerateInvoicesAction(
             "test-project",
@@ -84,6 +94,7 @@ class GenerateInvoicesActionTest extends BeamActionTestBase {
 
   @Test
   void testLaunchTemplateJob_withoutPublish() throws Exception {
+    setCursor(Instant.parse("2017-11-01T00:00:00Z"));
     action =
         new GenerateInvoicesAction(
             "test-project",
@@ -107,6 +118,7 @@ class GenerateInvoicesActionTest extends BeamActionTestBase {
 
   @Test
   void testCaughtIOException() throws IOException {
+    setCursor(Instant.parse("2017-11-01T00:00:00Z"));
     when(launch.execute()).thenThrow(new IOException("Pipeline error"));
     action =
         new GenerateInvoicesAction(
@@ -128,4 +140,58 @@ class GenerateInvoicesActionTest extends BeamActionTestBase {
     verify(emailUtils).sendAlertEmail("Pipeline Launch failed due to Pipeline error");
     cloudTasksHelper.assertNoTasksEnqueued("beam-reporting");
   }
+
+  @Test
+  void testFailure_cursorLagging() {
+    setCursor(Instant.parse("2017-10-31T23:59:59.999Z"));
+    action =
+        new GenerateInvoicesAction(
+            "test-project",
+            "test-region",
+            "staging_bucket",
+            "billing_bucket",
+            "REG-INV",
+            false,
+            YearMonth.of(2017, 10),
+            emailUtils,
+            cloudTasksUtils,
+            clock,
+            response,
+            dataflow);
+    action.run();
+    assertThat(response.getStatus()).isEqualTo(SC_INTERNAL_SERVER_ERROR);
+    assertThat(response.getPayload()).contains("Pipeline launch failed");
+    assertThat(response.getPayload()).contains("BillingRecurrence expansion cursor");
+    verify(emailUtils)
+        .sendAlertEmail(
+            startsWith("Pipeline Launch failed due to BillingRecurrence expansion cursor"));
+    cloudTasksHelper.assertNoTasksEnqueued("beam-reporting");
+  }
+
+  @Test
+  void testFailure_cursorMissing() {
+    // Do not set cursor, should default to START_INSTANT (1970)
+    action =
+        new GenerateInvoicesAction(
+            "test-project",
+            "test-region",
+            "staging_bucket",
+            "billing_bucket",
+            "REG-INV",
+            false,
+            YearMonth.of(2017, 10),
+            emailUtils,
+            cloudTasksUtils,
+            clock,
+            response,
+            dataflow);
+    action.run();
+    assertThat(response.getStatus()).isEqualTo(SC_INTERNAL_SERVER_ERROR);
+    assertThat(response.getPayload()).contains("Pipeline launch failed");
+    assertThat(response.getPayload()).contains("BillingRecurrence expansion cursor");
+    verify(emailUtils)
+        .sendAlertEmail(
+            startsWith("Pipeline Launch failed due to BillingRecurrence expansion cursor"));
+    cloudTasksHelper.assertNoTasksEnqueued("beam-reporting");
+  }
 }