Skip to content

Fix local storage #89

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
strug merged 4 commits into from
Aug 10, 2026
Merged

Fix local storage #89

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ef29c0b
Fix local count
strug Aug 6, 2026
bd418fd
Remove unrelated load tests and metrics
strug Aug 10, 2026
588ec10
Extract local payment field resolver
strug Aug 10, 2026
66e3263
Register local payment field resolver
strug Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions src/main/java/dev/vality/fraudbusters/config/payment/PaymentFraudoConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import dev.vality.damsel.wb_list.WbListServiceSrv;
import dev.vality.fraudbusters.fraud.constant.PaymentCheckedField;
import dev.vality.fraudbusters.fraud.localstorage.LocalPaymentFieldResolver;
import dev.vality.fraudbusters.fraud.localstorage.LocalResultStorageRepository;
import dev.vality.fraudbusters.fraud.localstorage.aggregator.LocalCountAggregatorDecorator;
import dev.vality.fraudbusters.fraud.localstorage.aggregator.LocalSumAggregatorDecorator;
Expand Down Expand Up @@ -99,6 +100,12 @@ public FieldResolver<PaymentModel, PaymentCheckedField> paymentModelFieldResolve
return new PaymentModelFieldResolver();
}

@Bean
public LocalPaymentFieldResolver localPaymentFieldResolver(
DatabasePaymentFieldResolver databasePaymentFieldResolver) {
return new LocalPaymentFieldResolver(databasePaymentFieldResolver);
}

