minor changes

docker-compose.yml

@@ -1,5 +1,3 @@
-version: '3.8'
-
 services:
   postgres:
     image: postgres:15

pom.xml

@@ -80,11 +80,22 @@
             <groupId>org.liquibase</groupId>
             <artifactId>liquibase-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
+            <version>4.12.0</version>
+        </dependency>
         <dependency>
             <groupId>com.h2database</groupId>
             <artifactId>h2</artifactId>
             <scope>runtime</scope>
         </dependency>
+        <dependency>
+            <groupId>io.github.resilience4j</groupId>
+            <artifactId>resilience4j-spring-boot2</artifactId>
+            <version>1.7.1</version>
+        </dependency>
+
     </dependencies>
 
     <build>

src/main/java/ivan/solscanbot/dto/external/portfolio/SingleTokenPortfolioResponseDto.java

@@ -6,10 +6,12 @@
 
 @Data
 public class SingleTokenPortfolioResponseDto {
+    @JsonProperty("token_name")
+    private String name;
     @JsonProperty("token_symbol")
-    private String tokenSymbol;
+    private String symbol;
     @JsonProperty("token_address")
-    private String tokenAddress;
+    private String address;
     @JsonProperty("balance")
     private String tokenBalance;
     @JsonProperty("value")

src/main/java/ivan/solscanbot/dto/internal/BalanceActivity.java

@@ -29,23 +29,22 @@ public class BalanceActivity {
     @Id
     @GeneratedValue(strategy = GenerationType.IDENTITY)
     private Long id;
-    @Column(nullable = false, name = "token_address")
-    private String tokenAddress;
-    @Column(name = "token_name")
-    private String tokenName;
-    @Column(nullable = false, name = "token_symbol")
-    private String tokenSymbol;
     @Column(nullable = false, name = "value_in_usd")
     private BigDecimal valueInUsd;
     @Column(nullable = false)
     private BigDecimal amount;
-    @Column(nullable = false)
-    private Date time;
+    @ManyToOne
+    @ToString.Exclude
+    @EqualsAndHashCode.Exclude
+    @JoinColumn(nullable = false, name = "token_id")
+    private Token token;
     @ManyToOne
     @ToString.Exclude
     @EqualsAndHashCode.Exclude
     @JoinColumn(nullable = false, name = "address_id")
     private MonitoredAddress monitoredAddress;
+    @Column(nullable = false)
+    private Date time;
     @Column(nullable = false, name = "is_deleted")
     private boolean isDeleted = false;
 }

src/main/java/ivan/solscanbot/dto/internal/Token.java

@@ -21,10 +21,10 @@ public class Token {
     @Id
     @GeneratedValue(strategy = GenerationType.IDENTITY)
     private Long id;
-    @Column(name = "token_symbol")
-    private String tokenSymbol;
-    @Column(unique = true, nullable = false, name = "token_address")
-    private String tokenAddress;
+    private String name;
+    private String symbol;
+    @Column(unique = true)
+    private String address;
     @Column(name = "is_deleted")
     private boolean isDeleted = false;
 }

src/main/java/ivan/solscanbot/mapper/ActivityMapper.java

@@ -3,9 +3,17 @@
 import ivan.solscanbot.config.MapperConfig;
 import ivan.solscanbot.dto.external.activity.SingleBalanceActivityResponseDto;
 import ivan.solscanbot.dto.internal.BalanceActivity;
+import ivan.solscanbot.dto.internal.Token;
 import org.mapstruct.Mapper;
 
 @Mapper(config = MapperConfig.class)
 public interface ActivityMapper {
     BalanceActivity toModel(SingleBalanceActivityResponseDto activityDto);
+
+    default BalanceActivity toModel(SingleBalanceActivityResponseDto activityDto,
+                                    Token token) {
+        BalanceActivity act = toModel(activityDto);
+        act.setToken(token);
+        return act;
+    }
 }

src/main/java/ivan/solscanbot/mapper/TokenMapper.java

@@ -1,11 +1,14 @@
 package ivan.solscanbot.mapper;
 
 import ivan.solscanbot.config.MapperConfig;
+import ivan.solscanbot.dto.external.meta.TokenMetaResponseDto;
 import ivan.solscanbot.dto.external.portfolio.SingleTokenPortfolioResponseDto;
 import ivan.solscanbot.dto.internal.Token;
 import org.mapstruct.Mapper;
 
 @Mapper(config = MapperConfig.class)
 public interface TokenMapper {
-    Token toModel(SingleTokenPortfolioResponseDto dto);
+    Token toModelFromPortfolioDto(SingleTokenPortfolioResponseDto dto);
+
+    Token toModelFromMetaDto(TokenMetaResponseDto dto);
 }

src/main/java/ivan/solscanbot/repository/TokenRepository.java

@@ -5,5 +5,5 @@
 import org.springframework.data.jpa.repository.JpaRepository;
 
 public interface TokenRepository extends JpaRepository<Token, Long> {
-    Optional<Token> findByTokenAddress(String tokenAddress);
+    Optional<Token> findByAddress(String address);
 }

src/main/java/ivan/solscanbot/service/MonitoringService.java

@@ -6,20 +6,24 @@
 import ivan.solscanbot.dto.external.meta.TokenMetaResponseDto;
 import ivan.solscanbot.dto.internal.BalanceActivity;
 import ivan.solscanbot.dto.internal.MonitoredAddress;
+import ivan.solscanbot.dto.internal.Token;
 import ivan.solscanbot.mapper.ActivityMapper;
+import ivan.solscanbot.mapper.TokenMapper;
 import ivan.solscanbot.repository.ActivityRepository;
 import ivan.solscanbot.repository.MonitoredAddressRepository;
+import ivan.solscanbot.repository.TokenRepository;
 import java.math.BigDecimal;
 import java.math.MathContext;
 import java.math.RoundingMode;
-import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -32,24 +36,25 @@ public class MonitoringService {
     private static final int CHUNK_SIZE = 20;
     private static final String SOLSCAN_ACCOUNT_URL = "https://solscan.io/account/";
     private static final String SOLSCAN_TOKEN_URL = "https://solscan.io/token/";
-    private static final String FIRST_TRACKING_ADDRESS =
-            "APZmQqyytWLMFioMsskqhWrGJCd9Fpo7L2f2YhdpSe6U";
 
     private final MonitoredAddressRepository addressRepository;
     private final TelegramBot telegramBot;
     private final SolScanServiceImpl solScanService;
     private final ActivityRepository activityRepository;
+    private final TokenRepository tokenRepository;
     private final ActivityMapper activityMapper;
+    private final TokenMapper tokenMapper;
+    private RateLimiter rateLimiter = RateLimiter.create(1.0);
 
-    @Scheduled(fixedDelay = 300000)
+    @Scheduled(fixedDelay = 60000)
     public void newActivityFound() {
-        RateLimiter rateLimiter = RateLimiter.create(5.0);
         addressRepository.findAll().forEach(address -> {
             try {
                 rateLimiter.acquire();
                 Set<BalanceActivity> activities = fetchAndProcessActivities(address);
                 if (!activities.isEmpty()) {
                     activityRepository.saveAll(activities);
+                    log.info("New activity found");
                     sendTelegramNotification(address, activities);
                 }
             } catch (Exception e) {
@@ -61,102 +66,127 @@ public void newActivityFound() {
     private Set<BalanceActivity> fetchAndProcessActivities(MonitoredAddress address) {
         Set<SingleBalanceActivityResponseDto> newActivities =
                 solScanService.getNewBalanceActivities(address.getAddress())
-                        .stream()
-                        .filter(act -> BigDecimal.ZERO.equals(act.getPreBalance()))
-                        .collect(Collectors.toSet());
-
+                .stream()
+                .filter(act -> BigDecimal.ZERO.equals(act.getPreBalance()))
+                .collect(Collectors.toSet());
+        log.info("Fetching new activities for address {}", address.getAddress());
+        if (newActivities.isEmpty()) {
+            return Collections.emptySet();
+        }
         List<String> tokenAddresses = newActivities.stream()
                 .map(SingleBalanceActivityResponseDto::getTokenAddress)
+                .distinct()
                 .toList();
 
         Map<String, TokenMetaResponseDto> metaMap = batchFetchTokenMetadata(tokenAddresses);
 
         return newActivities.stream()
-                .map(activityMapper::toModel)
-                .map(act -> enrichWithTokenMeta(act, metaMap.get(act.getTokenAddress())))
-                .filter(tok -> tok.getValueInUsd().compareTo(BigDecimal.valueOf(100)) > 0)
+                .map(dto -> {
+                    TokenMetaResponseDto meta = metaMap.get(dto.getTokenAddress());
+                    if (meta == null) {
+                        return Optional.<BalanceActivity>empty();
+                    }
+                    return enrichWithTokenMeta(dto, meta);
+                })
+                .filter(Optional::isPresent)
+                .map(Optional::get)
                 .peek(act -> act.setMonitoredAddress(address))
                 .collect(Collectors.toSet());
     }
 
     private Map<String, TokenMetaResponseDto> batchFetchTokenMetadata(List<String> tokenAddresses) {
         Map<String, TokenMetaResponseDto> metaMap = new ConcurrentHashMap<>();
-        List<String> addressList = new ArrayList<>(tokenAddresses);
-        for (int i = 0; i < addressList.size(); i += CHUNK_SIZE) {
-            List<String> chunk = addressList.subList(
-                    i, Math.min(i + CHUNK_SIZE, addressList.size()));
+        log.info("Fetching token metadata for addresses {}", tokenAddresses);
+        if (tokenAddresses.isEmpty()) {
+            return metaMap;
+        }
+
+        List<List<String>> chunks = partitionList(tokenAddresses);
+        chunks.parallelStream().forEach(chunk -> {
             try {
                 Map<String, TokenMetaResponseDto> chunkResults =
                         solScanService.getMetaMapFromAddresses(chunk);
                 metaMap.putAll(chunkResults);
             } catch (Exception e) {
-                log.error("Failed to fetch batch metadata for chunk: {}", chunk, e);
+                log.warn("Batch metadata fetch failed, falling back to individual requests", e);
                 fetchTokenMetadata(chunk, metaMap);
             }
-        }
+        });
+
         return metaMap;
     }
 
+    private <T> List<List<T>> partitionList(List<T> list) {
+        return IntStream.range(0, (list.size() + CHUNK_SIZE - 1) / CHUNK_SIZE)
+                .mapToObj(i -> list.subList(
+                        i * CHUNK_SIZE, Math.min(list.size(), (i + 1) * CHUNK_SIZE)))
+                .collect(Collectors.toList());
+    }
+
     private void fetchTokenMetadata(List<String> tokenAddresses,
                                     Map<String, TokenMetaResponseDto> metaMap) {
         tokenAddresses.forEach(address -> {
-            TokenMetaResponseDto meta = solScanService.getTokenMeta(address);
-            metaMap.put(address, meta);
+            try {
+                TokenMetaResponseDto meta = solScanService.getTokenMeta(address);
+                if (meta != null) {
+                    metaMap.put(address, meta);
+                }
+            } catch (Exception e) {
+                log.error("Failed to fetch metadata for token: {}", address, e);
+            }
         });
     }
 
-    private BalanceActivity enrichWithTokenMeta(BalanceActivity act, TokenMetaResponseDto meta) {
+    private Optional<BalanceActivity> enrichWithTokenMeta(
+            SingleBalanceActivityResponseDto dto,
+            TokenMetaResponseDto meta
+    ) {
         BigDecimal price = Optional.ofNullable(meta.getPrice()).orElse(BigDecimal.ZERO);
         int decimals = meta.getDecimals() > 0 ? meta.getDecimals() : 9;
-        BigDecimal normalizedAmount = act.getAmount()
+
+        BigDecimal normalizedAmount = dto.getAmount()
                 .divide(BigDecimal.TEN.pow(decimals), MathContext.DECIMAL32);
-        act.setValueInUsd(normalizedAmount.multiply(price))
-                .setTokenName(meta.getName())
-                .setTokenSymbol(meta.getSymbol());
-        return act;
+        BigDecimal valueInUsd = normalizedAmount.multiply(price);
+        if (valueInUsd.compareTo(BigDecimal.valueOf(1)) > 0) {
+            Token token = tokenRepository.findByAddress(meta.getAddress())
+                    .orElseGet(() -> tokenRepository.save(tokenMapper.toModelFromMetaDto(meta)));
+
+            BalanceActivity act = activityMapper.toModel(dto, token);
+            act.setValueInUsd(valueInUsd);
+            log.info("Saving the Token: {} and creating Balance Activity entity: {}.",
+                    token.getName(), act.getValueInUsd());
+            return Optional.of(act);
+        }
+        return Optional.empty();
     }
 
     private void sendTelegramNotification(MonitoredAddress address,
                                           Set<BalanceActivity> activities) {
-        AtomicInteger count = new AtomicInteger(1);
-        String tokens = activities.stream()
-                .map(act -> String.format(
-                        "%d. Token: %s\nUSD Value: $%s\nToken link: [%s](%s#balanceChanges)\n",
-                        count.getAndIncrement(),
-                        act.getTokenName(),
-                        act.getValueInUsd().setScale(2, RoundingMode.HALF_UP),
-                        shortenAddress(act.getTokenAddress()),
-                        SOLSCAN_TOKEN_URL + act.getTokenAddress()
-                ))
-                .collect(Collectors.joining());
+        String tokens = formatTokensMessage(activities);
         String message = String.format("New activity for address: [%s](%s#balanceChanges)\n%s",
                 shortenAddress(address.getAddress()),
                 SOLSCAN_ACCOUNT_URL + address.getAddress(),
                 tokens);
         telegramBot.sendNotification(address.getChatId(), message);
     }
 
+    private String formatTokensMessage(Set<BalanceActivity> balanceActivities) {
+        AtomicInteger counter = new AtomicInteger(1);
+        return balanceActivities.stream()
+                .map(act -> String.format(
+                        "%d. Token: %s\nUSD Value: $%s\nToken link: [%s](%s#balanceChanges)\n",
+                        counter.getAndIncrement(),
+                        act.getToken().getName(),
+                        act.getValueInUsd().setScale(2, RoundingMode.HALF_UP),
+                        shortenAddress(act.getToken().getAddress()),
+                        SOLSCAN_TOKEN_URL + act.getToken().getAddress()
+                ))
+                .collect(Collectors.joining());
+    }
+
     private String shortenAddress(String address) {
         return address.length() > 8
                 ? address.substring(0, 4) + "..." + address.substring(address.length() - 4)
                 : address;
     }
-
-    /*public void monitorAddress() {
-        try {
-            if (solScanService.newTokenTransfer(FIRST_TRACKING_ADDRESS)) {
-                Set<Long> ids =
-                        addressRepository.findAll()
-                                .stream()
-                                .map(MonitoredAddress::getChatId)
-                                .collect(Collectors.toSet());
-                log.info("New transfer detected, sending notification");
-                for (Long id : ids) {
-                    telegramBot.sendNotification(id, "!!!");
-                }
-            }
-        } catch (Exception e) {
-            log.error("Error in scheduled task", e);
-        }
-    }*/
 }