init
This commit is contained in:
+693
@@ -0,0 +1,693 @@
|
||||
package src
|
||||
|
||||
import (
|
||||
"AngkorWalletScanning/model"
|
||||
"AngkorWalletScanning/umlog"
|
||||
"AngkorWalletScanning/umsql"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/big"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
// Alchemy API Rate Limiter (무료 티어: 500 CU/s, eth_getLogs=75 CU → 약 6.67 req/s)
|
||||
var alchemyApiLimiter = rate.NewLimiter(rate.Limit(6.5), 7)
|
||||
|
||||
// HTTP 클라이언트 초기화 함수
|
||||
func initAlchemyHTTPClient() *http.Client {
|
||||
transport := &http.Transport{
|
||||
Dial: (&net.Dialer{
|
||||
Timeout: 30 * time.Second, // 10초 → 30초로 증가
|
||||
KeepAlive: 30 * time.Second,
|
||||
Resolver: &net.Resolver{
|
||||
PreferGo: false, // 시스템 DNS resolver 사용
|
||||
},
|
||||
}).Dial,
|
||||
TLSHandshakeTimeout: 15 * time.Second, // 10초 → 15초로 증가
|
||||
ResponseHeaderTimeout: 15 * time.Second, // 10초 → 15초로 증가
|
||||
ExpectContinueTimeout: 1 * time.Second,
|
||||
MaxIdleConns: 100,
|
||||
MaxIdleConnsPerHost: 10,
|
||||
IdleConnTimeout: 90 * time.Second,
|
||||
TLSClientConfig: &tls.Config{
|
||||
InsecureSkipVerify: false,
|
||||
},
|
||||
DisableKeepAlives: false,
|
||||
// DNS 캐시 시간 연장
|
||||
MaxConnsPerHost: 10,
|
||||
}
|
||||
|
||||
// 환경 변수 또는 시스템 프록시 사용
|
||||
if proxyURL := os.Getenv("HTTPS_PROXY"); proxyURL != "" {
|
||||
umlog.Debug("[Alchemy] Using proxy from HTTPS_PROXY: %s\n", proxyURL)
|
||||
if proxy, err := url.Parse(proxyURL); err == nil {
|
||||
transport.Proxy = http.ProxyURL(proxy)
|
||||
}
|
||||
} else if proxyURL := os.Getenv("HTTP_PROXY"); proxyURL != "" {
|
||||
umlog.Debug("[Alchemy] Using proxy from HTTP_PROXY: %s\n", proxyURL)
|
||||
if proxy, err := url.Parse(proxyURL); err == nil {
|
||||
transport.Proxy = http.ProxyURL(proxy)
|
||||
}
|
||||
} else {
|
||||
// 시스템 기본 프록시 사용
|
||||
transport.Proxy = http.ProxyFromEnvironment
|
||||
}
|
||||
|
||||
return &http.Client{
|
||||
Timeout: 60 * time.Second, // 30초 → 60초로 증가 (전체 요청 타임아웃)
|
||||
Transport: transport,
|
||||
}
|
||||
}
|
||||
|
||||
// HTTP 클라이언트 (타임아웃 및 연결 설정)
|
||||
var alchemyHTTPClient = initAlchemyHTTPClient()
|
||||
|
||||
// ERC20 Transfer Event Topic (Transfer 이벤트 시그니처 해시)
|
||||
const transferEventTopic = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
|
||||
|
||||
// Alchemy JSON-RPC Request/Response Structures
|
||||
type AlchemyRequest struct {
|
||||
JsonRPC string `json:"jsonrpc"`
|
||||
Method string `json:"method"`
|
||||
Params []interface{} `json:"params"`
|
||||
ID int `json:"id"`
|
||||
}
|
||||
|
||||
type AlchemyResponse struct {
|
||||
JsonRPC string `json:"jsonrpc"`
|
||||
ID int `json:"id"`
|
||||
Result map[string]interface{} `json:"result"`
|
||||
Error *AlchemyError `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
type AlchemyError struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
type AlchemyLog struct {
|
||||
Address string `json:"address"`
|
||||
Topics []string `json:"topics"`
|
||||
Data string `json:"data"`
|
||||
BlockNumber string `json:"blockNumber"`
|
||||
TransactionHash string `json:"transactionHash"`
|
||||
TransactionIndex string `json:"transactionIndex"`
|
||||
BlockHash string `json:"blockHash"`
|
||||
LogIndex string `json:"logIndex"`
|
||||
Removed bool `json:"removed"`
|
||||
}
|
||||
|
||||
type AlchemyTransferInfo struct {
|
||||
BlockNumber int64
|
||||
TransactionHash string
|
||||
From string
|
||||
To string
|
||||
Value string
|
||||
TokenAddress string
|
||||
Timestamp int64
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : Alchemy API를 통해 특정 주소의 ERC20 토큰 Transfer 이벤트 조회
|
||||
parameter
|
||||
- address : 조회할 지갑 주소
|
||||
- contractAddress : 토큰 컨트랙트 주소 (CYBX)
|
||||
- fromBlock : 시작 블록 (hex 형식)
|
||||
- toBlock : 종료 블록 (hex 형식, "latest" 가능)
|
||||
- netType : 네트워크 타입 ("MAINNET" 또는 "TESTNET")
|
||||
- CODE_STRC : 설정 정보
|
||||
return
|
||||
- []AlchemyTransferInfo : Transfer 이벤트 목록
|
||||
- error : 에러 정보
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func getAlchemyTokenTransfers(address string, contractAddress string, fromBlock string, toBlock string, CODE_STRC model.CodeStrc) ([]AlchemyTransferInfo, error) {
|
||||
// 입력 값 검증
|
||||
if address == "" || len(address) < 40 {
|
||||
return nil, fmt.Errorf("invalid wallet address: %s", address)
|
||||
}
|
||||
if contractAddress == "" || len(contractAddress) < 40 {
|
||||
return nil, fmt.Errorf("invalid contract address: %s", contractAddress)
|
||||
}
|
||||
|
||||
// Rate limiter 체크
|
||||
ctx := context.Background()
|
||||
err := alchemyApiLimiter.Wait(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("rate limiter error: %v", err)
|
||||
}
|
||||
|
||||
// eth_getLogs 요청 파라미터 구성
|
||||
// Transfer(address indexed from, address indexed to, uint256 value)
|
||||
// topic[0] = Transfer 이벤트 시그니처
|
||||
// topic[1] = from 주소 (32바이트 패딩)
|
||||
// topic[2] = to 주소 (32바이트 패딩)
|
||||
|
||||
// 입금 트랜잭션 필터: to = 내 주소
|
||||
paddedAddress := "0x" + strings.Repeat("0", 24) + strings.TrimPrefix(strings.ToLower(address), "0x")
|
||||
|
||||
filterParams := map[string]interface{}{
|
||||
"fromBlock": fromBlock,
|
||||
"toBlock": toBlock,
|
||||
"address": contractAddress,
|
||||
"topics": []interface{}{
|
||||
transferEventTopic, // Transfer 이벤트
|
||||
nil, // from (any)
|
||||
paddedAddress, // to (내 주소)
|
||||
},
|
||||
}
|
||||
|
||||
request := AlchemyRequest{
|
||||
JsonRPC: "2.0",
|
||||
Method: "eth_getLogs",
|
||||
Params: []interface{}{filterParams},
|
||||
ID: 1,
|
||||
}
|
||||
|
||||
// JSON 요청 생성
|
||||
requestBody, err := json.Marshal(request)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal request: %v", err)
|
||||
}
|
||||
|
||||
// HTTP POST 요청
|
||||
umlog.Debug("Request - Address: %s, From: %s, To: %s", address, fromBlock, toBlock)
|
||||
// umlog.Debug("[getAlchemyTokenTransfers] Padded address (topic[2]): %s\n", paddedAddress)
|
||||
// umlog.Debug("[getAlchemyTokenTransfers] Request body: %s", string(requestBody))
|
||||
req, err := http.NewRequest("POST", CODE_STRC.ALCHEMY_MAINNET, bytes.NewBuffer(requestBody))
|
||||
if err != nil {
|
||||
umlog.Warn("Failed to create request: %v\n", err)
|
||||
return nil, fmt.Errorf("failed to create request: %v", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
// 재시도 로직 (DNS 에러 대응)
|
||||
var resp *http.Response
|
||||
maxRetries := 3
|
||||
for attempt := 1; attempt <= maxRetries; attempt++ {
|
||||
resp, err = alchemyHTTPClient.Do(req)
|
||||
if err == nil {
|
||||
break // 성공
|
||||
}
|
||||
|
||||
// DNS 에러인 경우 재시도
|
||||
if attempt < maxRetries {
|
||||
umlog.Debug("Attempt %d failed: %v, retrying in 2s...", attempt, err)
|
||||
time.Sleep(2 * time.Second)
|
||||
// 요청 바디 재설정 (한 번 읽히면 소진되므로)
|
||||
req.Body = io.NopCloser(bytes.NewBuffer(requestBody))
|
||||
continue
|
||||
}
|
||||
|
||||
// 최종 실패
|
||||
umlog.Warn("HTTP request failed after %d attempts: %v\n", maxRetries, err)
|
||||
return nil, fmt.Errorf("http request failed: %v", err)
|
||||
}
|
||||
|
||||
defer resp.Body.Close()
|
||||
// umlog.Debug("Response status: %d", resp.StatusCode)
|
||||
|
||||
// 응답 읽기
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read response: %v", err)
|
||||
}
|
||||
|
||||
// umlog.Debug("Response body: %s\n", string(body))
|
||||
|
||||
// JSON 파싱
|
||||
var alchemyResp struct {
|
||||
JsonRPC string `json:"jsonrpc"`
|
||||
ID int `json:"id"`
|
||||
Result []AlchemyLog `json:"result"`
|
||||
Error *AlchemyError `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
err = json.Unmarshal(body, &alchemyResp)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse json: %v", err)
|
||||
}
|
||||
|
||||
// API 에러 체크
|
||||
if alchemyResp.Error != nil {
|
||||
return nil, fmt.Errorf("alchemy api error: %s (code: %d)", alchemyResp.Error.Message, alchemyResp.Error.Code)
|
||||
}
|
||||
|
||||
// Transfer 이벤트 파싱
|
||||
var transfers []AlchemyTransferInfo
|
||||
for _, log := range alchemyResp.Result {
|
||||
if len(log.Topics) < 3 {
|
||||
continue // Invalid Transfer event
|
||||
}
|
||||
umlog.Debug("Log entry: %+v\n", log)
|
||||
|
||||
// from 주소 추출 (topic[1])
|
||||
fromAddr := "0x" + log.Topics[1][26:] // 24개의 0 제거
|
||||
|
||||
// to 주소 추출 (topic[2])
|
||||
toAddr := "0x" + log.Topics[2][26:] // 24개의 0 제거
|
||||
|
||||
// value 추출 (data)
|
||||
value := strings.TrimPrefix(log.Data, "0x")
|
||||
|
||||
// 블록 번호 파싱
|
||||
blockNum := new(big.Int)
|
||||
blockNum.SetString(strings.TrimPrefix(log.BlockNumber, "0x"), 16)
|
||||
|
||||
// 타임스탬프는 별도 조회 필요 (블록 정보에서)
|
||||
timestamp := time.Now().Unix() // 기본값, 필요시 eth_getBlockByNumber 호출
|
||||
|
||||
transfers = append(transfers, AlchemyTransferInfo{
|
||||
BlockNumber: blockNum.Int64(),
|
||||
TransactionHash: log.TransactionHash,
|
||||
From: fromAddr,
|
||||
To: toAddr,
|
||||
Value: value,
|
||||
TokenAddress: log.Address,
|
||||
Timestamp: timestamp,
|
||||
})
|
||||
}
|
||||
|
||||
return transfers, nil
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 현재 블록 번호 조회 (Alchemy API)
|
||||
parameter
|
||||
- netType : 네트워크 타입 ("MAINNET" 또는 "TESTNET")
|
||||
- CODE_STRC : 설정 정보
|
||||
return
|
||||
- int64 : 현재 블록 번호
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func getCurrentBlockNumberAlchemy(CODE_STRC model.CodeStrc) int64 {
|
||||
// Alchemy API URL
|
||||
alchemyURL := CODE_STRC.ALCHEMY_MAINNET
|
||||
|
||||
if alchemyURL == "" {
|
||||
umlog.Error("Alchemy API URL not configured")
|
||||
return 0
|
||||
}
|
||||
|
||||
request := AlchemyRequest{
|
||||
JsonRPC: "2.0",
|
||||
Method: "eth_blockNumber",
|
||||
Params: []interface{}{},
|
||||
ID: 1,
|
||||
}
|
||||
|
||||
requestBody, err := json.Marshal(request)
|
||||
if err != nil {
|
||||
umlog.Warn("Failed to marshal request: %v\n", err)
|
||||
return 0
|
||||
}
|
||||
|
||||
umlog.Info("Requesting: %s", alchemyURL)
|
||||
req, err := http.NewRequest("POST", alchemyURL, bytes.NewBuffer(requestBody))
|
||||
if err != nil {
|
||||
umlog.Warn("Failed to create request: %v\n", err)
|
||||
return 0
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
resp, err := alchemyHTTPClient.Do(req)
|
||||
if err != nil {
|
||||
umlog.Warn("HTTP request failed: %v\n", err)
|
||||
return 0
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
|
||||
var result struct {
|
||||
JsonRPC string `json:"jsonrpc"`
|
||||
ID int `json:"id"`
|
||||
Result string `json:"result"`
|
||||
}
|
||||
|
||||
json.Unmarshal(body, &result)
|
||||
|
||||
blockNum := new(big.Int)
|
||||
blockNum.SetString(strings.TrimPrefix(result.Result, "0x"), 16)
|
||||
|
||||
return blockNum.Int64()
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 해당 트랜잭션에 대해 이미 알림을 보냈는지 확인
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- txHash : 트랜잭션 해시
|
||||
return
|
||||
- bool : 이미 알림 보냈으면 true
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func isTransactionNotified(walletDBConn *sql.DB, txHash string) bool {
|
||||
query := "SELECT COUNT(*) FROM ank_wallet_notification WHERE tx_hash = ?"
|
||||
|
||||
var count int
|
||||
res, _, err := umsql.SqlSelect(walletDBConn, query, txHash)
|
||||
if err != nil {
|
||||
umlog.Error("SELECT Error(%s): %s", err.Error(), query)
|
||||
return false
|
||||
}
|
||||
defer res.Close()
|
||||
|
||||
if res.Next() {
|
||||
if err := res.Scan(&count); err != nil {
|
||||
umlog.Error("SCAN Error(%s): %s", err.Error(), query)
|
||||
return false
|
||||
}
|
||||
}
|
||||
return count > 0
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 특정 지갑의 입금 트랜잭션 체크 (Alchemy API 사용)
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- userId : 사용자 ID
|
||||
- walletAddress : 지갑 주소
|
||||
- contractAddress : CYBX 토큰 컨트랙트 주소
|
||||
- lastCheckTime : 마지막 체크 시각
|
||||
- netType : 네트워크 타입
|
||||
- CODE_STRC : 설정 정보
|
||||
return
|
||||
- int : 새로 발견된 입금 건수
|
||||
- error : 에러 정보
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func checkWalletWithAlchemy(walletDBConn *sql.DB, header model.Header, userId int, walletAddress string, contractAddress string, lastCheckTime time.Time, currentBlock int64, CODE_STRC model.CodeStrc) (int, error) {
|
||||
// 현재 블록은 호출자로부터 받음 (배치 시작 시 한 번만 조회)
|
||||
|
||||
// 입력 값 검증
|
||||
if walletAddress == "" {
|
||||
umlog.Warn("User %d has empty wallet address, skipping\n", userId)
|
||||
return 0, nil
|
||||
}
|
||||
if contractAddress == "" {
|
||||
umlog.Warn("Empty contract address, skipping\n")
|
||||
return 0, fmt.Errorf("contract address is empty")
|
||||
}
|
||||
|
||||
// Alchemy 무료 티어 제한: 한 번에 최대 10 블록만 조회 가능
|
||||
const maxBlockRange = 9 // 10블록이 아니라 9블록 차이 (0부터 9까지 = 10개)
|
||||
|
||||
// 마지막 체크 시각을 블록 번호로 변환 (BSC는 약 3초당 1블록)
|
||||
timeDiff := time.Since(lastCheckTime)
|
||||
blocksToCheck := int64(timeDiff.Seconds() / 3)
|
||||
|
||||
// umlog.Debug("Time diff: %v, Calculated blocks: %d\n", timeDiff, blocksToCheck)
|
||||
|
||||
// Alchemy 무료 티어 제한 적용 (최소 1블록은 체크)
|
||||
if blocksToCheck < 1 {
|
||||
blocksToCheck = 1
|
||||
}
|
||||
if blocksToCheck > maxBlockRange {
|
||||
blocksToCheck = maxBlockRange
|
||||
umlog.Debug("Limiting block range to %d due to Alchemy free tier\n", maxBlockRange)
|
||||
}
|
||||
|
||||
startBlock := currentBlock - blocksToCheck
|
||||
if startBlock < 0 {
|
||||
startBlock = 0
|
||||
}
|
||||
|
||||
// 블록 번호를 hex로 변환
|
||||
fromBlockHex := fmt.Sprintf("0x%x", startBlock)
|
||||
toBlockHex := fmt.Sprintf("0x%x", currentBlock)
|
||||
|
||||
umlog.Debug("Checking blocks %d to %d (range: %d)\n", startBlock, currentBlock, blocksToCheck)
|
||||
|
||||
// Alchemy API로 Transfer 이벤트 조회
|
||||
transfers, err := getAlchemyTokenTransfers(walletAddress, contractAddress, fromBlockHex, toBlockHex, CODE_STRC)
|
||||
if err != nil {
|
||||
umlog.Warn("Failed to get transfers for user %d: %v\n", userId, err)
|
||||
return 0, err
|
||||
}
|
||||
|
||||
newTransactionCount := 0
|
||||
|
||||
for _, transfer := range transfers {
|
||||
umlog.Debug("transfer info: %v\n", transfer)
|
||||
|
||||
// 트랜잭션 시각 체크 (Timestamp 사용)
|
||||
txTime := time.Unix(transfer.Timestamp, 0)
|
||||
|
||||
// 마지막 체크 이후의 트랜잭션만 처리
|
||||
if txTime.Before(lastCheckTime) {
|
||||
continue
|
||||
}
|
||||
|
||||
// 이미 알림 보낸 트랜잭션인지 확인
|
||||
if isTransactionNotified(walletDBConn, transfer.TransactionHash) {
|
||||
continue
|
||||
}
|
||||
|
||||
// 금액 계산 (18 decimals for most tokens, CYBX는 확인 필요)
|
||||
amount := calculateTokenAmountFromHex(transfer.Value, "18")
|
||||
|
||||
// 알림 전송
|
||||
sendIncomingTransactionNotification(walletDBConn, header, userId, transfer.From, transfer.To, amount, transfer.TransactionHash)
|
||||
|
||||
// 알림 기록 저장
|
||||
saveNotificationRecord(walletDBConn, userId, transfer.TransactionHash, transfer.From, transfer.To, amount, "CYBX")
|
||||
|
||||
newTransactionCount++
|
||||
|
||||
umlog.Info("checkWalletWithAlchemy\nNew incoming tx detected - User: %d, From: %s, Amount: %s CYBX, Hash: %s",
|
||||
userId, transfer.From, amount, transfer.TransactionHash)
|
||||
}
|
||||
|
||||
return newTransactionCount, nil
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : Hex 값으로 토큰 금액 계산 (decimal 적용)
|
||||
parameter
|
||||
- hexValue : 토큰 raw value (hex 문자열)
|
||||
- decimal : 토큰 decimal
|
||||
return
|
||||
- string : 사람이 읽을 수 있는 금액
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func calculateTokenAmountFromHex(hexValue string, decimal string) string {
|
||||
// hex를 big.Int로 변환
|
||||
valueInt := new(big.Int)
|
||||
if strings.HasPrefix(hexValue, "0x") {
|
||||
hexValue = strings.TrimPrefix(hexValue, "0x")
|
||||
}
|
||||
valueInt.SetString(hexValue, 16)
|
||||
|
||||
// decimal을 int로 변환
|
||||
decimalInt, _ := strconv.Atoi(decimal)
|
||||
|
||||
// 10^decimal 계산
|
||||
divisor := new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(decimalInt)), nil)
|
||||
|
||||
// 나눗셈
|
||||
result := new(big.Float).Quo(
|
||||
new(big.Float).SetInt(valueInt),
|
||||
new(big.Float).SetInt(divisor),
|
||||
)
|
||||
|
||||
// 소수점 4자리까지 표시
|
||||
return result.Text('f', 4)
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 모든 활성 사용자의 지갑 체크 (Alchemy API 사용)
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- contractAddress : CYBX 토큰 컨트랙트 주소
|
||||
- batchSize : 한 번에 처리할 사용자 수
|
||||
- netType : 네트워크 타입
|
||||
- CODE_STRC : 설정 정보
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func checkAllWalletsWithAlchemy(walletDBConn *sql.DB, header model.Header, contractAddress string, batchSize int, CODE_STRC model.CodeStrc) {
|
||||
startTime := time.Now()
|
||||
|
||||
// 배치 시작 시 현재 블록 번호 한 번만 조회
|
||||
currentBlock := getCurrentBlockNumberAlchemy(CODE_STRC)
|
||||
if currentBlock == 0 {
|
||||
umlog.Error("Failed to get current block number")
|
||||
return
|
||||
}
|
||||
umlog.Debug("Current block: %d - will be used for all wallets in this batch", currentBlock)
|
||||
|
||||
// 활성 사용자 목록 조회 (최근 활동 순)
|
||||
query := `
|
||||
SELECT id, address, checkedAt
|
||||
FROM ank_wallet_user
|
||||
WHERE address > ''
|
||||
ORDER BY checkedAt ASC
|
||||
LIMIT ?
|
||||
`
|
||||
|
||||
rows, err := walletDBConn.Query(query, batchSize)
|
||||
if err != nil {
|
||||
umlog.Error("checkAllWalletsWithAlchemy\nDB query failed: %v", err)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
totalChecked := 0
|
||||
totalNewTransactions := 0
|
||||
|
||||
// 요청 분산 간격 계산 (batchSize개를 3초에 균등 분산)
|
||||
// 예: 12개 → 3000ms ÷ 12 = 250ms 간격
|
||||
delayBetweenRequests := time.Duration(3000/batchSize) * time.Millisecond
|
||||
|
||||
for rows.Next() {
|
||||
var userId int
|
||||
var walletAddress string
|
||||
var lastCheckStr string
|
||||
|
||||
err := rows.Scan(&userId, &walletAddress, &lastCheckStr)
|
||||
if err != nil {
|
||||
umlog.Error("checkAllWalletsWithAlchemy\nRow scan error: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
// 마지막 체크 시각 파싱
|
||||
lastCheckTime, _ := time.Parse("2006-01-02 15:04:05", lastCheckStr)
|
||||
|
||||
// 개별 지갑 체크 (현재 블록 번호를 파라미터로 전달)
|
||||
newTxCount, err := checkWalletWithAlchemy(walletDBConn, header, userId, walletAddress, contractAddress, lastCheckTime, currentBlock, CODE_STRC)
|
||||
if err != nil {
|
||||
// 에러 로그는 checkWalletWithAlchemy 내부에서 이미 출력됨
|
||||
continue
|
||||
}
|
||||
|
||||
// 마지막 체크 시각 업데이트
|
||||
updateLastCheckTime(walletDBConn, userId)
|
||||
|
||||
totalChecked++
|
||||
totalNewTransactions += newTxCount
|
||||
|
||||
// 다음 요청 전 대기 (마지막 요청 후에는 대기 불필요)
|
||||
if totalChecked < batchSize {
|
||||
time.Sleep(delayBetweenRequests)
|
||||
}
|
||||
}
|
||||
|
||||
elapsed := time.Since(startTime)
|
||||
|
||||
umlog.Info("checkAllWalletsWithAlchemy\nBatch completed - Checked: %d wallets, New transactions: %d, Time: %.2f seconds",
|
||||
totalChecked, totalNewTransactions, elapsed.Seconds())
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 트랜잭션 모니터링 시작 (Alchemy API 사용)
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- contractAddress : CYBX 토큰 컨트랙트 주소
|
||||
- checkInterval : 체크 주기
|
||||
- batchSize : 한 번에 처리할 사용자 수
|
||||
- netType : 네트워크 타입
|
||||
- CODE_STRC : 설정 정보
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func StartTransactionMonitorAlchemy(walletDBConn *sql.DB, header model.Header, contractAddress string, checkInterval time.Duration, batchSize int, CODE_STRC model.CodeStrc) {
|
||||
umlog.Debug("[StartTransactionMonitorAlchemy] Starting monitor - Interval: %v, Batch size: %d\n", checkInterval, batchSize)
|
||||
umlog.Debug("[StartTransactionMonitorAlchemy] Contract Address: %s\n", contractAddress)
|
||||
umlog.Debug("[StartTransactionMonitorAlchemy] Alchemy URL: %s\n", CODE_STRC.ALCHEMY_MAINNET)
|
||||
|
||||
ticker := time.NewTicker(checkInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
// 즉시 한 번 실행
|
||||
checkAllWalletsWithAlchemy(walletDBConn, header, contractAddress, batchSize, CODE_STRC)
|
||||
|
||||
// 주기적 실행
|
||||
for range ticker.C {
|
||||
checkAllWalletsWithAlchemy(walletDBConn, header, contractAddress, batchSize, CODE_STRC)
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 입금 알림 전송 (FCM Push + Telegram)
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- userId : 사용자 ID
|
||||
- fromAddress : 발신 주소
|
||||
- toAddress : 수신 주소
|
||||
- amount : 금액
|
||||
- txHash : 트랜잭션 해시
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func sendIncomingTransactionNotification(walletDBConn *sql.DB, header model.Header, userId int, fromAddress string, toAddress string, amount string, txHash string) {
|
||||
// FCM 토큰 조회
|
||||
var fcmToken, angkorId string
|
||||
query := "SELECT fcm_token, angkorid FROM ank_wallet_user WHERE id = ?"
|
||||
umlog.Debug("authCode count query: %s", query)
|
||||
rows, _, err := umsql.SqlSelect(walletDBConn, query, userId)
|
||||
if err != nil {
|
||||
umlog.Error("SELECT Error(%s): %s", err.Error(), query)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
if rows.Next() {
|
||||
if scanErr := rows.Scan(&fcmToken, &angkorId); scanErr != nil {
|
||||
umlog.Error("SELECT Scan Error(%s): %s", scanErr.Error(), query)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// FCM Push 알림 (FCM 토큰이 있는 경우)
|
||||
// if fcmToken.Valid && fcmToken.String != "" {
|
||||
// sendFCMPushNotification(fcmToken.String, amount, fromAddress, txHash)
|
||||
// }
|
||||
|
||||
// Telegram 알림 (관리자용)
|
||||
messageboxy := fmt.Sprintf(
|
||||
"💰 Deposit notification(from Alchemy)\nFrom: %s\nTo: %s\nAmount: %s CYBX\nTxHash: %s",
|
||||
fromAddress, toAddress, amount, txHash)
|
||||
message := map[string]string{"message": messageboxy}
|
||||
jsonData, err := json.Marshal(message)
|
||||
if err != nil {
|
||||
umlog.Warn("SendMessage Error(%s): %v", err.Error(), message)
|
||||
}
|
||||
SendMessage(header, angkorId, string(jsonData))
|
||||
}
|
||||
+149
@@ -0,0 +1,149 @@
|
||||
package src
|
||||
|
||||
import (
|
||||
"AngkorWalletScanning/model"
|
||||
"AngkorWalletScanning/umlog"
|
||||
"AngkorWalletScanning/umsql"
|
||||
"bytes"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : DB에서 코드값을 읽어온다.
|
||||
parameter
|
||||
- dbConn : DB 핸들러
|
||||
- key : AES 복호화 키
|
||||
return
|
||||
- int : 상태 코드
|
||||
- model.CodeStrc: 코드 구조체
|
||||
error code
|
||||
- 4001202 : DB 연결이 nil
|
||||
- 4001213 : DB 조회 오류
|
||||
- 4001215 : DB 스캔 오류
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func GetCodeInfo(dbConn *sql.DB, key string) (int, model.CodeStrc) {
|
||||
var status int = 200
|
||||
var code_strc model.CodeStrc
|
||||
|
||||
// DB 연결 검증
|
||||
if dbConn == nil {
|
||||
umlog.Error("DB connection is nil")
|
||||
return 4001202, code_strc
|
||||
}
|
||||
|
||||
// 매개변수 검증
|
||||
if len(key) == 0 {
|
||||
umlog.Error("Decryption key is empty")
|
||||
return 4001213, code_strc
|
||||
}
|
||||
|
||||
// SQL 쿼리 (매개변수 바인딩 사용)
|
||||
query := `SELECT code_name, AES_DECRYPT(UNHEX(code_value), ?) AS code_value
|
||||
FROM angkor_code
|
||||
WHERE code_number IN ('001001', '001002', '001010', '001011', '001003', '001004', '001005', '100001', '100002', '100003', '100004', '100005')`
|
||||
|
||||
res, _, err := umsql.SqlSelect(dbConn, query, key)
|
||||
if err != nil {
|
||||
umlog.Error("SELECT Error(%s): %s", err.Error(), query)
|
||||
return 4001213, code_strc
|
||||
}
|
||||
defer res.Close()
|
||||
|
||||
for res.Next() {
|
||||
var code_name, code_value string
|
||||
if err := res.Scan(&code_name, &code_value); err != nil {
|
||||
umlog.Error("SCAN Error(%s): %s", err.Error(), query)
|
||||
return 4001215, code_strc
|
||||
}
|
||||
|
||||
// 코드명에 따라 구조체 필드 설정
|
||||
switch code_name {
|
||||
case "CIPHERKEY":
|
||||
code_strc.CIPHERKEY = code_value
|
||||
case "CIPHERIVKEY":
|
||||
code_strc.CIPHERIVKEY = code_value
|
||||
case "MESSAGE_TOKEN":
|
||||
code_strc.MESSAGE_TOKEN = code_value
|
||||
case "MESSAGE_ID":
|
||||
code_strc.MESSAGE_ID = code_value
|
||||
case "SENDBIRD_KEY":
|
||||
code_strc.SENDBIRD_KEY = code_value
|
||||
case "SENDBIRD_TOKEN":
|
||||
code_strc.SENDBIRD_TOKEN = code_value
|
||||
case "SENDBIRD_VERSION":
|
||||
code_strc.SENDBIRD_VERSION = code_value
|
||||
case "ETHERSCAN":
|
||||
code_strc.ETHERSCAN_TOKEN = code_value
|
||||
case "CYBX_MAINNET":
|
||||
code_strc.CYBX_MAINNET = code_value
|
||||
case "CYBX_TESTNET":
|
||||
code_strc.CYBX_TESTNET = code_value
|
||||
case "ALCHEMY_MAINNET":
|
||||
code_strc.ALCHEMY_MAINNET = code_value
|
||||
case "ALCHEMY_TESTNET":
|
||||
code_strc.ALCHEMY_TESTNET = code_value
|
||||
default:
|
||||
umlog.Warn("Unknown code_name: %s", code_name)
|
||||
}
|
||||
}
|
||||
|
||||
return status, code_strc
|
||||
}
|
||||
|
||||
func SendMessage(header model.Header, angkorId string, message string) error {
|
||||
// Media 서버에서 사용자 정보 조회
|
||||
fullUrl := "https://aauth.angkorlifes.com/message/v1/chat"
|
||||
var sendBody = map[string]any{
|
||||
"grantType": "session",
|
||||
"angkorId": angkorId,
|
||||
"messageType": "WAL",
|
||||
"notiMessage": message,
|
||||
}
|
||||
sendBodyByte, err := json.Marshal(sendBody)
|
||||
if err != nil {
|
||||
umlog.Warn("JSON marshal error: %s", err.Error())
|
||||
return err
|
||||
}
|
||||
req, err := http.NewRequest("POST", fullUrl, bytes.NewBuffer(sendBodyByte))
|
||||
if err != nil {
|
||||
umlog.Warn("Error creating request: => %s : %s", fullUrl, err.Error())
|
||||
return err
|
||||
}
|
||||
//Content-Type 헤더 추가
|
||||
req.Header.Add("Content-Type", "application/json")
|
||||
req.Header.Add("AppKey", header.AppKey)
|
||||
req.Header.Add("Authorization", header.SecretKey)
|
||||
|
||||
client := &http.Client{}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
umlog.Warn("Error: => %s", err.Error())
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
umlog.Debug("StatusCode: => %d", resp.StatusCode)
|
||||
|
||||
// 응답 본문 읽기
|
||||
resBody, err := ioutil.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
umlog.Warn("Error reading response body: %s", err.Error())
|
||||
return err
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
// 실패 응답
|
||||
umlog.Warn("User info retrieval failed - Status: %d, Response: %s", resp.StatusCode, string(resBody))
|
||||
return fmt.Errorf("User info retrieval failed - Status: %s, Response: %s", strconv.Itoa(resp.StatusCode), string(resBody))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,464 @@
|
||||
package src
|
||||
|
||||
import (
|
||||
"AngkorWalletScanning/model"
|
||||
"AngkorWalletScanning/umlog"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/big"
|
||||
"net/http"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
// BSCScan API Rate Limiter (5 requests per second for free tier)
|
||||
var bscApiLimiter = rate.NewLimiter(rate.Limit(5), 5)
|
||||
|
||||
// BSCScan API Response Structures
|
||||
type BSCScanTransaction struct {
|
||||
BlockNumber string `json:"blockNumber"`
|
||||
TimeStamp string `json:"timeStamp"`
|
||||
Hash string `json:"hash"`
|
||||
From string `json:"from"`
|
||||
To string `json:"to"`
|
||||
Value string `json:"value"`
|
||||
ContractAddress string `json:"contractAddress"`
|
||||
TokenName string `json:"tokenName"`
|
||||
TokenSymbol string `json:"tokenSymbol"`
|
||||
TokenDecimal string `json:"tokenDecimal"`
|
||||
TransactionIndex string `json:"transactionIndex"`
|
||||
Gas string `json:"gas"`
|
||||
GasPrice string `json:"gasPrice"`
|
||||
GasUsed string `json:"gasUsed"`
|
||||
CumulativeGasUsed string `json:"cumulativeGasUsed"`
|
||||
Input string `json:"input"`
|
||||
Confirmations string `json:"confirmations"`
|
||||
}
|
||||
|
||||
type BSCScanResponse struct {
|
||||
Status string `json:"status"`
|
||||
Message string `json:"message"`
|
||||
Result []BSCScanTransaction `json:"result"`
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : BSCScan API를 통해 특정 주소의 토큰 트랜잭션 조회
|
||||
parameter
|
||||
- address : 조회할 지갑 주소
|
||||
- contractAddress : 토큰 컨트랙트 주소 (CYBX)
|
||||
- startBlock : 시작 블록 번호 (0 = 처음부터)
|
||||
- netType : 네트워크 타입 ("MAINNET" 또는 "TESTNET")
|
||||
return
|
||||
- []BSCScanTransaction : 트랜잭션 목록
|
||||
- error : 에러 정보
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func getBSCScanTokenTransactions(address string, contractAddress string, startBlock int64, netType string, CODE_STRC model.CodeStrc) ([]BSCScanTransaction, error) {
|
||||
// Rate limiter 체크 (5 requests per second)
|
||||
ctx := context.Background()
|
||||
err := bscApiLimiter.Wait(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("rate limiter error: %v", err)
|
||||
}
|
||||
|
||||
// chainid 결정 (MAINNET: 56, TESTNET: 97)
|
||||
chainId := "56" // BSC Mainnet
|
||||
if netType == "TESTNET" {
|
||||
chainId = "97" // BSC Testnet
|
||||
}
|
||||
|
||||
// Etherscan API V2 URL 구성
|
||||
url := fmt.Sprintf(
|
||||
"https://api.etherscan.io/v2/api?chainid=%s&module=account&action=tokentx&contractaddress=%s&address=%s&startblock=%d&endblock=999999999&sort=desc&apikey=%s",
|
||||
chainId,
|
||||
contractAddress,
|
||||
address,
|
||||
startBlock,
|
||||
CODE_STRC.ETHERSCAN_TOKEN,
|
||||
)
|
||||
|
||||
// HTTP GET 요청
|
||||
resp, err := http.Get(url)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("http request failed: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// 응답 읽기
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read response: %v", err)
|
||||
}
|
||||
|
||||
// JSON 파싱
|
||||
var bscResp BSCScanResponse
|
||||
err = json.Unmarshal(body, &bscResp)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse json: %v", err)
|
||||
}
|
||||
|
||||
// API 응답 상태 체크
|
||||
if bscResp.Status != "1" {
|
||||
if bscResp.Message == "No transactions found" {
|
||||
return []BSCScanTransaction{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("bscscan api error: %s", bscResp.Message)
|
||||
}
|
||||
|
||||
return bscResp.Result, nil
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 특정 지갑의 입금 트랜잭션 체크 및 알림 전송
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- userId : 사용자 ID
|
||||
- walletAddress : 지갑 주소
|
||||
- contractAddress : CYBX 토큰 컨트랙트 주소
|
||||
- lastCheckTime : 마지막 체크 시각
|
||||
- netType : 네트워크 타입 ("MAINNET" 또는 "TESTNET")
|
||||
return
|
||||
- int : 새로 발견된 입금 건수
|
||||
- error : 에러 정보
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func checkWalletWithAPI(walletDBConn *sql.DB, header model.Header, userId int, walletAddress string, contractAddress string, lastCheckTime time.Time, netType string, CODE_STRC model.CodeStrc) (int, error) {
|
||||
// 마지막 체크 시각을 블록 번호로 변환 (BSC는 약 3초당 1블록)
|
||||
timeDiff := time.Since(lastCheckTime)
|
||||
blocksToCheck := int64(timeDiff.Seconds() / 3) // 3초당 1블록
|
||||
|
||||
// 최소 100블록은 체크 (약 5분)
|
||||
if blocksToCheck < 100 {
|
||||
blocksToCheck = 100
|
||||
}
|
||||
|
||||
// 현재 블록에서 역산
|
||||
currentBlock := getCurrentBlockNumber(netType)
|
||||
startBlock := currentBlock - blocksToCheck
|
||||
if startBlock < 0 {
|
||||
startBlock = 0
|
||||
}
|
||||
|
||||
// BSCScan API로 트랜잭션 조회
|
||||
transactions, err := getBSCScanTokenTransactions(walletAddress, contractAddress, startBlock, netType, CODE_STRC)
|
||||
if err != nil {
|
||||
umlog.Error("checkWalletWithAPI\nFailed to get transactions for user %d: %v", userId, err)
|
||||
return 0, err
|
||||
}
|
||||
|
||||
newTransactionCount := 0
|
||||
|
||||
// 입금 트랜잭션 필터링 (To == 내 지갑 주소)
|
||||
for _, tx := range transactions {
|
||||
// 소문자로 변환하여 비교
|
||||
if strings.ToLower(tx.To) != strings.ToLower(walletAddress) {
|
||||
continue // 출금 트랜잭션은 무시
|
||||
}
|
||||
|
||||
// 트랜잭션 시각 체크
|
||||
txTimestamp, _ := strconv.ParseInt(tx.TimeStamp, 10, 64)
|
||||
txTime := time.Unix(txTimestamp, 0)
|
||||
|
||||
// 마지막 체크 이후의 트랜잭션만 처리
|
||||
if txTime.Before(lastCheckTime) {
|
||||
continue
|
||||
}
|
||||
|
||||
// 이미 알림 보낸 트랜잭션인지 확인
|
||||
if isTransactionNotified(walletDBConn, tx.Hash) {
|
||||
continue
|
||||
}
|
||||
|
||||
// 금액 계산 (토큰 decimal 적용)
|
||||
amount := calculateTokenAmount(tx.Value, tx.TokenDecimal)
|
||||
|
||||
// 알림 전송
|
||||
sendIncomingTransactionNotification(walletDBConn, header, userId, tx.From, walletAddress, amount, tx.Hash)
|
||||
|
||||
// 알림 기록 저장
|
||||
saveNotificationRecord(walletDBConn, userId, tx.Hash, tx.From, tx.To, amount, tx.TokenSymbol)
|
||||
|
||||
newTransactionCount++
|
||||
|
||||
umlog.Info("checkWalletWithAPI\nNew incoming tx detected - User: %d, From: %s, Amount: %s %s, Hash: %s",
|
||||
userId, tx.From, amount, tx.TokenSymbol, tx.Hash)
|
||||
}
|
||||
|
||||
return newTransactionCount, nil
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 현재 BSC 블록 번호 조회
|
||||
parameter
|
||||
- netType : 네트워크 타입 ("MAINNET" 또는 "TESTNET")
|
||||
return
|
||||
- int64 : 현재 블록 번호
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func getCurrentBlockNumber(netType string) int64 {
|
||||
apiKey := os.Getenv("BSCSCAN_API_KEY")
|
||||
if apiKey == "" {
|
||||
return 0
|
||||
}
|
||||
|
||||
// chainid 결정 (MAINNET: 56, TESTNET: 97)
|
||||
chainId := "56" // BSC Mainnet
|
||||
if netType == "TESTNET" {
|
||||
chainId = "97" // BSC Testnet
|
||||
}
|
||||
|
||||
url := fmt.Sprintf("https://api.etherscan.io/v2/api?chainid=%s&module=proxy&action=eth_blockNumber&apikey=%s", chainId, apiKey)
|
||||
|
||||
resp, err := http.Get(url)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
var result map[string]interface{}
|
||||
json.Unmarshal(body, &result)
|
||||
|
||||
if blockHex, ok := result["result"].(string); ok {
|
||||
blockNum := new(big.Int)
|
||||
blockNum.SetString(blockHex[2:], 16) // "0x" 제거 후 16진수 파싱
|
||||
return blockNum.Int64()
|
||||
}
|
||||
|
||||
return 0
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 토큰 금액 계산 (decimal 적용)
|
||||
parameter
|
||||
- value : 토큰 raw value (wei 단위)
|
||||
- decimal : 토큰 decimal
|
||||
return
|
||||
- string : 사람이 읽을 수 있는 금액
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func calculateTokenAmount(value string, decimal string) string {
|
||||
// value를 big.Int로 변환
|
||||
valueInt := new(big.Int)
|
||||
valueInt.SetString(value, 10)
|
||||
|
||||
// decimal을 int로 변환
|
||||
decimalInt, _ := strconv.Atoi(decimal)
|
||||
|
||||
// 10^decimal 계산
|
||||
divisor := new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(decimalInt)), nil)
|
||||
|
||||
// 나눗셈
|
||||
result := new(big.Float).Quo(
|
||||
new(big.Float).SetInt(valueInt),
|
||||
new(big.Float).SetInt(divisor),
|
||||
)
|
||||
|
||||
// 소수점 4자리까지 표시
|
||||
return result.Text('f', 4)
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 알림 기록을 DB에 저장
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- userId : 사용자 ID
|
||||
- txHash : 트랜잭션 해시
|
||||
- txFrom : 발신 주소
|
||||
- txTo : 수신 주소
|
||||
- amount : 금액
|
||||
- token : 토큰 심볼
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func saveNotificationRecord(walletDBConn *sql.DB, userId int, txHash string, txFrom string, txTo string, amount string, token string) {
|
||||
query := `
|
||||
INSERT INTO ank_wallet_notification
|
||||
(user_id, tx_hash, tx_from, tx_to, amount, token, notified_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, NOW())
|
||||
`
|
||||
|
||||
_, err := walletDBConn.Exec(query, userId, txHash, txFrom, txTo, amount, token)
|
||||
if err != nil {
|
||||
umlog.Error("saveNotificationRecord", fmt.Sprintf("Failed to save notification: %v", err))
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 사용자의 마지막 체크 시각 업데이트
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- userId : 사용자 ID
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func updateLastCheckTime(walletDBConn *sql.DB, userId int) {
|
||||
query := "UPDATE ank_wallet_user SET checkedAt = NOW() WHERE id = ?"
|
||||
|
||||
_, err := walletDBConn.Exec(query, userId)
|
||||
if err != nil {
|
||||
umlog.Error("updateLastCheckTime\nFailed to update last_check_time for user %d: %v", userId, err)
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : FCM Push 알림 전송
|
||||
parameter
|
||||
- fcmToken : FCM 토큰
|
||||
- amount : 입금 금액
|
||||
- fromAddress : 발신 주소
|
||||
- txHash : 트랜잭션 해시
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func sendFCMPushNotification(fcmToken string, amount string, fromAddress string, txHash string) {
|
||||
// TODO: Firebase Admin SDK를 사용하여 FCM Push 알림 구현
|
||||
// 여기서는 구조만 제공하고, 실제 구현은 프로젝트 환경에 맞게 작성 필요
|
||||
|
||||
umlog.Info("sendFCMPushNotification\nSending FCM push to token: %s, Amount: %s CYBX", fcmToken[:20]+"...", amount)
|
||||
|
||||
// Example:
|
||||
// message := &messaging.Message{
|
||||
// Token: fcmToken,
|
||||
// Notification: &messaging.Notification{
|
||||
// Title: "입금 알림",
|
||||
// Body: fmt.Sprintf("%s CYBX가 입금되었습니다.", amount),
|
||||
// },
|
||||
// Data: map[string]string{
|
||||
// "type": "incoming_transaction",
|
||||
// "amount": amount,
|
||||
// "from": fromAddress,
|
||||
// "txHash": txHash,
|
||||
// },
|
||||
// }
|
||||
//
|
||||
// _, err := fcmClient.Send(ctx, message)
|
||||
// if err != nil {
|
||||
// umlog.Error("sendFCMPushNotification", fmt.Sprintf("FCM send failed: %v", err))
|
||||
// }
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 모든 활성 사용자의 지갑 체크 (배치 처리)
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- contractAddress : CYBX 토큰 컨트랙트 주소
|
||||
- batchSize : 한 번에 처리할 사용자 수 (기본 300)
|
||||
- netType : 네트워크 타입 ("MAINNET" 또는 "TESTNET")
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func checkAllWalletsForIncomingTransactions(walletDBConn *sql.DB, header model.Header, contractAddress string, batchSize int, netType string, CODE_STRC model.CodeStrc) {
|
||||
startTime := time.Now()
|
||||
|
||||
// 활성 사용자 목록 조회 (최근 활동 순)
|
||||
query := `
|
||||
SELECT id, address, checkedAt
|
||||
FROM ank_wallet_user
|
||||
WHERE address > ''
|
||||
ORDER BY checkedAt ASC
|
||||
LIMIT ?
|
||||
`
|
||||
|
||||
rows, err := walletDBConn.Query(query, batchSize)
|
||||
if err != nil {
|
||||
umlog.Error("checkAllWalletsForIncomingTransactions\nDB query failed: %v", err)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
totalChecked := 0
|
||||
totalNewTransactions := 0
|
||||
|
||||
for rows.Next() {
|
||||
var userId int
|
||||
var walletAddress string
|
||||
var lastCheckStr string
|
||||
|
||||
err := rows.Scan(&userId, &walletAddress, &lastCheckStr)
|
||||
if err != nil {
|
||||
umlog.Error("checkAllWalletsForIncomingTransactions\nRow scan error: %v", err)
|
||||
continue
|
||||
}
|
||||
|
||||
// 마지막 체크 시각 파싱
|
||||
lastCheckTime, _ := time.Parse("2006-01-02 15:04:05", lastCheckStr)
|
||||
|
||||
// 개별 지갑 체크
|
||||
newTxCount, err := checkWalletWithAPI(walletDBConn, header, userId, walletAddress, contractAddress, lastCheckTime, netType, CODE_STRC)
|
||||
if err != nil {
|
||||
// 에러 로그는 checkWalletWithAPI 내부에서 이미 출력됨
|
||||
continue
|
||||
}
|
||||
|
||||
// 마지막 체크 시각 업데이트
|
||||
updateLastCheckTime(walletDBConn, userId)
|
||||
|
||||
totalChecked++
|
||||
totalNewTransactions += newTxCount
|
||||
}
|
||||
|
||||
elapsed := time.Since(startTime)
|
||||
|
||||
umlog.Info("checkAllWalletsForIncomingTransactions\nBatch completed - Checked: %d wallets, New transactions: %d, Time: %.2f seconds",
|
||||
totalChecked, totalNewTransactions, elapsed.Seconds())
|
||||
}
|
||||
|
||||
/*
|
||||
**************************************************************
|
||||
|
||||
desc : 트랜잭션 모니터링 시작 (백그라운드 고루틴)
|
||||
parameter
|
||||
- walletDBConn : Wallet DB 핸들러
|
||||
- contractAddress : CYBX 토큰 컨트랙트 주소
|
||||
- checkInterval : 체크 주기 (예: 5분)
|
||||
- batchSize : 한 번에 처리할 사용자 수
|
||||
- netType : 네트워크 타입 ("MAINNET" 또는 "TESTNET")
|
||||
|
||||
**************************************************************
|
||||
*/
|
||||
func StartTransactionMonitor(walletDBConn *sql.DB, header model.Header, contractAddress string, checkInterval time.Duration, batchSize int, netType string, CODE_STRC model.CodeStrc) {
|
||||
umlog.Info("StartTransactionMonitor: Transaction monitor started - Network: %s, Interval: %v, Batch size: %d", netType, checkInterval, batchSize)
|
||||
|
||||
ticker := time.NewTicker(checkInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
// 즉시 한 번 실행
|
||||
checkAllWalletsForIncomingTransactions(walletDBConn, header, contractAddress, batchSize, netType, CODE_STRC)
|
||||
|
||||
// // 주기적 실행
|
||||
// for range ticker.C {
|
||||
// checkAllWalletsForIncomingTransactions(walletDBConn, header, contractAddress, batchSize, netType, CODE_STRC)
|
||||
// }
|
||||
}
|
||||
Reference in New Issue
Block a user