Public Sequencer: Add a DevNet only HTTP endpoint to SV App, that creates free MemberTraffic contracts for the provided participantId, using its own test CC

apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/SvAppReference.scala

@@ -83,6 +83,12 @@ abstract class SvAppReference(
       httpCommand(HttpSvPublicAppClient.getSvOnboardingStatus(candidate))
     }
 
+  @Help.Summary("Buy member traffic for a new validator (DevNet only) (via client API)")
+  def devNetBuyMemberTraffic(participantId: ParticipantId): Unit =
+    consoleEnvironment.run {
+      httpCommand(HttpSvPublicAppClient.DevNetBuyMemberTraffic(participantId.toProtoPrimitive))
+    }
+
   @Help.Summary("Prepare a validator onboarding and return an onboarding secret (via client API)")
   def devNetOnboardValidatorPrepare(): String =
     consoleEnvironment.run {

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/PermissionedSynchronizerIntegrationTest.scala

@@ -100,5 +100,25 @@ class PermissionedSynchronizerIntegrationTest
         aliceValidatorBackend.onboardUser("TestUser")
       },
     )
+
+    clue("Sponser SV Buys Member Traffic for Bob in the DevNet") {
+      sv1Backend.devNetBuyMemberTraffic(bobValidatorBackend.participantClient.id)
+    }
+
+    clue("Verify Bob is granted ParticipantSynchronizerPermission") {
+      eventually() {
+        sv1ScanBackend.getParticipantSynchronizerPermission(
+          decentralizedSynchronizerId.toProtoPrimitive,
+          bobValidatorBackend.participantClient.id.toProtoPrimitive,
+        ) shouldBe Some(
+          SynchronizerPermissionState(None)
+        )
+      }
+    }
+
+    clue("Bob Validator starts and onboards correctly") {
+      bobValidatorBackend.startSync()
+      bobValidatorBackend.onboardUser("TestUserBob")
+    }
   }
 }

apps/sv/src/main/openapi/sv-internal.yaml

@@ -493,6 +493,31 @@ paths:
                 "$ref": "#/components/schemas/OnboardSvSequencerResponse"
         "400":
           $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/400"
+  /v0/devnet/onboard/validator/purchase-traffic:
+    post:
+      tags: [sv]
+      x-jvm-package: sv_public
+      description: "faucet for validator candidates to buy member traffic"
+      operationId: "devNetBuyMemberTraffic"
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              "$ref": "#/components/schemas/DevNetBuyMemberTrafficRequest"
+      responses:
+        "200":
+          description: ok
+          content:
+            text/plain:
+              schema:
+                type: string
+        "400":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/400"
+        "500":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/500"
+        "501":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/501"
   /v0/devnet/onboard/validator/prepare:
     post:
       tags: [sv]
@@ -721,6 +746,14 @@ components:
             Human-readable alias of the validator party decoded from the stored secret.
           type: string
 
+    DevNetBuyMemberTrafficRequest:
+      type: object
+      required:
+        - participant_id
+      properties:
+        participant_id:
+          type: string
+
     PrepareValidatorOnboardingRequest:
       type: object
       required:

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/SvApp.scala

