chore: refactor/simplify MessagingService-related layers creation

Signed-off-by: Benjamin Voiturier <[email protected]>
hyperledger · Oct 8, 2024 · bb5ea8f · bb5ea8f
1 parent e4c4260
commit bb5ea8f
Show file tree

Hide file tree

Showing 14 changed files with 107 additions and 168 deletions.
diff --git a/cloud-agent/service/server/src/main/resources/application.conf b/cloud-agent/service/server/src/main/resources/application.conf
@@ -248,7 +248,6 @@ agent {
         authApiKey = "default"
         authApiKey = ${?DEFAULT_WALLET_AUTH_API_KEY}
     }
-    inMemoryQueueCapacity = 1000
     messagingService {
         connectFlow {
           consumerCount = 5
@@ -280,6 +279,7 @@ agent {
         statusListSync {
           consumerCount = 5
         }
+        inMemoryQueueCapacity = 1000
         kafkaEnabled = false
         kafkaEnabled = ${?DEFAULT_KAFKA_ENABLED}
         kafka {

diff --git a/cloud-agent/service/server/src/main/scala/org/hyperledger/identus/agent/server/MainApp.scala b/cloud-agent/service/server/src/main/scala/org/hyperledger/identus/agent/server/MainApp.scala
@@ -63,11 +63,7 @@ import org.hyperledger.identus.pollux.sql.repository.{
 }
 import org.hyperledger.identus.presentproof.controller.PresentProofControllerImpl
 import org.hyperledger.identus.resolvers.DIDResolver
-import org.hyperledger.identus.shared.messaging.kafka.{
-  InMemoryMessagingService,
-  ZKafkaMessagingServiceImpl,
-  ZKafkaProducerImpl
-}
+import org.hyperledger.identus.shared.messaging
 import org.hyperledger.identus.shared.messaging.WalletIdAndRecordId
 import org.hyperledger.identus.shared.models.WalletId
 import org.hyperledger.identus.system.controller.SystemControllerImpl
@@ -172,46 +168,6 @@ object MainApp extends ZIOAppDefault {
       )
       _ <- preMigrations
       _ <- migrations
-      appConfig <- ZIO.service[AppConfig].provide(SystemModule.configLayer)
-      messagingServiceLayer <-
-        if (appConfig.agent.messagingService.kafkaEnabled) {
-          val kafkaConfig = appConfig.agent.messagingService.kafka.get
-          ZIO.succeed(
-            ZKafkaMessagingServiceImpl.layer(
-              kafkaConfig.bootstrapServers.split(',').toList,
-              kafkaConfig.consumers.autoCreateTopics,
-              kafkaConfig.consumers.maxPollRecords,
-              kafkaConfig.consumers.maxPollInterval,
-              kafkaConfig.consumers.pollTimeout,
-              kafkaConfig.consumers.rebalanceSafeCommits
-            )
-          )
-        } else {
-          ZIO.succeed(
-            ZLayer.succeed(appConfig.agent.inMemoryQueueCapacity) >>> InMemoryMessagingService.messagingServiceLayer
-          )
-        }
-      messageProducerLayer <-
-        if (appConfig.agent.messagingService.kafkaEnabled) {
-          ZIO.succeed(
-            ZKafkaProducerImpl.layer[UUID, WalletIdAndRecordId]
-          )
-        } else {
-          ZIO.succeed(
-            InMemoryMessagingService.producerLayer[UUID, WalletIdAndRecordId]
-          )
-        }
-      syncDIDStateProducerLayer <-
-        if (appConfig.agent.messagingService.kafkaEnabled) {
-          ZIO.succeed(
-            ZKafkaProducerImpl.layer[WalletId, WalletId]
-          )
-        } else {
-          ZIO.succeed(
-            InMemoryMessagingService.producerLayer[WalletId, WalletId]
-          )
-        }
-
       app <- CloudAgentApp.run
         .provide(
           DidCommX.liveLayer,
@@ -296,9 +252,11 @@ object MainApp extends ZIOAppDefault {
           // HTTP client
           SystemModule.zioHttpClientLayer,
           Scope.default,
-          messagingServiceLayer,
-          messageProducerLayer,
-          syncDIDStateProducerLayer
+          // Messaging Service
+          ZLayer.fromZIO(ZIO.service[AppConfig].map(_.agent.messagingService)),
+          messaging.MessagingService.serviceLayer,
+          messaging.MessagingService.producerLayer[UUID, WalletIdAndRecordId],
+          messaging.MessagingService.producerLayer[WalletId, WalletId]
         )
     } yield app
 

diff --git a/...service/server/src/main/scala/org/hyperledger/identus/agent/server/config/AppConfig.scala b/...service/server/src/main/scala/org/hyperledger/identus/agent/server/config/AppConfig.scala
@@ -5,8 +5,8 @@ import org.hyperledger.identus.iam.authentication.AuthenticationConfig
 import org.hyperledger.identus.pollux.vc.jwt.*
 import org.hyperledger.identus.shared.db.DbConfig
 import org.hyperledger.identus.shared.messaging.MessagingServiceConfig
-import zio.config.magnolia.*
 import zio.Config
+import zio.config.magnolia.*
 
 import java.net.URL
 import java.time.Duration
@@ -165,7 +165,6 @@ final case class AgentConfig(
     secretStorage: SecretStorageConfig,
     webhookPublisher: WebhookPublisherConfig,
     defaultWallet: DefaultWalletConfig,
-    inMemoryQueueCapacity: Int,
     messagingService: MessagingServiceConfig
 ) {
   def validate: Either[String, Unit] =

diff --git a/...c/test/scala/org/hyperledger/identus/oid4vci/domain/OIDCCredentialIssuerServiceSpec.scala b/...c/test/scala/org/hyperledger/identus/oid4vci/domain/OIDCCredentialIssuerServiceSpec.scala
@@ -9,25 +9,19 @@ import org.hyperledger.identus.castor.core.service.{DIDService, MockDIDService}
 import org.hyperledger.identus.oid4vci.http.{ClaimDescriptor, CredentialDefinition, Localization}
 import org.hyperledger.identus.oid4vci.service.{OIDCCredentialIssuerService, OIDCCredentialIssuerServiceImpl}
 import org.hyperledger.identus.oid4vci.storage.InMemoryIssuanceSessionService
-import org.hyperledger.identus.pollux.core.model.oid4vci.CredentialConfiguration
 import org.hyperledger.identus.pollux.core.model.CredentialFormat
-import org.hyperledger.identus.pollux.core.repository.{
-  CredentialRepository,
-  CredentialRepositoryInMemory,
-  CredentialStatusListRepositoryInMemory
-}
+import org.hyperledger.identus.pollux.core.model.oid4vci.CredentialConfiguration
+import org.hyperledger.identus.pollux.core.repository.{CredentialRepositoryInMemory, CredentialStatusListRepositoryInMemory}
 import org.hyperledger.identus.pollux.core.service.*
 import org.hyperledger.identus.pollux.core.service.uriResolvers.ResourceUrlResolver
 import org.hyperledger.identus.pollux.vc.jwt.PrismDidResolver
-import org.hyperledger.identus.shared.messaging.kafka.InMemoryMessagingService
-import org.hyperledger.identus.shared.messaging.WalletIdAndRecordId
+import org.hyperledger.identus.shared.messaging.{MessagingService, MessagingServiceConfig, WalletIdAndRecordId}
 import org.hyperledger.identus.shared.models.{WalletAccessContext, WalletId}
-import zio.{Clock, Random, URLayer, ZIO, ZLayer}
-import zio.json.*
 import zio.json.ast.Json
 import zio.mock.MockSpecDefault
 import zio.test.*
 import zio.test.Assertion.*
+import zio.{Clock, Random, URLayer, ZIO, ZLayer}
 
 import java.net.URI
 import java.time.Instant
@@ -56,8 +50,8 @@ object OIDCCredentialIssuerServiceSpec
       GenericSecretStorageInMemory.layer,
       LinkSecretServiceImpl.layer,
       CredentialServiceImpl.layer,
-      (ZLayer.succeed(100) >>> InMemoryMessagingService.messagingServiceLayer >>>
-        InMemoryMessagingService.producerLayer[UUID, WalletIdAndRecordId]).orDie,
+      (MessagingServiceConfig.inMemoryLayer >>> MessagingService.serviceLayer >>>
+        MessagingService.producerLayer[UUID, WalletIdAndRecordId]).orDie,
       OIDCCredentialIssuerServiceImpl.layer
     )
 

diff --git a/...c/test/scala/org/hyperledger/identus/connect/core/service/ConnectionServiceImplSpec.scala b/...c/test/scala/org/hyperledger/identus/connect/core/service/ConnectionServiceImplSpec.scala
@@ -1,22 +1,19 @@
 package org.hyperledger.identus.connect.core.service
 
 import io.circe.syntax.*
+import org.hyperledger.identus.connect.core.model.ConnectionRecord.*
 import org.hyperledger.identus.connect.core.model.error.ConnectionServiceError
 import org.hyperledger.identus.connect.core.model.error.ConnectionServiceError.InvalidStateForOperation
-import org.hyperledger.identus.connect.core.model.ConnectionRecord
-import org.hyperledger.identus.connect.core.model.ConnectionRecord.*
 import org.hyperledger.identus.connect.core.repository.ConnectionRepositoryInMemory
 import org.hyperledger.identus.mercury.model.{DidId, Message}
 import org.hyperledger.identus.mercury.protocol.connection.ConnectionResponse
-import org.hyperledger.identus.shared.messaging.kafka.InMemoryMessagingService
+import org.hyperledger.identus.shared.messaging
 import org.hyperledger.identus.shared.messaging.WalletIdAndRecordId
-import org.hyperledger.identus.shared.models.*
 import org.hyperledger.identus.shared.models.{WalletAccessContext, WalletId}
 import zio.*
 import zio.test.*
 import zio.test.Assertion.*
 
-import java.time.Instant
 import java.util.UUID
 
 object ConnectionServiceImplSpec extends ZIOSpecDefault {
@@ -315,8 +312,9 @@ object ConnectionServiceImplSpec extends ZIOSpecDefault {
       }
     ).provide(
       connectionServiceLayer,
-      ZLayer.succeed(100) >>> InMemoryMessagingService.messagingServiceLayer,
-      InMemoryMessagingService.producerLayer[UUID, WalletIdAndRecordId],
+      messaging.MessagingServiceConfig.inMemoryLayer,
+      messaging.MessagingService.serviceLayer,
+      messaging.MessagingService.producerLayer[UUID, WalletIdAndRecordId],
       ZLayer.succeed(WalletAccessContext(WalletId.random)),
     )
   }

diff --git a/...st/scala/org/hyperledger/identus/connect/core/service/ConnectionServiceNotifierSpec.scala b/...st/scala/org/hyperledger/identus/connect/core/service/ConnectionServiceNotifierSpec.scala
@@ -7,9 +7,8 @@ import org.hyperledger.identus.event.notification.*
 import org.hyperledger.identus.mercury.model.DidId
 import org.hyperledger.identus.mercury.protocol.connection.{ConnectionRequest, ConnectionResponse}
 import org.hyperledger.identus.mercury.protocol.invitation.v2.Invitation
-import org.hyperledger.identus.shared.messaging.kafka.{ZKafkaMessagingServiceImpl, ZKafkaProducerImpl}
+import org.hyperledger.identus.shared.messaging
 import org.hyperledger.identus.shared.messaging.WalletIdAndRecordId
-import org.hyperledger.identus.shared.models.*
 import org.hyperledger.identus.shared.models.{WalletAccessContext, WalletId}
 import zio.*
 import zio.mock.Expectation
@@ -154,8 +153,9 @@ object ConnectionServiceNotifierSpec extends ZIOSpecDefault {
             inviteeExpectations.toLayer
         ) >>> ConnectionServiceNotifier.layer,
         ZLayer.succeed(WalletAccessContext(WalletId.random)),
-        ZKafkaMessagingServiceImpl.layer(List("localhost:29092")),
-        ZKafkaProducerImpl.layer[UUID, WalletIdAndRecordId]
+        messaging.MessagingServiceConfig.inMemoryLayer,
+        messaging.MessagingService.serviceLayer,
+        messaging.MessagingService.producerLayer[UUID, WalletIdAndRecordId]
       )
     )
   }

