feat(prism-agent): simple event mechanism using webhook

build.sbt

@@ -150,7 +150,7 @@ lazy val D_Connect = new {
 
   // Dependency Modules
   private lazy val baseDependencies: Seq[ModuleID] =
-    Seq(D.zio, D.zioTest, D.zioTestSbt, D.zioTestMagnolia, D.testcontainersPostgres, logback)
+    Seq(D.zio, D.zioTest, D.zioTestSbt, D.zioTestMagnolia, D.zioMock, D.testcontainersPostgres, logback)
 
   // Project Dependencies
   lazy val coreDependencies: Seq[ModuleID] =
@@ -223,6 +223,7 @@ lazy val D_Pollux = new {
     D.zioTest,
     D.zioTestSbt,
     D.zioTestMagnolia,
+    D.zioMock,
     D.munit,
     D.munitZio,
     prismCrypto,
@@ -280,6 +281,17 @@ lazy val D_Pollux_VC_JWT = new {
   lazy val polluxVcJwtDependencies: Seq[ModuleID] = baseDependencies
 }
 
+lazy val D_EventNotification = new {
+  val zio = "dev.zio" %% "zio" % V.zio
+  val zioConcurrent = "dev.zio" %% "zio-concurrent" % V.zio
+  val zioTest = "dev.zio" %% "zio-test" % V.zio % Test
+  val zioTestSbt = "dev.zio" %% "zio-test-sbt" % V.zio % Test
+  val zioTestMagnolia = "dev.zio" %% "zio-test-magnolia" % V.zio % Test
+
+  val zioDependencies: Seq[ModuleID] = Seq(zio, zioConcurrent, zioTest, zioTestSbt, zioTestMagnolia)
+  val baseDependencies: Seq[ModuleID] = zioDependencies
+}
+
 lazy val D_PrismAgent = new {
 
   // Added here to make prism-crypto works.
@@ -632,7 +644,7 @@ lazy val polluxCore = project
   )
   .dependsOn(shared)
   .dependsOn(polluxVcJWT)
-  .dependsOn(protocolIssueCredential, protocolPresentProof, resolver, agentDidcommx)
+  .dependsOn(protocolIssueCredential, protocolPresentProof, resolver, agentDidcommx, eventNotification)
 
 lazy val polluxDoobie = project
   .in(file("pollux/lib/sql-doobie"))
@@ -687,7 +699,7 @@ lazy val connectCore = project
     Test / publishArtifact := true
   )
   .dependsOn(shared)
-  .dependsOn(protocolConnection, protocolReportProblem)
+  .dependsOn(protocolConnection, protocolReportProblem, eventNotification)
 
 lazy val connectDoobie = project
   .in(file("connect/lib/sql-doobie"))
@@ -699,6 +711,17 @@ lazy val connectDoobie = project
   .dependsOn(shared)
   .dependsOn(connectCore % "compile->compile;test->test")
 
+// ############################
+// #### Event Notification ####
+// ############################
+
+lazy val eventNotification = project
+  .in(file("event-notification"))
+  .settings(
+    name := "event-notification",
+    libraryDependencies ++= D_EventNotification.baseDependencies
+  )
+
 // #####################
 // #### Prism Agent ####
 // #####################
@@ -711,8 +734,11 @@ lazy val prismAgentWalletAPI = project
     name := "prism-agent-wallet-api",
     libraryDependencies ++= D_PrismAgent.keyManagementDependencies
   )
-  .dependsOn(agentDidcommx)
-  .dependsOn(castorCore)
+  .dependsOn(
+    agentDidcommx,
+    castorCore,
+    eventNotification
+  )
 
 lazy val prismAgentServer = project
   .in(file("prism-agent/service/server"))
@@ -741,7 +767,8 @@ lazy val prismAgentServer = project
     polluxAnoncreds,
     connectCore,
     connectDoobie,
-    castorCore
+    castorCore,
+    eventNotification
   )
 
 // ##################
