Fixing BusinessEventPublisher

Author: FalkWolsky

server/api-service/lowcoder-server/src/main/java/org/lowcoder/api/framework/filter/ApiEventFilter.java

@@ -1,3 +1,4 @@
+
 package org.lowcoder.api.framework.filter;
 
 import com.google.common.hash.Hashing;
@@ -17,6 +18,7 @@
 import org.springframework.web.server.WebFilterChain;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
+import reactor.util.context.ContextView;
 
 import java.nio.charset.StandardCharsets;
 import java.util.Optional;
@@ -40,7 +42,7 @@ public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
                     String token = contextView.get(VISITOR_TOKEN);
                     ((Mono<OrgMember>) contextView.get(CURRENT_ORG_MEMBER))
                             .flatMap(orgMember -> {
-                                emitEvent(exchange.getRequest(), token, orgMember);
+                                emitEvent(exchange.getRequest(), token, orgMember, contextView);
                                 return Mono.empty();
                             })
                             .subscribeOn(Schedulers.boundedElastic())
@@ -51,7 +53,7 @@ public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
         );
     }
 
-    private void emitEvent(ServerHttpRequest request, String token, OrgMember orgMember) {
+    private void emitEvent(ServerHttpRequest request, String token, OrgMember orgMember, ContextView contextView) {
         MultiValueMap<String, String> headers = writableHttpHeaders(request.getHeaders());
         headers.remove("Cookie");
         Optional<String> ipAddressOptional = headers.remove("X-Real-IP").stream().findFirst();
@@ -69,7 +71,7 @@ private void emitEvent(ServerHttpRequest request, String token, OrgMember orgMem
                 .queryParams(request.getQueryParams())
                 .ipAddress(ipAddress)
                 .build();
-        event.populateDetails();
+        event.populateDetails(contextView);
 
         log.debug("API call event emitted for '{}' from org '{}' on URI: {}", orgMember.getUserId(), orgMember.getUserId(), request.getURI().getPath());
         applicationEventPublisher.publishEvent(event);

server/api-service/lowcoder-server/src/main/java/org/lowcoder/api/framework/filter/GlobalContextFilter.java

@@ -16,6 +16,7 @@
 import org.springframework.http.HttpMethod;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.stereotype.Component;
+import org.springframework.util.MultiValueMap;
 import org.springframework.web.server.ServerWebExchange;
 import org.springframework.web.server.WebFilter;
 import org.springframework.web.server.WebFilterChain;
@@ -32,6 +33,7 @@
 import static org.lowcoder.sdk.constants.Authentication.isAnonymousUser;
 import static org.lowcoder.sdk.constants.GlobalContext.*;
 import static org.lowcoder.sdk.util.IDUtils.generate;
+import static org.springframework.http.HttpHeaders.writableHttpHeaders;
 
 @Component
 @RequiredArgsConstructor
@@ -107,6 +109,11 @@ private Map<String, Object> buildContextMap(ServerWebExchange serverWebExchange,
         contextMap.put(CURRENT_ORG_MEMBER, orgMemberService.getCurrentOrgMember(visitorId).cache());
         contextMap.put(VISITOR_TOKEN, cookieHelper.getCookieToken(serverWebExchange));
         contextMap.put(DOMAIN, UriUtils.getRefererDomainFromRequest(serverWebExchange));
+
+        //Analytics related fields
+        MultiValueMap<String, String> headers = writableHttpHeaders(request.getHeaders());
+        headers.remove("Cookie");
+        contextMap.put(HEADERS, headers);
         return contextMap;
     }
 

server/api-service/lowcoder-server/src/main/java/org/lowcoder/api/util/ApiCallEventPublisher.java

@@ -48,19 +48,20 @@ public void deleteMapping(){}
     @Pointcut("@annotation(org.springframework.web.bind.annotation.PatchMapping)")
     public void patchMapping(){}
 
+    @SuppressWarnings("unchecked")
     @Around("(getMapping() || postMapping() || putMapping() || deleteMapping() || patchMapping())")
     public Object handleAPICallEvent(ProceedingJoinPoint joinPoint) throws Throwable {
 
         return sessionUserService.getVisitorToken()
                 .zipWith(sessionUserService.getVisitorOrgMemberCacheSilent().defaultIfEmpty(OrgMember.NOT_EXIST))
                 .zipWith(ReactiveRequestContextHolder.getRequest())
-                .doOnNext(
+                .delayUntil(
                         tuple -> {
                             String token = tuple.getT1().getT1();
                             OrgMember orgMember = tuple.getT1().getT2();
                             ServerHttpRequest request = tuple.getT2();
                             if (orgMember == OrgMember.NOT_EXIST) {
-                                return;
+                                return Mono.empty();
                             }
                             MultiValueMap<String, String> headers = writableHttpHeaders(request.getHeaders());
                             headers.remove("Cookie");
@@ -77,8 +78,11 @@ public Object handleAPICallEvent(ProceedingJoinPoint joinPoint) throws Throwable
                                     .queryParams(request.getQueryParams())
                                     .ipAddress(ipAddress)
                                     .build();
-                            event.populateDetails();
-                            applicationEventPublisher.publishEvent(event);
+                            return Mono.deferContextual(contextView -> {
+                                event.populateDetails(contextView);
+                                applicationEventPublisher.publishEvent(event);
+                                return Mono.empty();
+                            });
                         })
                 .onErrorResume(throwable -> {
                     log.error("handleAPICallEvent error {} for: {} ", joinPoint.getSignature().getName(), EventType.API_CALL_EVENT, throwable);
@@ -87,4 +91,4 @@ public Object handleAPICallEvent(ProceedingJoinPoint joinPoint) throws Throwable
                 .then((Mono) joinPoint.proceed());
     }
 
-}
+}