diff --git a/...-notification/src/test/scala/org/hyperledger/identus/messaging/MessagingServiceTest.scala b/...-notification/src/test/scala/org/hyperledger/identus/messaging/MessagingServiceTest.scala
@@ -1,9 +1,9 @@
 package org.hyperledger.identus.messaging
 
+import org.hyperledger.identus.shared.messaging
 import org.hyperledger.identus.shared.messaging.{Message, MessagingService, Serde}
-import org.hyperledger.identus.shared.messaging.kafka.ZKafkaMessagingServiceImpl
-import zio.{durationInt, Random, Schedule, Scope, URIO, ZIO, ZIOAppArgs, ZIOAppDefault, ZLayer}
 import zio.json.{DecoderOps, DeriveJsonDecoder, DeriveJsonEncoder, EncoderOps, JsonDecoder, JsonEncoder}
+import zio.{Random, Schedule, Scope, URIO, ZIO, ZIOAppArgs, ZIOAppDefault, ZLayer, durationInt}
 
 import java.nio.charset.StandardCharsets
 import java.util.UUID
@@ -36,7 +36,8 @@ object MessagingServiceTest extends ZIOAppDefault {
       _ <- ZIO.never
     } yield ()
     effect.provide(
-      ZKafkaMessagingServiceImpl.layer(List("localhost:29092")),
+      messaging.MessagingServiceConfig.inMemoryLayer,
+      messaging.MessagingService.serviceLayer,
       ZLayer.succeed("Sample 'R' passed to handler")
     )
   }

