feat: serve the registration for explicit disclosure

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

@@ -24,6 +24,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{
   OpenMiningRound,
 }
 import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.AnsRules
+import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
 import org.lfdecentralizedtrust.splice.config.NetworkAppClientConfig
 import org.lfdecentralizedtrust.splice.environment.SpliceConsoleEnvironment
 import org.lfdecentralizedtrust.splice.http.v0.definitions
@@ -196,6 +197,14 @@ abstract class ScanAppReference(
       httpCommand(HttpScanAppClient.LookupTransferPreapprovalByParty(party))
     }
 
+  @Help.Summary("Lookup the RegisteredSynchronizer for a dedicated synchronizer id")
+  def lookupSynchronizerRegistration(
+      synchronizerId: String
+  ): Option[ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer]] =
+    consoleEnvironment.run {
+      httpCommand(HttpScanAppClient.LookupSynchronizerRegistration(synchronizerId))
+    }
+
   @Help.Summary("Lookup a TransferCommandCounter by the receiver party")
   def lookupTransferCommandCounterByParty(
       party: PartyId

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

@@ -3,7 +3,6 @@
 
 package org.lfdecentralizedtrust.splice.integration.tests
 
-import com.daml.ledger.javaapi.data.CreatedEvent
 import com.digitalasset.canton.SynchronizerAlias
 import com.digitalasset.canton.config.RequireTypes.NonNegativeLong
 import com.digitalasset.canton.data.CantonTimestamp
@@ -19,9 +18,7 @@ import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.{
   IntegrationTest,
   SpliceTestConsoleEnvironment,
 }
-import org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.ContractState
 import org.lfdecentralizedtrust.splice.util.{
-  Contract,
   ContractWithState,
   DisclosedContracts,
   SynchronizerFeesTestUtil,
@@ -84,10 +81,8 @@ class SyncOperatorTrafficIntegrationTest
         }
       }
 
