Top Users and Top Consumers in v7.0.0

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

@@ -4874,7 +4874,13 @@ object APIUtil extends MdcLoggable with CustomJsonFormats{
 
   val allowedAnswerTransactionRequestChallengeAttempts = APIUtil.getPropsAsIntValue("answer_transactionRequest_challenge_allowed_attempts").openOr(3)
 
-  lazy val allStaticResourceDocs = (code.api.util.http4s.Http4sResourceDocAggregation.v600
+  // Base is the v7 aggregation — the newest OBP-standard surface, which already contains the
+  // v6.0.0-and-older aggregation plus the v7-only endpoints (deduped by URL/method). Basing on
+  // the v6 aggregation silently excluded v7-only operation ids from everything that resolves
+  // operation ids through this list (api-collection endpoint validation, top-apis lookups, ...).
+  // ResourceDocRegistryParityTest pins the invariant that every per-standard surface the
+  // resource-docs dispatcher can serve is contained here.
+  lazy val allStaticResourceDocs = (code.api.v7_0_0.Http4s700.allResourceDocs
     ++ OBP_UKOpenBanking_200.allResourceDocs
     ++ OBP_UKOpenBanking_310.allResourceDocs
     ++ OBP_UKOpenBanking_401.allResourceDocs
@@ -4885,7 +4891,11 @@ object APIUtil extends MdcLoggable with CustomJsonFormats{
     //    ++ code.api.MxOF.CNBV9_1_0_0.allResourceDocs
     //    ++ code.api.MxOF.OBP_MXOF_1_0_0.allResourceDocs
     //    ++ code.api.BahrainOBF.v1_0_0.ApiCollector.allResourceDocs
-    ++ code.api.berlin.group.v1_3.OBP_BERLIN_GROUP_1_3.allResourceDocs).toList
+    ++ code.api.berlin.group.v1_3.OBP_BERLIN_GROUP_1_3.allResourceDocs
+    // BGv2 was missing here even though /resource-docs/BGv2 serves it, so a BGv2 operation id
+    // (e.g. BGv2-getAccountDetails) failed the getAllResourceDocs membership check that
+    // api-collection-endpoints (and anything else resolving operation ids) relies on.
+    ++ code.api.berlin.group.v2.Http4sBGv2.resourceDocs).toList
 
   def allDynamicResourceDocs= (DynamicEntityHelper.doc ++ DynamicEndpointHelper.doc ++ DynamicEndpoints.dynamicResourceDocs).toList
 

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

@@ -417,6 +417,8 @@ object ErrorMessages {
 
   val CheckbookOrderNotFound = "OBP-30041: CheckbookOrder not found for Account. "
   val GetTopApisError = "OBP-30042: Could not get the top apis from database.  "
+  val GetTopUsersError = "OBP-30551: Could not get the top users from database.  "
+  val GetTopConsumersError = "OBP-30552: Could not get the top consumers from database.  "
   val GetMetricsTopConsumersError = "OBP-30045: Could not get the top consumers from database.  "
   val GetAggregateMetricsError = "OBP-30043: Could not get the aggregate metrics from database.  "
 

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

@@ -141,6 +141,8 @@ object Migration extends MdcLoggable {
       renameCustomerRoleNames()
       addUniqueIndexOnResourceUserUserId()
       addIndexOnMappedMetricUserId()
+      addCompositeIndexOnMetricUserIdDate()
+      addIndexOnResourceUserCreatedByConsentId()
       alterRoleNameLength()
       alterConsentRequestColumnConsumerIdLength()
       alterMappedConsentColumnConsumerIdLength()
@@ -681,6 +683,20 @@ object Migration extends MdcLoggable {
         MigrationOfUserIdIndexes.addIndexOnMappedMetricUserId(name)
       }
     }
+
+    private def addCompositeIndexOnMetricUserIdDate(): Boolean = {
+      val name = nameOf(addCompositeIndexOnMetricUserIdDate)
+      runOnce(name) {
+        MigrationOfActivityDashboardIndexes.addCompositeIndexOnMetricUserIdDate(name)
+      }
+    }
+
+    private def addIndexOnResourceUserCreatedByConsentId(): Boolean = {
+      val name = nameOf(addIndexOnResourceUserCreatedByConsentId)
+      runOnce(name) {
+        MigrationOfActivityDashboardIndexes.addIndexOnResourceUserCreatedByConsentId(name)
+      }
+    }
 
     private def alterRoleNameLength(): Boolean = {
       val name = nameOf(alterRoleNameLength)

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

@@ -0,0 +1,150 @@
+package code.api.util.migration
+
+import code.api.util.APIUtil
+import code.api.util.migration.Migration.{DbFunction, saveLog}
+import code.metrics.MappedMetric
+import code.model.dataAccess.ResourceUser
+import net.liftweb.common.Full
+import net.liftweb.mapper.Schemifier
+
+/**
+ * Indexes for the self-service metrics reads behind the Activity dashboards.
+ *
+ * Declared on the models too (MappedMetric.dbIndexes / ResourceUser.dbIndexes) so fresh
+ * deploys get them from Lift's Schemifier; existing databases get them here, under the
+ * migration framework's control — the composite metric index can take a while to build on
+ * a large table, which is exactly the kind of DDL ops should schedule, not boot should
+ * spring on them.
+ */
+object MigrationOfActivityDashboardIndexes {
+
+  /**
+   * Composite index on Metric(userid, date_c).
+   *
+   * Serves /my/metrics (locks on user ids, filters/sorts/limits on date) and top-users
+   * (groups by userid over a date range) in one pass: equality on user + range and order
+   * on date. The single-column metric_userid_idx (MigrationOfUserIdIndexes) still needs a
+   * per-user sort for these queries and becomes redundant once this exists.
+   * Note: The table name is "Metric" (capital M) and the date column is "date_c".
+   */
+  def addCompositeIndexOnMetricUserIdDate(name: String): Boolean = {
+    DbFunction.tableExists(MappedMetric) 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") =>
+                () =>
+                  """
+                    |-- Check if index exists, if not create it
+                    |IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = 'metric_userid_date_idx' AND object_id = OBJECT_ID('Metric'))
+                    |BEGIN
+                    |    CREATE INDEX metric_userid_date_idx ON Metric(userid, date_c);
+                    |END
+                  """.stripMargin
+              case Full(dbDriver) if dbDriver.contains("com.mysql.cj.jdbc.Driver") =>
+                () =>
+                  """
+                    |-- MySQL: Create index (will fail silently if exists in some versions)
+                    |CREATE INDEX metric_userid_date_idx ON Metric(userid, date_c);
+                  """.stripMargin
+              case _ => // Default (H2, PostgreSQL, etc.)
+                () =>
+                  """
+                    |CREATE INDEX IF NOT EXISTS metric_userid_date_idx ON Metric(userid, date_c);
+                  """.stripMargin
+            }
+          }
+
+        val endDate = System.currentTimeMillis()
+        val comment: String =
+          s"""Added composite index on Metric(userid, date_c)
+             |Executed SQL:
+             |$executedSql
+             |Serves /my/metrics (user ids + date range/order) and top-users (group by userid over a date range).
+             |Note: Table name is "Metric" (capital M); the date column is "date_c".
+             |""".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"""${MappedMetric._dbTableNameLC} table does not exist. Skipping index creation.""".stripMargin
+        saveLog(name, commitId, isSuccessful, startDate, endDate, comment)
+        isSuccessful
+    }
+  }
+
+  /**
+   * Index on resourceuser.createdbyconsentid.
+   *
+   * The delegation registry: consent-agent fan-down (/my/metrics, /my/banks) and
+   * CallContext.effectiveHumanUserId look up agent users by the consent that minted them.
+   * Unindexed this is a full scan of resourceuser on every such request, which matters on
+   * consent-heavy instances where every consent mints a user row.
+   */
+  def addIndexOnResourceUserCreatedByConsentId(name: String): Boolean = {
+    DbFunction.tableExists(ResourceUser) 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") =>
+                () =>
+                  """
+                    |-- Check if index exists, if not create it
+                    |IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = 'resourceuser_createdbyconsentid_idx' AND object_id = OBJECT_ID('resourceuser'))
+                    |BEGIN
+                    |    CREATE INDEX resourceuser_createdbyconsentid_idx ON resourceuser(createdbyconsentid);
+                    |END
+                  """.stripMargin
+              case Full(dbDriver) if dbDriver.contains("com.mysql.cj.jdbc.Driver") =>
+                () =>
+                  """
+                    |-- MySQL: Create index (will fail silently if exists in some versions)
+                    |CREATE INDEX resourceuser_createdbyconsentid_idx ON resourceuser(createdbyconsentid);
+                  """.stripMargin
+              case _ => // Default (H2, PostgreSQL, etc.)
+                () =>
+                  """
+                    |CREATE INDEX IF NOT EXISTS resourceuser_createdbyconsentid_idx ON resourceuser(createdbyconsentid);
+                  """.stripMargin
+            }
+          }
+
+        val endDate = System.currentTimeMillis()
+        val comment: String =
+          s"""Added index on resourceuser.createdbyconsentid
+             |Executed SQL:
+             |$executedSql
+             |Serves the consent-agent delegation fan-down (/my/metrics, /my/banks, effectiveHumanUserId).
+             |""".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"""${ResourceUser._dbTableNameLC} table does not exist. Skipping index creation.""".stripMargin
+        saveLog(name, commitId, isSuccessful, startDate, endDate, comment)
+        isSuccessful
+    }
+  }
+}