diff --git a/...src/test/scala/org/hyperledger/identus/messaging/kafka/InMemoryMessagingServiceSpec.scala b/...src/test/scala/org/hyperledger/identus/messaging/kafka/InMemoryMessagingServiceSpec.scala
@@ -1,16 +1,14 @@
 package org.hyperledger.identus.messaging.kafka
 
-import org.hyperledger.identus.messaging.*
-import org.hyperledger.identus.shared.messaging.{Consumer, Message, Producer}
-import org.hyperledger.identus.shared.messaging.kafka.InMemoryMessagingService
+import org.hyperledger.identus.shared.messaging.*
 import zio.*
 import zio.test.*
 import zio.test.Assertion.*
 
 object InMemoryMessagingServiceSpec extends ZIOSpecDefault {
-  val testLayer = ZLayer.succeed(100) >+> InMemoryMessagingService.messagingServiceLayer >+>
-    InMemoryMessagingService.producerLayer[String, String] >+>
-    InMemoryMessagingService.consumerLayer[String, String]("test-group")
+  val testLayer = MessagingServiceConfig.inMemoryLayer >+> MessagingService.serviceLayer >+>
+    MessagingService.producerLayer[String, String] >+>
+    MessagingService.consumerLayer[String, String]("test-group")
 
   def spec = suite("InMemoryMessagingServiceSpec")(
     test("should produce and consume messages") {

diff --git a/.../test/scala/org/hyperledger/identus/pollux/core/service/CredentialServiceSpecHelper.scala b/.../test/scala/org/hyperledger/identus/pollux/core/service/CredentialServiceSpecHelper.scala
@@ -3,24 +3,17 @@ package org.hyperledger.identus.pollux.core.service
 import io.circe.Json
 import org.hyperledger.identus.agent.walletapi.memory.GenericSecretStorageInMemory
 import org.hyperledger.identus.agent.walletapi.service.ManagedDIDService
-import org.hyperledger.identus.agent.walletapi.storage.GenericSecretStorage
 import org.hyperledger.identus.castor.core.model.did.PrismDID
 import org.hyperledger.identus.castor.core.service.DIDService
 import org.hyperledger.identus.mercury.model.{AttachmentDescriptor, DidId}
 import org.hyperledger.identus.mercury.protocol.issuecredential.*
 import org.hyperledger.identus.pollux.core.model.*
 import org.hyperledger.identus.pollux.core.model.presentation.Options
-import org.hyperledger.identus.pollux.core.repository.{
-  CredentialDefinitionRepositoryInMemory,
-  CredentialRepositoryInMemory,
-  CredentialStatusListRepositoryInMemory
-}
+import org.hyperledger.identus.pollux.core.repository.{CredentialDefinitionRepositoryInMemory, CredentialRepositoryInMemory, CredentialStatusListRepositoryInMemory}
 import org.hyperledger.identus.pollux.prex.{ClaimFormat, Ldp, PresentationDefinition}
 import org.hyperledger.identus.pollux.vc.jwt.*
 import org.hyperledger.identus.shared.http.UriResolver
-import org.hyperledger.identus.shared.messaging.kafka.InMemoryMessagingService
-import org.hyperledger.identus.shared.messaging.WalletIdAndRecordId
-import org.hyperledger.identus.shared.models.*
+import org.hyperledger.identus.shared.messaging.{MessagingService, MessagingServiceConfig, WalletIdAndRecordId}
 import org.hyperledger.identus.shared.models.{WalletAccessContext, WalletId}
 import zio.*
 
@@ -44,8 +37,8 @@ trait CredentialServiceSpecHelper {
       credentialDefinitionServiceLayer,
       GenericSecretStorageInMemory.layer,
       LinkSecretServiceImpl.layer,
-      (ZLayer.succeed(100) >>> InMemoryMessagingService.messagingServiceLayer >>>
-        InMemoryMessagingService.producerLayer[UUID, WalletIdAndRecordId]).orDie,
+      (MessagingServiceConfig.inMemoryLayer >>> MessagingService.serviceLayer >>>
+        MessagingService.producerLayer[UUID, WalletIdAndRecordId]).orDie,
       CredentialServiceImpl.layer
     )
 

diff --git a/...est/scala/org/hyperledger/identus/pollux/core/service/PresentationServiceSpecHelper.scala b/...est/scala/org/hyperledger/identus/pollux/core/service/PresentationServiceSpecHelper.scala
@@ -1,11 +1,10 @@
 package org.hyperledger.identus.pollux.core.service
 
-import com.nimbusds.jose.jwk.*
 import org.hyperledger.identus.agent.walletapi.memory.GenericSecretStorageInMemory
 import org.hyperledger.identus.castor.core.model.did.DID
-import org.hyperledger.identus.mercury.{AgentPeerService, PeerDID}
 import org.hyperledger.identus.mercury.model.{AttachmentDescriptor, DidId}
 import org.hyperledger.identus.mercury.protocol.presentproof.*
+import org.hyperledger.identus.mercury.{AgentPeerService, PeerDID}
 import org.hyperledger.identus.pollux.core.model.*
 import org.hyperledger.identus.pollux.core.model.error.PresentationError
 import org.hyperledger.identus.pollux.core.repository.*
@@ -14,9 +13,7 @@ import org.hyperledger.identus.pollux.core.service.uriResolvers.ResourceUrlResol
 import org.hyperledger.identus.pollux.vc.jwt.*
 import org.hyperledger.identus.shared.crypto.KmpSecp256k1KeyOps
 import org.hyperledger.identus.shared.http.UriResolver
-import org.hyperledger.identus.shared.messaging.kafka.InMemoryMessagingService
-import org.hyperledger.identus.shared.messaging.WalletIdAndRecordId
-import org.hyperledger.identus.shared.models.*
+import org.hyperledger.identus.shared.messaging.{MessagingService, MessagingServiceConfig, WalletIdAndRecordId}
 import org.hyperledger.identus.shared.models.{WalletAccessContext, WalletId}
 import zio.*
 
@@ -46,8 +43,8 @@ trait PresentationServiceSpecHelper {
     linkSecretLayer,
     PresentationRepositoryInMemory.layer,
     CredentialRepositoryInMemory.layer,
-    (ZLayer.succeed(100) >>> InMemoryMessagingService.messagingServiceLayer >>>
-      InMemoryMessagingService.producerLayer[UUID, WalletIdAndRecordId]).orDie,
+    (MessagingServiceConfig.inMemoryLayer >>> MessagingService.serviceLayer >>>
+      MessagingService.producerLayer[UUID, WalletIdAndRecordId]).orDie,
   ) ++ defaultWalletLayer
 
   def createIssuer(did: String): Issuer = {

diff --git a/shared/core/src/main/scala/org/hyperledger/identus/shared/messaging/MessagingService.scala b/shared/core/src/main/scala/org/hyperledger/identus/shared/messaging/MessagingService.scala
@@ -1,6 +1,7 @@
 package org.hyperledger.identus.shared.messaging
 
-import zio.{durationInt, Cause, Duration, EnvironmentTag, RIO, Task, URIO, ZIO}
+import org.hyperledger.identus.shared.messaging.kafka.{InMemoryMessagingService, ZKafkaMessagingServiceImpl}
+import zio.{durationInt, Cause, Duration, EnvironmentTag, RIO, RLayer, Task, URIO, URLayer, ZIO, ZLayer}
 
 import java.time.Instant
 trait MessagingService {
@@ -69,6 +70,30 @@ object MessagingService {
   )(implicit kSerde: Serde[K], vSerde: Serde[V]): RIO[HR & Producer[K, V] & MessagingService, Unit] =
     consumeWithRetryStrategy(groupId, handler, Seq(RetryStep(topicName, consumerCount, 0.seconds, None)))
 
+  val serviceLayer: URLayer[MessagingServiceConfig, MessagingService] =
+    ZLayer
+      .service[MessagingServiceConfig]
+      .flatMap(config =>
+        if (config.get.kafkaEnabled) ZKafkaMessagingServiceImpl.layer
+        else InMemoryMessagingService.layer
+      )
+
+  def producerLayer[K: EnvironmentTag, V: EnvironmentTag](implicit
+      kSerde: Serde[K],
+      vSerde: Serde[V]
+  ): RLayer[MessagingService, Producer[K, V]] = ZLayer.fromZIO(for {
+    messagingService <- ZIO.service[MessagingService]
+    producer <- messagingService.makeProducer[K, V]()
+  } yield producer)
+
+  def consumerLayer[K: EnvironmentTag, V: EnvironmentTag](groupId: String)(implicit
+      kSerde: Serde[K],
+      vSerde: Serde[V]
+  ): RLayer[MessagingService, Consumer[K, V]] = ZLayer.fromZIO(for {
+    messagingService <- ZIO.service[MessagingService]
+    consumer <- messagingService.makeConsumer[K, V](groupId)
+  } yield consumer)
+
 }
 
 case class Message[K, V](key: K, value: V, offset: Long, timestamp: Long)