@@ -118,7 +118,7 @@ class SvApp(
     ) {
 
   override def packagesForJsonDecoding: Seq[DarResource] =
-    super.packagesForJsonDecoding ++ DarResources.dsoGovernance.all ++ DarResources.validatorLifecycle.all ++ DarResources.amuletNameService.all
+    super.packagesForJsonDecoding ++ DarResources.dsoGovernance.all ++ DarResources.validatorLifecycle.all ++ DarResources.amuletNameService.all ++ DarResources.wallet.all
 
   override def preInitializeBeforeLedgerConnection()(implicit tc: TraceContext): Future[Unit] = {
     val participantAdminConnection = new ParticipantAdminConnection(

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/admin/api/client/commands/HttpSvPublicAppClient.scala

@@ -158,6 +158,28 @@ object HttpSvPublicAppClient {
     }
   }
 
+  case class DevNetBuyMemberTraffic(participantId: String)
+      extends BaseCommandPublic[http.DevNetBuyMemberTrafficResponse, Unit] {
+
+    override def submitRequest(
+        client: Client,
+        headers: List[HttpHeader],
+    ): EitherT[Future, Either[
+      Throwable,
+      HttpResponse,
+    ], http.DevNetBuyMemberTrafficResponse] =
+      client.devNetBuyMemberTraffic(
+        body = definitions.DevNetBuyMemberTrafficRequest(participantId),
+        headers = headers,
+      )
+
+    override def handleOk()(implicit
+        decoder: TemplateJsonDecoder
+    ) = { case http.DevNetBuyMemberTrafficResponse.OK(_) =>
+      Right(())
+    }
+  }
+
   case class DevNetOnboardValidatorPrepare()
       extends BaseCommandPublic[http.DevNetOnboardValidatorPrepareResponse, String] {
 

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/admin/http/HttpSvPublicHandler.scala

@@ -307,6 +307,40 @@ class HttpSvPublicHandler(
     }
   }
 
+  /** Intended use: Used by validator candidates to buy member traffic using SV's DevNet faucet
+    *
+    * Protection: Rate limiting, endpoint only used for DevNet
+    */
+  override def devNetBuyMemberTraffic(
+      respond: r0.DevNetBuyMemberTrafficResponse.type
+  )(
+      body: definitions.DevNetBuyMemberTrafficRequest
+  )(extracted: TraceContext): Future[r0.DevNetBuyMemberTrafficResponse] = {
+    implicit val traceContext: TraceContext = extracted
+    withSpan(s"$workflowId.devNetBuyMemberTraffic") { _ => _ =>
+      if (isDevNet && config.permissionedSynchronizer) {
+        for {
+          participantId <- ParticipantId.fromProtoPrimitive(
+            body.participantId,
+            "participant_id",
+          ) match {
+            case Right(id) => Future.successful(id)
+            case Left(err) =>
+              Future.failed(HttpErrorHandler.badRequest(s"Invalid participant ID: $err"))
+          }
+          _ <- devNetTapAndBuyMemberTraffic(participantId)
+
+        } yield r0.DevNetBuyMemberTrafficResponseOK("Success")
+      } else {
+        Future.failed(
+          HttpErrorHandler.notImplemented(
+            "Traffic purchasing self-service is only available in DevNet when permissioned synchronizer is enabled."
+          )
+        )
+      }
+    }
+  }
+
   /** Intended use: Used by other validators to get a free onboarding secret
     *
     * Protection: Rate limiting, endpoint only used for DevNet
@@ -850,6 +884,55 @@ class HttpSvPublicHandler(
         .yieldUnit()
     } yield ()
 
+  private def devNetTapAndBuyMemberTraffic(
+      participantId: ParticipantId
+  )(implicit tc: TraceContext): Future[Unit] = {
+    for {
+      dsoRules <- dsoStore.getDsoRules()
+
+      svWalletInstall <- retryProvider.retryForClientCalls(
+        "wait_for_wallet_install",
+        "Wait for SV WalletAppInstall contract to be ingested",
+        for {
+          svWalletInstallOpt <- svStoreWithIngestion.store.lookupWalletAppInstallByEndUser(svParty)
+          install <- svWalletInstallOpt match {
+            case Some(install) => Future.successful(install)
+            case None =>
+              Future.failed(
+                HttpErrorHandler.internalServerError(
+                  "SV WalletAppInstall contract not found."
+                )
+              )
+          }
+        } yield install,
+        logger,
+      )
+
+      cmd = svWalletInstall.contractId.exerciseWalletAppInstall_CreateBuyTrafficRequest(
+        participantId.toProtoPrimitive,
+        dsoRules.payload.config.decentralizedSynchronizer.activeSynchronizerId,
+        dsoStore.domainMigrationId.toInt,
+        config.devNetPublicSetupTrafficAmount,
+        clock.now.plus(java.time.Duration.ofMinutes(5)).toInstant,
+        s"devnet-onboard-${participantId.toProtoPrimitive}-${clock.now.toInstant.toEpochMilli}",
+      )
+
+      _ = logger.info(s"Creating BuyTrafficRequest by $svUserName for $participantId")
+
+      _ <- dsoStoreWithIngestion
+        .connection(SpliceLedgerConnectionPriority.Medium)
+        .submit(
+          actAs = Seq(svParty),
+          readAs = Seq(dsoParty),
+          update = cmd,
+        )
+        .withSynchronizerId(dsoRules.domain)
+        .noDedup
+        .yieldUnit()
+
+    } yield ()
+  }
+
   private def startSvOnboarding(
       candidateName: String,
       candidateParty: PartyId,

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/config/SvAppConfig.scala

@@ -485,6 +485,7 @@ case class SvAppBackendConfig(
     // where intentional overlap is required.
     instanceLockEnabled: Boolean = true,
     minMemberTrafficToOnboardValidator: Long = 100000L,
+    devNetPublicSetupTrafficAmount: Long = 10000000L,
 ) extends SpliceBackendConfig {
 
   def allIgnoredAmuletVersions: Set[String] =

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/NodeInitializerUtil.scala

@@ -95,6 +95,7 @@ trait NodeInitializerUtil extends NamedLogging with Spanning with SynchronizerNo
     config.automation.ingestion,
     config.parameters.defaultLimit,
     acsStoreDescriptorUserVersion,
+    Some(config),
   )
 
   protected def newSvSvAutomationService(

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/SvSvStore.scala

@@ -19,16 +19,19 @@ import org.lfdecentralizedtrust.splice.util.{Contract, TemplateJsonDecoder}
 import com.digitalasset.canton.lifecycle.CloseContext
 import com.digitalasset.canton.logging.NamedLoggerFactory
 import com.digitalasset.canton.resource.DbStorage
-import com.digitalasset.canton.topology.ParticipantId
+import com.digitalasset.canton.topology.{ParticipantId, PartyId}
 import com.digitalasset.canton.tracing.TraceContext
+import org.lfdecentralizedtrust.splice.codegen.java.splice.wallet.install.WalletAppInstall
 import org.lfdecentralizedtrust.splice.config.IngestionConfig
 import org.lfdecentralizedtrust.splice.store.db.AcsInterfaceViewRowData
+import org.lfdecentralizedtrust.splice.sv.config.SvAppBackendConfig
 
 import scala.concurrent.{ExecutionContext, Future}
 
 /* Store used by the SV app for filtering contracts visible to the SV party. */
 trait SvSvStore extends AppStore {
 
+  def config: Option[SvAppBackendConfig]
   protected val outerLoggerFactory: NamedLoggerFactory
 
   override protected lazy val loggerFactory: NamedLoggerFactory =
@@ -38,7 +41,7 @@ trait SvSvStore extends AppStore {
       : org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.ContractFilter[
         org.lfdecentralizedtrust.splice.sv.store.db.SvTables.SvAcsStoreRowData,
         AcsInterfaceViewRowData.NoInterfacesIngested,
-      ] = SvSvStore.contractFilter(key)
+      ] = SvSvStore.contractFilter(key, config.map(_.permissionedSynchronizer).getOrElse(false))
 
   def lookupValidatorOnboardingBySecretWithOffset(
       secret: String
@@ -53,6 +56,19 @@ trait SvSvStore extends AppStore {
   ] =
     lookupValidatorOnboardingBySecretWithOffset(secret).map(_.value)
 
+  def lookupWalletAppInstallByEndUserWithOffset(
+      endUserParty: PartyId
+  )(implicit tc: TraceContext): Future[
+    QueryResult[Option[Contract[WalletAppInstall.ContractId, WalletAppInstall]]]
+  ]
+
+  def lookupWalletAppInstallByEndUser(
+      endUserParty: PartyId
+  )(implicit tc: TraceContext): Future[
+    Option[Contract[WalletAppInstall.ContractId, WalletAppInstall]]
+  ] =
+    lookupWalletAppInstallByEndUserWithOffset(endUserParty).map(_.value)
+
   def lookupUsedSecretWithOffset(
       secret: String
   )(implicit tc: TraceContext): Future[
@@ -103,6 +119,7 @@ object SvSvStore {
       ingestionConfig: IngestionConfig,
       defaultLimit: Limit,
       acsStoreDescriptorUserVersion: Option[Long] = None,
+      svBackendconfig: Option[SvAppBackendConfig] = None,
   )(implicit
       ec: ExecutionContext,
       templateJsonDecoder: TemplateJsonDecoder,
@@ -118,40 +135,57 @@ object SvSvStore {
       ingestionConfig,
       acsStoreDescriptorUserVersion,
       defaultLimit = defaultLimit,
+      config = svBackendconfig,
     )
 
   /** Contract filter of an sv acs store for a specific acs party. */
-  def contractFilter(key: SvStore.Key): MultiDomainAcsStore.ContractFilter[
+  def contractFilter(
+      key: SvStore.Key,
+      enablePermissionedSynchronizer: Boolean,
+  ): MultiDomainAcsStore.ContractFilter[
     SvAcsStoreRowData,
     AcsInterfaceViewRowData.NoInterfacesIngested,
   ] = {
     import MultiDomainAcsStore.mkFilter
     val sv = key.svParty.toProtoPrimitive
 
-    MultiDomainAcsStore.SimpleContractFilter(
-      key.svParty,
-      Map(
-        mkFilter(vo.ValidatorOnboarding.COMPANION)(co => co.payload.sv == sv) { contract =>
-          SvAcsStoreRowData(
-            contract,
-            contractExpiresAt = Some(Timestamp.assertFromInstant(contract.payload.expiresAt)),
-            onboardingSecret = Some(contract.payload.candidateSecret),
-          )
-        },
-        mkFilter(vo.UsedSecret.COMPANION)(co => co.payload.sv == sv) { contract =>
-          SvAcsStoreRowData(
-            contract,
-            onboardingSecret = Some(contract.payload.secret),
-          )
-        },
-        mkFilter(so.SvOnboardingConfirmed.COMPANION)(co => co.payload.svParty == sv) { contract =>
+    val svFilters = Map(
+      mkFilter(vo.ValidatorOnboarding.COMPANION)(co => co.payload.sv == sv) { contract =>
+        SvAcsStoreRowData(
+          contract,
+          contractExpiresAt = Some(Timestamp.assertFromInstant(contract.payload.expiresAt)),
+          onboardingSecret = Some(contract.payload.candidateSecret),
+        )
+      },
+      mkFilter(vo.UsedSecret.COMPANION)(co => co.payload.sv == sv) { contract =>
+        SvAcsStoreRowData(
+          contract,
+          onboardingSecret = Some(contract.payload.secret),
+        )
+      },
+      mkFilter(so.SvOnboardingConfirmed.COMPANION)(co => co.payload.svParty == sv) { contract =>
+        SvAcsStoreRowData(
+          contract,
+          contractExpiresAt = Some(Timestamp.assertFromInstant(contract.payload.expiresAt)),
+          svCandidateName = Some(contract.payload.svName),
+        )
+      },
+    )
+
+    val finalFilters = if (enablePermissionedSynchronizer) {
+      svFilters + mkFilter(WalletAppInstall.COMPANION)(co => co.payload.endUserParty == sv) {
+        contract =>
           SvAcsStoreRowData(
-            contract,
-            contractExpiresAt = Some(Timestamp.assertFromInstant(contract.payload.expiresAt)),
-            svCandidateName = Some(contract.payload.svName),
+            contract
           )
-        },
-      ),
+      }
+    } else {
+      svFilters
+    }
+
+    MultiDomainAcsStore.SimpleContractFilter(
+      key.svParty,
+      finalFilters,
     )
   }
 }