feat: Create tee requisition fetcher

[jojijac0b]

Create Requisition Fetcher for EDP Aggregator

[wfa-reviewable]

This change is 

[Marco-Premier]

Reviewed 5 of 5 files at r2, all commit messages.
Reviewable status: all files reviewed, 2 unresolved discussions (waiting on @jojijac0b and @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 71 at r2 (raw file):

          try {
            listRequisitions(listRequisitionsRequest {
              parent = kingdomConfig.dataProvider

I think we use filters (cross-media-measurement-api/src/main/proto/wfa/measurement/api/v2alpha/requisitions_service.proto at main · world-federation-of-advertisers/cross-media-measurement-api) here to fetch only for UNFULFILLED requisitions (https://github.com/world-federation-of-advertisers/cross-media-measurement-api/blob/main/src/main/proto/wfa/measurement/api/v2alpha/requisition.proto#L181)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 80 at r2 (raw file):

      }
      .flattenConcat()
      .filter { it.updateTime.seconds > storageConfig.lastUpdate.seconds }

NIT move filter before flattenConcat to have the second running on fewer elements

[SanjayVas]

feat: Create tee requisition fetcher

This isn't actually a TEE application, right? It's just the fetcher for the EDP Aggregator

Reviewed 2 of 9 files at r3, all commit messages.
Reviewable status: 2 of 9 files reviewed, 8 unresolved discussions (waiting on @jojijac0b, @kungfucraig, @Marco-Premier, and @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcherRunner.kt line 31 at r3 (raw file):

import org.wfanet.measurement.gcloud.gcs.GcsStorageClient

class RequisitionFetcherRunner : HttpFunction {

Why is this a cloud function? What is it triggered by? I thought that this would just be a cron job. Apologies, I cannot find the appropriate spot where this is covered in the design doc.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcherRunner.kt line 32 at r3 (raw file):

class RequisitionFetcherRunner : HttpFunction {
  override fun service(request: HttpRequest?, response: HttpResponse?) {

I believe these params are non-nullable. IntelliJ will just default to nullables for implementing Java interfaces when it cannot determine the nullability.

Code quote:

HttpRequest?

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 28 at r3 (raw file):

import org.wfanet.measurement.gcloud.gcs.GcsStorageClient

// 1. Polls for new requisitions

Use KDoc syntax following https://developer.android.com/kotlin/style-guide#documentation

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 55 at r3 (raw file):

        ListRequisitionsRequestKt.filter {
          states += Requisition.State.UNFULFILLED
          measurementStates += Measurement.State.AWAITING_REQUISITION_FULFILLMENT

nit: now that we have the WITHDRAWN Requisition state, it should no longer be necessary to filter by Measurement state as well

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 68 at r3 (raw file):

  private suspend fun storeRequisitions(requisitions: List<Requisition>) {
    for (requisition in requisitions) {
      val blobUri = "gs://${gcsBucket}/${requisition.name}"

GcsStorageClient takes a blob key, not a blob URI. The bucket is already bound to the client.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 80 at r3 (raw file):

  companion object {
    val logger: Logger = Logger.getLogger(this::class.java.name)

private

[kungfucraig]

Reviewed 3 of 9 files at r3, all commit messages.
Reviewable status: 4 of 9 files reviewed, 14 unresolved discussions (waiting on @jojijac0b, @Marco-Premier, @SanjayVas, and @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 28 at r3 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

Use KDoc syntax following https://developer.android.com/kotlin/style-guide#documentation

Also please document the arguments. I wonder what the bucket format is. I also wonder whether the dataProviderName is a resource name.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 17 at r3 (raw file):

 */

package org.wfanet.measurement.securecomputation.requisitions

Can you use a package name of "requisitionfetcher"?

What's more, given that this is not a TEE App, it probably shouldn't live in "securecomputation."

We should create an EDP Aggregator directory that is child of measurement and start putting specific EDP Aggregator components in it.

The securecomputation directory should be reserved for common infra.

@SanjayVas @stevenwarejones Wdyt?

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 42 at r3 (raw file):

    if (requisitions.isEmpty()) {
      logger.fine("No unfulfilled requisitions for $dataProviderName. Polling again later...")

Log the number of requisitions received and pull this out of the if statement.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcher.kt line 67 at r3 (raw file):

  private suspend fun storeRequisitions(requisitions: List<Requisition>) {
    for (requisition in requisitions) {

Do you expect the fulfiller to process requisitions one by one or in batch?

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcherRunner.kt line 37 at r3 (raw file):

  companion object {
    val clientCerts = runBlocking { getClientCerts() }

I wonder if we should be building this with an assumption that consent signaling gets deprecated. @SanjayVas @stevenwarejones wdyt?

Maybe we can implement just part of it, decryption and signing but not verification?

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/RequisitionFetcherRunner.kt line 40 at r3 (raw file):

    val publicChannel =
      buildMutualTlsChannel(System.getenv("TARGET"), clientCerts, System.getenv("CERT_HOST"))

We've usually used flags not environment variables for passing this kind of information.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/requisitions/BUILD.bazel line 4 at r3 (raw file):

load("@wfa_rules_kotlin_jvm//kotlin:defs.bzl", "kt_jvm_library")

package(default_visibility = ["//visibility:public"])

Default visibility should be private. We don't want anything taking dependencies on TEE Apps.