OC related

obp-api/src/main/scala/bootstrap/liftweb/Boot.scala

@@ -135,7 +135,8 @@ import code.transactionRequestAttribute.TransactionRequestAttribute
 import code.transactionStatusScheduler.TransactionRequestStatusScheduler
 import code.transaction_types.MappedTransactionType
 import code.transactionattribute.MappedTransactionAttribute
-import code.bankconnectors.opencorridor.{OpenCorridorBankBroker, OpenCorridorOutbox, OpenCorridorOutboxRelay}
+import code.amqpbroker.AmqpBankBroker
+import code.messageoutbox.{MessageOutbox, MessageOutboxRelay}
 import code.transactionrequests.{MappedTransactionRequest, MappedTransactionRequestTypeCharge, TransactionRequestReasons}
 import code.usercustomerlinks.MappedUserCustomerLink
 import code.customerlinks.CustomerLink
@@ -564,7 +565,7 @@ class Boot extends MdcLoggable {
     // Open Corridor: the transactional-outbox relay publishing Interface C messages
     // (credit notifications + settlement instructions) to the banks' own vhosts.
     if (APIUtil.getPropsAsBoolValue("open_corridor_enabled", false)) {
-      OpenCorridorOutboxRelay.start(APIUtil.getPropsAsLongValue("open_corridor.outbox_relay_interval", 10L))
+      MessageOutboxRelay.start(APIUtil.getPropsAsLongValue("open_corridor.outbox_relay_interval", 10L))
     }
     APIUtil.getPropsAsLongValue("database_messages_scheduler_interval") match {
       case Full(i) => DatabaseDriverScheduler.start(i)
@@ -994,8 +995,8 @@ object ToSchemify extends MdcLoggable {
     MappedCounterpartyWhereTag,
     MappedTransactionRequest,
     TransactionRequestAttribute,
-    OpenCorridorBankBroker,
-    OpenCorridorOutbox,
+    AmqpBankBroker,
+    MessageOutbox,
     MappedMetric,
     MetricArchive,
     MetricsArchiveRun,

obp-api/src/main/scala/code/amqpbroker/AmqpBankBroker.scala

@@ -0,0 +1,103 @@
+package code.amqpbroker
+
+import net.liftweb.common.Box
+import net.liftweb.mapper._
+
+/**
+ * Per-bank AMQP broker coordinates — where OBP-API publishes messages destined
+ * for a bank's own infrastructure.
+ *
+ * Named by transport, not by consumer: the fields (host/port/vhost/credentials)
+ * are AMQP 0-9-1 concepts, and any feature that needs to push AMQP messages to
+ * a specific bank resolves its coordinates here. The first consumer is Open
+ * Corridor Interface C: each onboarded bank's Bank Node consumes on its OWN
+ * vhost (e.g. `/bank.ke.01.kcs`) with its own credentials — permission
+ * isolation is enforced at the broker level, and publishing is keyed by
+ * bank_id through this registry (populated at onboarding).
+ *
+ * Transport coordinates only: the bank's on-chain settlement address is NOT
+ * stored here — it is the CARDANO account routing on the bank's
+ * OBP-INCOMING-SETTLEMENT-ACCOUNT.
+ */
+class AmqpBankBroker extends LongKeyedMapper[AmqpBankBroker] with IdPK {
+  def getSingleton = AmqpBankBroker
+
+  object BankId extends MappedString(this, 255) {
+    override def dbColumnName = "bank_id"
+  }
+  object Host extends MappedString(this, 255) {
+    override def dbColumnName = "host"
+  }
+  object Port extends MappedInt(this) {
+    override def dbColumnName = "port"
+    override def defaultValue = 5672
+  }
+  object VirtualHost extends MappedString(this, 255) {
+    override def dbColumnName = "virtual_host"
+  }
+  object Username extends MappedString(this, 255) {
+    override def dbColumnName = "username"
+  }
+  /** Write-only: accepted on registration, never echoed by any endpoint. */
+  object Password extends MappedString(this, 255) {
+    override def dbColumnName = "password"
+  }
+  object UseSsl extends MappedBoolean(this) {
+    override def dbColumnName = "use_ssl"
+    override def defaultValue = false
+  }
+  object CreatedAt extends MappedDateTime(this) {
+    override def dbColumnName = "created_at"
+    override def defaultValue = new java.util.Date()
+  }
+  object UpdatedAt extends MappedDateTime(this) {
+    override def dbColumnName = "updated_at"
+    override def defaultValue = new java.util.Date()
+  }
+
+  def bankId: String = BankId.get
+  def host: String = Host.get
+  def port: Int = Port.get
+  def virtualHost: String = VirtualHost.get
+  def username: String = Username.get
+  def password: String = Password.get
+  def useSsl: Boolean = UseSsl.get
+
+  override def save: Boolean = {
+    UpdatedAt(new java.util.Date())
+    super.save
+  }
+}
+
+object AmqpBankBroker extends AmqpBankBroker with LongKeyedMetaMapper[AmqpBankBroker] {
+  override def dbTableName = "amqp_bank_broker"
+
+  override def dbIndexes: List[BaseIndex[AmqpBankBroker]] = UniqueIndex(BankId) :: super.dbIndexes
+
+  def findByBankId(bankId: String): Box[AmqpBankBroker] =
+    AmqpBankBroker.find(By(AmqpBankBroker.BankId, bankId))
+
+  /** Upsert the broker coordinates for a bank (one row per bank, enforced by the unique index). */
+  def upsert(
+    bankId: String,
+    host: String,
+    port: Int,
+    virtualHost: String,
+    username: String,
+    password: String,
+    useSsl: Boolean
+  ): AmqpBankBroker = {
+    val row = findByBankId(bankId).getOrElse(AmqpBankBroker.create.BankId(bankId))
+    row
+      .Host(host)
+      .Port(port)
+      .VirtualHost(virtualHost)
+      .Username(username)
+      .Password(password)
+      .UseSsl(useSsl)
+      .saveMe()
+  }
+
+  def deleteByBankId(bankId: String): Boolean =
+    AmqpBankBroker.bulkDelete_!!(By(AmqpBankBroker.BankId, bankId))
+}

obp-api/src/main/scala/code/api/util/APIUtil.scala

@@ -1723,6 +1723,15 @@ object APIUtil extends MdcLoggable with CustomJsonFormats{
       _autoValidateRoles = false
       this
     }
+
+    /**
+     * Whether the declared roles are enforced automatically (Lift wrappedWithAuthCheck /
+     * http4s ResourceDocMiddleware). False when the endpoint declared
+     * disableAutoValidateRoles() because enforcement is conditional and done in the
+     * handler — the roles stay in the doc for the catalog / API Explorer / entitlement
+     * requests.
+     */
+    def isAutoValidateRoles: Boolean = _autoValidateRoles
     private var _autoValidateAuthenticate = true
     def disableAutoValidateAuthenticate(): ResourceDoc = {
       _autoValidateAuthenticate = false
@@ -2292,7 +2301,8 @@ object APIUtil extends MdcLoggable with CustomJsonFormats{
                   bankId,
                   userId,
                   role.toString,
-                  "create_just_in_time_entitlements"
+                  "create_just_in_time_entitlements",
+                  grantedByUserId = Some(userId)
                 )
                 logger.info(s"Just in Time Entitlements: $addedEntitlement")
                 addedEntitlement.isDefined
@@ -2347,7 +2357,8 @@ object APIUtil extends MdcLoggable with CustomJsonFormats{
                 hasEntitlement("", userId, ApiRole.canCreateEntitlementAtAnyBank)) &&
                 roles.forall { role =>
                   val addedEntitlement = Entitlement.entitlement.vend.addEntitlement(
-                    bankId, userId, role.toString, "create_just_in_time_entitlements"
+                    bankId, userId, role.toString, "create_just_in_time_entitlements",
+                    grantedByUserId = Some(userId)
                   )
                   logger.info(s"Just in Time Entitlements: $addedEntitlement")
                   addedEntitlement.isDefined

obp-api/src/main/scala/code/api/util/ApiRole.scala

@@ -212,16 +212,27 @@ object ApiRole extends MdcLoggable{
   case class CanAttachOpenCorridorPromise(requiresBankId: Boolean = true) extends ApiRole
   lazy val canAttachOpenCorridorPromise = CanAttachOpenCorridorPromise()
 
-  // Open Corridor: operator role for registering each onboarded bank's RabbitMQ broker
-  // coordinates (host/port/vhost/credentials) in the per-bank publish registry.
-  case class CanConfigureOpenCorridorBroker(requiresBankId: Boolean = false) extends ApiRole
-  lazy val canConfigureOpenCorridorBroker = CanConfigureOpenCorridorBroker()
+  // Operator role for registering each onboarded bank's AMQP broker coordinates
+  // (host/port/vhost/credentials) in the per-bank publish registry. Transport
+  // registry, not corridor-specific; Open Corridor Interface C is the first consumer.
+  case class CanConfigureAmqpBankBroker(requiresBankId: Boolean = false) extends ApiRole
+  lazy val canConfigureAmqpBankBroker = CanConfigureAmqpBankBroker()
 
   // Open Corridor: operator role for the settle-pair trigger — nets a bank pair's
   // PENDING promises, posts the net Transaction and enqueues the Interface C messages.
-  case class CanSettleOpenCorridor(requiresBankId: Boolean = false) extends ApiRole
+  case class CanSettleOpenCorridor(requiresBankId: Boolean = true) extends ApiRole
   lazy val canSettleOpenCorridor = CanSettleOpenCorridor()
 
+  // Operator role for reading the generic message outbox (delivery states,
+  // sticky failures) across all outbox types.
+  case class CanGetMessageOutbox(requiresBankId: Boolean = false) extends ApiRole
+  lazy val canGetMessageOutbox = CanGetMessageOutbox()
+
+  // Operator role for re-queuing a STICKY message-outbox row after
+  // reconciliation — flips it back to PENDING for the relay to redeliver.
+  case class CanRetryMessageOutbox(requiresBankId: Boolean = false) extends ApiRole
+  lazy val canRetryMessageOutbox = CanRetryMessageOutbox()
+
   case class CanAddSocialMediaHandle(requiresBankId: Boolean = true) extends ApiRole
   lazy val canAddSocialMediaHandle = CanAddSocialMediaHandle()
 

obp-api/src/main/scala/code/api/util/ErrorMessages.scala

@@ -889,10 +889,14 @@ object ErrorMessages {
   val OpenCorridorPromiseTypeMismatch = "OBP-40051: The Transaction Request is not of type OPEN_CORRIDOR_PROMISE."
   val OpenCorridorPromiseNotPending = "OBP-40052: The Open Corridor promise Transaction Request is not in PENDING status."
   val OpenCorridorPromiseEvidenceConflict = "OBP-40053: Open Corridor promise evidence is already attached to this Transaction Request with different values. Evidence cannot be overwritten."
-  val OpenCorridorBankBrokerNotConfigured = "OBP-40054: No Open Corridor broker is configured for this bank. Register the bank's RabbitMQ coordinates first."
+  val AmqpBankBrokerNotConfigured = "OBP-40054: No AMQP broker is configured for this bank. Register the bank's AMQP coordinates first."
   val OpenCorridorPublishFailed = "OBP-40055: Could not publish the Open Corridor message to the bank's broker or no reply arrived in time."
   val OpenCorridorSettlementAddressMissing = "OBP-40056: The creditor bank has no settlement address registered in its Open Corridor broker registration, so the settlement instruction cannot be addressed."
   val OpenCorridorDisabled = "OBP-40057: Open Corridor is not enabled on this API instance. Set open_corridor_enabled=true in the props."
+  val OpenCorridorSettlementNotFound = "OBP-40058: No Open Corridor settlement with this SETTLEMENT_ID exists for this bank."
+  val OpenCorridorSameBankNotAllowed = "OBP-40061: OPEN_CORRIDOR is inter-bank: the beneficiary bank must differ from the sending bank. Use an ordinary payment for intra-bank transfers."
+  val MessageOutboxRowNotFound = "OBP-40059: No message outbox row with this OUTBOX_ID exists."
+  val MessageOutboxRowNotSticky = "OBP-40060: The message outbox row is not STICKY. Only STICKY rows can be re-queued; PENDING rows retry automatically."
   // Exceptions (OBP-50XXX)
   val UnknownError = "OBP-50000: Unknown Error."
   val FutureTimeoutException = "OBP-50001: Future Timeout Exception."

obp-api/src/main/scala/code/api/util/http4s/ResourceDocMiddleware.scala

@@ -331,8 +331,11 @@ object ResourceDocMiddleware extends MdcLoggable {
   private def authorizeRoles(resourceDoc: ResourceDoc, pathParams: Map[String, String], ctx: ValidationContext): Validation[ValidationContext] = {
     import DSL._
 
+    // Docs may declare roles purely for the catalog and enforce them conditionally in
+    // the handler (disableAutoValidateRoles, e.g. create-account's "role or self") —
+    // mirror Lift's isNeedCheckRoles = _autoValidateRoles && rolesForCheck.nonEmpty.
     resourceDoc.roles match {
-      case Some(roles) if roles.nonEmpty =>
+      case Some(roles) if roles.nonEmpty && resourceDoc.isAutoValidateRoles =>
         ctx.user match {
           case Full(user) =>
             val bankId = pathParams.getOrElse("BANK_ID", "")

obp-api/src/main/scala/code/api/util/migration/Migration.scala

@@ -136,6 +136,7 @@ object Migration extends MdcLoggable {
       alterMetricColumnUrlLength()
       alterMetricArchiveColumnCorrelationidLength()
       alterCounterpartyLimitFieldType()
+      alterTransactionRequestAttributeValueType()
       changeTypeOfAudFieldAtConsumerTable()
       renameCustomerRoleNames()
       addUniqueIndexOnResourceUserUserId()
@@ -579,6 +580,13 @@ object Migration extends MdcLoggable {
       }
     }
 
+    private def alterTransactionRequestAttributeValueType(): Boolean = {
+      val name = nameOf(alterTransactionRequestAttributeValueType)
+      runOnce(name) {
+        MigrationOfTransactionRequestAttributeValueType.alterColumnValueType(name)
+      }
+    }
+
     private def alterMappedCounterpartyDescriptionLength(): Boolean = {
       val name = nameOf(alterMappedCounterpartyDescriptionLength)
       runOnce(name) {

obp-api/src/main/scala/code/api/util/migration/MigrationOfTransactionRequestAttributeValueType.scala

@@ -0,0 +1,56 @@
+package code.api.util.migration
+
+import code.api.util.APIUtil
+import code.api.util.migration.Migration.{DbFunction, saveLog}
+import code.transactionRequestAttribute.TransactionRequestAttribute
+import net.liftweb.common.Full
+import net.liftweb.mapper.Schemifier
+
+object MigrationOfTransactionRequestAttributeValueType {
+
+  def alterColumnValueType(name: String): Boolean = {
+    DbFunction.tableExists(TransactionRequestAttribute) match {
+      case true =>
+        val startDate = System.currentTimeMillis()
+        val commitId: String = APIUtil.gitCommit
+        var isSuccessful = false
+
+        val executedSql =
+          DbFunction.maybeWrite(true, Schemifier.infoF _) {
+            APIUtil.getPropsValue("db.driver") match {
+              case Full(dbDriver) if dbDriver.contains("com.microsoft.sqlserver.jdbc.SQLServerDriver") =>
+                () =>
+                  """
+                    |-- Open Corridor promise evidence (preimage JSON) exceeds varchar(255)
+                    |ALTER TABLE transactionrequestattribute ALTER COLUMN value VARCHAR(MAX);
+                    |""".stripMargin
+              case _ =>
+                () =>
+                  """
+                    |-- Open Corridor promise evidence (preimage JSON) exceeds varchar(255)
+                    |ALTER TABLE transactionrequestattribute ALTER COLUMN value TYPE text;
+                    |""".stripMargin
+            }
+          }
+
+        val endDate = System.currentTimeMillis()
+        val comment: String =
+          s"""Executed SQL:
+             |$executedSql
+             |""".stripMargin
+        isSuccessful = true
+        saveLog(name, commitId, isSuccessful, startDate, endDate, comment)
+        isSuccessful
+
+      case false =>
+        val startDate = System.currentTimeMillis()
+        val commitId: String = APIUtil.gitCommit
+        val isSuccessful = false
+        val endDate = System.currentTimeMillis()
+        val comment: String =
+          s"""${TransactionRequestAttribute._dbTableNameLC} table does not exist""".stripMargin
+        saveLog(name, commitId, isSuccessful, startDate, endDate, comment)
+        isSuccessful
+    }
+  }
+}

obp-api/src/main/scala/code/api/v2_0_0/Http4s200.scala

@@ -1213,7 +1213,10 @@ object Http4s200 {
               !hasEntitlement(body.bank_id, userId, role)
             }
             addedEntitlement <- Future {
-              unboxFull(Entitlement.entitlement.vend.addEntitlement(body.bank_id, userId, body.role_name))
+              // Audit only (no wire change): record who granted.
+              unboxFull(Entitlement.entitlement.vend.addEntitlement(
+                body.bank_id, userId, body.role_name,
+                grantedByUserId = Some(user.userId)))
             }
           } yield JSONFactory200.createEntitlementJSON(addedEntitlement)
         }

obp-api/src/main/scala/code/api/v4_0_0/Http4s400.scala

@@ -10569,7 +10569,7 @@ object Http4s400 {
         List(apiTagAccount),
         Some(List(canCreateAccount)),
         http4sPartialFunction = Some(addAccount)
-      )
+      ).disableAutoValidateRoles() // Lift parity: "role or self-create" is enforced in the handler
 
       staticResourceDocs += ResourceDoc(
         implementedInApiVersion,

obp-api/src/main/scala/code/api/v5_0_0/Http4s500.scala

@@ -468,12 +468,14 @@ object Http4s500 {
             _ <- entitlementsByBank.exists(_.roleName == CanCreateEntitlementAtOneBank.toString()) match {
               case true  => Future.successful(())
               case false => Future(Entitlement.entitlement.vend.addEntitlement(
-                postJson.id.getOrElse(""), cc.userId, CanCreateEntitlementAtOneBank.toString()))
+                postJson.id.getOrElse(""), cc.userId, CanCreateEntitlementAtOneBank.toString(),
+                grantedByUserId = Some(cc.userId)))
             }
             _ <- entitlementsByBank.exists(_.roleName == CanReadDynamicResourceDocsAtOneBank.toString()) match {
               case true  => Future.successful(())
               case false => Future(Entitlement.entitlement.vend.addEntitlement(
-                postJson.id.getOrElse(""), cc.userId, CanReadDynamicResourceDocsAtOneBank.toString()))
+                postJson.id.getOrElse(""), cc.userId, CanReadDynamicResourceDocsAtOneBank.toString(),
+                grantedByUserId = Some(cc.userId)))
             }
           } yield JSONFactory500.createBankJSON500(success)
         }
@@ -655,7 +657,8 @@ object Http4s500 {
       List(apiTagAccount, apiTagOnboarding),
       Some(List(canCreateAccount)),
       http4sPartialFunction = Some(createAccount)
-    )
+    ) // Lift parity: unlike v4 addAccount, the v5 Lift doc did NOT disableAutoValidateRoles —
+      // canCreateAccount is always enforced; the inline "role or self-create" check is a safety net.
 
     // ─── createUserAuthContext (POST /users/USER_ID/auth-context → 201) ─────