-      val registration = clue("the DSO registers the synchronizer to this operator") {
-        // The purchase is submitted from alice's participant, which hosts neither the DSO nor
-        // the operator, so the registration must be disclosed with its created-event blob.
-        val tx = sv1Backend.participantClientWithAdminToken.ledger_api_extensions.commands
+      clue("the DSO registers the synchronizer to this operator") {
+        sv1Backend.participantClientWithAdminToken.ledger_api_extensions.commands
           .submitJava(
             actAs = Seq(dsoParty),
             readAs = Seq(dsoParty),
@@ -100,16 +95,14 @@ class SyncOperatorTrafficIntegrationTest
               .asScala
               .toSeq,
             userId = sv1Backend.config.ledgerApiUser,
-            includeCreatedEventBlob = true,
           )
-        val contract = tx.getEventsById.values.asScala
-          .collect { case ev: CreatedEvent => ev }
-          .flatMap(Contract.fromCreatedEvent(RegisteredSynchronizer.COMPANION)(_))
-          .loneElement
-        ContractWithState(
-          contract,
-          ContractState.Assigned(SynchronizerId.tryFromString(tx.getSynchronizerId)),
-        )
+      }
+
+      // Alice's participant hosts neither the DSO nor the operator, so the registration has to
+      // be disclosed, and Scan is the only source of its created-event blob.
+      sv1ScanBackend.lookupSynchronizerRegistration("dedicated::does-not-exist") shouldBe None
+      val registration = eventually() {
+        sv1ScanBackend.lookupSynchronizerRegistration(synchronizerId.toProtoPrimitive).value
       }
 
       val aliceParty = onboardWalletUser(aliceWalletClient, aliceValidatorBackend)

apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V075__scan_acs_store_registered_synchronizer.sql

@@ -0,0 +1,4 @@
+-- The synchronizer id from a RegisteredSynchronizer, so Scan can serve a registration by
+-- synchronizer id without a JSON extraction on every candidate row.
+alter table scan_acs_store
+  add column registered_synchronizer_id text;

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/SqlIndexInitializationTrigger.scala

@@ -264,6 +264,15 @@ object SqlIndexInitializationTrigger {
           where reward_beneficiary_is_observer = false
         """,
       ),
+    IndexAction
+      .Create(
+        indexName = "scan_acs_store_sid_mid_pn_tid_rsid",
+        createAction = sqlu"""
+          create index concurrently if not exists scan_acs_store_sid_mid_pn_tid_rsid
+          on scan_acs_store (store_id, migration_id, package_name, template_id_qualified_name, registered_synchronizer_id)
+          where registered_synchronizer_id is not null
+        """,
+      ),
     IndexAction
       .Create(
         indexName = "scan_txlog_store_sid_effat_en_vot",

apps/common/src/test/scala/org/lfdecentralizedtrust/splice/automation/SqlIndexInitializationTriggerStoreTest.scala

@@ -42,6 +42,7 @@ class SqlIndexInitializationTriggerStoreTest
     "updt_hist_crea_hi_mi_ci_import_updates",
     "updt_hist_tran_hi_eth",
     "dso_acs_store_sid_mid_pn_tid_rbio",
+    "scan_acs_store_sid_mid_pn_tid_rsid",
     "scan_txlog_store_sid_effat_en_vot",
   )
 

apps/scan/src/main/openapi/scan.yaml

@@ -1124,6 +1124,32 @@ paths:
         "404":
           $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/404"
 
+  /v0/synchronizers/{synchronizer_id}/registration:
+    get:
+      tags: [internal, scan]
+      x-jvm-package: scan
+      operationId: "lookupSynchronizerRegistration"
+      description: |
+        Get the RegisteredSynchronizer authorizing Amulet-funded traffic for a dedicated
+        synchronizer.
+      parameters:
+        - name: "synchronizer_id"
+          description: |
+            The synchronizer ID whose registration to look up.
+          in: "path"
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: ok
+          content:
+            application/json:
+              schema:
+                "$ref": "#/components/schemas/LookupSynchronizerRegistrationResponse"
+        "404":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/404"
+
   /v0/transfer-command-counter/{party}:
     get:
       tags: [internal, scan]
@@ -3531,6 +3557,14 @@ components:
       properties:
         transfer_preapproval:
           "$ref": "../../../../common/src/main/openapi/common-internal.yaml#/components/schemas/ContractWithState"
+    LookupSynchronizerRegistrationResponse:
+      description: A Daml contract of template `Splice.DecentralizedSynchronizer:RegisteredSynchronizer`.
+      type: object
+      required: ["registration"]
+      properties:
+        registration:
+          "$ref": "../../../../common/src/main/openapi/common-internal.yaml#/components/schemas/ContractWithState"
+
     LookupTransferCommandCounterByPartyResponse:
       description: A Daml contract of template `Splice.ExternalPartyAmuletRules:TransferCommandCounter`.
       type: object

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala

@@ -104,6 +104,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationi
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationinstructionv2
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.transferinstructionv1
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.transferinstructionv2
+import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
 import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{
   DsoRules,
   DsoRules_CloseVoteRequestResult,
@@ -483,6 +484,12 @@ class BftScanConnection(
   ): Future[Option[LookupTransferCommandStatusResponse]] =
     bftCall(_.lookupTransferCommandStatus(sender, nonce), "lookupTransferCommandStatus")
 
+  override def lookupSynchronizerRegistration(synchronizerId: String)(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+  ): Future[Option[ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer]]] =
+    bftCall(_.lookupSynchronizerRegistration(synchronizerId), "lookupSynchronizerRegistration")
+
   override def lookupTransferPreapprovalByParty(receiver: PartyId)(implicit
       ec: ExecutionContext,
       tc: TraceContext,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/ScanConnection.scala

@@ -20,6 +20,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{
   OpenMiningRound,
 }
 import org.lfdecentralizedtrust.splice.codegen.java.splice.types.Round
+import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
 import org.lfdecentralizedtrust.splice.config.UpgradesConfig
 import org.lfdecentralizedtrust.splice.environment.*
 import org.lfdecentralizedtrust.splice.http.HttpClient
@@ -308,6 +309,11 @@ trait ScanConnection
       tc: TraceContext,
   ): Future[Option[ContractWithState[TransferPreapproval.ContractId, TransferPreapproval]]]
 
+  def lookupSynchronizerRegistration(synchronizerId: String)(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+  ): Future[Option[ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer]]]
+
   def listVoteRequestResults(
       filters: VoteResultsFilters,
       limit: Int,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/SingleScanConnection.scala

@@ -87,6 +87,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationv
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationv2
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationinstructionv1
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.allocationinstructionv2
+import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
 import org.lfdecentralizedtrust.splice.http.v0.definitions.HoldingsSummaryRequest.RecordTimeMatch
 import org.lfdecentralizedtrust.splice.metrics.ScanConnectionMetrics
 import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient.BftSequencer
@@ -490,6 +491,15 @@ class SingleScanConnection private[client] (
       HttpScanAppClient.GetMigrationId(),
     )
 
+  override def lookupSynchronizerRegistration(synchronizerId: String)(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+  ): Future[Option[ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer]]] =
+    runHttpCmd(
+      config.adminApi.url,
+      HttpScanAppClient.LookupSynchronizerRegistration(synchronizerId),
+    )
+
   override def lookupTransferPreapprovalByParty(receiver: PartyId)(implicit
       ec: ExecutionContext,
       tc: TraceContext,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala

@@ -32,6 +32,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{
 }
 import org.lfdecentralizedtrust.splice.codegen.java.splice.ans as ansCodegen
 import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.AnsRules
+import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
 import org.lfdecentralizedtrust.splice.config.SpliceInstanceNamesConfig
 import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as http}
 import org.lfdecentralizedtrust.tokenstandard.{
@@ -602,6 +603,30 @@ object HttpScanAppClient {
     }
   }
 
+  case class LookupSynchronizerRegistration(
+      synchronizerId: String
+  ) extends InternalBaseCommand[http.LookupSynchronizerRegistrationResponse, Option[
+        ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer]
+      ]] {
+
+    override def submitRequest(
+        client: ScanClient,
+        headers: List[HttpHeader],
+    ) = client.lookupSynchronizerRegistration(synchronizerId, headers)
+
+    override def handleOk()(implicit
+        decoder: TemplateJsonDecoder
+    ) = {
+      case http.LookupSynchronizerRegistrationResponse.OK(response) =>
+        ContractWithState
+          .fromHttp(RegisteredSynchronizer.COMPANION)(response.registration)
+          .map(Some(_))
+          .leftMap(_.toString)
+      case http.LookupSynchronizerRegistrationResponse.NotFound(_) =>
+        Right(None)
+    }
+  }
+
   case class LookupTransferCommandCounterByParty(
       party: PartyId
   ) extends InternalBaseCommand[http.LookupTransferCommandCounterByPartyResponse, Option[

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala

@@ -1185,6 +1185,32 @@ class HttpScanHandler(
     }
   }
 
+  override def lookupSynchronizerRegistration(
+      respond: ScanResource.LookupSynchronizerRegistrationResponse.type
+  )(
+      synchronizerId: String
+  )(extracted: TraceContext): Future[ScanResource.LookupSynchronizerRegistrationResponse] = {
+    implicit val tc = extracted
+    withSpan(s"$workflowId.lookupSynchronizerRegistration") { _ => _ =>
+      store
+        .lookupSynchronizerRegistration(synchronizerId)
+        .map {
+          case Some(c) =>
+            v0.ScanResource.LookupSynchronizerRegistrationResponse.OK(
+              definitions.LookupSynchronizerRegistrationResponse(
+                c.toHttp
+              )
+            )
+          case None =>
+            v0.ScanResource.LookupSynchronizerRegistrationResponse.NotFound(
+              definitions.ErrorResponse(
+                s"No RegisteredSynchronizer found for synchronizer id: $synchronizerId"
+              )
+            )
+        }
+    }
+  }
+
   override def lookupTransferCommandCounterByParty(
       respond: ScanResource.LookupTransferCommandCounterByPartyResponse.type
   )(

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/CachingScanStore.scala

@@ -17,6 +17,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.{
   TransferPreapproval,
 }
 import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.{AnsEntry, AnsRules}
+import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
 import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.svstate.SvNodeState
 import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{
   DsoRules_CloseVoteRequestResult,
@@ -181,6 +182,11 @@ class CachingScanStore(
       store.lookupTransferPreapprovalByParty,
     ).get(partyId)
 
+  override def lookupSynchronizerRegistration(synchronizerId: String)(implicit
+      tc: TraceContext
+  ): Future[Option[ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer]]] =
+    store.lookupSynchronizerRegistration(synchronizerId)
+
   override def lookupTransferCommandCounterByParty(partyId: PartyId)(implicit
       tc: TraceContext
   ): Future[Option[ContractWithState[TransferCommandCounter.ContractId, TransferCommandCounter]]] =

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala

@@ -204,6 +204,21 @@ trait ScanStore
     ]]
   ]
 
+  /** The registration authorizing Amulet-funded traffic for `synchronizerId`, if any.
+    *
+    * Comes from the ACS store, so the result carries the created event blob the buyer needs to
+    * disclose it. If more than one is live, returns one of them under a stable total order; no
+    * supersession is implied.
+    */
+  def lookupSynchronizerRegistration(
+      synchronizerId: String
+  )(implicit tc: TraceContext): Future[
+    Option[ContractWithState[
+      splice.decentralizedsynchronizer.RegisteredSynchronizer.ContractId,
+      splice.decentralizedsynchronizer.RegisteredSynchronizer,
+    ]]
+  ]
+
   def lookupTransferCommandCounterByParty(
       partyId: PartyId
   )(implicit tc: TraceContext): Future[
@@ -408,8 +423,14 @@ object ScanStore {
               pkgVersionSupport
                 .supportsDedicatedSynchronizers(Seq(key.dsoParty), now)(tc)
           },
-        )(
-          ScanAcsStoreRowData(_)
+        )(contract =>
+          // A String, not a SynchronizerId: DsoRules_RegisterSynchronizer only checks the id is
+          // non-empty, so tryFromString here would throw on a governance typo and take down the
+          // ingestion pipeline. The query compares it as text.
+          ScanAcsStoreRowData(
+            contract,
+            registeredSynchronizerId = Some(contract.payload.synchronizerId),
+          )
         ),
         mkFilter(splice.validatorlicense.ValidatorLicense.COMPANION)(co => co.payload.dso == dso) {
           contract =>

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala

@@ -22,7 +22,10 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.{
   TransferPreapproval,
 }
 import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.{AnsEntry, AnsRules}
-import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.MemberTraffic
+import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.{
+  MemberTraffic,
+  RegisteredSynchronizer,
+}
 import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.svstate.SvNodeState
 import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{
   DsoRules_CloseVoteRequestResult,
@@ -365,6 +368,33 @@ class DbScanStore(
     } yield contractWithStateFromRow(TransferPreapproval.COMPANION)(row)).value
   }
 
+  override def lookupSynchronizerRegistration(
+      synchronizerId: String
+  )(implicit tc: TraceContext): Future[
+    Option[ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer]]
+  ] = waitUntilAcsIngested {
+    (for {
+      row <- storage
+        .querySingle(
+          selectFromAcsTableWithState(
+            ScanTables.acsTableName,
+            acsStoreId,
+            domainMigrationId,
+            RegisteredSynchronizer.COMPANION,
+            additionalWhere = sql"""
+                and registered_synchronizer_id = ${lengthLimited(synchronizerId)}
+            """,
+            // Uniqueness is not enforced on-ledger, so pick a total order: every Scan must
+            // return the same row for bftCall to agree.
+            orderLimit = sql"""
+                order by contract_id limit 1
+            """,
+          ).headOption,
+          "lookupSynchronizerRegistration",
+        )
+    } yield contractWithStateFromRow(RegisteredSynchronizer.COMPANION)(row)).value
+  }
+
   override def lookupTransferCommandCounterByParty(
       partyId: PartyId
   )(implicit tc: TraceContext): Future[