OPS-382:Change elastic to opensearch, add shop and wallets search (#32)

---------

Co-authored-by: Alex Romanov <ex1tus@yandex.ru>
Co-authored-by: Anatoly Karlov <karleowne@gmail.com>

.github/workflows/build.yml

@@ -7,4 +7,4 @@ on:
 
 jobs:
   build:
-    uses: valitydev/java-workflow/.github/workflows/maven-service-build.yml@v1
+    uses: valitydev/java-workflow/.github/workflows/maven-service-build.yml@v2

.github/workflows/deploy.yml

@@ -7,7 +7,7 @@ on:
 
 jobs:
   build-and-deploy:
-    uses: valitydev/java-workflow/.github/workflows/maven-service-deploy.yml@v1
+    uses: valitydev/java-workflow/.github/workflows/maven-service-deploy.yml@v2
     secrets:
       github-token: ${{ secrets.GITHUB_TOKEN }}
       mm-webhook-url: ${{ secrets.MATTERMOST_WEBHOOK_URL }}

pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>dev.vality</groupId>
         <artifactId>service-parent-pom</artifactId>
-        <version>1.0.18</version>
+        <version>2.1.8</version>
     </parent>
 
     <artifactId>deanonimus</artifactId>
@@ -27,14 +27,6 @@
 
     <dependencies>
         <!--dev.vality-->
-        <dependency>
-            <groupId>dev.vality.woody</groupId>
-            <artifactId>woody-thrift</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>dev.vality</groupId>
-            <artifactId>shared-resources</artifactId>
-        </dependency>
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>kafka-common-lib</artifactId>
@@ -63,7 +55,7 @@
         <dependency>
             <groupId>dev.vality</groupId>
             <artifactId>deanonimus-proto</artifactId>
-            <version>1.36-94fb655</version>
+            <version>1.40-2a02d87</version>
         </dependency>
         <!--spring-->
         <dependency>
@@ -84,22 +76,14 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.springframework.data</groupId>
-            <artifactId>spring-data-elasticsearch</artifactId>
-            <!--Compatible with opensearch -->
-            <version>4.2.1</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.elasticsearch.client</groupId>
-                    <artifactId>elasticsearch-rest-high-level-client</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
 
         <!--third party-->
         <dependency>
@@ -118,7 +102,7 @@
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
-            <version>1.18.22</version>
+            <version>1.18.26</version>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
@@ -126,19 +110,34 @@
             <version>3.12.0</version>
         </dependency>
         <dependency>
-            <groupId>org.elasticsearch</groupId>
+            <groupId>org.opensearch</groupId>
-            <artifactId>elasticsearch</artifactId>
+            <artifactId>opensearch</artifactId>
-            <version>7.12.1</version>
+            <version>2.11.0</version>
         </dependency>
         <dependency>
-            <groupId>org.elasticsearch.client</groupId>
+            <groupId>org.opensearch.client</groupId>
-            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+            <artifactId>opensearch-rest-client</artifactId>
-            <version>7.12.1</version>
+            <version>2.11.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.opensearch.client</groupId>
+            <artifactId>opensearch-java</artifactId>
+            <version>2.8.0</version>
+        </dependency>
+        <dependency>
+            <groupId>jakarta.json</groupId>
+            <artifactId>jakarta.json-api</artifactId>
+            <version>2.1.3</version>
         </dependency>
         <dependency>
             <groupId>software.amazon.msk</groupId>
             <artifactId>aws-msk-iam-auth</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.jetbrains</groupId>
+            <artifactId>annotations</artifactId>
+            <version>24.1.0</version>
+        </dependency>
 
         <!--test-->
         <dependency>
@@ -154,19 +153,25 @@
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility</artifactId>
-            <version>4.0.3</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>kafka</artifactId>
-            <version>1.16.2</version>
+            <version>1.17.6</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
-            <artifactId>elasticsearch</artifactId>
+            <artifactId>testcontainers</artifactId>
-            <version>1.16.2</version>
+            <version>1.17.6</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>1.17.6</version>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 
@@ -200,12 +205,12 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-remote-resources-plugin</artifactId>
-                <version>1.6.0</version>
+                <version>3.0.0</version>
                 <dependencies>
                     <dependency>
                         <groupId>org.apache.maven.shared</groupId>
                         <artifactId>maven-filtering</artifactId>
-                        <version>1.3</version>
+                        <version>3.3.1</version>
                     </dependency>
                 </dependencies>
                 <configuration>

src/main/java/dev/vality/deanonimus/config/ElasticsearchConfig.java

@@ -1,9 +0,0 @@
-package dev.vality.deanonimus.config;
-
-import org.springframework.context.annotation.Configuration;
-import org.springframework.data.elasticsearch.config.ElasticsearchConfigurationSupport;
-
-@Configuration
-public class ElasticsearchConfig extends ElasticsearchConfigurationSupport {
-
-}

src/main/java/dev/vality/deanonimus/config/OpenSearchClientConfig.java

@@ -0,0 +1,74 @@
+package dev.vality.deanonimus.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import lombok.SneakyThrows;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.opensearch.client.RestClient;
+import org.opensearch.client.RestClientBuilder;
+import org.opensearch.client.json.jackson.JacksonJsonpMapper;
+import org.opensearch.client.opensearch.OpenSearchClient;
+import org.opensearch.client.transport.rest_client.RestClientTransport;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.Resource;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.InputStream;
+import java.security.KeyStore;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+
+@Configuration
+public class OpenSearchClientConfig {
+
+    @Bean(destroyMethod = "close")
+    public RestClient restClient(OpenSearchProperties openSearchProperties) {
+        var httpHost = new HttpHost(openSearchProperties.getHostname(),
+                openSearchProperties.getPort(),
+                openSearchProperties.getSslEnabled() ? "https" : "http");
+        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
+        if (openSearchProperties.getSslEnabled()) {
+            final var credentialsProvider = new BasicCredentialsProvider();
+            credentialsProvider.setCredentials(
+                    AuthScope.ANY,
+                    new UsernamePasswordCredentials(
+                            openSearchProperties.getUsername(),
+                            openSearchProperties.getPassword()));
+            restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
+                    .setDefaultCredentialsProvider(credentialsProvider)
+                    .setSSLContext(sslContext(openSearchProperties.getCertificate()))
+                    .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE));
+        }
+        return restClientBuilder.build();
+    }
+
+    @Bean
+    public OpenSearchClient openSearchClient(RestClient restClient) {
+        var transport = new RestClientTransport(restClient,
+                new JacksonJsonpMapper(new ObjectMapper().registerModule(new JavaTimeModule())));
+        return new OpenSearchClient(transport);
+    }
+
+    @SneakyThrows
+    private SSLContext sslContext(Resource certificate) {
+        var tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+        try (InputStream pKeyFileStream = certificate.getInputStream()) {
+            var cf = CertificateFactory.getInstance("X.509");
+            var caCert = (X509Certificate) cf.generateCertificate(pKeyFileStream);
+            var ks = KeyStore.getInstance(KeyStore.getDefaultType());
+            ks.load(null);
+            ks.setCertificateEntry("caCert", caCert);
+            tmf.init(ks);
+        }
+        var sslContext = SSLContext.getInstance("TLS");
+        sslContext.init(null, tmf.getTrustManagers(), null);
+        return sslContext;
+    }
+
+}

src/main/java/dev/vality/deanonimus/config/OpenSearchProperties.java

@@ -0,0 +1,19 @@
+package dev.vality.deanonimus.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.Resource;
+
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "opensearch")
+public class OpenSearchProperties {
+
+    private String username;
+    private String password;
+    private String hostname;
+    private Integer port;
+    private Resource certificate;
+    private Boolean sslEnabled;
+}

src/main/java/dev/vality/deanonimus/constant/OpenSearchConstants.java

@@ -1,7 +1,9 @@
 package dev.vality.deanonimus.constant;
 
-public class ElasticsearchConstants {
+public class OpenSearchConstants {
+    public static final String PARTY_INDEX = "party";
     public static final String SHOP_INDEX = "shops";
+    public static final String WALLET_INDEX = "wallets";
     public static final String CONTRACT_INDEX = "contracts";
     public static final String CONTRACTOR_INDEX = "contractors";
 }

src/main/java/dev/vality/deanonimus/converter/ContractorConverter.java

@@ -2,9 +2,9 @@ package dev.vality.deanonimus.converter;
 
 import dev.vality.damsel.deanonimus.*;
 import dev.vality.deanonimus.domain.Contractor;
-import org.jetbrains.annotations.NotNull;
 import org.springframework.stereotype.Component;
 
+import javax.validation.constraints.NotNull;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -35,7 +35,6 @@ public class ContractorConverter {
             case legal_entity -> legal_entity(convertLegalEntity(contractor));
             case private_entity -> private_entity(new PrivateEntity());
             case registered_user -> registered_user(new RegisteredUser(contractor.getRegisteredUserEmail()));
-            default -> throw new IllegalArgumentException("No such contractorType: " + contractor.getType());
         };
     }
 
@@ -43,7 +42,6 @@ public class ContractorConverter {
         return switch (contractor.getLegalEntity()) {
             case international_legal_entity -> buildInternationalLegalEntity(contractor);
             case russian_legal_entity -> buildRussianLegalEntity(contractor);
-            default -> throw new IllegalArgumentException("No such legalEntity " + contractor.getLegalEntity());
         };
     }
 

src/main/java/dev/vality/deanonimus/converter/SearchHitConverter.java

@@ -3,7 +3,8 @@ package dev.vality.deanonimus.converter;
 import dev.vality.damsel.deanonimus.SearchHit;
 import dev.vality.deanonimus.domain.Party;
 import lombok.RequiredArgsConstructor;