@@ -822,6 +849,7 @@ lazy val aggregatedProjects: Seq[ProjectReference] = Seq(
   prismAgentWalletAPI,
   prismAgentServer,
   mediator,
+  eventNotification,
 )
 
 lazy val root = project

connect/lib/core/src/main/scala/io/iohk/atala/connect/core/service/ConnectionServiceNotifier.scala

@@ -0,0 +1,89 @@
+package io.iohk.atala.connect.core.service
+
+import io.iohk.atala.connect.core.model.ConnectionRecord
+import io.iohk.atala.connect.core.model.error.ConnectionServiceError
+import io.iohk.atala.event.notification.{Event, EventNotificationService}
+import io.iohk.atala.mercury.model.DidId
+import io.iohk.atala.mercury.protocol.connection.{ConnectionRequest, ConnectionResponse}
+import zio.{IO, URLayer, ZIO, ZLayer}
+
+import java.util.UUID
+
+class ConnectionServiceNotifier(
+    svc: ConnectionService,
+    eventNotificationService: EventNotificationService
+) extends ConnectionService {
+
+  private val connectionUpdatedEvent = "ConnectionUpdated"
+
+  override def createConnectionInvitation(
+      label: Option[String],
+      pairwiseDID: DidId
+  ): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.createConnectionInvitation(label, pairwiseDID))
+
+  override def receiveConnectionInvitation(invitation: String): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.receiveConnectionInvitation(invitation))
+
+  override def acceptConnectionInvitation(
+      recordId: UUID,
+      pairwiseDid: DidId
+  ): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.acceptConnectionInvitation(recordId, pairwiseDid))
+
+  override def markConnectionRequestSent(recordId: UUID): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.markConnectionRequestSent(recordId))
+
+  override def receiveConnectionRequest(request: ConnectionRequest): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.receiveConnectionRequest(request))
+
+  override def acceptConnectionRequest(recordId: UUID): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.acceptConnectionRequest(recordId))
+
+  override def markConnectionResponseSent(recordId: UUID): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.markConnectionResponseSent(recordId))
+
+  override def receiveConnectionResponse(response: ConnectionResponse): IO[ConnectionServiceError, ConnectionRecord] =
+    notifyOnSuccess(svc.receiveConnectionResponse(response))
+
+  private[this] def notifyOnSuccess(effect: IO[ConnectionServiceError, ConnectionRecord]) =
+    for {
+      record <- effect
+      _ <- notify(record)
+    } yield record
+
+  private[this] def notify(record: ConnectionRecord) = {
+    val result = for {
+      producer <- eventNotificationService.producer[ConnectionRecord]("Connect")
+      _ <- producer.send(Event(connectionUpdatedEvent, record))
+    } yield ()
+    result.catchAll(e => ZIO.logError(s"Notification service error: $e"))
+  }
+
+  override def getConnectionRecord(recordId: UUID): IO[ConnectionServiceError, Option[ConnectionRecord]] =
+    svc.getConnectionRecord(recordId)
+
+  override def getConnectionRecordByThreadId(thid: String): IO[ConnectionServiceError, Option[ConnectionRecord]] =
+    svc.getConnectionRecordByThreadId(thid)
+
+  override def deleteConnectionRecord(recordId: UUID): IO[ConnectionServiceError, Int] =
+    svc.deleteConnectionRecord(recordId)
+
+  override def reportProcessingFailure(recordId: UUID, failReason: Option[String]): IO[ConnectionServiceError, Unit] =
+    svc.reportProcessingFailure(recordId, failReason)
+
+  override def getConnectionRecords(): IO[ConnectionServiceError, Seq[ConnectionRecord]] =
+    svc.getConnectionRecords()
+
+  override def getConnectionRecordsByStates(
+      ignoreWithZeroRetries: Boolean,
+      limit: Int,
+      states: ConnectionRecord.ProtocolState*
+  ): IO[ConnectionServiceError, Seq[ConnectionRecord]] =
+    svc.getConnectionRecordsByStates(ignoreWithZeroRetries, limit, states: _*)
+}
+
+object ConnectionServiceNotifier {
+  val layer: URLayer[ConnectionService & EventNotificationService, ConnectionService] =
+    ZLayer.fromFunction(ConnectionServiceNotifier(_, _))
+}