obp-api/src/main/scala/code/api/v6_0_0/JSONFactory6.0.0.scala

@@ -451,6 +451,7 @@ case class TopApiJsonV600(
 
 case class TopApisJsonV600(top_apis: List[TopApiJsonV600])
 
+
 case class MetricJsonV600(
     user_id: String,
     url: String,

obp-api/src/main/scala/code/api/v7_0_0/Http4s700.scala

@@ -8,7 +8,7 @@
 import code.api.ResourceDocs1_4_0.SwaggerDefinitionsJSON._
 import code.api.util.APIUtil.{EmptyBody, _}
 import code.api.util.{APIUtil, ApiRole, CallContext, CustomJsonFormats, Glossary, NewStyle}
-import code.api.util.ApiRole.{canAttachOpenCorridorPromise, canConfigureAmqpBankBroker, canGetMessageOutbox, canRetryMessageOutbox, canSettleOpenCorridor, canCreateAccount, canCreateEntitlementAtAnyBank, canCreateEntitlementAtOneBank, canCreateMetricsArchiveRun, canCreateOrganisation, canCreateRoutingScheme, canCreateTestEmail, canCreateUtilityVendResult, canDeleteEntitlementAtAnyBank, canDeleteOrganisation, canDeleteRoutingScheme, canDeleteSchedulerJobLock, canGetAccountAccessTrace, canGetAnyOrganisation, canGetAnyUser, canGetCacheConfig, canGetCacheInfo, canGetCacheNamespaces, canGetConnectorHealth, canGetCustomersAtOneBank, canGetDatabasePoolInfo, canGetMetricsDiagnostics, canGetMigrations, canGetSchedulerJobLocks, canUpdateBankSupportedRoutingScheme, canUpdateOrganisation, canUpdateRoutingScheme, canUpdateSystemView}
+import code.api.util.ApiRole.{canAttachOpenCorridorPromise, canConfigureAmqpBankBroker, canGetMessageOutbox, canRetryMessageOutbox, canSettleOpenCorridor, canCreateAccount, canCreateEntitlementAtAnyBank, canCreateEntitlementAtOneBank, canCreateMetricsArchiveRun, canCreateOrganisation, canCreateRoutingScheme, canCreateTestEmail, canCreateUtilityVendResult, canDeleteEntitlementAtAnyBank, canDeleteOrganisation, canDeleteRoutingScheme, canDeleteSchedulerJobLock, canGetAccountAccessTrace, canGetAnyOrganisation, canGetAnyUser, canGetCacheConfig, canGetCacheInfo, canGetCacheNamespaces, canGetConnectorHealth, canGetCustomersAtOneBank, canGetDatabasePoolInfo, canGetMetricsDiagnostics, canGetMigrations, canGetSchedulerJobLocks, canReadMetrics, canUpdateBankSupportedRoutingScheme, canUpdateOrganisation, canUpdateRoutingScheme, canUpdateSystemView}
 import code.api.util.CommonsEmailWrapper
 import code.model.dataAccess.{AuthUser, BankAccountCreation, MappedBank, ResourceUser}
 import code.consent.Consents
@@ -1209,6 +1209,164 @@
       http4sPartialFunction = Some(getMyMetrics)
     )
 
+    // ─── getTopUsers ──────────────────────────────────────────────────────────────
+
+    val getTopUsers: HttpRoutes[IO] = HttpRoutes.of[IO] {
+      case req @ GET -> `prefixPath` / "management" / "metrics" / "top-users" =>
+        EndpointHelpers.withUser(req) { (_, cc) =>
+          for {
+            httpParams <- NewStyle.function.extractHttpParamsFromUrl(req.uri.renderString)
+            (obpQueryParams, callContext) <- APIUtil.createQueriesByHttpParamsFuture(httpParams, cc.callContext)
+            topUsers <- APIMetrics.apiMetrics.vend.getTopUsersFuture(obpQueryParams) map {
+              APIUtil.unboxFullOrFail(_, callContext, GetTopUsersError)
+            }
+          } yield JSONFactory700.createTopUsersJsonV700(topUsers)
+        }
+    }
+
+    resourceDocs += ResourceDoc(
+      implementedInApiVersion,
+      nameOf(getTopUsers),
+      "GET",
+      "/management/metrics/top-users",
+      "Get Top Users",
+      s"""Get the users behind the API traffic: one row per distinct user with their call count,
+        |sorted by count descending.
+        |
+        |**On-behalf-of aware**: calls made under a Consent (e.g. by an agent or a TPP) are
+        |attributed to the granting (on-behalf-of) user, resolved via the consent table — not to
+        |the consent's technical shadow user. Anonymous calls are excluded. For a given window and
+        |filters the number of distinct users listed here therefore matches the
+        |`distinct_user_count` field of GET /management/aggregate-metrics.
+        |
+        |require CanReadMetrics role
+        |
+        |Should be able to filter on the following fields
+        |
+        |eg: /management/metrics/top-users?from_date=$DateWithMsExampleString&to_date=$DateWithMsExampleString&limit=50
+        |
+        |1 from_date (defaults to one year ago) eg:from_date=$DateWithMsExampleString
+        |
+        |2 to_date (defaults to the current date) eg:to_date=$DateWithMsExampleString
+        |
+        |3 consumer_id  (if null ignore)
+        |
+        |4 user_id (if null ignore)
+        |
+        |5 anon (if null ignore) only support two value : true (return where user_id is null) or false (return where user_id is not null)
+        |
+        |6 url (if null ignore), note: can not contain '&'.
+        |
+        |7 app_name (if null ignore)
+        |
+        |8 implemented_by_partial_function (if null ignore)
+        |
+        |9 implemented_in_version (if null ignore)
+        |
+        |10 verb (if null ignore)
+        |
+        |11 correlation_id (if null ignore)
+        |
+        |12 limit (defaults to 50) eg: limit=200
+        |
+      """.stripMargin,
+      EmptyBody,
+      JSONFactory700.TopUsersJsonV700(List(
+        JSONFactory700.TopUserJsonV700(1000, "9ca9a7e4-6d02-40e3-a129-0b2bf89de9b1", "felixsmith"),
+        JSONFactory700.TopUserJsonV700(250, "8ca8a7e4-6d02-48e3-a029-0b2bf89de9f0", "susan.uk.29@example.com")
+      )),
+      List(
+        $AuthenticatedUserIsRequired,
+        UserHasMissingRoles,
+        InvalidFilterParameterFormat,
+        GetTopUsersError,
+        UnknownError
+      ),
+      apiTagMetric :: apiTagUser :: Nil,
+      Some(canReadMetrics :: Nil),
+      http4sPartialFunction = Some(getTopUsers)
+    )
+
+    // ─── getTopConsumers ──────────────────────────────────────────────────────────
+
+    val getTopConsumers: HttpRoutes[IO] = HttpRoutes.of[IO] {
+      case req @ GET -> `prefixPath` / "management" / "metrics" / "top-consumers" =>
+        EndpointHelpers.withUser(req) { (_, cc) =>
+          for {
+            httpParams <- NewStyle.function.extractHttpParamsFromUrl(req.uri.renderString)
+            (obpQueryParams, callContext) <- APIUtil.createQueriesByHttpParamsFuture(httpParams, cc.callContext)
+            topConsumers <- APIMetrics.apiMetrics.vend.getTopConsumersByConsumerIdFuture(obpQueryParams) map {
+              APIUtil.unboxFullOrFail(_, callContext, GetTopConsumersError)
+            }
+          } yield JSONFactory700.createTopConsumersJsonV700(topConsumers)
+        }
+    }
+
+    resourceDocs += ResourceDoc(
+      implementedInApiVersion,
+      nameOf(getTopConsumers),
+      "GET",
+      "/management/metrics/top-consumers",
+      "Get Top Consumers",
+      s"""Get the Consumers (apps) behind the API traffic: one row per distinct consumer with its
+        |call count, sorted by count descending.
+        |
+        |Unlike the v3.1.0 version — which joins metric rows to consumers by APP NAME, dropping
+        |calls whose app name no longer matches a consumer and double-counting duplicate names —
+        |this groups by the consumer id stored on each metric row. For a given window and filters
+        |the number of distinct consumers listed here therefore matches the
+        |`distinct_consumer_count` field of GET /management/aggregate-metrics. Calls that carried
+        |no consumer are excluded. `app_name` and `developer_email` are empty when the consumer
+        |row no longer exists.
+        |
+        |require CanReadMetrics role
+        |
+        |Should be able to filter on the following fields
+        |
+        |eg: /management/metrics/top-consumers?from_date=$DateWithMsExampleString&to_date=$DateWithMsExampleString&limit=50
+        |
+        |1 from_date (defaults to one year ago) eg:from_date=$DateWithMsExampleString
+        |
+        |2 to_date (defaults to the current date) eg:to_date=$DateWithMsExampleString
+        |
+        |3 consumer_id  (if null ignore)
+        |
+        |4 user_id (if null ignore)
+        |
+        |5 anon (if null ignore) only support two value : true (return where user_id is null) or false (return where user_id is not null)
+        |
+        |6 url (if null ignore), note: can not contain '&'.
+        |
+        |7 app_name (if null ignore)
+        |
+        |8 implemented_by_partial_function (if null ignore)
+        |
+        |9 implemented_in_version (if null ignore)
+        |
+        |10 verb (if null ignore)
+        |
+        |11 correlation_id (if null ignore)
+        |
+        |12 limit (defaults to 50) eg: limit=200
+        |
+      """.stripMargin,
+      EmptyBody,
+      JSONFactory700.TopConsumersJsonV700(List(
+        JSONFactory700.TopConsumerJsonV700(1000, "7uy8a7e4-6d02-40e3-a129-0b2bf89de8uh", "API-EXPLORER", "developer@example.com"),
+        JSONFactory700.TopConsumerJsonV700(250, "8uy8a7e4-6d02-40e3-a129-0b2bf89de8uh", "API-Manager", "manager@example.com")
+      )),
+      List(
+        $AuthenticatedUserIsRequired,
+        UserHasMissingRoles,
+        InvalidFilterParameterFormat,
+        GetTopConsumersError,
+        UnknownError
+      ),
+      apiTagMetric :: apiTagApi :: Nil,
+      Some(canReadMetrics :: Nil),
+      http4sPartialFunction = Some(getTopConsumers)
+    )
+
     // ── Trading Endpoints ──────────────────────────────────────────────────
 
     // Route: POST /obp/v7.0.0/banks/BANK_ID/accounts/ACCOUNT_ID/views/VIEW_ID/trading/offers