feat: Create tee requisition fetcher

src/main/k8s/testing/secretfiles/BUILD.bazel

@@ -15,6 +15,7 @@ package(
         "//src/test/kotlin/org/wfanet/measurement/kingdom/batch:__subpackages__",
         "//src/test/kotlin/org/wfanet/measurement/loadtest:__subpackages__",
         "//src/test/kotlin/org/wfanet/measurement/reporting:__subpackages__",
+        "//src/test/kotlin/org/wfanet/measurement/securecomputation/requisitions:__subpackages__",
         "//src/test/kotlin/org/wfanet/panelmatch/integration:__subpackages__",
     ],
 )

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/BUILD.bazel

@@ -0,0 +1,35 @@
+load("@rules_java//java:defs.bzl", "java_binary")
+load("@wfa_rules_kotlin_jvm//kotlin:defs.bzl", "kt_jvm_library")
+
+package(default_visibility = ["//visibility:public"])
+
+kt_jvm_library(
+    name = "requisition_fetcher",
+    srcs = ["RequisitionFetcher.kt"],
+    deps = [
+        "//src/main/proto/wfa/measurement/api/v2alpha:requisitions_service_kt_jvm_grpc_proto",
+        "@wfa_common_jvm//imports/java/com/google/cloud/functions:functions_framework_api",
+        "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common/grpc",
+        "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/gcloud/gcs",
+    ],
+)
+
+java_binary(
+    name = "RequisitionFetcher",
+    main_class = "org.wfanet.measurement.securecomputation.requisitions.RequisitionFetcher",
+    runtime_deps = [":requisition_fetcher"],
+)
+
+kt_jvm_library(
+    name = "requisition_fetcher_runner",
+    srcs = ["RequisitionFetcherRunner.kt"],
+    deps = [
+        ":requisition_fetcher",
+    ],
+)
+
+java_binary(
+    name = "RequisitionFetcherRunner",
+    main_class = "org.wfanet.measurement.securecomputation.requisitions.RequisitionFetcherRunner",
+    runtime_deps = [":requisition_fetcher_runner"],
+)

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2025 The Cross-Media Measurement Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.wfanet.measurement.securecomputation.requisitions
+
+import io.grpc.StatusException
+import java.util.logging.Logger
+import org.wfanet.measurement.api.v2alpha.ListRequisitionsRequestKt
+import org.wfanet.measurement.api.v2alpha.Measurement
+import org.wfanet.measurement.api.v2alpha.Requisition
+import org.wfanet.measurement.api.v2alpha.RequisitionsGrpcKt.RequisitionsCoroutineStub
+import org.wfanet.measurement.api.v2alpha.listRequisitionsRequest
+import org.wfanet.measurement.gcloud.gcs.GcsStorageClient
+
+// 1. Polls for new requisitions
+// 2. Stores new requisitions into Google Cloud Storage
+class RequisitionFetcher(
+  private val requisitionsStub: RequisitionsCoroutineStub,
+  private val gcsStorageClient: GcsStorageClient,
+  private val gcsBucket: String,
+  private val dataProviderName: String,
+) {
+  suspend fun executeRequisitionFetchingWorkflow() {
+    logger.info("Executing requisitionFetchingWorkflow for $dataProviderName...")
+
+    val requisitions = fetchRequisitions()
+
+    if (requisitions.isEmpty()) {
+      logger.fine("No unfulfilled requisitions for $dataProviderName. Polling again later...")
+      return
+    }
+
+    storeRequisitions(requisitions)
+  }
+
+  private suspend fun fetchRequisitions(): List<Requisition> {
+    val request = listRequisitionsRequest {
+      parent = dataProviderName
+      filter =
+        ListRequisitionsRequestKt.filter {
+          states += Requisition.State.UNFULFILLED
+          measurementStates += Measurement.State.AWAITING_REQUISITION_FULFILLMENT
+        }
+    }
+
+    try {
+      return requisitionsStub.listRequisitions(request).requisitionsList
+    } catch (e: StatusException) {
+      throw Exception("Error listing requisitions", e)
+    }
+  }
+
+  private suspend fun storeRequisitions(requisitions: List<Requisition>) {
+    for (requisition in requisitions) {
+      val blobUri = "gs://${gcsBucket}/${requisition.name}"
+
+      // Only stores the requisition if it does not already exist in the GCS bucket by checking if
+      // the blob URI(created
+      // using the requisition name, ensuring uniqueness) is populated.
+      if (gcsStorageClient.getBlob(blobUri) == null) {
+        gcsStorageClient.writeBlob(blobUri, requisition.toByteString())
+      }
+    }
+  }
+
+  companion object {
+    val logger: Logger = Logger.getLogger(this::class.java.name)
+  }
+}

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcherRunner.kt

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2025 The Cross-Media Measurement Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.wfanet.measurement.securecomputation.requisitions
+
+import com.google.cloud.functions.HttpFunction
+import com.google.cloud.functions.HttpRequest
+import com.google.cloud.functions.HttpResponse
+import com.google.cloud.storage.StorageOptions
+import java.io.File
+import kotlinx.coroutines.runBlocking
+import org.wfanet.measurement.api.v2alpha.RequisitionsGrpcKt.RequisitionsCoroutineStub
+import org.wfanet.measurement.common.crypto.SigningCerts
+import org.wfanet.measurement.common.grpc.buildMutualTlsChannel
+import org.wfanet.measurement.common.toByteArray
+import org.wfanet.measurement.gcloud.gcs.GcsStorageClient
+
+class RequisitionFetcherRunner : HttpFunction {
+  override fun service(request: HttpRequest?, response: HttpResponse?) {
+    runBlocking { requisitionFetcher.executeRequisitionFetchingWorkflow() }
+  }
+
+  companion object {
+    val clientCerts = runBlocking { getClientCerts() }
+
+    val publicChannel =
+      buildMutualTlsChannel(System.getenv("TARGET"), clientCerts, System.getenv("CERT_HOST"))
+
+    val requisitionsStub = RequisitionsCoroutineStub(publicChannel)
+    val requisitionsStorageClient =
+      GcsStorageClient(
+        StorageOptions.newBuilder()
+          .setProjectId(System.getenv("REQUISITIONS_GCS_PROJECT_ID"))
+          .build()
+          .service,
+        System.getenv("REQUISITIONS_GCS_BUCKET"),
+      )
+
+    val requisitionFetcher =
+      RequisitionFetcher(
+        requisitionsStub,
+        requisitionsStorageClient,
+        System.getenv("DATAPROVIDER_NAME"),
+        System.getenv("REQUISITIONS_GCS_BUCKET"),
+      )
+
+    private suspend fun getClientCerts(): SigningCerts {
+      val authenticationStorageClient =
+        GcsStorageClient(
+          StorageOptions.newBuilder()
+            .setProjectId(System.getenv("AUTHENTICATION_GCS_PROJECT_ID"))
+            .build()
+            .service,
+          System.getenv("AUTHENTICATION_GCS_BUCKET"),
+        )
+
+      val certBlob =
+        authenticationStorageClient.getBlob(
+          "gs://${System.getenv("AUTHENTICATION_GCS_BUCKET")}/${System.getenv("CERT_FILE_PATH")}"
+        )
+      val privateKeyBlob =
+        authenticationStorageClient.getBlob(
+          "gs://${System.getenv("AUTHENTICATION_GCS_BUCKET")}/${System.getenv("PRIVATE_KEY_FILE_PATH")}"
+        )
+      val certCollectionBlob =
+        authenticationStorageClient.getBlob(
+          "gs://${System.getenv("AUTHENTICATION_GCS_BUCKET")}/${System.getenv("CERT_COLLECTION_FILE_PATH")}"
+        )
+
+      val certFile = File.createTempFile("cert", ".pem")
+      certFile.writeBytes(checkNotNull(certBlob).read().toByteArray())
+
+      val privateKeyFile = File.createTempFile("private_key", ".key")
+      privateKeyFile.writeBytes(checkNotNull(privateKeyBlob).read().toByteArray())
+
+      val certCollectionFile = File.createTempFile("cert_collection", ".pem")
+      certCollectionFile.writeBytes(checkNotNull(certCollectionBlob).read().toByteArray())
+
+      return SigningCerts.fromPemFiles(
+        certificateFile = certFile,
+        privateKeyFile = privateKeyFile,
+        trustedCertCollectionFile = certCollectionFile,
+      )
+    }
+  }
+}

