public Mono<ByteBuf> handleRequest0(
            UserSessionWrapper sessionWrapper,
            ByteBuf serviceRequestBuffer) {
        // Check if it is a heartbeat request
        if (!serviceRequestBuffer.isReadable()) {
            serviceRequestBuffer.release();
            ServiceAvailability serviceAvailability = serverStatusManager.getServiceAvailability();
            if (!serviceAvailability.isAvailable()) {
                return Mono.just(ClientMessageEncoder.encodeResponse(System.currentTimeMillis(),
                        HEARTBEAT_FAILURE_REQUEST_ID,
                        ResponseStatusCode.SERVER_UNAVAILABLE,
                        serviceAvailability.reason()));
            }
            return handleHeartbeatRequest(sessionWrapper);
        }
        // Parse and handle service requests
        long requestTime = System.currentTimeMillis();
        long startTime = System.nanoTime();
        int requestSize = serviceRequestBuffer.readableBytes();
        SimpleTurmsRequest request;
        SimpleTurmsRequest tempRequest;
        Mono<TurmsNotification> notificationMono = null;
        try {
            tempRequest = TurmsRequestParser
                    .parseSimpleRequest(ProtoDecoder.newInputStream(serviceRequestBuffer));
        } catch (Exception e) {
            serviceRequestBuffer.release();
            tempRequest = UNRECOGNIZED_REQUEST;
            UserSession session = sessionWrapper.getUserSession();
            if (session != null) {
                blocklistService.tryBlockUserIdForCorruptedRequest(session.getUserId());
            }
            blocklistService.tryBlockIpForCorruptedRequest(sessionWrapper.getIp());
            notificationMono = Mono.error(
                    ResponseException.get(ResponseStatusCode.INVALID_REQUEST, e.getMessage()));
        }
        request = tempRequest;
        TurmsRequest.KindCase requestType = request.type();
        TracingContext tracingContext = supportsTracing(requestType)
                ? new TracingContext()
                : TracingContext.NOOP;
        // Check if we can log to avoid logging DeleteSessionRequest twice
        boolean canLogRequest = true;
        UserSession session = sessionWrapper.getUserSession();
        if (session != null) {
            if (!session.hasPermission(requestType)) {
                notificationMono = UNAUTHORIZED_REQUEST_ERROR_MONO;
            }
            if (requestType == DELETE_SESSION_REQUEST) {
                canLogRequest = session.acquireDeleteSessionRequestLoggingLock();
            }
        }
        boolean finalCanLogRequest = canLogRequest;
        if (notificationMono == null) {
            notificationMono = handleServiceRequest(sessionWrapper,
                    request,
                    serviceRequestBuffer,
                    tracingContext);
        }
        return notificationMono
                // Metrics and logging
                .name(TURMS_CLIENT_REQUEST)
                .tag(TURMS_CLIENT_REQUEST_TAG_TYPE, requestType.name())
                .metrics()
                .onErrorResume(throwable -> {
                    ThrowableInfo info = ThrowableInfo.get(throwable);
                    if (info.code()
                            .isServerError()) {
                        tracingContext.updateThreadContext();
                        LOGGER.error("Failed to handle the service request: {}",
                                request,
                                throwable);
                    }
                    return Mono.just(NotificationFactory.create(info, request.requestId()));
                })
                .map(notification -> {
                    TurmsRequest.KindCase type = request.type();
                    // TODO: exclude the error caused by the inactive of turms-service
                    if (ResponseStatusCode.isServerError(notification.getCode())
                            || apiLoggingContext.shouldLogRequest(type) && finalCanLogRequest) {
                        try (TracingCloseableContext ignored = tracingContext.asCloseable()) {
                            UserSession userSession = sessionWrapper.getUserSession();
                            Integer version = null;
                            Long userId = null;
                            Integer sessionId = null;
                            DeviceType deviceType = null;
                            if (userSession != null) {
                                version = userSession.getVersion();
                                userId = userSession.getUserId();
                                sessionId = userSession.getId();
                                deviceType = userSession.getDeviceType();
                            }
                            ClientApiLogging.log(sessionId,
                                    userId,
                                    deviceType,
                                    version,
                                    sessionWrapper.getIpStr(),
                                    request.requestId(),
                                    type,
                                    requestSize,
                                    requestTime,
                                    notification,
                                    (System.nanoTime() - startTime) / DateTimeUtil.NANOS_PER_MILLI);
                        }
                    }
                    return ProtoEncoder.getDirectByteBuffer(notification);
                })
                .contextWrite(context -> {
                    TracingContext ctx = context.get(TracingContext.CTX_KEY_NAME);
                    ctx.setTraceId(tracingContext.getTraceId());
                    return context;
                });
    }
@implNote 1. If a throwable instance is thrown due to the failure of handling the client
          request, the method should recover it to {@link TurmsNotification}. In other words,
          the method should never return MonoError, and it should be considered as a bug if
          it occurs. 2. The method ensures {@param serviceRequestBuffer} will be released by
          1
turms-gateway/src/main/java/im/turms/gateway/access/client/common/ClientRequestDispatcher.java