cronProcessLog.ts

// cronProcessLog.ts
import config from './config';
import { Database, wkSign } from './common';
import * as mysql from 'mysql2';
import * as crypto from 'crypto';
import needle from 'needle';

class CronProcessor {
    private config: typeof config;
    private db: mysql.Connection;
    
    constructor() {
        // Load config
        this.config = config;
        
        // Set timezone (Node.js uses system timezone, but we can format dates accordingly)
        // Note: For proper timezone handling, consider using a library like date-fns-tz
        
        // Get database connection
        this.db = Database.getInstance().getConnection();
    }
    
    public async processWarmKeyLogs(): Promise<string> {
        const code = "process_warmkey";
        let logsProcessed = 0;
        
        try {
            // 1. Call WarmKey API
            const logs = await this.queryWarmKeyLogs();
            logsProcessed = logs.length;
            
            if (logs.length === 0) {
                return "No new logs to process.";
            }
            
            // 2. Start transaction
            try {
                await Database.getInstance().beginTransaction();
                
                // 3. Process logs
                await this.processLogs(logs);
                
                // 4. Commit
                await Database.getInstance().commit();
                
                return "Cron job completed successfully. Processed " + logsProcessed + " logs.";
                
            } catch (e: any) {
                // Handle errors within transaction
                await Database.getInstance().rollBack();
                throw e; // Re-throw to outer catch block
            }
            
        } catch (e: any) {
            if (e.code && (e.code.startsWith('ER_') || e.code.startsWith('ECONNREFUSED'))) {
                return "Database Error: " + e.message + " (Processed: " + logsProcessed + " logs)";
            } else {
                return "Error: " + e.message + " (Processed: " + logsProcessed + " logs)";
            }
        }
    }
    