@Bean
public InListFinder<PaymentModel, PaymentCheckedField> paymentInListFinder(
WbListServiceSrv.Iface wbListServiceSrv,
Expand Down Expand Up @@ -139,6 +146,7 @@ public CountPaymentAggregator<PaymentModel, PaymentCheckedField> countResultAggr
RefundRepository refundRepository,
ChargebackRepository chargebackRepository,
DatabasePaymentFieldResolver databasePaymentFieldResolver,
LocalPaymentFieldResolver localPaymentFieldResolver,
TimeBoundaryService timeBoundaryService) {

CountAggregatorImpl countAggregatorDecorator = new CountAggregatorImpl(
Expand All @@ -150,7 +158,7 @@ public CountPaymentAggregator<PaymentModel, PaymentCheckedField> countResultAggr
);
return new LocalCountAggregatorDecorator(
countAggregatorDecorator,
databasePaymentFieldResolver,
localPaymentFieldResolver,
localResultStorageRepository,
timeBoundaryService
);
Expand All @@ -163,6 +171,7 @@ public SumPaymentAggregator<PaymentModel, PaymentCheckedField> sumResultAggregat
RefundRepository refundRepository,
ChargebackRepository chargebackRepository,
DatabasePaymentFieldResolver databasePaymentFieldResolver,
LocalPaymentFieldResolver localPaymentFieldResolver,
TimeBoundaryService timeBoundaryService) {

SumAggregatorImpl sumAggregator = new SumAggregatorImpl(
Expand All @@ -174,7 +183,7 @@ public SumPaymentAggregator<PaymentModel, PaymentCheckedField> sumResultAggregat
);
return new LocalSumAggregatorDecorator(
sumAggregator,
databasePaymentFieldResolver,
localPaymentFieldResolver,
localResultStorageRepository,
timeBoundaryService
);
Expand All @@ -185,13 +194,14 @@ public UniqueValueAggregator<PaymentModel, PaymentCheckedField> uniqueValueResul
LocalResultStorageRepository localResultStorageRepository,
PaymentRepository fraudResultRepository,
DatabasePaymentFieldResolver databasePaymentFieldResolver,
LocalPaymentFieldResolver localPaymentFieldResolver,
TimeBoundaryService timeBoundaryService) {
UniqueValueAggregatorImpl uniqueValueAggregator =
new UniqueValueAggregatorImpl(databasePaymentFieldResolver, fraudResultRepository, timeBoundaryService);

return new LocalUniqueValueAggregatorDecorator(
uniqueValueAggregator,
databasePaymentFieldResolver,
localPaymentFieldResolver,
localResultStorageRepository,
timeBoundaryService
);
Expand Down
37 changes: 37 additions & 0 deletions src/main/java/dev/vality/fraudbusters/fraud/localstorage/LocalPaymentFieldResolver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
package dev.vality.fraudbusters.fraud.localstorage;

import dev.vality.fraudbusters.fraud.constant.PaymentCheckedField;
import dev.vality.fraudbusters.fraud.model.FieldModel;
import dev.vality.fraudbusters.fraud.model.PaymentModel;
import dev.vality.fraudbusters.fraud.payment.resolver.DatabasePaymentFieldResolver;
import lombok.RequiredArgsConstructor;
import org.jetbrains.annotations.NotNull;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@RequiredArgsConstructor
public class LocalPaymentFieldResolver {

private final DatabasePaymentFieldResolver databasePaymentFieldResolver;

public FieldModel resolve(PaymentCheckedField field, PaymentModel model) {
FieldModel fieldModel = databasePaymentFieldResolver.resolve(field, model);
return new FieldModel(resolveName(field), fieldModel.getValue());
}

public String resolveName(PaymentCheckedField field) {
return field.name();
}

@NotNull
public List<FieldModel> resolveListFields(PaymentModel model, List<PaymentCheckedField> fields) {
if (fields != null) {
return fields.stream()
.map(field -> resolve(field, model))
.collect(Collectors.toList());
}
return new ArrayList<>();
}
}
37 changes: 26 additions & 11 deletions src/main/java/dev/vality/fraudbusters/fraud/localstorage/LocalResultStorageRepository.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,8 @@ public Integer countOperationByFieldWithGroupBy(
List<FieldModel> fieldModels) {
List<CheckedPayment> checkedPayments = localStorage.get();
int count = (int) checkedPayments.stream()
.filter(checkedPayment -> filterPaymentByValue(from, to, fieldModels, checkedPayment))
.filter(checkedPayment -> filterPaymentByValue(
fieldName, value, from, to, fieldModels, checkedPayment))
.count();
log.debug("LocalResultStorageRepository countOperationByFieldWithGroupBy: {}", count);
return count;
Expand All @@ -57,7 +58,8 @@ public Long sumOperationByFieldWithGroupBy(
List<FieldModel> fieldModels) {
List<CheckedPayment> checkedPayments = localStorage.get();
long sum = checkedPayments.stream()
.filter(checkedPayment -> filterPaymentByValue(from, to, fieldModels, checkedPayment))
.filter(checkedPayment -> filterPaymentByValue(
fieldName, value, from, to, fieldModels, checkedPayment))
.mapToLong(CheckedPayment::getAmount)
.sum();
log.debug("LocalResultStorageRepository sumOperationByFieldWithGroupBy: {}", sum);
Expand Down Expand Up @@ -88,7 +90,8 @@ public Integer uniqCountOperationWithGroupBy(
List<FieldModel> fieldModels) {
List<CheckedPayment> checkedPayments = localStorage.get();
long count = checkedPayments.stream()
.filter(checkedPayment -> filterPaymentByValue(from, to, fieldModels, checkedPayment))
.filter(checkedPayment -> filterPaymentByValue(
fieldNameBy, value, from, to, fieldModels, checkedPayment))
.map(checkedPayment -> paymentFieldValueResolver.resolve(fieldNameCount, checkedPayment))
.distinct()
.count();
Expand All @@ -97,12 +100,15 @@ public Integer uniqCountOperationWithGroupBy(
}

private boolean filterPaymentByValue(
String fieldName,
Object value,
Long from,
Long to,
List<FieldModel> fieldModels,
CheckedPayment checkedPayment) {
return checkedPayment.getEventTime() >= from
&& checkedPayment.getEventTime() <= to
&& paymentFieldValueFilter.filter(fieldName, value, checkedPayment)
&& fieldModels.stream()
.allMatch(fieldModel -> paymentFieldValueFilter.filter(
fieldModel.getName(),
Expand All @@ -121,6 +127,8 @@ public Integer countOperationSuccessWithGroupBy(
List<CheckedPayment> checkedPayments = localStorage.get();
return (int) checkedPayments.stream()
.filter(checkedPayment -> filterByStatusAndFields(
fieldName,
value,
from,
to,
fieldModels,
Expand All @@ -136,6 +144,8 @@ public Integer countOperationPendingWithGroupBy(String fieldName, Object value,
List<CheckedPayment> checkedPayments = localStorage.get();
return (int) checkedPayments.stream()
.filter(checkedPayment -> filterByStatusAndFields(
fieldName,
value,
from,
to,
fieldModels,
Expand All @@ -156,6 +166,8 @@ public Integer countOperationErrorWithGroupBy(
List<CheckedPayment> checkedPayments = localStorage.get();
return (int) checkedPayments.stream()
.filter(checkedPayment -> filterByStatusAndFields(
fieldName,
value,
from,
to,
fieldModels,
Expand All @@ -177,6 +189,8 @@ public Integer countOperationErrorWithGroupBy(
List<CheckedPayment> checkedPayments = localStorage.get();
return (int) checkedPayments.stream()
.filter(checkedPayment -> filterByStatusAndFields(
fieldName,
value,
from,
to,
fieldModels,
Expand All @@ -195,6 +209,8 @@ public Long sumOperationSuccessWithGroupBy(
List<CheckedPayment> checkedPayments = localStorage.get();
return checkedPayments.stream()
.filter(checkedPayment -> filterByStatusAndFields(
fieldName,
value,
from,
to,
fieldModels,
Expand All @@ -216,6 +232,8 @@ public Long sumOperationErrorWithGroupBy(
List<CheckedPayment> checkedPayments = localStorage.get();
return checkedPayments.stream()
.filter(checkedPayment -> filterByStatusAndFields(
fieldName,
value,
from,
to,
fieldModels,
Expand All @@ -236,6 +254,8 @@ public Long sumOperationErrorWithGroupBy(String fieldName,
List<CheckedPayment> checkedPayments = localStorage.get();
return checkedPayments.stream()
.filter(checkedPayment -> filterByStatusAndFields(
fieldName,
value,
from,
to,
fieldModels,
Expand All @@ -247,19 +267,14 @@ public Long sumOperationErrorWithGroupBy(String fieldName,
}

private boolean filterByStatusAndFields(
String fieldName,
Object value,
Long from,
Long to,
List<FieldModel> fieldModels,
CheckedPayment checkedPayment,
PaymentStatus paymentStatus) {
return checkedPayment.getEventTime() >= from
&& checkedPayment.getEventTime() <= to
&& fieldModels.stream()
.allMatch(fieldModel -> paymentFieldValueFilter.filter(
fieldModel.getName(),
fieldModel.getValue(),
checkedPayment
))
return filterPaymentByValue(fieldName, value, from, to, fieldModels, checkedPayment)
&& paymentStatus.name().equals(checkedPayment.getPaymentStatus());
}

Expand Down
28 changes: 14 additions & 14 deletions .../dev/vality/fraudbusters/fraud/localstorage/aggregator/LocalCountAggregatorDecorator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,11 @@
import dev.vality.fraudbusters.exception.RuleFunctionException;
import dev.vality.fraudbusters.fraud.AggregateGroupingFunction;
import dev.vality.fraudbusters.fraud.constant.PaymentCheckedField;
import dev.vality.fraudbusters.fraud.localstorage.LocalPaymentFieldResolver;
import dev.vality.fraudbusters.fraud.localstorage.LocalResultStorageRepository;
import dev.vality.fraudbusters.fraud.model.FieldModel;
import dev.vality.fraudbusters.fraud.model.PaymentModel;
import dev.vality.fraudbusters.fraud.payment.aggregator.clickhouse.CountAggregatorImpl;
import dev.vality.fraudbusters.fraud.payment.resolver.DatabasePaymentFieldResolver;
import dev.vality.fraudbusters.service.TimeBoundaryService;
import dev.vality.fraudbusters.util.TimestampUtil;
import dev.vality.fraudo.model.TimeWindow;
Expand All @@ -25,7 +25,7 @@
public class LocalCountAggregatorDecorator implements CountPaymentAggregator<PaymentModel, PaymentCheckedField> {

private final CountAggregatorImpl countAggregator;
private final DatabasePaymentFieldResolver databasePaymentFieldResolver;
private final LocalPaymentFieldResolver localPaymentFieldResolver;
private final LocalResultStorageRepository localStorageRepository;
private final TimeBoundaryService timeBoundaryService;

Expand All @@ -36,11 +36,11 @@ public Integer count(
TimeWindow timeWindow,
List<PaymentCheckedField> list) {
Integer count = countAggregator.count(checkedField, paymentModel, timeWindow, list);
FieldModel resolve = databasePaymentFieldResolver.resolve(checkedField, paymentModel);
FieldModel resolve = localPaymentFieldResolver.resolve(checkedField, paymentModel);
Instant timestamp = TimestampUtil.instantFromPaymentModel(paymentModel);
TimeBound timeBound = timeBoundaryService.getBoundary(timestamp, timeWindow);
Integer localCount = localStorageRepository.countOperationByField(
checkedField.name(),
resolve.getName(),
resolve.getValue(),
timeBound.getLeft().getEpochSecond(),
timeBound.getRight().getEpochSecond()
Expand Down Expand Up @@ -74,10 +74,10 @@ public Integer countError(
Integer countError = countAggregator.countError(checkedField, paymentModel, timeWindow, errorCode, list);
Instant timestamp = TimestampUtil.instantFromPaymentModel(paymentModel);
TimeBound timeBound = timeBoundaryService.getBoundary(timestamp, timeWindow);
FieldModel resolve = databasePaymentFieldResolver.resolve(checkedField, paymentModel);
List<FieldModel> eventFields = databasePaymentFieldResolver.resolveListFields(paymentModel, list);
FieldModel resolve = localPaymentFieldResolver.resolve(checkedField, paymentModel);
List<FieldModel> eventFields = localPaymentFieldResolver.resolveListFields(paymentModel, list);
Integer localCount = localStorageRepository.countOperationErrorWithGroupBy(
checkedField.name(),
resolve.getName(),
resolve.getValue(),
timeBound.getLeft().getEpochSecond(),
timeBound.getRight().getEpochSecond(),
Expand Down Expand Up @@ -105,10 +105,10 @@ public Integer countError(PaymentCheckedField paymentCheckedField, PaymentModel
Integer countError = countAggregator.countError(paymentCheckedField, paymentModel, timeWindow, list);
Instant timestamp = TimestampUtil.instantFromPaymentModel(paymentModel);
TimeBound timeBound = timeBoundaryService.getBoundary(timestamp, timeWindow);
FieldModel resolve = databasePaymentFieldResolver.resolve(paymentCheckedField, paymentModel);
List<FieldModel> eventFields = databasePaymentFieldResolver.resolveListFields(paymentModel, list);
FieldModel resolve = localPaymentFieldResolver.resolve(paymentCheckedField, paymentModel);
List<FieldModel> eventFields = localPaymentFieldResolver.resolveListFields(paymentModel, list);
Integer localCount = localStorageRepository.countOperationErrorWithGroupBy(
paymentCheckedField.name(),
resolve.getName(),
resolve.getValue(),
timeBound.getLeft().getEpochSecond(),
timeBound.getRight().getEpochSecond(),
Expand Down Expand Up @@ -171,13 +171,13 @@ private Integer getCount(
try {
Instant timestamp = TimestampUtil.instantFromPaymentModel(paymentModel);
TimeBound timeBound = timeBoundaryService.getBoundary(timestamp, timeWindow);
FieldModel resolve = databasePaymentFieldResolver.resolve(checkedField, paymentModel);
List<FieldModel> eventFields = databasePaymentFieldResolver.resolveListFields(paymentModel, list);
FieldModel resolve = localPaymentFieldResolver.resolve(checkedField, paymentModel);
List<FieldModel> eventFields = localPaymentFieldResolver.resolveListFields(paymentModel, list);
Integer count = aggregateFunction.accept(
resolve.getName(),
resolve.getValue(),
timeBound.getLeft().toEpochMilli(),
timeBound.getRight().toEpochMilli(),
timeBound.getLeft().getEpochSecond(),
timeBound.getRight().getEpochSecond(),
eventFields
);
log.debug(
Expand Down
Loading
Loading