src/test/kotlin/org/wfanet/measurement/securecomputation/requisitions/BUILD.bazel

@@ -0,0 +1,38 @@
+load("@rules_java//java:defs.bzl", "java_test")
+load("@wfa_rules_kotlin_jvm//kotlin:defs.bzl", "kt_jvm_test")
+
+package(
+    default_testonly = True,
+    default_visibility = ["//visibility:public"],
+)
+
+kt_jvm_test(
+    name = "requisition_fetcher_test",
+    srcs = ["RequisitionFetcherTest.kt"],
+    data = [
+        "//src/main/k8s/testing/secretfiles:all_configs",
+        "//src/main/k8s/testing/secretfiles:all_der_files",
+        "//src/main/k8s/testing/secretfiles:all_tink_keysets",
+        "//src/main/k8s/testing/secretfiles:edp_trusted_certs.pem",
+    ],
+    test_class = "org.wfanet.measurement.securecomputation.requisitions.RequisitionFetcherTest",
+    deps = [
+        "//src/main/kotlin/org/wfanet/measurement/api/v2alpha/testing",
+        "//src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions:requisition_fetcher",
+        "//src/main/proto/wfa/measurement/api/v2alpha:requisitions_service_kt_jvm_grpc_proto",
+        "//src/main/proto/wfa/measurement/api/v2alpha/event_templates/testing:test_event_kt_jvm_proto",
+        "@wfa_common_jvm//imports/java/com/google/cloud/storage/contrib/nio",
+        "@wfa_common_jvm//imports/java/com/google/common/truth",
+        "@wfa_common_jvm//imports/java/org/junit",
+        "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common/crypto/testing",
+        "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common/testing",
+        "@wfa_consent_signaling_client//src/main/kotlin/org/wfanet/measurement/consent/client/duchy",
+        "@wfa_consent_signaling_client//src/main/kotlin/org/wfanet/measurement/consent/client/measurementconsumer",
+    ],
+)
+
+java_test(
+    name = "RequisitionFetcherTest",
+    test_class = "org.wfanet.measurement.securecomputation.requisitions.RequisitionFetcherTest",
+    runtime_deps = [":requisition_fetcher_test"],
+)