connect/lib/core/src/main/scala/io/iohk/atala/connect/core/service/MockConnectionService.scala

@@ -0,0 +1,80 @@
+package io.iohk.atala.connect.core.service
+
+import io.iohk.atala.connect.core.model.ConnectionRecord
+import io.iohk.atala.connect.core.model.error.ConnectionServiceError
+import io.iohk.atala.mercury.model.DidId
+import io.iohk.atala.mercury.protocol.connection.{ConnectionRequest, ConnectionResponse}
+import zio.mock.{Mock, Proxy}
+import zio.{IO, URLayer, ZIO, ZLayer, mock}
+
+import java.util.UUID
+
+object MockConnectionService extends Mock[ConnectionService] {
+
+  object CreateConnectionInvitation extends Effect[(Option[String], DidId), ConnectionServiceError, ConnectionRecord]
+  object ReceiveConnectionInvitation extends Effect[String, ConnectionServiceError, ConnectionRecord]
+  object AcceptConnectionInvitation extends Effect[(UUID, DidId), ConnectionServiceError, ConnectionRecord]
+  object MarkConnectionRequestSent extends Effect[UUID, ConnectionServiceError, ConnectionRecord]
+  object ReceiveConnectionRequest extends Effect[ConnectionRequest, ConnectionServiceError, ConnectionRecord]
+  object AcceptConnectionRequest extends Effect[UUID, ConnectionServiceError, ConnectionRecord]
+  object MarkConnectionResponseSent extends Effect[UUID, ConnectionServiceError, ConnectionRecord]
+  object ReceiveConnectionResponse extends Effect[ConnectionResponse, ConnectionServiceError, ConnectionRecord]
+
+  override val compose: URLayer[mock.Proxy, ConnectionService] = ZLayer {
+    for {
+      proxy <- ZIO.service[Proxy]
+    } yield new ConnectionService {
+      override def createConnectionInvitation(
+          label: Option[String],
+          pairwiseDID: DidId
+      ): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(CreateConnectionInvitation, label, pairwiseDID)
+
+      override def receiveConnectionInvitation(invitation: String): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(ReceiveConnectionInvitation, invitation)
+
+      override def acceptConnectionInvitation(
+          recordId: UUID,
+          pairwiseDid: DidId
+      ): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(AcceptConnectionInvitation, recordId, pairwiseDid)
+
+      override def markConnectionRequestSent(recordId: UUID): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(MarkConnectionRequestSent, recordId)
+
+      override def receiveConnectionRequest(request: ConnectionRequest): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(ReceiveConnectionRequest, request)
+
+      override def acceptConnectionRequest(recordId: UUID): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(AcceptConnectionRequest, recordId)
+
+      override def markConnectionResponseSent(recordId: UUID): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(MarkConnectionResponseSent, recordId)
+
+      override def receiveConnectionResponse(
+          response: ConnectionResponse
+      ): IO[ConnectionServiceError, ConnectionRecord] =
+        proxy(ReceiveConnectionResponse, response)
+
+      override def getConnectionRecords(): IO[ConnectionServiceError, Seq[ConnectionRecord]] = ???
+
+      override def getConnectionRecordsByStates(
+          ignoreWithZeroRetries: Boolean,
+          limit: Int,
+          states: ConnectionRecord.ProtocolState*
+      ): IO[ConnectionServiceError, Seq[ConnectionRecord]] = ???
+
+      override def getConnectionRecord(recordId: UUID): IO[ConnectionServiceError, Option[ConnectionRecord]] = ???
+
+      override def getConnectionRecordByThreadId(thid: String): IO[ConnectionServiceError, Option[ConnectionRecord]] =
+        ???
+
+      override def deleteConnectionRecord(recordId: UUID): IO[ConnectionServiceError, Int] = ???
+
+      override def reportProcessingFailure(
+          recordId: UUID,
+          failReason: Option[String]
+      ): IO[ConnectionServiceError, Unit] = ???
+    }
+  }
+}