feat(approvals): live Approvals badge counts over one SSE stream
Fineract E2E Tests / E2E Tests (Shard 8 of 10) (push) Waiting to run
Fineract Docker Builds / build (false, docker-compose-postgresql.yml, postgresql) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 9 of 10) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-1) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-2) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-3) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-4) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-5) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-1) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-2) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 10 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 1 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 3 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 4 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 5 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 6 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 7 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 2 of 10) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-3) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-4) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-5) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-1) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-2) (push) Waiting to run
Fineract Docker Builds / build (true, docker-compose.yml, mariadb) (push) Waiting to run
Fineract Documentation build / build (push) Waiting to run
Fineract Build & Cucumber tests (without E2E tests) / verify (main, build-core) (push) Waiting to run
Fineract Build & Cucumber tests (without E2E tests) / verify (main, cucumber) (push) Waiting to run
Fineract Build & Cucumber tests (without E2E tests) / verify (progressive-loan, build-progressive-loan) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-3) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-4) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-5) (push) Waiting to run
Fineract Liquibase Only mode - PostgreSQL / test (push) Waiting to run
Fineract Messaging Smoke Tests / Smoke Test with ActiveMQ (push) Waiting to run
Fineract Messaging Smoke Tests / Smoke Test with Kafka (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 8 of 10) (push) Waiting to run
Fineract Docker Builds / build (false, docker-compose-postgresql.yml, postgresql) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 9 of 10) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-1) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-2) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-3) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-4) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MariaDB / test (test-core-5) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-1) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-2) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 10 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 1 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 3 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 4 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 5 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 6 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 7 of 10) (push) Waiting to run
Fineract E2E Tests / E2E Tests (Shard 2 of 10) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-3) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-4) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - MySQL / test (test-core-5) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-1) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-2) (push) Waiting to run
Fineract Docker Builds / build (true, docker-compose.yml, mariadb) (push) Waiting to run
Fineract Documentation build / build (push) Waiting to run
Fineract Build & Cucumber tests (without E2E tests) / verify (main, build-core) (push) Waiting to run
Fineract Build & Cucumber tests (without E2E tests) / verify (main, cucumber) (push) Waiting to run
Fineract Build & Cucumber tests (without E2E tests) / verify (progressive-loan, build-progressive-loan) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-3) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-4) (push) Waiting to run
Fineract Cargo & Unit- & Integration tests - PostgreSQL / test (test-core-5) (push) Waiting to run
Fineract Liquibase Only mode - PostgreSQL / test (push) Waiting to run
Fineract Messaging Smoke Tests / Smoke Test with ActiveMQ (push) Waiting to run
Fineract Messaging Smoke Tests / Smoke Test with Kafka (push) Waiting to run
One endpoint now answers how many requests wait on every Approval Center tab, so the screen no longer makes a call per tab. - GET /v1/approvals/counts: snapshot, numbers only (held 3s, fresh=true skips). - GET /v1/approvals/counts/stream: Server-Sent Events. A `counts` event is written the moment the numbers change; one shared 1s check serves every open screen and callers counted alike share one query. Keep-alive ping every 15s, stream recycled every 5 minutes. - All 26 tabs are counted by one statement of scalar sub-selects, each the same table/status/office scope/exclusions as the tab's own API. A tab the caller may not see is not asked (permissions mirrored, disbursement for managers only). - Layered: thin API resource, ApprovalCountsReadPlatformService (+Impl), ApprovalCountsStreamService, ApprovalCounterRegistry (one line per tab). - Adds jersey-media-sse (SseEventSink / Sse). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -81,6 +81,8 @@ dependencies {
|
||||
'org.apache.avro:avro',
|
||||
'jakarta.ws.rs:jakarta.ws.rs-api',
|
||||
'org.glassfish.jersey.media:jersey-media-multipart',
|
||||
// Server-Sent Events (SseEventSink, Sse): the Approvals badge is pushed to the screen over one.
|
||||
'org.glassfish.jersey.media:jersey-media-sse',
|
||||
|
||||
'org.springframework:spring-jms',
|
||||
'org.springframework:spring-context-support',
|
||||
|
||||
+67
@@ -0,0 +1,67 @@
|
||||
package org.apache.fineract.portfolio.approvals.api;
|
||||
|
||||
import jakarta.servlet.http.HttpServletResponse;
|
||||
import jakarta.ws.rs.DefaultValue;
|
||||
import jakarta.ws.rs.GET;
|
||||
import jakarta.ws.rs.Path;
|
||||
import jakarta.ws.rs.Produces;
|
||||
import jakarta.ws.rs.QueryParam;
|
||||
import jakarta.ws.rs.core.CacheControl;
|
||||
import jakarta.ws.rs.core.Context;
|
||||
import jakarta.ws.rs.core.MediaType;
|
||||
import jakarta.ws.rs.core.Response;
|
||||
import jakarta.ws.rs.sse.Sse;
|
||||
import jakarta.ws.rs.sse.SseEventSink;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
|
||||
import org.apache.fineract.portfolio.approvals.service.ApprovalCountsReadPlatformService;
|
||||
import org.apache.fineract.portfolio.approvals.service.ApprovalCountsStreamService;
|
||||
import org.apache.fineract.useradministration.domain.AppUser;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* How many requests wait on every Approval Center tab, for the badge on each tab and the Total on the Approvals menu.
|
||||
* Numbers only: opening a tab still loads that tab's own list. The counting is in
|
||||
* {@link ApprovalCountsReadPlatformService}; the live stream is in {@link ApprovalCountsStreamService}.
|
||||
*/
|
||||
@Path("/v1/approvals/counts")
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
@Produces({ MediaType.APPLICATION_JSON })
|
||||
public class ApprovalCountsApiResource {
|
||||
|
||||
private final PlatformSecurityContext context;
|
||||
private final ApprovalCountsReadPlatformService readService;
|
||||
private final ApprovalCountsStreamService streamService;
|
||||
|
||||
/**
|
||||
* The counts as they are now, answered at once. {@code fresh=true} skips the few seconds the last answer is held
|
||||
* for. Screens that can stream use {@link #stream} instead; this is for a one-off read and as a fallback.
|
||||
*/
|
||||
@GET
|
||||
public Response counts(@QueryParam("fresh") @DefaultValue("false") final boolean fresh) {
|
||||
final AppUser user = this.context.authenticatedUser();
|
||||
final CacheControl noStore = new CacheControl();
|
||||
noStore.setNoStore(true);
|
||||
return Response.ok(this.readService.retrieveHeld(user, fresh).toJson()).cacheControl(noStore).build();
|
||||
}
|
||||
|
||||
/**
|
||||
* Server-Sent Events: the connection stays open and a {@code counts} event is written the moment the numbers change,
|
||||
* so a screen neither polls nor reloads. The first event carries the numbers as they stand.
|
||||
*/
|
||||
@GET
|
||||
@Path("stream")
|
||||
@Produces(MediaType.SERVER_SENT_EVENTS)
|
||||
public void stream(@Context final SseEventSink sink, @Context final Sse sse, @Context final HttpServletResponse response) {
|
||||
// Ask reverse proxies not to hold the events back.
|
||||
response.setHeader("X-Accel-Buffering", "no");
|
||||
response.setHeader("Cache-Control", "no-cache, no-store");
|
||||
try {
|
||||
this.streamService.open(this.readService.planFor(this.context.authenticatedUser()), sink, sse);
|
||||
} catch (final RuntimeException e) {
|
||||
sink.close();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
+14
@@ -0,0 +1,14 @@
|
||||
package org.apache.fineract.portfolio.approvals.data;
|
||||
|
||||
/**
|
||||
* One Approval Center tab and the question that counts what waits on it.
|
||||
*
|
||||
* @param key
|
||||
* the tab's route segment, which is also the key its badge reads on the screen
|
||||
* @param permission
|
||||
* the permission the tab is shown under (READ_* is also met by ALL_FUNCTIONS_READ); null means every caller sees
|
||||
* the tab
|
||||
* @param sql
|
||||
* a scalar sub-select answering "how many", written with the named parameters of {@link ApprovalCountsPlan}
|
||||
*/
|
||||
public record ApprovalCounter(String key, String permission, String sql) {}
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
package org.apache.fineract.portfolio.approvals.data;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* What is waiting for approval: the total, the number for each tab that has anything, and a version that is equal
|
||||
* exactly when the numbers are, which is how a change is told from a repeat.
|
||||
*/
|
||||
public record ApprovalCountsData(long total, Map<String, Long> counts, String version) {
|
||||
|
||||
private static final Gson GSON = new Gson();
|
||||
|
||||
public static ApprovalCountsData of(final Map<String, Long> counts) {
|
||||
final long total = counts.values().stream().mapToLong(Long::longValue).sum();
|
||||
return new ApprovalCountsData(total, counts, Integer.toHexString(counts.hashCode()));
|
||||
}
|
||||
|
||||
public String toJson() {
|
||||
final Map<String, Object> json = new LinkedHashMap<>();
|
||||
json.put("total", this.total);
|
||||
json.put("counts", this.counts);
|
||||
json.put("version", this.version);
|
||||
return GSON.toJson(json);
|
||||
}
|
||||
}
|
||||
+22
@@ -0,0 +1,22 @@
|
||||
package org.apache.fineract.portfolio.approvals.data;
|
||||
|
||||
import java.util.List;
|
||||
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
|
||||
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
|
||||
|
||||
/**
|
||||
* What one caller is counted for: the tabs they may see and the values (office scope, user, checker rights) those
|
||||
* tabs' queries are asked with. It is worked out on the request thread, because it reads the user's roles, and can then
|
||||
* be used from any thread.
|
||||
*
|
||||
* @param tenant
|
||||
* the tenant the counts are read from
|
||||
* @param visible
|
||||
* the tabs this caller may see, in the order the Approval Center shows them
|
||||
* @param params
|
||||
* the named parameters the tabs' queries use
|
||||
* @param group
|
||||
* equal for callers who are counted identically, so they can share one query
|
||||
*/
|
||||
public record ApprovalCountsPlan(FineractPlatformTenant tenant, List<ApprovalCounter> visible, MapSqlParameterSource params,
|
||||
String group) {}
|
||||
+122
@@ -0,0 +1,122 @@
|
||||
package org.apache.fineract.portfolio.approvals.service;
|
||||
|
||||
import java.util.List;
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCounter;
|
||||
|
||||
/**
|
||||
* Every Approval Center tab that carries a badge, and the question that counts what waits on it. Each question is the
|
||||
* one the tab asks of its own API (same table, same status, same office scope, same entries left out). To badge a new
|
||||
* tab, add one line here and one badge on the screen.
|
||||
*
|
||||
* The queries use these named parameters, supplied per caller: {@code :h} (office hierarchy pattern), {@code :ho} (head
|
||||
* office), {@code :uid} (user) and {@code :superChecker} (may check every maker-checker command).
|
||||
*/
|
||||
public final class ApprovalCounterRegistry {
|
||||
|
||||
private ApprovalCounterRegistry() {}
|
||||
|
||||
/** Entities whose maker-checker requests are shown under Product Approval, not in the Checker Inbox. */
|
||||
private static final String PRODUCT_ENTITIES = "'LOANPRODUCT','SAVINGSPRODUCT','CURRENTPRODUCT','SHAREPRODUCT',"
|
||||
+ "'FIXEDDEPOSITPRODUCT','RECURRINGDEPOSITPRODUCT','CHARGE'";
|
||||
|
||||
/** Product edits that are staged in m_saving_product_temp, so Product Approval already lists them from there. */
|
||||
private static final String STAGED_PRODUCT_ENTITIES = "'SAVINGSPRODUCT','CURRENTPRODUCT','SHAREPRODUCT',"
|
||||
+ "'FIXEDDEPOSITPRODUCT','RECURRINGDEPOSITPRODUCT'";
|
||||
|
||||
/** Accounts whose UPDATE is staged for Pending Account Edits, so the Checker Inbox leaves it out. */
|
||||
private static final String STAGED_ACCOUNT_ENTITIES = "'SAVINGSACCOUNT','CURRENTACCOUNT'";
|
||||
|
||||
private static final String DDS_ENTITIES = "'PARTY','AGENT','PARTYTRANSACTION','DDSACCOUNTTYPE','COMMISSIONSLAB',"
|
||||
+ "'DDSBATCH','DDSCOLLECTIONENTRY'";
|
||||
|
||||
/** Maker-checker commands awaiting a checker, limited to the offices and checker rights the caller has. */
|
||||
private static final String MC_PENDING = "FROM m_portfolio_command_source aud WHERE aud.status = 2 "
|
||||
+ "AND (:ho OR EXISTS (SELECT 1 FROM m_office o2 WHERE o2.id = aud.office_id AND o2.hierarchy LIKE :h)) "
|
||||
+ "AND (:superChecker OR EXISTS (SELECT 1 FROM m_permission p JOIN m_role_permission rp ON rp.permission_id = p.id "
|
||||
+ "JOIN m_appuser_role ur ON ur.role_id = rp.role_id AND ur.appuser_id = :uid "
|
||||
+ "WHERE p.code LIKE '%\\_CHECKER' AND REPLACE(p.action_name, '_CHECKER', '') = aud.action_name "
|
||||
+ "AND p.entity_name = aud.entity_name)) ";
|
||||
|
||||
/** Where the person who raised a loan or account sits, for the office scope the tab applies. */
|
||||
private static final String CLIENT_OFFICE = "EXISTS (SELECT 1 FROM m_office o WHERE o.id = COALESCE(c.office_id, g.office_id) "
|
||||
+ "AND o.hierarchy LIKE :h)";
|
||||
|
||||
/** The one tab whose list is open to managers and head cashiers only. */
|
||||
public static final String DISBURSEMENT_REQUESTS = "disbursement-requests";
|
||||
|
||||
/** The tabs, in the order the Approval Center shows them. A null permission means every caller sees the tab. */
|
||||
public static final List<ApprovalCounter> ALL = List.of(
|
||||
// ── Fineract's own maker-checker and pending lists ────────────────────────────
|
||||
new ApprovalCounter("checker-inbox", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) " + MC_PENDING + "AND upper(aud.entity_name) NOT IN (" + PRODUCT_ENTITIES + ") "
|
||||
+ "AND NOT (upper(aud.entity_name) IN (" + STAGED_ACCOUNT_ENTITIES + ") AND upper(aud.action_name) = 'UPDATE')"),
|
||||
new ApprovalCounter("client-approval", "READ_CLIENT",
|
||||
"SELECT count(*) FROM m_client c WHERE c.status_enum = 100 "
|
||||
+ "AND EXISTS (SELECT 1 FROM m_office o WHERE o.id = c.office_id AND o.hierarchy LIKE :h)"),
|
||||
new ApprovalCounter("loan-approval", "READ_LOAN",
|
||||
"SELECT count(*) FROM m_loan l LEFT JOIN m_client c ON c.id = l.client_id LEFT JOIN m_group g ON g.id = l.group_id "
|
||||
+ "WHERE l.loan_status_id = 100 AND " + CLIENT_OFFICE),
|
||||
new ApprovalCounter("loan-disbursal", "READ_LOAN",
|
||||
"SELECT count(*) FROM m_loan l LEFT JOIN m_client c ON c.id = l.client_id LEFT JOIN m_group g ON g.id = l.group_id "
|
||||
+ "WHERE l.loan_status_id = 200 AND " + CLIENT_OFFICE),
|
||||
new ApprovalCounter("reschedule-loan", "RESCHEDULE_LOAN",
|
||||
"SELECT count(*) FROM m_loan_reschedule_request WHERE status_enum = 100"),
|
||||
new ApprovalCounter("pending-account", "ALL_FUNCTIONS_READ",
|
||||
"SELECT (SELECT count(*) FROM m_savings_account s LEFT JOIN m_client c ON c.id = s.client_id "
|
||||
+ "LEFT JOIN m_group g ON g.id = s.group_id WHERE s.status_enum = 100 "
|
||||
+ "AND COALESCE(s.terminate_true, false) = false AND " + CLIENT_OFFICE
|
||||
+ ") + (SELECT count(*) FROM m_current_account ca WHERE ca.status_enum = 100 "
|
||||
+ "AND COALESCE(ca.terminate_true, false) = false "
|
||||
+ "AND EXISTS (SELECT 1 FROM m_office o WHERE o.id = ca.office_id AND o.hierarchy LIKE :h))"),
|
||||
new ApprovalCounter("pending-share-accounts", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_share_account WHERE status_enum = 100"),
|
||||
new ApprovalCounter("product-approval", "ALL_FUNCTIONS_READ",
|
||||
"SELECT (SELECT count(*) FROM m_saving_product_temp WHERE status = 'PENDING' "
|
||||
+ "AND product_type IN ('SAVINGS','CURRENT','SHARE','FD','RD')) + (SELECT count(*) " + MC_PENDING
|
||||
+ "AND upper(aud.entity_name) IN (" + PRODUCT_ENTITIES + ") "
|
||||
+ "AND NOT (upper(aud.entity_name) IN (" + STAGED_PRODUCT_ENTITIES + ") AND upper(aud.action_name) = 'UPDATE'))"),
|
||||
new ApprovalCounter("pending-savings-account-edits", "ALL_FUNCTIONS_READ",
|
||||
"SELECT (SELECT count(*) FROM m_savings_account_pending_edit WHERE status = 'PENDING') "
|
||||
+ "+ (SELECT count(*) FROM m_current_account_pending_edit WHERE status = 'PENDING') "
|
||||
+ "+ (SELECT count(*) FROM m_share_account_pending_edit WHERE status = 'PENDING')"),
|
||||
new ApprovalCounter("pending-client-edits", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_client_pending_edit pe JOIN m_client c ON c.id = pe.client_id WHERE pe.status = 'PENDING'"),
|
||||
new ApprovalCounter("pending-user-edits", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_user_pending_request WHERE status = 'PENDING'"),
|
||||
// ── Requests raised by clerks, cashiers and managers ──────────────────────────
|
||||
new ApprovalCounter("withdrawal-requests", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_clerk_withdrawal_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("cash-transfers", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_cash_transfer_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("voucher-corrections", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_voucher_correction WHERE status IN ('UNPASS_PENDING','CORRECTION_PENDING')"),
|
||||
new ApprovalCounter("share-withdrawal-requests", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_clerk_share_withdrawal_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("dividend-requests", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_share_product_dividend_pay_out p JOIN m_share_product sp ON sp.id = p.product_id "
|
||||
+ "WHERE p.status = 100"),
|
||||
new ApprovalCounter("share-addition-requests", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_share_account_transactions t JOIN m_share_account sa ON sa.id = t.account_id "
|
||||
+ "JOIN m_client c ON c.id = sa.client_id JOIN m_share_product sp ON sp.id = sa.product_id "
|
||||
+ "WHERE t.is_active = true AND t.status_enum = 100 AND t.type_enum = 500"),
|
||||
new ApprovalCounter("deposit-requests", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_cashier_deposit_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter(DISBURSEMENT_REQUESTS, "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_clerk_disbursement_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("dds-approvals", null,
|
||||
"SELECT count(*) FROM m_portfolio_command_source c JOIN m_permission p "
|
||||
+ "ON p.code = c.action_name || '_' || c.entity_name AND p.can_maker_checker = true "
|
||||
+ "WHERE c.status = 2 AND c.entity_name IN (" + DDS_ENTITIES + ")"),
|
||||
new ApprovalCounter("chequebook-requests", "READ_CHEQUEBOOKPENDINGREQUEST",
|
||||
"SELECT count(*) FROM m_chequebook_pending_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("closure-requests", "ALL_FUNCTIONS_READ",
|
||||
"SELECT count(*) FROM m_clerk_closure_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("dd-issue-approvals", null,
|
||||
"SELECT count(*) FROM m_sponsor_dd_issue_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("dd-reprint-approvals", null,
|
||||
"SELECT count(*) FROM m_sponsor_dd_reprint_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("dd-revoke-approvals", null,
|
||||
"SELECT count(*) FROM m_sponsor_dd_revoke_request WHERE status = 'PENDING'"),
|
||||
new ApprovalCounter("risk-change-approvals", null,
|
||||
"SELECT count(*) FROM m_client_risk_change_request WHERE status = 'PENDING'"));
|
||||
}
|
||||
+24
@@ -0,0 +1,24 @@
|
||||
package org.apache.fineract.portfolio.approvals.service;
|
||||
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCountsData;
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCountsPlan;
|
||||
import org.apache.fineract.useradministration.domain.AppUser;
|
||||
|
||||
/** Reads how many requests wait on each Approval Center tab, in one statement. */
|
||||
public interface ApprovalCountsReadPlatformService {
|
||||
|
||||
/**
|
||||
* Works out what the caller is counted for. Call it on the request thread: it reads the user's roles and the
|
||||
* tenant of the current request.
|
||||
*/
|
||||
ApprovalCountsPlan planFor(AppUser user);
|
||||
|
||||
/** Counts now, for the tabs in the plan. The plan's tenant must be the current tenant of the calling thread. */
|
||||
ApprovalCountsData retrieve(ApprovalCountsPlan plan);
|
||||
|
||||
/**
|
||||
* Counts for the caller, from the answer held for a few seconds when there is one, so many screens asking together
|
||||
* cost the database one query. {@code fresh} skips the held answer.
|
||||
*/
|
||||
ApprovalCountsData retrieveHeld(AppUser user, boolean fresh);
|
||||
}
|
||||
+114
@@ -0,0 +1,114 @@
|
||||
package org.apache.fineract.portfolio.approvals.service;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.stream.Collectors;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
|
||||
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCounter;
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCountsData;
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCountsPlan;
|
||||
import org.apache.fineract.useradministration.domain.AppUser;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
|
||||
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* All the visible tabs' questions go to the database as one statement (one scalar sub-select per tab), so the cost is
|
||||
* one connection and one plan instead of a call per tab. A tab the caller may not see is simply not asked.
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class ApprovalCountsReadPlatformServiceImpl implements ApprovalCountsReadPlatformService {
|
||||
|
||||
/** Screens asking together read one answer instead of each running the statement. */
|
||||
private static final long HOLD_MILLIS = 3_000L;
|
||||
|
||||
private record Held(long at, ApprovalCountsData data) {}
|
||||
|
||||
private final JdbcTemplate jdbcTemplate;
|
||||
private final Map<String, Held> held = new ConcurrentHashMap<>();
|
||||
|
||||
@Override
|
||||
public ApprovalCountsPlan planFor(final AppUser user) {
|
||||
final List<ApprovalCounter> visible = new ArrayList<>();
|
||||
for (final ApprovalCounter counter : ApprovalCounterRegistry.ALL) {
|
||||
if (isVisible(user, counter)) {
|
||||
visible.add(counter);
|
||||
}
|
||||
}
|
||||
final String hierarchy = user.getOffice().getHierarchy();
|
||||
final boolean superChecker = !user.hasNotPermissionForAnyOf("ALL_FUNCTIONS", "CHECKER_SUPER_USER");
|
||||
final MapSqlParameterSource params = new MapSqlParameterSource().addValue("h", hierarchy + "%")
|
||||
.addValue("ho", ".".equals(hierarchy)).addValue("uid", user.getId()).addValue("superChecker", superChecker);
|
||||
final FineractPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
|
||||
// Callers counted the same way share one query: only a limited checker's numbers depend on who they are.
|
||||
final String group = tenant.getTenantIdentifier() + '|' + hierarchy + '|' + superChecker + '|'
|
||||
+ (superChecker ? "" : user.getId()) + '|' + visible.stream().map(ApprovalCounter::key).collect(Collectors.joining(","));
|
||||
return new ApprovalCountsPlan(tenant, visible, params, group);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ApprovalCountsData retrieve(final ApprovalCountsPlan plan) {
|
||||
final Map<String, Long> counts = new LinkedHashMap<>();
|
||||
final List<ApprovalCounter> visible = plan.visible();
|
||||
if (!visible.isEmpty()) {
|
||||
final StringBuilder sql = new StringBuilder("SELECT ");
|
||||
for (int i = 0; i < visible.size(); i++) {
|
||||
sql.append(i == 0 ? "" : ", ").append('(').append(visible.get(i).sql()).append(") AS c").append(i);
|
||||
}
|
||||
final Map<String, Object> row = new NamedParameterJdbcTemplate(this.jdbcTemplate).queryForMap(sql.toString(),
|
||||
plan.params());
|
||||
for (int i = 0; i < visible.size(); i++) {
|
||||
final long n = row.get("c" + i) instanceof Number number ? number.longValue() : 0L;
|
||||
if (n > 0) {
|
||||
counts.put(visible.get(i).key(), n);
|
||||
}
|
||||
}
|
||||
}
|
||||
return ApprovalCountsData.of(counts);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ApprovalCountsData retrieveHeld(final AppUser user, final boolean fresh) {
|
||||
final ApprovalCountsPlan plan = planFor(user);
|
||||
final String holdKey = plan.tenant().getTenantIdentifier() + ':' + user.getId();
|
||||
final long now = System.currentTimeMillis();
|
||||
final Held recent = this.held.get(holdKey);
|
||||
if (!fresh && recent != null && now - recent.at() < HOLD_MILLIS) {
|
||||
return recent.data();
|
||||
}
|
||||
final ApprovalCountsData data = retrieve(plan);
|
||||
this.held.put(holdKey, new Held(now, data));
|
||||
if (this.held.size() > 500) {
|
||||
this.held.entrySet().removeIf(e -> now - e.getValue().at() >= HOLD_MILLIS);
|
||||
}
|
||||
return data;
|
||||
}
|
||||
|
||||
/** The tab's own visibility rule: its permission (READ_* is also met by ALL_FUNCTIONS_READ) and, for one tab, its role. */
|
||||
private boolean isVisible(final AppUser user, final ApprovalCounter counter) {
|
||||
if (counter.permission() != null) {
|
||||
final boolean allowed = counter.permission().startsWith("READ_")
|
||||
? user.hasAnyPermission(counter.permission(), "ALL_FUNCTIONS_READ")
|
||||
: user.hasAnyPermission(counter.permission());
|
||||
if (!allowed) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return !ApprovalCounterRegistry.DISBURSEMENT_REQUESTS.equals(counter.key()) || isChecker(user);
|
||||
}
|
||||
|
||||
private boolean isChecker(final AppUser user) {
|
||||
if (user.hasAnyPermission("ALL_FUNCTIONS")) {
|
||||
return true;
|
||||
}
|
||||
return user.getRoles().stream().map(r -> r.getName() == null ? "" : r.getName().toLowerCase().trim())
|
||||
.anyMatch(n -> n.contains("manager") || n.contains("head cashier"));
|
||||
}
|
||||
}
|
||||
+145
@@ -0,0 +1,145 @@
|
||||
package org.apache.fineract.portfolio.approvals.service;
|
||||
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import jakarta.ws.rs.sse.OutboundSseEvent;
|
||||
import jakarta.ws.rs.sse.Sse;
|
||||
import jakarta.ws.rs.sse.SseEventSink;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCountsData;
|
||||
import org.apache.fineract.portfolio.approvals.data.ApprovalCountsPlan;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* Keeps the Approvals badge live. A screen listens on a Server-Sent Events stream and is written to the moment its
|
||||
* numbers change, so it neither polls nor reloads.
|
||||
*
|
||||
* One shared check, every {@value #CHECK_MILLIS} ms, serves every open screen, and screens counted the same way share
|
||||
* its query. A comment line every {@value #PING_MILLIS} ms keeps proxies from closing an idle connection and tells the
|
||||
* server when a screen has gone. A stream ends after {@value #LIFETIME_MILLIS} ms so the screen reconnects with a clean
|
||||
* connection, which bounds what a browser accumulates on one.
|
||||
*/
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class ApprovalCountsStreamService {
|
||||
|
||||
/** How often the open screens are looked at: the longest a change waits before it is pushed. */
|
||||
private static final long CHECK_MILLIS = 1_000L;
|
||||
|
||||
private static final long PING_MILLIS = 15_000L;
|
||||
|
||||
private static final long LIFETIME_MILLIS = 300_000L;
|
||||
|
||||
/** A screen listening, and the version of the numbers it was last sent. */
|
||||
private static final class Listener {
|
||||
|
||||
private final ApprovalCountsPlan plan;
|
||||
private final SseEventSink sink;
|
||||
private final Sse sse;
|
||||
private final long expiresAt;
|
||||
private volatile String version;
|
||||
private volatile long lastSent;
|
||||
|
||||
private Listener(final ApprovalCountsPlan plan, final SseEventSink sink, final Sse sse, final long expiresAt) {
|
||||
this.plan = plan;
|
||||
this.sink = sink;
|
||||
this.sse = sse;
|
||||
this.expiresAt = expiresAt;
|
||||
}
|
||||
}
|
||||
|
||||
private final ApprovalCountsReadPlatformService readService;
|
||||
private final Set<Listener> listeners = ConcurrentHashMap.newKeySet();
|
||||
private final ScheduledExecutorService ticker = Executors.newSingleThreadScheduledExecutor(runnable -> {
|
||||
final Thread thread = new Thread(runnable, "approval-counts-stream");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
});
|
||||
|
||||
@PostConstruct
|
||||
void startTicker() {
|
||||
this.ticker.scheduleWithFixedDelay(this::tick, CHECK_MILLIS, CHECK_MILLIS, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
void stopTicker() {
|
||||
this.ticker.shutdownNow();
|
||||
this.listeners.forEach(this::close);
|
||||
}
|
||||
|
||||
/** Sends the numbers as they stand, then keeps the screen informed. Call it on the request thread. */
|
||||
public void open(final ApprovalCountsPlan plan, final SseEventSink sink, final Sse sse) {
|
||||
final Listener listener = new Listener(plan, sink, sse, System.currentTimeMillis() + LIFETIME_MILLIS);
|
||||
try {
|
||||
publish(listener, this.readService.retrieve(plan));
|
||||
} catch (final RuntimeException e) {
|
||||
sink.close();
|
||||
throw e;
|
||||
}
|
||||
this.listeners.add(listener);
|
||||
}
|
||||
|
||||
/** Looks at the numbers for every open screen and writes to the ones whose numbers moved. */
|
||||
private void tick() {
|
||||
if (this.listeners.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
final long now = System.currentTimeMillis();
|
||||
final Map<String, ApprovalCountsData> computed = new HashMap<>();
|
||||
for (final Listener listener : this.listeners) {
|
||||
try {
|
||||
if (listener.sink.isClosed() || now >= listener.expiresAt) {
|
||||
close(listener);
|
||||
continue;
|
||||
}
|
||||
ApprovalCountsData data = computed.get(listener.plan.group());
|
||||
if (data == null) {
|
||||
ThreadLocalContextUtil.setTenant(listener.plan.tenant());
|
||||
data = this.readService.retrieve(listener.plan);
|
||||
computed.put(listener.plan.group(), data);
|
||||
}
|
||||
if (!data.version().equals(listener.version)) {
|
||||
publish(listener, data);
|
||||
} else if (now - listener.lastSent >= PING_MILLIS) {
|
||||
listener.lastSent = now;
|
||||
send(listener, listener.sse.newEventBuilder().comment("ping").build());
|
||||
}
|
||||
} catch (final RuntimeException e) {
|
||||
close(listener);
|
||||
} finally {
|
||||
ThreadLocalContextUtil.clearTenant();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void publish(final Listener listener, final ApprovalCountsData data) {
|
||||
listener.version = data.version();
|
||||
listener.lastSent = System.currentTimeMillis();
|
||||
send(listener, listener.sse.newEvent("counts", data.toJson()));
|
||||
}
|
||||
|
||||
private void send(final Listener listener, final OutboundSseEvent event) {
|
||||
listener.sink.send(event).whenComplete((sent, failure) -> {
|
||||
if (failure != null) {
|
||||
close(listener);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void close(final Listener listener) {
|
||||
this.listeners.remove(listener);
|
||||
try {
|
||||
listener.sink.close();
|
||||
} catch (final RuntimeException ignored) {
|
||||
// The screen is already gone.
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user