-import org.springframework.data.elasticsearch.core.SearchHits;
+import org.opensearch.client.opensearch.core.SearchResponse;
+import org.opensearch.client.opensearch.core.search.Hit;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -16,14 +17,14 @@ public class SearchHitConverter {
 
     private final PartyConverter partyConverter;
 
-    public List<SearchHit> convert(SearchHits<Party> searchHits) {
+    public List<SearchHit> convert(SearchResponse<Party> searchHits) {
-        return searchHits.stream()
+        return searchHits.hits().hits().stream()
                 .map(this::convertSearchHit)
                 .collect(toList());
     }
 
-    private SearchHit convertSearchHit(org.springframework.data.elasticsearch.core.SearchHit<Party> partySearchHit) {
+    private SearchHit convertSearchHit(Hit<Party> partySearchHit) {
-        return new SearchHit(partySearchHit.getScore(), partyConverter.convert(partySearchHit.getContent()));
+        return new SearchHit(partySearchHit.score(), partyConverter.convert(partySearchHit.source()));
     }
 
 }

src/main/java/dev/vality/deanonimus/converter/SearchHitShopConverter.java

@@ -3,29 +3,30 @@ package dev.vality.deanonimus.converter;
 import dev.vality.damsel.deanonimus.SearchShopHit;
 import dev.vality.deanonimus.domain.Party;
 import lombok.RequiredArgsConstructor;
-import org.springframework.data.elasticsearch.core.SearchHit;
+import org.opensearch.client.opensearch.core.SearchResponse;
-import org.springframework.data.elasticsearch.core.SearchHits;
+import org.opensearch.client.opensearch.core.search.Hit;
 import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
 import java.util.List;
 
-import static java.util.stream.Collectors.toList;
-
 @Component
 @RequiredArgsConstructor
 public class SearchHitShopConverter {
 
-    private final ShopListConverter converter;
+    private final ShopListConverter shopListConverter;
+    private final PartyConverter partyConverter;
 
-    public List<SearchShopHit> convert(SearchHits<Party> searchHits) {
+
+    public List<SearchShopHit> convert(SearchResponse<Party> searchHits) {
         List<SearchShopHit> hits = new ArrayList<>();
-        for (SearchHit<Party> searchHit : searchHits) {
+        for (Hit<Party> searchHit : searchHits.hits().hits()) {
-            hits.addAll(converter.convert(searchHit.getContent().getShops()).values()
+            hits.addAll(shopListConverter.convert(searchHit.source().getShops()).values()
                     .stream()
-                    .map(shop -> new SearchShopHit(searchHit.getScore(), shop))
+                    .map(shop -> new SearchShopHit(searchHit.score(), shop, partyConverter.convert(searchHit.source())))
-                    .collect(toList()));
+                    .toList());
         }
+
         return hits;
     }
 }

src/main/java/dev/vality/deanonimus/converter/SearchHitWalletConverter.java

@@ -0,0 +1,35 @@
+package dev.vality.deanonimus.converter;
+
+import dev.vality.damsel.deanonimus.SearchWalletHit;
+import dev.vality.deanonimus.domain.Party;
+import lombok.RequiredArgsConstructor;
+import org.opensearch.client.opensearch.core.SearchResponse;
+import org.opensearch.client.opensearch.core.search.Hit;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Component
+@RequiredArgsConstructor
+public class SearchHitWalletConverter {
+
+    private final WalletListConverter walletListConverter;
+    private final PartyConverter partyConverter;
+
+
+    public List<SearchWalletHit> convert(SearchResponse<Party> searchHits) {
+        List<SearchWalletHit> hits = new ArrayList<>();
+        for (Hit<Party> searchHit : searchHits.hits().hits()) {
+            hits.addAll(walletListConverter.convert(searchHit.source().getWallets()).values()
+                    .stream()
+                    .map(wallet -> new SearchWalletHit(
+                            searchHit.score(),
+                            wallet,
+                            partyConverter.convert(searchHit.source())))
+                    .toList());
+        }
+
+        return hits;
+    }
+}

src/main/java/dev/vality/deanonimus/converter/WalletListConverter.java

@@ -0,0 +1,30 @@
+package dev.vality.deanonimus.converter;
+
+import dev.vality.damsel.deanonimus.Wallet;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+@Component
+@RequiredArgsConstructor
+public class WalletListConverter {
+
+    public Map<String, Wallet> convert(List<dev.vality.deanonimus.domain.Wallet> wallets) {
+        return Optional.ofNullable(wallets).orElse(Collections.emptyList())
+                .stream()
+                .map(this::convertToEntity)
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (a, b) -> a));
+    }
+
+    private Map.Entry<String, Wallet> convertToEntity(dev.vality.deanonimus.domain.Wallet walletDomain) {
+        return Map.entry(walletDomain.getId(), new Wallet(
+                walletDomain.getId(),
+                walletDomain.getName()
+        ));
+    }
+}

src/main/java/dev/vality/deanonimus/db/PartyRepository.java

@@ -1,11 +0,0 @@
-package dev.vality.deanonimus.db;
-
-import dev.vality.deanonimus.domain.Party;
-import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
-import org.springframework.stereotype.Repository;
-
-@Repository
-public interface PartyRepository extends ElasticsearchRepository<Party, String> {
-
-
-}

src/main/java/dev/vality/deanonimus/db/SearchDao.java

@@ -1,10 +1,10 @@
 package dev.vality.deanonimus.db;
 
 import dev.vality.deanonimus.domain.Party;
-import org.springframework.data.elasticsearch.core.SearchHits;
+import org.opensearch.client.opensearch.core.SearchResponse;
 
 public interface SearchDao {
 
-    SearchHits<Party> searchParty(String text);
+    SearchResponse<Party> searchParty(String text);
 
 }

src/main/java/dev/vality/deanonimus/db/SearchDaoImpl.java

@@ -2,19 +2,15 @@ package dev.vality.deanonimus.db;
 
 import dev.vality.deanonimus.domain.Party;
 import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.lucene.search.join.ScoreMode;
+import org.opensearch.client.opensearch.OpenSearchClient;
-import org.elasticsearch.index.query.QueryBuilder;
+import org.opensearch.client.opensearch._types.query_dsl.*;
+import org.opensearch.client.opensearch.core.SearchResponse;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.data.domain.PageRequest;
-import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
-import org.springframework.data.elasticsearch.core.SearchHits;
-import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
-import org.springframework.data.elasticsearch.core.query.Query;
 import org.springframework.stereotype.Component;
 
-import static dev.vality.deanonimus.constant.ElasticsearchConstants.*;
+import static dev.vality.deanonimus.constant.OpenSearchConstants.*;
-import static org.elasticsearch.index.query.QueryBuilders.*;
 
 @Slf4j
 @Component
@@ -24,59 +20,91 @@ public class SearchDaoImpl implements SearchDao {
     @Value("${data.response.limit}")
     Integer responseLimit;
 
-    private final ElasticsearchRestTemplate elasticsearchRestTemplate;
+    private final OpenSearchClient openSearchClient;
 
+
+    @SneakyThrows
     @Override
-    public SearchHits<Party> searchParty(String text) {
+    public SearchResponse<Party> searchParty(String text) {
 
-        QueryBuilder builder = boolQuery()
+        BoolQuery queryBuilder = new BoolQuery.Builder()
-                .should(searchPartyFields(text))
+                .should(searchPartyFields(text),
-                .should(searchShopFields(text))
+                        searchShopFields(text),
-                .should(searchContractFields(text))
+                        searchContractFields(text),
-                .should(searchContractorFields(text));
+                        searchContractorFields(text),
-
+                        searchWalletFields(text))
-        Query searchQuery = new NativeSearchQueryBuilder()
-                .withQuery(builder)
-                .withPageable(PageRequest.of(0, responseLimit))
                 .build();
 
-        return elasticsearchRestTemplate.search(searchQuery, Party.class);
+        return openSearchClient.search(s -> s
+                        .size(responseLimit)
+                        .query(new Query.Builder()
+                                .bool(queryBuilder)
+                                .build()),
+                Party.class);
     }
 
-    private QueryBuilder searchContractorFields(String text) {
+    private Query searchContractorFields(String text) {
-        return nestedQuery(CONTRACTOR_INDEX,
+        return new NestedQuery.Builder()
-                multiMatchQuery(text,
+                .path(CONTRACTOR_INDEX)
-                        "contractors.id",
+                .query(new Query(new MultiMatchQuery.Builder()
-                        "contractors.registeredUserEmail",
+                        .fields("contractors.id",
-                        "contractors.russianLegalEntityRegisteredName",
+                                "contractors.registeredUserEmail",
-                        "contractors.russianLegalEntityInn",
+                                "contractors.russianLegalEntityRegisteredName",
-                        "contractors.russianLegalEntityRussianBankAccount",
+                                "contractors.russianLegalEntityInn",
-                        "contractors.internationalLegalEntityLegalName",
+                                "contractors.russianLegalEntityRussianBankAccount",
-                        "contractors.internationalLegalEntityTradingName"), ScoreMode.Total);
+                                "contractors.internationalLegalEntityLegalName",
+                                "contractors.internationalLegalEntityTradingName")
+                        .query(text)
+                        .type(TextQueryType.Phrase)
+                        .build()))
+                .build().query();
     }
 
-    private QueryBuilder searchContractFields(String text) {
+    private Query searchContractFields(String text) {
-        return nestedQuery(CONTRACT_INDEX,
+        return new NestedQuery.Builder()
-                multiMatchQuery(text,
+                .path(CONTRACT_INDEX)
-                        "contracts.id",
+                .query(new Query(new MultiMatchQuery.Builder()
-                        "contracts.legalAgreementId",
+                        .fields("contracts.id",
-                        "contracts.reportActSignerFullName"), ScoreMode.Total);
+                                "contracts.legalAgreementId",
+                                "contracts.reportActSignerFullName")
+                        .query(text)
+                        .type(TextQueryType.Phrase)
+                        .build()))
+                .build().query();
     }
 
-
+    private Query searchPartyFields(String text) {
-    private QueryBuilder searchPartyFields(String text) {
+        return new Query(new MultiMatchQuery.Builder()
-        return multiMatchQuery(text,
+                .fields("id",
-                "id",
+                        "email")
-                "email"
+                .query(text)
-        );
+                .type(TextQueryType.Phrase)
+                .build());
     }
 
-    private QueryBuilder searchShopFields(String text) {
+    private Query searchShopFields(String text) {
-        return nestedQuery(SHOP_INDEX,
+        return new NestedQuery.Builder()
-                multiMatchQuery(text,
+                .path(SHOP_INDEX)
-                        "shops.id",
+                .query(new Query(new MultiMatchQuery.Builder()
-                        "shops.locationUrl"
+                        .fields("shops.id",
-                ), ScoreMode.Total);
+                                "shops.locationUrl",
+                                "shops.detailsName")
+                        .query(text)
+                        .type(TextQueryType.Phrase)
+                        .build()))
+                .build().query();
+    }
+
+    private Query searchWalletFields(String text) {
+        return new NestedQuery.Builder()
+                .path(WALLET_INDEX)
+                .query(new Query(new MultiMatchQuery.Builder()
+                        .fields("wallets.id",
+                                "wallets.name")
+                        .query(text)
+                        .type(TextQueryType.Phrase)
+                        .build()))
+                .build().query();
     }
 
 }

src/main/java/dev/vality/deanonimus/domain/Contract.java

@@ -4,8 +4,6 @@ import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.springframework.data.elasticsearch.annotations.Field;
-import org.springframework.data.elasticsearch.annotations.FieldType;
 
 import java.time.LocalDateTime;
 
@@ -14,7 +12,6 @@ import java.time.LocalDateTime;
 @NoArgsConstructor
 @AllArgsConstructor
 public class Contract {
-    @Field(type = FieldType.Keyword)
     private String id;
     private String contractorId;
     private String partyId;
@@ -23,8 +20,6 @@ public class Contract {
     private LocalDateTime validUntil;
     private ContractStatus status;
     private Integer termsId;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "standard")
     private String legalAgreementId;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "standard")
     private String reportActSignerFullName;
 }

src/main/java/dev/vality/deanonimus/domain/Contractor.java

@@ -4,36 +4,27 @@ import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.springframework.data.elasticsearch.annotations.Field;
-import org.springframework.data.elasticsearch.annotations.FieldType;
 
 @Data
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
 public class Contractor {
-    @Field(type = FieldType.Keyword)
     private String id;
     private String partyId;
     private ContractorType type;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "standard")
     private String registeredUserEmail;
     private LegalEntity legalEntity;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "standard")
     private String russianLegalEntityRegisteredName;
     private String russianLegalEntityRegisteredNumber;
-    @Field(type = FieldType.Keyword)
     private String russianLegalEntityInn;
     private String russianLegalEntityActualAddress;
     private String russianLegalEntityPostAddress;
-    @Field(type = FieldType.Keyword)
     private String russianLegalEntityRussianBankAccount;
     private String russianLegalEntityRussianBankName;
     private String russianLegalEntityRussianBankPostAccount;
     private String russianLegalEntityRussianBankBik;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "standard")
     private String internationalLegalEntityLegalName;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "standard")
     private String internationalLegalEntityTradingName;
     private String internationalLegalEntityRegisteredAddress;
     private String internationalLegalEntityActualAddress;

src/main/java/dev/vality/deanonimus/domain/Party.java

@@ -4,11 +4,6 @@ import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.springframework.data.annotation.Id;
-import org.springframework.data.elasticsearch.annotations.Document;
-import org.springframework.data.elasticsearch.annotations.Field;
-import org.springframework.data.elasticsearch.annotations.FieldType;
-import org.springframework.data.elasticsearch.annotations.Setting;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -18,25 +13,18 @@ import java.util.Optional;
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
-@Document(indexName = "party")
-@Setting(settingPath = "/settings/autocomplete-analyzer.json")
 public class Party {
 
-    @Id
-    @Field(type = FieldType.Keyword)
     private String id;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "standard")
     private String email;
 
     private Blocking blocking;
     private Suspension suspension;
 
-    @Field(type = FieldType.Nested, store = true)
     private List<Contractor> contractors;
-    @Field(type = FieldType.Nested, store = true)
     private List<Contract> contracts;
-    @Field(type = FieldType.Nested, store = true)
     private List<Shop> shops;
+    private List<Wallet> wallets;
 
     public void addShop(Shop shop) {
         if (this.shops == null) {
@@ -59,6 +47,13 @@ public class Party {
         this.contractors.add(contractor);
     }
 
+    public void addWallet(Wallet wallet) {
+        if (this.wallets == null) {
+            this.wallets = new ArrayList<>();
+        }
+        this.wallets.add(wallet);
+    }
+
     public Optional<Shop> getShopById(String id) {
         return this.shops.stream().filter(shop -> shop.getId().equals(id)).findFirst();
     }
@@ -66,9 +61,4 @@ public class Party {
     public Optional<Contract> getContractById(String id) {
         return this.contracts.stream().filter(contract -> contract.getId().equals(id)).findFirst();
     }
-
-    public Optional<Contractor> getContractorById(String id) {
-        return this.contractors.stream().filter(contractor -> contractor.getId().equals(id)).findFirst();
-    }
-
 }

src/main/java/dev/vality/deanonimus/domain/Shop.java

@@ -4,21 +4,17 @@ import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
-import org.springframework.data.elasticsearch.annotations.Field;
-import org.springframework.data.elasticsearch.annotations.FieldType;
 
 @Data
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
 public class Shop {
-    @Field(type = FieldType.Keyword)
     private String id;
     private Blocking blocking;
     private Suspension suspension;
     private String detailsName;
     private String detailsDescription;
-    @Field(type = FieldType.Text, analyzer = "autocomplete", searchAnalyzer = "write_url_analyzer")
     private String locationUrl;
     private Integer categoryId;
     private String accountCurrencyCode;

src/main/java/dev/vality/deanonimus/domain/Wallet.java

@@ -0,0 +1,15 @@
+package dev.vality.deanonimus.domain;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class Wallet {
+    private String id;
+    private String name;
+}

src/main/java/dev/vality/deanonimus/handler/DeanonimusServiceHandler.java

@@ -3,15 +3,15 @@ package dev.vality.deanonimus.handler;
 import dev.vality.damsel.deanonimus.DeanonimusSrv;
 import dev.vality.damsel.deanonimus.SearchHit;
 import dev.vality.damsel.deanonimus.SearchShopHit;
+import dev.vality.damsel.deanonimus.SearchWalletHit;
 import dev.vality.deanonimus.converter.SearchHitConverter;
 import dev.vality.deanonimus.converter.SearchHitShopConverter;
+import dev.vality.deanonimus.converter.SearchHitWalletConverter;
 import dev.vality.deanonimus.db.SearchDao;
 import dev.vality.deanonimus.domain.Party;
-import dev.vality.deanonimus.domain.Shop;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.thrift.TException;
+import org.opensearch.client.opensearch.core.SearchResponse;
-import org.springframework.data.elasticsearch.core.SearchHits;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -23,21 +23,30 @@ public class DeanonimusServiceHandler implements DeanonimusSrv.Iface {
 
     private final SearchHitConverter searchHitConverter;
     private final SearchHitShopConverter searchHitShopConverter;
+    private final SearchHitWalletConverter searchHitWalletConverter;
     private final SearchDao searchDao;
 
     @Override
-    public List<SearchHit> searchParty(String text) throws TException {
+    public List<SearchHit> searchParty(String text) {
         log.info("Incoming request for party with text: {}", text);
-        SearchHits<Party> searchHits = searchDao.searchParty(text);
+        SearchResponse<Party> searchHits = searchDao.searchParty(text);
         log.info("Found party: {}", searchHits);
         return searchHitConverter.convert(searchHits);
     }
 
     @Override
-    public List<SearchShopHit> searchShopText(String text) throws TException {
+    public List<SearchShopHit> searchShopText(String text) {
         log.info("Incoming request for shop with text: {}", text);
-        SearchHits<Party> searchHits = searchDao.searchParty(text);
+        SearchResponse<Party> searchHits = searchDao.searchParty(text);
         log.info("Found shop: {}", searchHits);
         return searchHitShopConverter.convert(searchHits);
     }
+
+    @Override
+    public List<SearchWalletHit> searchWalletText(String text) {
+        log.info("Incoming request for wallets with text: {}", text);
+        SearchResponse<Party> searchHits = searchDao.searchParty(text);
+        log.info("Found wallet: {}", searchHits);
+        return searchHitWalletConverter.convert(searchHits);
+    }
 }

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/PartyManagementHandler.java

@@ -3,4 +3,5 @@ package dev.vality.deanonimus.kafka.handler.party.management;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.machinegun.eventsink.MachineEvent;
 
-public interface PartyManagementHandler extends Handler<PartyChange, MachineEvent> {}
+public interface PartyManagementHandler extends Handler<PartyChange, MachineEvent> {
+}

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/contract/ContractCreatedHandler.java

@@ -3,12 +3,11 @@ package dev.vality.deanonimus.kafka.handler.party.management.contract;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.ContractEffectUnit;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.domain.Contract;
 import dev.vality.deanonimus.domain.ContractStatus;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.deanonimus.util.ContractUtil;
 import dev.vality.geck.common.util.TBaseUtil;
 import dev.vality.geck.common.util.TypeUtil;
@@ -27,7 +26,7 @@ import java.util.UUID;
 @RequiredArgsConstructor
 public class ContractCreatedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -73,9 +72,9 @@ public class ContractCreatedHandler extends AbstractClaimChangedHandler {
         }
         String contractorId = initContractorId(contractCreated);
         contract.setContractorId(contractorId);
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
         party.addContract(contract);
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
         log.info("End contract created handling, sequenceId={}, partyId={}, contractId={}, changeId={}",
                 sequenceId, partyId, contractId, changeId);
     }

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/contract/ContractLegalAgreementBoundHandler.java

@@ -4,12 +4,11 @@ import dev.vality.damsel.domain.LegalAgreement;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.ContractEffectUnit;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
 import dev.vality.deanonimus.db.exception.ContractNotFoundException;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.domain.Contract;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -24,7 +23,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ContractLegalAgreementBoundHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -46,12 +45,11 @@ public class ContractLegalAgreementBoundHandler extends AbstractClaimChangedHand
         String partyId = event.getSourceId();
         log.info("Start contract legal agreement bound handling, sequenceId={}, partyId={}, contractId={}, changeId={}",
                 sequenceId, partyId, contractId, changeId);
-        Party party = partyRepository.findById(partyId)
+        Party party = openSearchService.findPartyById(partyId);
-                .orElseThrow(() -> new PartyNotFoundException(partyId));
         Contract contract = party.getContractById(contractId)
                 .orElseThrow(() -> new ContractNotFoundException(contractId));
         contract.setLegalAgreementId(legalAgreement.getLegalAgreementId());
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End contract legal agreement bound handling, sequenceId={}, partyId={}, contractId={}, changeId={}",
                 sequenceId, partyId, contractId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/contract/ContractReportPreferencesChangedHandler.java

@@ -4,12 +4,11 @@ import dev.vality.damsel.domain.ReportPreferences;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.ContractEffectUnit;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
 import dev.vality.deanonimus.db.exception.ContractNotFoundException;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.domain.Contract;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.deanonimus.util.ContractUtil;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
@@ -25,7 +24,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ContractReportPreferencesChangedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -52,7 +51,7 @@ public class ContractReportPreferencesChangedHandler extends AbstractClaimChange
                         """,
                 sequenceId, partyId, contractId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
 
         Contract contract =
                 party.getContractById(contractId).orElseThrow(() -> new ContractNotFoundException(contractId));
@@ -63,12 +62,12 @@ public class ContractReportPreferencesChangedHandler extends AbstractClaimChange
             ContractUtil.setNullReportPreferences(contract);
         }
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info(
                 """
-                        End contract report preferences changed handling, 
+                        End contract report preferences changed handling,
-                        sequenceId={}, partyId={}, contractId={}, changeId={}
+                         sequenceId={}, partyId={}, contractId={}, changeId={}
                         """,
                 sequenceId, partyId, contractId, changeId);
     }

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/contractor/ContractorCreatedHandler.java

@@ -4,10 +4,9 @@ import dev.vality.damsel.domain.PartyContractor;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.ContractorEffectUnit;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.deanonimus.util.ContractorUtil;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
@@ -23,7 +22,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ContractorCreatedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -45,13 +44,13 @@ public class ContractorCreatedHandler extends AbstractClaimChangedHandler {
         String partyId = event.getSourceId();
         log.info("Start contractor created handling, eventId={}, partyId={}, contractorId={}", eventId, partyId,
                 contractorId);
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
         dev.vality.deanonimus.domain.Contractor contractor =
                 ContractorUtil.convertContractor(partyId, contractorCreated, contractorId);
 
         party.addContractor(contractor);
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
         log.info("End contractor created handling, eventId={}, partyId={}, contractorId={}", eventId, partyId,
                 contractorId);
     }

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/party/PartyBlockingHandler.java

@@ -2,10 +2,9 @@ package dev.vality.deanonimus.kafka.handler.party.management.party;
 
 import dev.vality.damsel.domain.Blocking;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.kafka.handler.party.management.PartyManagementHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.geck.filter.Filter;
 import dev.vality.geck.filter.PathConditionFilter;
 import dev.vality.geck.filter.condition.IsNullCondition;
@@ -22,7 +21,8 @@ import org.springframework.transaction.annotation.Transactional;
 @RequiredArgsConstructor
 public class PartyBlockingHandler implements PartyManagementHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
+
     private final Filter filter = new PathConditionFilter(new PathConditionRule(
             "party_blocking",
             new IsNullCondition().not()));
@@ -35,7 +35,7 @@ public class PartyBlockingHandler implements PartyManagementHandler {
         String partyId = event.getSourceId();
         log.info("Start party blocking handling, sequenceId={}, partyId={}, changeId={}", sequenceId, partyId,
                 changeId);
-        Party partySource = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party partySource = openSearchService.findPartyById(partyId);
 
         if (partyBlocking.isSetUnblocked()) {
             partySource.setBlocking(dev.vality.deanonimus.domain.Blocking.unblocked);
@@ -43,7 +43,7 @@ public class PartyBlockingHandler implements PartyManagementHandler {
             partySource.setBlocking(dev.vality.deanonimus.domain.Blocking.blocked);
         }
 
-        partyRepository.save(partySource);
+        openSearchService.updateParty(partySource);
         log.info("End party blocking handling, sequenceId={}, partyId={}, changeId={}", sequenceId, partyId, changeId);
     }
 

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/party/PartyCreatedHandler.java

@@ -2,11 +2,11 @@ package dev.vality.deanonimus.kafka.handler.party.management.party;
 
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.PartyCreated;
-import dev.vality.deanonimus.db.PartyRepository;
 import dev.vality.deanonimus.domain.Blocking;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Suspension;
 import dev.vality.deanonimus.kafka.handler.party.management.PartyManagementHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.geck.filter.Filter;
 import dev.vality.geck.filter.PathConditionFilter;
 import dev.vality.geck.filter.condition.IsNullCondition;
@@ -23,7 +23,7 @@ import org.springframework.transaction.annotation.Transactional;
 @RequiredArgsConstructor
 public class PartyCreatedHandler implements PartyManagementHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
     private final Filter filter = new PathConditionFilter(new PathConditionRule(
             "party_created",
             new IsNullCondition().not()));
@@ -41,7 +41,7 @@ public class PartyCreatedHandler implements PartyManagementHandler {
         party.setBlocking(Blocking.unblocked);
         party.setSuspension(Suspension.active);
 
-        partyRepository.save(party);
+        openSearchService.createParty(party);
         log.info("Party has been saved, sequenceId={}, partyId={}, changeId={}", sequenceId, partyId, changeId);
     }
 

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/party/PartySuspensionHandler.java

@@ -2,10 +2,9 @@ package dev.vality.deanonimus.kafka.handler.party.management.party;
 
 import dev.vality.damsel.domain.Suspension;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.kafka.handler.party.management.PartyManagementHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.geck.filter.Filter;
 import dev.vality.geck.filter.PathConditionFilter;
 import dev.vality.geck.filter.condition.IsNullCondition;
@@ -26,7 +25,7 @@ public class PartySuspensionHandler implements PartyManagementHandler {
             "party_suspension",
             new IsNullCondition().not()));
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -35,15 +34,15 @@ public class PartySuspensionHandler implements PartyManagementHandler {
         Suspension partySuspension = change.getPartySuspension();
         String partyId = event.getSourceId();
         log.info("Start party suspension handling, eventId={}, partyId={}, changeId={}", sequenceId, partyId, changeId);
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
 
         if (partySuspension.isSetActive()) {
             party.setSuspension(dev.vality.deanonimus.domain.Suspension.active);
         } else if (partySuspension.isSetSuspended()) {
             party.setSuspension(dev.vality.deanonimus.domain.Suspension.suspended);
         }
+        openSearchService.updateParty(party);
 
-        partyRepository.save(party);
         log.info("End party suspension handling, eventId={}, partyId={}, changeId={}", sequenceId, partyId, changeId);
     }
 

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopAccountCreatedHandler.java

@@ -4,11 +4,10 @@ import dev.vality.damsel.domain.ShopAccount;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.deanonimus.util.ShopUtil;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
@@ -24,7 +23,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopAccountCreatedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -46,12 +45,12 @@ public class ShopAccountCreatedHandler extends AbstractClaimChangedHandler {
         log.info("Start shop accountCreated handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
 
         ShopUtil.fillShopAccount(party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId)),
                 accountCreated);
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop accountCreated handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopBlockingHandler.java

@@ -2,12 +2,11 @@ package dev.vality.deanonimus.kafka.handler.party.management.shop;
 
 import dev.vality.damsel.domain.Blocking;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.PartyManagementHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.geck.filter.Filter;
 import dev.vality.geck.filter.PathConditionFilter;
 import dev.vality.geck.filter.condition.IsNullCondition;
@@ -24,7 +23,7 @@ import org.springframework.transaction.annotation.Transactional;
 @RequiredArgsConstructor
 public class ShopBlockingHandler implements PartyManagementHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
     private final Filter filter = new PathConditionFilter(new PathConditionRule(
             "shop_blocking",
             new IsNullCondition().not()));
@@ -39,11 +38,11 @@ public class ShopBlockingHandler implements PartyManagementHandler {
         log.info("Start shop blocking handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
 
         initBlockingFields(blocking, party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId)));
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop blocking handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopCategoryChangedHandler.java

@@ -3,12 +3,11 @@ package dev.vality.deanonimus.kafka.handler.party.management.shop;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -23,7 +22,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopCategoryChangedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -45,12 +44,12 @@ public class ShopCategoryChangedHandler extends AbstractClaimChangedHandler {
         log.info("Start shop categoryId changed handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
         Shop shop = party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId));
 
         shop.setCategoryId(categoryId);
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop categoryId changed handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopContractChangedHandler.java

@@ -4,12 +4,11 @@ import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ShopContractChanged;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -24,7 +23,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopContractChangedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -46,13 +45,13 @@ public class ShopContractChangedHandler extends AbstractClaimChangedHandler {
         log.info("Start shop contractChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
         Shop shop = party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId));
 
         shop.setContractId(contractChanged.getContractId());
         shop.setPayoutToolId(contractChanged.getPayoutToolId());
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop contractChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopCreatedHandler.java

@@ -4,13 +4,12 @@ package dev.vality.deanonimus.kafka.handler.party.management.shop;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.domain.Blocking;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.domain.Suspension;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.deanonimus.util.ShopUtil;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
@@ -26,7 +25,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopCreatedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -48,13 +47,13 @@ public class ShopCreatedHandler extends AbstractClaimChangedHandler {
         log.info("Start shop created handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
 
         Shop shop = fillShopInfo(shopCreated, shopId);
 
         party.addShop(shop);
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop created handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopDetailsChangedHandler.java

@@ -4,12 +4,11 @@ import dev.vality.damsel.domain.ShopDetails;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -24,7 +23,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopDetailsChangedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -46,14 +45,14 @@ public class ShopDetailsChangedHandler extends AbstractClaimChangedHandler {
         log.info("Start shop detailsChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
 
         Shop shop = party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId));
 
         shop.setDetailsName(detailsChanged.getName());
         shop.setDetailsDescription(detailsChanged.getDescription());
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop detailsChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopLocationChangedHandler.java

@@ -4,12 +4,11 @@ import dev.vality.damsel.domain.ShopLocation;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -24,7 +23,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopLocationChangedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -45,7 +44,7 @@ public class ShopLocationChangedHandler extends AbstractClaimChangedHandler {
         String partyId = event.getSourceId();
         log.info("Start shop locationChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
 
         Shop shop = party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId));
 
@@ -55,7 +54,7 @@ public class ShopLocationChangedHandler extends AbstractClaimChangedHandler {
             throw new IllegalArgumentException("Illegal shop location " + locationChanged);
         }
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop locationChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopPayoutScheduleChangedHandler.java

@@ -4,12 +4,11 @@ import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ScheduleChanged;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -24,7 +23,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopPayoutScheduleChangedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -46,7 +45,7 @@ public class ShopPayoutScheduleChangedHandler extends AbstractClaimChangedHandle
         log.info("Start shop payoutScheduleChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
         Shop shop = party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId));
 
         if (payoutScheduleChanged.isSetSchedule()) {
@@ -55,7 +54,7 @@ public class ShopPayoutScheduleChangedHandler extends AbstractClaimChangedHandle
             shop.setPayoutScheduleId(null);
         }
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop payoutScheduleChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopPayoutToolChangedHandler.java

@@ -3,12 +3,11 @@ package dev.vality.deanonimus.kafka.handler.party.management.shop;
 import dev.vality.damsel.payment_processing.ClaimEffect;
 import dev.vality.damsel.payment_processing.PartyChange;
 import dev.vality.damsel.payment_processing.ShopEffectUnit;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.machinegun.eventsink.MachineEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -23,7 +22,7 @@ import java.util.List;
 @RequiredArgsConstructor
 public class ShopPayoutToolChangedHandler extends AbstractClaimChangedHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
 
     @Override
     @Transactional(propagation = Propagation.REQUIRED)
@@ -45,12 +44,12 @@ public class ShopPayoutToolChangedHandler extends AbstractClaimChangedHandler {
         log.info("Start shop payoutToolChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
         Shop shop = party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId));
 
         shop.setPayoutToolId(payoutToolChanged);
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop payoutToolChanged handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/shop/ShopSuspensionHandler.java

@@ -2,12 +2,11 @@ package dev.vality.deanonimus.kafka.handler.party.management.shop;
 
 import dev.vality.damsel.domain.Suspension;
 import dev.vality.damsel.payment_processing.PartyChange;
-import dev.vality.deanonimus.db.PartyRepository;
-import dev.vality.deanonimus.db.exception.PartyNotFoundException;
 import dev.vality.deanonimus.db.exception.ShopNotFoundException;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.domain.Shop;
 import dev.vality.deanonimus.kafka.handler.party.management.PartyManagementHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
 import dev.vality.geck.filter.Filter;
 import dev.vality.geck.filter.PathConditionFilter;
 import dev.vality.geck.filter.condition.IsNullCondition;
@@ -24,7 +23,7 @@ import org.springframework.transaction.annotation.Transactional;
 @RequiredArgsConstructor
 public class ShopSuspensionHandler implements PartyManagementHandler {
 
-    private final PartyRepository partyRepository;
+    private final OpenSearchService openSearchService;
     private final Filter filter = new PathConditionFilter(new PathConditionRule(
             "shop_suspension",
             new IsNullCondition().not()));
@@ -39,7 +38,7 @@ public class ShopSuspensionHandler implements PartyManagementHandler {
         log.info("Start shop suspension handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);
 
-        Party party = partyRepository.findById(partyId).orElseThrow(() -> new PartyNotFoundException(partyId));
+        Party party = openSearchService.findPartyById(partyId);
         Shop shop = party.getShopById(shopId).orElseThrow(() -> new ShopNotFoundException(shopId));
 
         if (suspension.isSetActive()) {
@@ -48,7 +47,7 @@ public class ShopSuspensionHandler implements PartyManagementHandler {
             shop.setSuspension(dev.vality.deanonimus.domain.Suspension.suspended);
         }
 
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
 
         log.info("End shop suspension handling, sequenceId={}, partyId={}, shopId={}, changeId={}",
                 sequenceId, partyId, shopId, changeId);

src/main/java/dev/vality/deanonimus/kafka/handler/party/management/wallet/WalletCreatedHandler.java

@@ -0,0 +1,63 @@
+package dev.vality.deanonimus.kafka.handler.party.management.wallet;
+
+import dev.vality.damsel.payment_processing.ClaimEffect;
+import dev.vality.damsel.payment_processing.PartyChange;
+import dev.vality.damsel.payment_processing.WalletEffectUnit;
+import dev.vality.deanonimus.domain.Party;
+import dev.vality.deanonimus.domain.Wallet;
+import dev.vality.deanonimus.kafka.handler.party.management.AbstractClaimChangedHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
+import dev.vality.machinegun.eventsink.MachineEvent;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class WalletCreatedHandler extends AbstractClaimChangedHandler {
+
+    private final OpenSearchService openSearchService;
+
+    @Override
+    @Transactional(propagation = Propagation.REQUIRED)
+    public void handle(PartyChange change, MachineEvent event, Integer changeId) {
+        List<ClaimEffect> claimEffects = getClaimStatus(change).getAccepted().getEffects();
+        for (ClaimEffect claimEffect : claimEffects) {
+            if (claimEffect.isSetWalletEffect() && claimEffect.getWalletEffect().getEffect().isSetCreated()) {
+                handleEvent(event, changeId, claimEffect);
+            }
+        }
+    }
+
+    private void handleEvent(MachineEvent event, Integer changeId, ClaimEffect e) {
+        long sequenceId = event.getEventId();
+        WalletEffectUnit walletEffect = e.getWalletEffect();
+        dev.vality.damsel.domain.Wallet walletCreated = walletEffect.getEffect().getCreated();
+        String walletId = walletEffect.getId();
+        String partyId = event.getSourceId();
+        log.info("Start wallet created handling, sequenceId={}, partyId={}, walletId={}, changeId={}",
+                sequenceId, partyId, walletId, changeId);
+
+        Party party = openSearchService.findPartyById(partyId);
+        Wallet wallet = fillWalletInfo(walletCreated, walletId);
+        party.addWallet(wallet);
+
+        openSearchService.updateParty(party);
+
+        log.info("End wallet created handling, sequenceId={}, partyId={}, walletId={}, changeId={}",
+                sequenceId, partyId, walletId, changeId);
+    }
+
+    private Wallet fillWalletInfo(dev.vality.damsel.domain.Wallet walletCreated, String walletId) {
+        Wallet wallet = new Wallet();
+        wallet.setId(walletId);
+        wallet.setName(walletCreated.getName());
+
+        return wallet;
+    }
+}

src/main/java/dev/vality/deanonimus/service/OpenSearchService.java

@@ -0,0 +1,51 @@
+package dev.vality.deanonimus.service;
+
+import dev.vality.deanonimus.domain.Party;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.opensearch.client.opensearch.OpenSearchClient;
+import org.opensearch.client.opensearch.core.GetRequest;
+import org.opensearch.client.opensearch.core.IndexRequest;
+import org.opensearch.client.opensearch.core.UpdateRequest;
+import org.springframework.stereotype.Service;
+
+import static dev.vality.deanonimus.constant.OpenSearchConstants.PARTY_INDEX;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class OpenSearchService {
+
+    private final OpenSearchClient openSearchClient;
+
+    @SneakyThrows
+    public Party findPartyById(String partyId) {
+        return openSearchClient.get(new GetRequest.Builder()
+                        .index(PARTY_INDEX)
+                        .id(partyId)
+                        .build(),
+                Party.class).source();
+    }
+
+    @SneakyThrows
+    public void updateParty(Party party) {
+        openSearchClient.update(
+                new UpdateRequest.Builder<Party, Party>()
+                        .index(PARTY_INDEX)
+                        .id(party.getId())
+                        .doc(party)
+                        .build(),
+                Party.class);
+    }
+
+    @SneakyThrows
+    public Party createParty(Party party) {
+        openSearchClient.index(new IndexRequest.Builder<Party>()
+                .index(PARTY_INDEX)
+                .id(party.getId())
+                .document(party)
+                .build());
+        return party;
+    }
+}

src/main/resources/application.yml

@@ -1,11 +1,11 @@
 server:
-  port: '@server.port@'
+  port: '${server.port}'
 
 management:
   security:
     flag: false
   server:
-    port: '@management.port@'
+    port: '${management.port}'
   metrics:
     export:
       prometheus:
@@ -28,8 +28,6 @@ spring:
   output:
     ansi:
       enabled: always
-  elasticsearch:
-    uris: "http://localhost:9200"
   kafka:
     bootstrap-servers: localhost:9092
     client-id: deanonimus
@@ -42,6 +40,14 @@ spring:
         max.poll.interval.ms: 30000
         session.timeout.ms: 30000
 
+opensearch:
+  username: none
+  password: none
+  hostname: localhost
+  port: 9200
+  certificate: none
+  sslEnabled: false
+
 info:
   version: '@project.version@'
   stage: dev

src/test/java/dev/vality/deanonimus/AbstractIntegrationTest.java

@@ -31,12 +31,8 @@ public abstract class AbstractIntegrationTest {
     @DynamicPropertySource
     static void containersProps(DynamicPropertyRegistry registry) {
         registry.add("spring.kafka.bootstrap-servers", KafkaContainerExtension.KAFKA::getBootstrapServers);
-        registry.add("spring.elasticsearch.rest.uris", () -> {
+        registry.add("opensearch.hostname", () -> OpensearchContainerExtension.OPENSEARCH.getHost());
-            return "http://" +
+        registry.add("opensearch.port", () -> OpensearchContainerExtension.OPENSEARCH.getFirstMappedPort());
-                    OpensearchContainerExtension.OPENSEARCH.getHost() +
-                    ":" +
-                    OpensearchContainerExtension.OPENSEARCH.getFirstMappedPort();
-        });
     }
 
     public static void sendMessages(List<SinkEvent> sinkEvents) {

src/test/java/dev/vality/deanonimus/PartyFlowGenerator.java

@@ -16,53 +16,23 @@ import lombok.NoArgsConstructor;
 import java.io.IOException;
 import java.time.Instant;
 import java.time.LocalDateTime;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class PartyFlowGenerator {
 
-    public static final String PARTY_EMAIL = "testPartyEmail";
+    private static final String PARTY_EMAIL = "testPartyEmail";
-    public static final String SOURCE_NS = "source_ns";
+    private static final String SOURCE_NS = "source_ns";
-    public static final String PARTY_BLOCK_REASON = "testPartyBlockReason";
+    private static final String PARTY_BLOCK_REASON = "testPartyBlockReason";
-    public static final String SHOP_BLOCK_REASON = "testShopBlockReason";
+    private static final Long PARTY_REVISION_ID = 12345L;
-    public static final String SHOP_UNBLOCK_REASON = "testShopUnblockReason";
+    private static final Long CLAIM_ID = 524523L;
-    public static final Long PARTY_REVISION_ID = 12345L;
+    private static final Integer REVISION_ID = 431531;
-    public static final Long CLAIM_ID = 524523L;
+    private static final String WALLET_ID = "345435435";
-    public static final Integer REVISION_ID = 431531;
+    private static final String CONTRACTOR_ID = "563462";
-    public static final Integer CATEGORY_ID = 542432;
+    private static final String INN = "213123123123";
-    public static final String CONTRACT_ID = "142534";
-    public static final String PAYOUT_ID = "654635";
-    public static final String DETAILS_NAME = "testDetailsName";
-    public static final String DETAILS_DESCRIPTION = "testDescription";
-    public static final Integer SCHEDULE_ID = 15643653;
-    public static final String PAYOUT_TOOL_ID = "654635";
-    public static final String CURRENCY_SYMBOL = "RUB";
-    public static final Long SETTLEMENT_ID = 245234L;
-    public static final String CONTRACTOR_ID = "563462";
-    public static final Long SHOP_ACCOUNT_PAYOUT = 5425234L;
-    public static final String INN = "213123123123";
-
-    public static List<SinkEvent> generatePartyFlow(String partyId, String shopId) throws IOException {
-        List<SinkEvent> sinkEvents = new ArrayList<>();
-        Long sequenceId = 0L;
-        sinkEvents.add(buildSinkEvent(buildMessagePartyCreated(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessagePartyBlocking(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessagePartySuspension(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessagePartyRevisionChanged(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildContractorCreated(sequenceId++, buildPartyContractor(partyId), partyId)));
-        sinkEvents.add(buildSinkEvent(buildContractorIdentificationLevelChanged(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopCreated(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopBlocking(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopSuspension(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopCategoryChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopContractChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopDetailsChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopPayoutScheduleChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopPayoutToolChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopAccountCreated(sequenceId++, partyId, shopId)));
-
-        return sinkEvents;
-    }
 
     public static List<SinkEvent> generatePartyContractorFlow(String partyId) throws IOException {
         List<SinkEvent> sinkEvents = new ArrayList<>();
@@ -73,161 +43,19 @@ public class PartyFlowGenerator {
         sinkEvents.add(buildSinkEvent(buildMessagePartyRevisionChanged(sequenceId++, partyId)));
         sinkEvents.add(buildSinkEvent(
                 buildContractorCreated(sequenceId++, buildRussianLegalPartyContractor(partyId), partyId)));
-        sinkEvents.add(buildSinkEvent(buildContractorIdentificationLevelChanged(sequenceId++, partyId)));
+        sinkEvents.add(buildSinkEvent(buildContractorIdentificationLevelChanged(sequenceId, partyId)));
 
         return sinkEvents;
     }
 
-    public static List<SinkEvent> generateShopFlow(String partyId, String shopId) throws IOException {
+    public static List<SinkEvent> generateWalletFlow(String partyId) throws IOException {
         List<SinkEvent> sinkEvents = new ArrayList<>();
-        Long sequenceId = 0L;
+        long sequenceId = 0L;
         sinkEvents.add(buildSinkEvent(buildMessagePartyCreated(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopCreated(sequenceId++, partyId, shopId)));
+        sinkEvents.add(buildSinkEvent(buildMessageWalletCreated(sequenceId, partyId, WALLET_ID)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopBlocking(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopPayoutToolChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopSuspension(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopCategoryChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopContractChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopDetailsChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopPayoutScheduleChanged(sequenceId++, partyId, shopId)));
-        sinkEvents.add(buildSinkEvent(buildMessageShopAccountCreated(sequenceId++, partyId, shopId)));
-
-        sinkEvents.add(buildSinkEvent(buildMessageShopSuspension(
-                sequenceId++, partyId,
-                buildSuspendedShopSuspension(TypeUtil.temporalToString(LocalDateTime.now()), shopId))));
-
         return sinkEvents;
     }
 
-    public static List<SinkEvent> generatePartyFlowWithCount(int count, String lastPartyId, PartyContractor contractor)
-            throws IOException {
-        List<SinkEvent> sinkEvents = new ArrayList<>();
-        Long sequenceId = 0L;
-        for (int i = 0; i < count; i++) {
-            if (i == count - 1) {
-                sinkEvents.add(buildSinkEvent(buildMessagePartyCreated(sequenceId++, lastPartyId)));
-                sinkEvents.add(buildSinkEvent(buildMessagePartyBlocking(sequenceId++, lastPartyId)));
-                sinkEvents.add(buildSinkEvent(buildContractorCreated(sequenceId++, contractor, lastPartyId)));
-                sinkEvents.add(buildSinkEvent(buildContractorIdentificationLevelChanged(sequenceId++, lastPartyId)));
-            } else {
-                String partyId = UUID.randomUUID().toString();
-                sinkEvents.add(buildSinkEvent(buildMessagePartyCreated(sequenceId++, partyId)));
-                sinkEvents.add(buildSinkEvent(
-                        buildContractorCreated(sequenceId++, buildRussianLegalPartyContractor(partyId), partyId)));
-                sinkEvents.add(buildSinkEvent(buildContractorIdentificationLevelChanged(sequenceId++, partyId)));
-                sinkEvents.add(buildSinkEvent(buildMessagePartyBlocking(sequenceId++, partyId)));
-                sinkEvents.add(buildSinkEvent(buildMessagePartySuspension(sequenceId++, partyId)));
-                sinkEvents.add(buildSinkEvent(buildMessagePartyRevisionChanged(sequenceId++, partyId)));
-            }
-        }
-
-        return sinkEvents;
-    }
-
-    public static List<SinkEvent> generatePartyFlowWithMultiplePartyChange(int count,
-                                                                           String lastPartyId,
-                                                                           PartyContractor partyContractor)
-            throws IOException {
-        List<SinkEvent> sinkEvents = new ArrayList<>();
-        Long sequenceId = 0L;
-        for (int i = 0; i < count; i++) {
-            if (i == count - 1) {
-                PartyChange partyChange = buildContractorCreatedPartyChange(partyContractor);
-                sinkEvents.add(buildSinkEvent(buildMultiPartyChange(sequenceId, lastPartyId, partyChange)));
-            } else {
-                String partyId = UUID.randomUUID().toString();
-                sinkEvents.add(buildSinkEvent(buildMultiPartyChange(sequenceId, partyId, null)));
-            }
-        }
-
-        return sinkEvents;
-    }
-
-    public static List<SinkEvent> generatePartyFlowWithMultiplePartyShopChange(int count,
-                                                                               String lastPartyId,
-                                                                               String lastShopId,
-                                                                               PartyChange customPartyChange)
-            throws IOException {
-        List<SinkEvent> sinkEvents = new ArrayList<>();
-        Long sequenceId = 0L;
-        for (int i = 0; i < count; i++) {
-            if (i == count - 1) {
-                sinkEvents.add(buildSinkEvent(
-                        buildMultiShopChange(sequenceId, lastPartyId, lastShopId, customPartyChange)));
-            } else {
-                String partyId = UUID.randomUUID().toString();
-                String shopId = UUID.randomUUID().toString();
-                sinkEvents.add(buildSinkEvent(buildMultiShopChange(sequenceId, partyId, shopId, null)));
-            }
-        }
-
-        return sinkEvents;
-    }
-
-    public static List<SinkEvent> generatePartyFlowWithMultipleShopInOneChange(String lastPartyId,
-                                                                               String lastShopId,
-                                                                               PartyChange customPartyChange)
-            throws IOException {
-        List<SinkEvent> sinkEvents = new ArrayList<>();
-        Long sequenceId = 0L;
-        sinkEvents.add(buildSinkEvent(
-                buildMultiShopChangeDifferentShopId(sequenceId, lastPartyId, lastShopId, customPartyChange)));
-
-        return sinkEvents;
-    }
-
-    public static List<SinkEvent> generatePartyFlowWithContract(String partyId, LegalEntity legalEntity)
-            throws IOException {
-        List<SinkEvent> sinkEvents = new ArrayList<>();
-        Long sequenceId = 0L;
-        sinkEvents.add(buildSinkEvent(buildMessagePartyCreated(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessagePartyBlocking(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessagePartySuspension(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildMessagePartyRevisionChanged(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(
-                buildContractorCreated(sequenceId++, buildRussianLegalPartyContractor(partyId), partyId)));
-        sinkEvents.add(buildSinkEvent(buildContractorIdentificationLevelChanged(sequenceId++, partyId)));
-        sinkEvents.add(buildSinkEvent(buildContractContractorCreated(sequenceId++, partyId, legalEntity)));
-
-        return sinkEvents;
-    }
-
-    public static MachineEvent buildContractContractorCreated(Long sequenceId, String partyId,
-                                                              LegalEntity legalEntity) {
-        Contract contract = new Contract();
-        contract.setId("testContractId");
-        contract.setContractorId("testContractorId");
-        contract.setCreatedAt(TypeUtil.temporalToString(LocalDateTime.now()));
-        contract.setStatus(ContractStatus.active(new ContractActive()));
-        contract.setAdjustments(Collections.emptyList());
-        contract.setPayoutTools(Collections.emptyList());
-        TermSetHierarchyRef termSetHierarchyRef = new TermSetHierarchyRef();
-        termSetHierarchyRef.setId(12345);
-        contract.setTerms(termSetHierarchyRef);
-        Contractor contractor = new Contractor();
-        contractor.setLegalEntity(legalEntity);
-        contract.setContractor(contractor);
-        ContractEffect contractEffect = new ContractEffect();
-        contractEffect.setCreated(contract);
-        ContractEffectUnit contractEffectUnit = new ContractEffectUnit();
-        contractEffectUnit.setContractId("testContractId");
-        contractEffectUnit.setEffect(contractEffect);
-        ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setContractEffect(contractEffectUnit);
-        Claim claim = buildClaimCreated(claimEffect);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setClaimCreated(claim);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static RussianLegalEntity buildRussianLegalEntity() throws IOException {
-        RussianLegalEntity russianLegalEntity = new RussianLegalEntity();
-        russianLegalEntity = new MockTBaseProcessor(MockMode.ALL)
-                .process(russianLegalEntity, new TBaseHandler<>(RussianLegalEntity.class));
-        russianLegalEntity.setInn(INN);
-        return russianLegalEntity;
-    }
-
     public static MachineEvent buildMessagePartyCreated(Long sequenceId, String partyId) {
         PartyChange partyChange = buildPartyCreatedPartyChange(partyId);
         return buildMachineEvent(partyId, sequenceId, partyChange);
@@ -276,190 +104,30 @@ public class PartyFlowGenerator {
         return partyChange;
     }
 
-    public static MachineEvent buildMessageShopBlocking(Long sequenceId, String partyId, String shopId) {
+    public static MachineEvent buildMessageWalletCreated(Long sequenceId, String partyId, String walletId)
-        PartyChange partyChange = buildShopBlockingPartyChange(shopId);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static MachineEvent buildMessageShopSuspension(Long sequenceId, String partyId, String shopId) {
-        PartyChange partyChange = buildShopSuspensionPartyChange(shopId);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static MachineEvent buildMessageShopSuspension(Long sequenceId, String partyId,
-                                                          ShopSuspension shopSuspension) {
-        PartyChange partyChange = new PartyChange();
-        partyChange.setShopSuspension(shopSuspension);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static PartyChange buildShopBlockingPartyChange(String shopId) {
-        ShopBlocking shopBlocking = buildShopBlocking(shopId);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setShopBlocking(shopBlocking);
-        return partyChange;
-    }
-
-    public static PartyChange buildShopSuspensionPartyChange(String shopId) {
-        ShopSuspension shopSuspension =
-                buildActiveShopSuspension(TypeUtil.temporalToString(LocalDateTime.now()), shopId);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setShopSuspension(shopSuspension);
-        return partyChange;
-    }
-
-    public static MachineEvent buildMessageShopCreated(Long sequenceId, String partyId, String shopId)
             throws IOException {
-        PartyChange partyChange = buildShopCreatedPartyChange(shopId);
+        PartyChange partyChange = buildWalletCreatedPartyChange(walletId);
         return buildMachineEvent(partyId, sequenceId, partyChange);
     }
 
-    public static PartyChange buildShopCreatedPartyChange(String shopId) throws IOException {
+    public static PartyChange buildWalletCreatedPartyChange(String walletd) throws IOException {
-        Shop shop = buildShopCreated();
+        WalletEffectUnit walletEffectUnit = new WalletEffectUnit();
-        ShopEffectUnit shopEffectUnit = new ShopEffectUnit();
+        walletEffectUnit.setId(walletd);
-        shopEffectUnit.setShopId(shopId);
+        WalletEffect shopEffect = new WalletEffect();
-        ShopEffect shopEffect = new ShopEffect();
+        shopEffect.setCreated(buildWalletCreated());
-        shopEffect.setCreated(buildShopCreated());
+        walletEffectUnit.setEffect(shopEffect);
-        shopEffectUnit.setEffect(shopEffect);
         ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setShopEffect(shopEffectUnit);
+        claimEffect.setWalletEffect(walletEffectUnit);
-        Claim claim = buildClaimCreated(claimEffect);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setClaimCreated(claim);
-
-        return partyChange;
-    }
-
-    public static MachineEvent buildMessageShopCategoryChanged(Long sequenceId, String partyId, String shopId) {
-        PartyChange partyChange = buildShopCategoryPartyChange(shopId);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static PartyChange buildShopCategoryPartyChange(String shopId) {
-        ShopEffectUnit shopEffectUnit = new ShopEffectUnit();
-        shopEffectUnit.setShopId(shopId);
-        ShopEffect shopEffect = new ShopEffect();
-        CategoryRef categoryRef = new CategoryRef();
-        categoryRef.setId(CATEGORY_ID);
-        shopEffect.setCategoryChanged(categoryRef);
-        shopEffectUnit.setEffect(shopEffect);
-        ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setShopEffect(shopEffectUnit);
         Claim claim = buildClaimCreated(claimEffect);
         PartyChange partyChange = new PartyChange();
         partyChange.setClaimCreated(claim);
         return partyChange;
     }
 
-    public static MachineEvent buildMessageShopContractChanged(Long sequenceId, String partyId, String shopId) {
+    public static MachineEvent buildContractorCreated(
-        PartyChange partyChange = buildShopContractPartyChange(shopId);
+            Long sequenceId,
-        return buildMachineEvent(partyId, sequenceId, partyChange);
+            PartyContractor partyContractor,
-    }
+            String partyId) {
-
-    public static PartyChange buildShopContractPartyChange(String shopId) {
-        ShopEffectUnit shopEffectUnit = new ShopEffectUnit();
-        shopEffectUnit.setShopId(shopId);
-        ShopContractChanged shopContractChanged = new ShopContractChanged();
-        shopContractChanged.setContractId(CONTRACT_ID);
-        shopContractChanged.setPayoutToolId(PAYOUT_ID);
-        ShopEffect shopEffect = new ShopEffect();
-        shopEffect.setContractChanged(shopContractChanged);
-        shopEffectUnit.setEffect(shopEffect);
-        ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setShopEffect(shopEffectUnit);
-        Claim claim = buildClaimCreated(claimEffect);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setClaimCreated(claim);
-        return partyChange;
-    }
-
-    public static MachineEvent buildMessageShopDetailsChanged(Long sequenceId, String partyId, String shopId) {
-        PartyChange partyChange = buildShopDetailsPartyChange(shopId);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static PartyChange buildShopDetailsPartyChange(String shopId) {
-        ShopEffectUnit shopEffectUnit = new ShopEffectUnit();
-        shopEffectUnit.setShopId(shopId);
-        ShopDetails shopDetails = new ShopDetails();
-        shopDetails.setName(DETAILS_NAME);
-        shopDetails.setDescription(DETAILS_DESCRIPTION);
-        ShopEffect shopEffect = new ShopEffect();
-        shopEffect.setDetailsChanged(shopDetails);
-        shopEffectUnit.setEffect(shopEffect);
-        ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setShopEffect(shopEffectUnit);
-        Claim claim = buildClaimCreated(claimEffect);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setClaimCreated(claim);
-        return partyChange;
-    }
-
-    public static MachineEvent buildMessageShopPayoutScheduleChanged(Long sequenceId, String partyId, String shopdId) {
-        PartyChange partyChange = buildShopPayouScheduleChangedPartyChange(shopdId);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static PartyChange buildShopPayouScheduleChangedPartyChange(String shopdId) {
-        ShopEffectUnit shopEffectUnit = new ShopEffectUnit();
-        shopEffectUnit.setShopId(shopdId);
-        ScheduleChanged scheduleChanged = new ScheduleChanged();
-        scheduleChanged.setSchedule(new BusinessScheduleRef(SCHEDULE_ID));
-        ShopEffect shopEffect = new ShopEffect();
-        shopEffect.setPayoutScheduleChanged(scheduleChanged);
-        shopEffectUnit.setEffect(shopEffect);
-        ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setShopEffect(shopEffectUnit);
-        Claim claim = buildClaimCreated(claimEffect);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setClaimCreated(claim);
-        return partyChange;
-    }
-
-    public static MachineEvent buildMessageShopPayoutToolChanged(Long sequenceId, String partyId, String shopdId) {
-        PartyChange partyChange = buildShopPayoutToolChangedPartyChange(shopdId, PAYOUT_TOOL_ID);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static PartyChange buildShopPayoutToolChangedPartyChange(String shopdId, String payoutToolId) {
-        ShopEffectUnit shopEffectUnit = new ShopEffectUnit();
-        shopEffectUnit.setShopId(shopdId);
-        ShopEffect shopEffect = new ShopEffect();
-        shopEffect.setPayoutToolChanged(payoutToolId);
-        shopEffectUnit.setEffect(shopEffect);
-        ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setShopEffect(shopEffectUnit);
-        Claim claim = buildClaimCreated(claimEffect);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setClaimCreated(claim);
-        return partyChange;
-    }
-
-    public static MachineEvent buildMessageShopAccountCreated(Long sequenceId, String partyId, String shopdId) {
-        PartyChange partyChange = buildShopAccountCreatedPartyChange(shopdId);
-        return buildMachineEvent(partyId, sequenceId, partyChange);
-    }
-
-    public static PartyChange buildShopAccountCreatedPartyChange(String shopdId) {
-        ShopEffectUnit shopEffectUnit = new ShopEffectUnit();
-        shopEffectUnit.setShopId(shopdId);
-        ShopAccount shopAccount = new ShopAccount();
-        shopAccount.setCurrency(new CurrencyRef(CURRENCY_SYMBOL));
-        shopAccount.setPayout(SHOP_ACCOUNT_PAYOUT);
-        shopAccount.setSettlement(SETTLEMENT_ID);
-        ShopEffect shopEffect = new ShopEffect();
-        shopEffect.setAccountCreated(shopAccount);
-        shopEffectUnit.setEffect(shopEffect);
-        ClaimEffect claimEffect = new ClaimEffect();
-        claimEffect.setShopEffect(shopEffectUnit);
-        Claim claim = buildClaimCreated(claimEffect);
-        PartyChange partyChange = new PartyChange();
-        partyChange.setClaimCreated(claim);
-        return partyChange;
-    }
-
-    public static MachineEvent buildContractorCreated(Long sequenceId, PartyContractor partyContractor, String partyId)
-            throws IOException {
         PartyChange partyChange = buildContractorCreatedPartyChange(partyContractor);
         return buildMachineEvent(partyId, sequenceId, partyChange);
     }
@@ -497,95 +165,6 @@ public class PartyFlowGenerator {
         return partyChange;
     }
 
-    public static MachineEvent buildMultiPartyChange(Long sequenceId, String partyId, PartyChange customPartyChange)
-            throws IOException {
-        PartyChange partyCreatedChange = buildPartyCreatedPartyChange(partyId);
-        PartyChange partyRevisionChange = buildPartyRevisionChangedPartyChange();
-        PartyChange partyBlockingChange = buildPartyBlockingPartyChange();
-        PartyChange contractorCreatedPartyChange = buildContractorCreatedPartyChange(buildPartyContractor(partyId));
-        PartyChange identificationLevelChangedPartyChange = buildContractorIdentificationLevelChangedPartyChange();
-        if (customPartyChange == null) {
-            return buildMachineEvent(partyId, sequenceId, partyCreatedChange, partyRevisionChange,
-                    partyBlockingChange, contractorCreatedPartyChange, identificationLevelChangedPartyChange);
-        } else {
-            return buildMachineEvent(partyId, sequenceId, partyCreatedChange, partyRevisionChange,
-                    partyBlockingChange, contractorCreatedPartyChange, identificationLevelChangedPartyChange,
-                    customPartyChange);
-        }
-    }
-
-    public static MachineEvent buildMultiShopChange(Long sequenceId, String partyId, String shopId,
-                                                    PartyChange customPartyChange) throws IOException {
-        PartyChange shopCreatedPartyChange = buildShopCreatedPartyChange(shopId);
-        PartyChange shopBlockingPartyChange = buildShopBlockingPartyChange(shopId);
-        PartyChange shopCategoryPartyChange = buildShopCategoryPartyChange(shopId);
-        PartyChange shopContractPartyChange = buildShopContractPartyChange(shopId);
-        PartyChange shopDetailsPartyChange = buildShopDetailsPartyChange(shopId);
-        PartyChange shopSuspensionPartyChange = buildShopSuspensionPartyChange(shopId);
-        PartyChange shopAccountCreatedPartyChange = buildShopAccountCreatedPartyChange(shopId);
-        PartyChange shopPayoutToolChangedPartyChange = buildShopPayoutToolChangedPartyChange(shopId, PAYOUT_TOOL_ID);
-        PartyChange shopPayouScheduleChangedPartyChange = buildShopPayouScheduleChangedPartyChange(shopId);
-        if (customPartyChange == null) {
-            return buildMachineEvent(partyId, sequenceId, shopCreatedPartyChange, shopBlockingPartyChange,
-                    shopCategoryPartyChange, shopContractPartyChange, shopDetailsPartyChange, shopSuspensionPartyChange,
-                    shopAccountCreatedPartyChange, shopPayoutToolChangedPartyChange,
-                    shopPayouScheduleChangedPartyChange);
-        } else {
-            return buildMachineEvent(partyId, sequenceId, shopCreatedPartyChange, shopBlockingPartyChange,
-                    shopCategoryPartyChange, shopContractPartyChange, shopDetailsPartyChange, shopSuspensionPartyChange,
-                    shopAccountCreatedPartyChange, shopPayoutToolChangedPartyChange,
-                    shopPayouScheduleChangedPartyChange, customPartyChange);
-        }
-    }
-
-    public static MachineEvent buildMultiShopChangeDifferentShopId(Long sequenceId, String partyId, String shopId,
-                                                                   PartyChange customPartyChange) throws IOException {
-        String firstShopId = UUID.randomUUID().toString();
-        PartyChange shopCreatedPartyChange = buildShopCreatedPartyChange(firstShopId);
-        PartyChange shopBlockingPartyChange = buildShopBlockingPartyChange(firstShopId);
-        String secondShopId = UUID.randomUUID().toString();
-        PartyChange shopCreatedPartyChangeSec = buildShopCreatedPartyChange(secondShopId);
-        PartyChange shopCategoryPartyChangeSec = buildShopCategoryPartyChange(secondShopId);
-        PartyChange shopContractPartyChangeSec = buildShopContractPartyChange(secondShopId);
-        String thirdShopId = UUID.randomUUID().toString();
-        PartyChange shopCreatedPartyChangeThird = buildShopCreatedPartyChange(thirdShopId);
-        PartyChange shopDetailsPartyChangeThird = buildShopDetailsPartyChange(thirdShopId);
-        PartyChange shopSuspensionPartyChangeThird = buildShopSuspensionPartyChange(thirdShopId);
-        String fourthShopId = shopId != null ? shopId : UUID.randomUUID().toString();
-        PartyChange shopCreatedPartyChangeFourth = buildShopCreatedPartyChange(fourthShopId);
-        PartyChange shopAccountCreatedPartyChangeFourth = buildShopAccountCreatedPartyChange(fourthShopId);
-        PartyChange shopPayoutToolChangedPartyChangeFourth =
-                buildShopPayoutToolChangedPartyChange(fourthShopId, PAYOUT_TOOL_ID);
-        PartyChange shopPayouScheduleChangedPartyChangeFourth = buildShopPayouScheduleChangedPartyChange(fourthShopId);
-        if (customPartyChange == null) {
-            return buildMachineEvent(partyId, sequenceId, shopCreatedPartyChange, shopBlockingPartyChange,
-                    shopCreatedPartyChangeSec, shopCategoryPartyChangeSec, shopContractPartyChangeSec,
-                    shopCreatedPartyChangeThird,
-                    shopDetailsPartyChangeThird, shopSuspensionPartyChangeThird, shopCreatedPartyChangeFourth,
-                    shopAccountCreatedPartyChangeFourth,
-                    shopPayoutToolChangedPartyChangeFourth, shopPayouScheduleChangedPartyChangeFourth);
-        } else {
-            return buildMachineEvent(partyId, sequenceId, shopCreatedPartyChange, shopBlockingPartyChange,
-                    shopCreatedPartyChangeSec, shopCategoryPartyChangeSec, shopContractPartyChangeSec,
-                    shopCreatedPartyChangeThird,
-                    shopDetailsPartyChangeThird, shopSuspensionPartyChangeThird, shopCreatedPartyChangeFourth,
-                    shopAccountCreatedPartyChangeFourth,
-                    shopPayoutToolChangedPartyChangeFourth, shopPayouScheduleChangedPartyChangeFourth,
-                    customPartyChange);
-        }
-    }
-
-    public static PartyContractor buildPartyContractor(String partyId) throws IOException {
-        PartyContractor partyContractor = new PartyContractor();
-        partyContractor.setId(partyId);
-        partyContractor.setStatus(ContractorIdentificationLevel.full);
-        Contractor contractor = new Contractor();
-        contractor = new MockTBaseProcessor(MockMode.ALL).process(contractor, new TBaseHandler<>(Contractor.class));
-        partyContractor.setContractor(contractor);
-        partyContractor.setIdentityDocuments(Collections.emptyList());
-        return partyContractor;
-    }
-
     public static PartyContractor buildRussianLegalPartyContractor(String partyId) throws IOException {
         PartyContractor partyContractor = new PartyContractor();
         partyContractor.setId(partyId);
@@ -611,33 +190,12 @@ public class PartyFlowGenerator {
                 TypeUtil.temporalToString(LocalDateTime.now()));
     }
 
-    public static Shop buildShopCreated() throws IOException {
+    public static Wallet buildWalletCreated() throws IOException {
-        Shop shop = new Shop();
+        Wallet wallet = new Wallet();
-        shop = new MockTBaseProcessor(MockMode.ALL).process(shop, new TBaseHandler<>(Shop.class));
+        wallet = new MockTBaseProcessor(MockMode.ALL).process(wallet, new TBaseHandler<>(Wallet.class));
-        shop.setCreatedAt(TypeUtil.temporalToString(LocalDateTime.now()));
+        wallet.setCreatedAt(TypeUtil.temporalToString(LocalDateTime.now()));
-        Blocking blocking = new Blocking();
+        wallet.setSuspension(buildPartySuspension());
-        blocking.setBlocked(new Blocked(SHOP_BLOCK_REASON, TypeUtil.temporalToString(LocalDateTime.now())));
+        return wallet;
-        shop.setBlocking(blocking);
-        shop.setSuspension(buildPartySuspension());
-        return shop;
-    }
-
-    public static ShopSuspension buildActiveShopSuspension(String since, String shopId) {
-        Suspension suspension = new Suspension();
-        suspension.setActive(new Active(since));
-        return new ShopSuspension(shopId, suspension);
-    }
-
-    public static ShopSuspension buildSuspendedShopSuspension(String since, String shopId) {
-        Suspension suspension = new Suspension();
-        suspension.setSuspended(new Suspended(since));
-        return new ShopSuspension(shopId, suspension);
-    }
-
-    public static ShopBlocking buildShopBlocking(String shopId) {
-        Blocking blocking = new Blocking();
-        blocking.setUnblocked(new Unblocked(SHOP_UNBLOCK_REASON, TypeUtil.temporalToString(LocalDateTime.now())));
-        return new ShopBlocking(shopId, blocking);
     }
 
     public static PartyRevisionChanged buildPartyRevisionChanged() {
@@ -681,5 +239,4 @@ public class PartyFlowGenerator {
         sinkEvent.setEvent(machineEvent);
         return sinkEvent;
     }
-
 }

src/test/java/dev/vality/deanonimus/ReadTest.java

@@ -2,12 +2,14 @@ package dev.vality.deanonimus;
 
 import dev.vality.damsel.deanonimus.SearchHit;
 import dev.vality.damsel.deanonimus.SearchShopHit;
-import dev.vality.deanonimus.db.PartyRepository;
+import dev.vality.damsel.deanonimus.SearchWalletHit;
 import dev.vality.deanonimus.domain.Party;
 import dev.vality.deanonimus.handler.DeanonimusServiceHandler;
+import dev.vality.deanonimus.service.OpenSearchService;
+import lombok.SneakyThrows;
 import org.apache.thrift.TException;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.opensearch.client.opensearch.OpenSearchClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 
@@ -21,7 +23,10 @@ public class ReadTest extends AbstractIntegrationTest {
     Integer responseLimit;
 
     @Autowired
-    PartyRepository partyRepository;
+    OpenSearchService openSearchService;
+
+    @Autowired
+    OpenSearchClient client;
 
     @Autowired
     DeanonimusServiceHandler deanonimusServiceHandler;
@@ -32,18 +37,16 @@ public class ReadTest extends AbstractIntegrationTest {
     private static final String URL = "http://url.com";
     private static final String CONTRACT = "contract";
     private static final String CONTRACTOR = "contractor";
+    private static final String WALLET = "wallet";
+    private static final String WALLET_NAME = "wallet_name";
     private static final String INN = "1234234123";
     private static final String ACCOUNT = "9999999999";
 
-    @BeforeEach
-    void setUp() {
-        partyRepository.deleteAll();
-    }
 
     @Test
     void searchByPartyId() throws TException {
         givenParty(PARTY, EMAIL);
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(PARTY);
 
         assertFalse(searchHits.isEmpty());
@@ -58,7 +61,7 @@ public class ReadTest extends AbstractIntegrationTest {
         givenParty(PARTY + "-test-rofl", EMAIL + "3");
         givenParty(PARTY + "-test-ricardo", EMAIL + "4");
         givenParty(PARTY + "-test-milos", EMAIL + "5");
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(PARTY + "-test-lol");
 
         assertEquals(1, searchHits.size());
@@ -67,7 +70,7 @@ public class ReadTest extends AbstractIntegrationTest {
     @Test
     void searchByPartyEmail() throws TException {
         givenParty(PARTY, EMAIL);
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(EMAIL);
 
         assertFalse(searchHits.isEmpty());
@@ -79,7 +82,7 @@ public class ReadTest extends AbstractIntegrationTest {
     void searchByShopUrl() throws TException {
         Party party = givenParty(PARTY, EMAIL);
         givenShop(party, SHOP, URL);
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(URL);
 
         assertFalse(searchHits.isEmpty());
@@ -92,7 +95,7 @@ public class ReadTest extends AbstractIntegrationTest {
     void searchByShopId() throws TException {
         Party party = givenParty(PARTY, EMAIL);
         givenShop(party, SHOP, URL);
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(SHOP);
 
         assertFalse(searchHits.isEmpty());
@@ -105,7 +108,7 @@ public class ReadTest extends AbstractIntegrationTest {
     void searchShopByShopId() throws TException {
         Party party = givenParty(PARTY, EMAIL);
         givenShop(party, SHOP, URL);
-
+        refreshIndices();
         List<SearchShopHit> searchShopHits = deanonimusServiceHandler.searchShopText(SHOP);
 
         assertFalse(searchShopHits.isEmpty());
@@ -117,7 +120,7 @@ public class ReadTest extends AbstractIntegrationTest {
     void searchShopByShopUrl() throws TException {
         Party party = givenParty(PARTY, EMAIL);
         givenShop(party, SHOP, URL);
-
+        refreshIndices();
         List<SearchShopHit> searchHits = deanonimusServiceHandler.searchShopText(URL);
 
         assertFalse(searchHits.isEmpty());
@@ -125,12 +128,36 @@ public class ReadTest extends AbstractIntegrationTest {
                 .anyMatch(partySearchHit -> partySearchHit.getShop().getLocation().getUrl().contains(URL)));
     }
 
+    @Test
+    void searchWalletByName() throws TException {
+        Party party = givenParty(PARTY, EMAIL);
+        givenWallet(party, WALLET, WALLET_NAME);
+        refreshIndices();
+        List<SearchWalletHit> searchHits = deanonimusServiceHandler.searchWalletText(WALLET_NAME);
+
+        assertFalse(searchHits.isEmpty());
+        assertTrue(searchHits.stream()
+                .anyMatch(partySearchHit -> partySearchHit.getWallet().getName().contains(WALLET_NAME)));
+    }
+
+    @Test
+    void searchWalletById() throws TException {
+        Party party = givenParty(PARTY, EMAIL);
+        givenWallet(party, WALLET, WALLET_NAME);
+        refreshIndices();
+        List<SearchWalletHit> searchHits = deanonimusServiceHandler.searchWalletText(WALLET);
+
+        assertFalse(searchHits.isEmpty());
+        assertTrue(searchHits.stream()
+                .anyMatch(partySearchHit -> partySearchHit.getWallet().getId().contains(WALLET)));
+    }
+
     @Test
     void searchByContractorEmail() throws TException {
         Party party = givenParty(PARTY, null);
         givenRegisteredUserContractor(party, CONTRACTOR, EMAIL);
 
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(EMAIL);
 
         assertFalse(searchHits.isEmpty());
@@ -144,7 +171,7 @@ public class ReadTest extends AbstractIntegrationTest {
         Party party = givenParty(PARTY, null);
         givenRussianContractor(party, CONTRACTOR, "ООО \"ЧИ ИЛИ НЕ ЧИ\"", INN, ACCOUNT);
 
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("ЧИ");
 
         assertFalse(searchHits.isEmpty());
@@ -159,8 +186,8 @@ public class ReadTest extends AbstractIntegrationTest {
         Party party = givenParty(PARTY, null);
         givenRussianContractor(party, CONTRACTOR, "ООО \"ЧИ ИЛИ НЕ ЧИ\"", INN, ACCOUNT);
 
-
+        refreshIndices();
-        List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("ЧИ ДА");
+        List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("ЧИ ИЛИ");
 
         assertFalse(searchHits.isEmpty());
         assertTrue(searchHits.stream()
@@ -173,7 +200,7 @@ public class ReadTest extends AbstractIntegrationTest {
         Party party = givenParty(PARTY, null);
         givenRussianContractor(party, CONTRACTOR, "ООО \"ЧИ ИЛИ НЕ ЧИ\"", INN, ACCOUNT);
 
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("ДА");
 
         assertTrue(searchHits.isEmpty());
@@ -184,7 +211,7 @@ public class ReadTest extends AbstractIntegrationTest {
         Party party = givenParty(PARTY, null);
         givenRussianContractor(party, CONTRACTOR, "ООО \"ЧИ ИЛИ НЕ ЧИ\"", INN, ACCOUNT);
 
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(INN);
 
         assertFalse(searchHits.isEmpty());
@@ -199,7 +226,7 @@ public class ReadTest extends AbstractIntegrationTest {
         Party party = givenParty(PARTY, null);
         givenRussianContractor(party, CONTRACTOR, "ООО \"ЧИ ИЛИ НЕ ЧИ\"", INN, ACCOUNT);
 
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty(INN.substring(0, 6));
 
         assertTrue(searchHits.isEmpty());
@@ -210,7 +237,7 @@ public class ReadTest extends AbstractIntegrationTest {
         Party party = givenParty(PARTY, null);
         givenInternationalContractor(party, CONTRACTOR, "SoMe LeGaL NaMe", "ANOTHER TRADING NAME");
 
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("legal");
 
         assertFalse(searchHits.isEmpty());
@@ -224,7 +251,7 @@ public class ReadTest extends AbstractIntegrationTest {
         Party party = givenParty(PARTY, null);
         givenContract(party, CONTRACT, 123, "ДГ-123432", "Василий Пупкин");
 
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("ДГ");
 
         assertFalse(searchHits.isEmpty());
@@ -239,7 +266,7 @@ public class ReadTest extends AbstractIntegrationTest {
             Party party = givenParty(i + "", i + EMAIL.substring(EMAIL.indexOf("@")));
             givenShop(party, 9 - i + "", URL + i);
         }
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("1");
 
         assertEquals(2, searchHits.size());
@@ -253,7 +280,7 @@ public class ReadTest extends AbstractIntegrationTest {
             Party party = givenParty(i + "", EMAIL);
             givenShop(party, 29 - i + "", URL + i);
         }
-
+        refreshIndices();
         List<SearchHit> searchHits = deanonimusServiceHandler.searchParty("email");
 
         assertEquals((long) responseLimit, searchHits.size());
@@ -269,7 +296,7 @@ public class ReadTest extends AbstractIntegrationTest {
                 null,
                 null,
                 null));
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
     }
 
     private void givenRussianContractor(Party party,
@@ -284,16 +311,21 @@ public class ReadTest extends AbstractIntegrationTest {
                 russianLegalEntityRussianBankAccount,
                 null,
                 null));
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
     }
 
     private void givenShop(Party party, String id, String url) {
         party.addShop(TestData.shop(id, url));
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
+    }
+
+    private void givenWallet(Party party, String id, String name) {
+        party.addWallet(TestData.wallet(id, name));
+        openSearchService.updateParty(party);
     }
 
     private Party givenParty(String id, String email) {
-        return partyRepository.save(TestData.party(id, email));
+        return openSearchService.createParty(TestData.party(id, email));
     }
 
     private void givenInternationalContractor(Party party,
@@ -307,7 +339,7 @@ public class ReadTest extends AbstractIntegrationTest {
                 null,
                 internationalLegalEntityLegalName,
                 internationalLegalEntityTradingName));
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
     }
 
     private void givenContract(Party party,
@@ -319,7 +351,11 @@ public class ReadTest extends AbstractIntegrationTest {
                 termsId,
                 legalAgreementId,
                 reportActSignerFullName));
-        partyRepository.save(party);
+        openSearchService.updateParty(party);
     }
 
+    @SneakyThrows
+    private void refreshIndices() {
+        client.indices().refresh();
+    }
 }

src/test/java/dev/vality/deanonimus/TestData.java

@@ -15,7 +15,7 @@ public abstract class TestData {
 
     private static final Map.Entry<FieldHandler, String[]> timeFields = Map.entry(
             structHandler -> structHandler.value(Instant.now().toString()),
-            new String[] {"created_at", "at", "due"}
+            new String[]{"created_at", "at", "due"}
     );
 
     private static final MockTBaseProcessor mockTBaseProcessor = new MockTBaseProcessor(MockMode.ALL, 15, 1);
@@ -38,10 +38,17 @@ public abstract class TestData {
                 .build();
     }
 
+    public static dev.vality.deanonimus.domain.Wallet wallet(String id, String name) {
+        return dev.vality.deanonimus.domain.Wallet.builder()
+                .id(id)
+                .name(name)
+                .build();
+    }
+
     public static dev.vality.deanonimus.domain.Contract contract(String id,
-                                                                   Integer termsId,
+                                                                 Integer termsId,
-                                                                   String legalAgreementId,
+                                                                 String legalAgreementId,
-                                                                   String reportActSignerFullName) {
+                                                                 String reportActSignerFullName) {
         return dev.vality.deanonimus.domain.Contract.builder()
                 .id(id)
                 .status(ContractStatus.active)
@@ -52,12 +59,12 @@ public abstract class TestData {
     }
 
     public static dev.vality.deanonimus.domain.Contractor contractor(String id,
-                                                                       String registeredUserEmail,
+                                                                     String registeredUserEmail,
-                                                                       String russianLegalEntityRegisteredName,
+                                                                     String russianLegalEntityRegisteredName,
-                                                                       String russianLegalEntityRegisteredInn,
+                                                                     String russianLegalEntityRegisteredInn,
-                                                                       String russianLegalEntityRussianBankAccount,
+                                                                     String russianLegalEntityRussianBankAccount,
-                                                                       String internationalLegalEntityLegalName,
+                                                                     String internationalLegalEntityLegalName,
-                                                                       String internationalLegalEntityTradingName) {
+                                                                     String internationalLegalEntityTradingName) {
         return dev.vality.deanonimus.domain.Contractor.builder()
                 .id(id)
                 .type(getContractorType(registeredUserEmail, russianLegalEntityRegisteredInn,

src/test/java/dev/vality/deanonimus/WriteTest.java

@@ -1,8 +1,9 @@
 package dev.vality.deanonimus;
 
 
-import dev.vality.deanonimus.db.PartyRepository;
 import dev.vality.deanonimus.domain.Blocking;
+import dev.vality.deanonimus.service.OpenSearchService;
+import lombok.SneakyThrows;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 
@@ -15,15 +16,27 @@ import static org.awaitility.Awaitility.await;
 public class WriteTest extends AbstractIntegrationTest {
 
     @Autowired
-    private PartyRepository partyRepository;
+    OpenSearchService openSearchService;
 
     @Test
     void onPartyCreatedElasticHaveIt() throws IOException {
 
         sendMessages(generatePartyContractorFlow(TestData.SOURCE_ID_ONE));
+        sleep();
+        await().until(() -> openSearchService.findPartyById(TestData.SOURCE_ID_ONE),
+                party -> party != null && party.getId().equals(TestData.SOURCE_ID_ONE)
+        );
 
-        await().until(() -> partyRepository.findById(TestData.SOURCE_ID_ONE),
+    }
-                partyOptional -> partyOptional.isPresent() && partyOptional.get().getId().equals(TestData.SOURCE_ID_ONE)
+
+    @Test
+    void onPartyCreatedWalletFlowElasticHaveIt() throws IOException {
+
+        sendMessages(generateWalletFlow(TestData.SOURCE_ID_ONE));
+        sleep();
+        await().until(() -> openSearchService.findPartyById(TestData.SOURCE_ID_ONE),
+                party -> party != null && party.getId().equals(TestData.SOURCE_ID_ONE)
+                        && !party.getWallets().isEmpty()
         );
 
     }
@@ -36,13 +49,17 @@ public class WriteTest extends AbstractIntegrationTest {
                         buildSinkEvent(buildMessagePartyBlocking(0L, TestData.SOURCE_ID_ONE))
                 )
         );
-
+        sleep();
-        await().until(() -> partyRepository.findById(TestData.SOURCE_ID_ONE),
+        await().until(() -> openSearchService.findPartyById(TestData.SOURCE_ID_ONE),
-                partyOptional -> partyOptional.isPresent()
+                partyOptional -> partyOptional != null
-                        && partyOptional.get().getId().equals(TestData.SOURCE_ID_ONE)
+                        && partyOptional.getId().equals(TestData.SOURCE_ID_ONE)
-                        && partyOptional.get().getBlocking().equals(Blocking.blocked)
+                        && partyOptional.getBlocking().equals(Blocking.blocked)
         );
 
     }
 
+    @SneakyThrows
+    private void sleep() {
+        Thread.sleep(5000L);
+    }
 }

src/test/java/dev/vality/deanonimus/extension/OpensearchContainerExtension.java

@@ -9,8 +9,9 @@ import org.testcontainers.utility.DockerImageName;
 
 public class OpensearchContainerExtension implements BeforeAllCallback, AfterAllCallback {
 
+
     public static GenericContainer<?> OPENSEARCH = new GenericContainer<>(
-            DockerImageName.parse("opensearchproject/opensearch").withTag("1.2.4")
+            DockerImageName.parse("opensearchproject/opensearch:2.0.0")
     );
 
     @Override