    private async queryWarmKeyLogs(): Promise<any[]> {
        const api_url = this.config.warmkey.api.url;
        const method = '/paymentV1/queryLog';
        const api_key = this.config.warmkey.api.key;
        const nonce = String(Math.round(Date.now()));
        
        // Get last log ID
        return new Promise((resolve, reject) => {
            this.db.query("SELECT log_id FROM wk_query_log ORDER BY log_id DESC LIMIT 1", [], async (error: mysql.QueryError | null, results: any[]) => {
                if (error) {
                    reject(error);
                    return;
                }
                
                let log_id = (results.length > 0 && results[0].log_id) ? results[0].log_id : 1;
                log_id++;
                
                const payload = { 'log_id': log_id };
                const headers: any = { 'api_key': api_key, 'nonce': nonce };
                
                // Generate signature
                const to_sign = crypto.createHash('sha256').update(JSON.stringify({'header': headers, 'payload': payload}).replace(/\//g, '\\/')).digest();
                const api_secret_key = this.config.warmkey.api_secret_key;
                
                const signature = wkSign(to_sign, api_secret_key);
                if (signature.error) {
                    reject(new Error("Signature error: " + signature.error));
                    return;
                }
                
                headers["signature"] = signature.signature;
                
                // Make API call
                needle.post(
                    api_url + method,
                    JSON.stringify({'header': headers, 'payload': payload}).replace(/\//g, '\\/'),
                    {
                        headers: { 'Content-Type': 'application/json' },
                        timeout: 30000
                    },
                    (err, resp) => {
                        if (err) {
                            reject(new Error("API Error: " + err.message));
                            return;
                        }
                        
                        let result: any;
                        try {
                            result = typeof resp.body === 'string' ? JSON.parse(resp.body) : resp.body;
                        } catch (e) {
                            reject(new Error("Invalid JSON response"));
                            return;
                        }
                        
                        if (!result.code || result.code != 100) {
                            reject(new Error("API returned error: " + (result.message || 'Unknown error')));
                            return;
                        }
                        
                        resolve(result.result || []);
                    }
                );
            });
        });
    }
    
    private async processLogs(logs: any[]): Promise<void> {
        for (const log of logs) {

            console.log(log);
            // Log the processed record
            await this.logQuery(log);
            
            // Process different log types
            switch (log.log_type) {
                case 'withdrawal_req_created':
                    await this.processWithdrawalCreated(log);
                    break;

                case 'deposit_link_created':
                    await this.processDepositLink(log);
                    break;
                    
                case 'deposit_tx':
                    await this.processDepositTransaction(log);
                    break;
                    
                case 'deposit_fundout':
                    await this.processFundOut(log);
                    break;
                    
                case 'withdrawal_status_changed':
                    await this.processWithdrawalStatus(log);
                    break;
            }
        }
    }
    
    private async processDepositLink(log: any): Promise<void> {
        const path = log.deposit.path.replace(/\\/g, '');
        const path_arr = path.split('/');
        const user_id = parseInt(path_arr[1]);
        
        const addresses = log.deposit.address || {};
        
        const updateData: { [key: string]: string } = {};
        
        for (const network in addresses) {
            const fieldName = 'addr_' + network.toLowerCase() + '_address';
            updateData[fieldName] = addresses[network];
        }
        
        // Check if record exists
        return new Promise((resolve, reject) => {
            this.db.query("SELECT addr_id FROM wk_address WHERE addr_relative_path = ? LIMIT 1", [path], async (error: mysql.QueryError | null, results: any[]) => {
                if (error) {
                    reject(error);
                    return;
                }
                
                const exists = results.length > 0 && results[0].addr_id;
                
                if (exists) {
                    // UPDATE
                    await this.updateAddressRecord(path, user_id, updateData);
                } else {
                    // INSERT
                    await this.insertAddressRecord(path, user_id, updateData);
                }
                
                resolve();
            });
        });
    }
    
    private async updateAddressRecord(path: string, user_id: number, updateData: { [key: string]: string }): Promise<void> {
        const setClauses: string[] = [];
        const params: any[] = [];
        
        for (const field in updateData) {
            setClauses.push(`${field} = IF(${field} = '' OR ${field} IS NULL, ?, ${field})`);
            params.push(updateData[field]);
        }
        
        params.push(new Date().toISOString().slice(0, 19).replace('T', ' '));
        params.push(path);
        params.push(user_id);
        
        const sql = `UPDATE wk_address 
                SET ${setClauses.join(', ')}, addr_mdate = ?
                WHERE addr_relative_path = ? AND user_id = ?`;
        
        return new Promise((resolve, reject) => {
            this.db.query(sql, params, (error: mysql.QueryError | null) => {
                if (error) {
                    reject(error);
                } else {
                    resolve();
                }
            });
        });
    }
    
    private async insertAddressRecord(path: string, user_id: number, insertData: { [key: string]: string }): Promise<void> {
        const data: any = { ...insertData };
        data.addr_relative_path = path;
        data.user_id = user_id;
        data.addr_cdate = new Date().toISOString().slice(0, 19).replace('T', ' ');
        
        const columns = Object.keys(data).join(', ');
        const placeholders = Object.keys(data).map(() => '?').join(', ');
        const values = Object.values(data);
        
        const sql = `INSERT INTO wk_address (${columns}) VALUES (${placeholders})`;
        
        return new Promise((resolve, reject) => {
            this.db.query(sql, values, (error: mysql.QueryError | null) => {
                if (error) {
                    reject(error);
                } else {
                    resolve();
                }
            });
        });
    }
    
    private async processDepositTransaction(log: any): Promise<void> {
        const path = log.deposit.path.replace(/\\/g, '');
        const path_arr = path.split('/');
        const user_id = parseInt(path_arr[1]);
        
        const chain = log.deposit_tx.network;
        const from = log.deposit_tx.sender;
        const to = log.deposit_tx.recipient;
        const position = log.deposit_tx.position;
        const tx_hash = log.deposit_tx.transaction_hash;
        const coin = log.deposit_tx.coin_symbol;
        const amount = log.deposit_tx.amount;
        const now = new Date().toISOString().slice(0, 19).replace('T', ' ');
        
        return new Promise((resolve, reject) => {
            this.db.query(`
                INSERT INTO wk_deposit (bcd_chain, user_id, bcd_sender, bcd_receiver, bcd_amount, bcd_position, bcd_txhash, bcd_cdate, bcd_token)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            `, [chain, user_id, from, to, amount, position, tx_hash, now, coin], async (error: mysql.QueryError | null) => {
                if (error) {
                    reject(error);
                    return;
                }
                
                // Update wk_user coin balance
                if (user_id) {
                    const coin_name = coin.toLowerCase();
                    
                    this.db.query(
                        `UPDATE wk_user SET user_${coin_name}_balance = user_${coin_name}_balance + ?, user_mdate = ? WHERE user_id = ?`,
                        [amount, now, user_id],
                        (updateError: mysql.QueryError | null) => {
                            if (updateError) {
                                reject(updateError);
                            } else {
                                resolve();
                            }
                        }
                    );
                } else {
                    resolve();
                }
            });
        });
    }

    private async processWithdrawalCreated(log: any): Promise<void> {
        // process withdrawal request is received by WarmKey
    }
    
    private async processFundOut(log: any): Promise<void> {
        // insert into table if necessary
        // Empty implementation as in PHP
    }
    
    private async processWithdrawalStatus(log: any): Promise<void> {
        const status = log.withdrawal.status;
        const mw_id = log.withdrawal.unique_id;
        const hash = log.withdrawal.transaction_hash;
        const position = log.withdrawal.position;
        const now = new Date().toISOString().slice(0, 19).replace('T', ' ');
        
        return new Promise((resolve, reject) => {
            if (status == 'success') {
                this.db.query(`
                    UPDATE wk_withdrawal 
                    SET mw_status = ?, mw_txhash = ?, mw_pddate = ?, mw_position = ? 
                    WHERE mw_id = ?
                `, [status, hash, now, position, mw_id], (error: mysql.QueryError | null) => {
                    if (error) {
                        reject(error);
                    } else {
                        resolve();
                    }
                });
            } else {
                this.db.query(`
                    UPDATE wk_withdrawal 
                    SET mw_status = ?, mw_remark = ? 
                    WHERE mw_id = ?
                `, [status, log.withdrawal.status_message, mw_id], (error: mysql.QueryError | null) => {
                    if (error) {
                        reject(error);
                    } else {
                        resolve();
                    }
                });
            }
        });
    }
    
    private async logQuery(log: any): Promise<void> {
        const now = new Date().toISOString().slice(0, 19).replace('T', ' ');
        
        return new Promise((resolve, reject) => {
            this.db.query(`
                INSERT INTO wk_query_log (log_id, log_type, log_response, log_cdate)
                VALUES (?, ?, ?, ?)
            `, [log.log_id, log.log_type, JSON.stringify(log), now], (error: mysql.QueryError | null) => {
                if (error) {
                    reject(error);
                } else {
                    resolve();
                }
            });
        });
    }
}

// Run cron job
(async () => {
    try {
        const cron = new CronProcessor();
        const result = await cron.processWarmKeyLogs();
        console.log(result);
    } catch (error: any) {
        console.error("Error:", error.message);
        process.exit(1);
    } finally {
        // Close database connection
        await Database.getInstance().close();
        process.exit(0);
    }
})();