nâng cấp ws renew tránh cao tải

KittyFallWs/KittyFallWs/sql/20260713_result_request_inbox_online.sql

@@ -0,0 +1,172 @@
+-- ============================================================================
+-- KittyFallWs - durable resultRequest inbox
+-- Oracle online-safe migration
+--
+-- Muc tieu:
+--   1. Tao ledger chong xu ly trung khi resultRequest duoc doc lai tu file.
+--   2. Ho tro doi soat/replay theo EVENT_ID va TRANSACTION_ID.
+--   3. Khong ALTER/LOCK cac bang nghiep vu dang nong nhu REG_INFO, CHARGE_LOG.
+--
+-- Dac tinh trien khai:
+--   - Chi tao object moi.
+--   - Co the chay lai; cac object da ton tai se duoc bo qua.
+--   - Khong DML vao bang nghiep vu hien tai.
+--   - Oracle DDL tu dong COMMIT. Nen backup schema metadata truoc khi chay.
+--
+-- Cach chay (SQL*Plus/SQLcl) bang schema ma KittyFallWs dang ket noi:
+--   sqlplus <user>/<password>@<service> @20260713_result_request_inbox_online.sql
+-- ============================================================================
+
+SET DEFINE OFF
+SET SERVEROUTPUT ON
+SET FEEDBACK ON
+SET VERIFY OFF
+WHENEVER SQLERROR EXIT SQL.SQLCODE ROLLBACK
+
+PROMPT [1/4] Creating RESULT_REQUEST_INBOX when missing...
+
+DECLARE
+    v_count NUMBER;
+BEGIN
+    SELECT COUNT(*)
+      INTO v_count
+      FROM USER_TABLES
+     WHERE TABLE_NAME = 'RESULT_REQUEST_INBOX';
+
+    IF v_count = 0 THEN
+        EXECUTE IMMEDIATE q'[
+            CREATE TABLE RESULT_REQUEST_INBOX
+            (
+                EVENT_ID          VARCHAR2(64 CHAR)   NOT NULL,
+                TRANSACTION_ID    VARCHAR2(128 CHAR),
+                SERVICE_ID        VARCHAR2(100 CHAR)  NOT NULL,
+                MSISDN            VARCHAR2(32 CHAR)   NOT NULL,
+                REQUEST_TIME      TIMESTAMP(6)        NOT NULL,
+                CHARGE_TIME_RAW   VARCHAR2(64 CHAR),
+                PARAMS_VALUE      VARCHAR2(32 CHAR),
+                MODE_VALUE        VARCHAR2(32 CHAR),
+                AMOUNT_VALUE      VARCHAR2(64 CHAR),
+                COMMAND_VALUE     VARCHAR2(100 CHAR),
+                PAYLOAD_HASH      VARCHAR2(64 CHAR)   NOT NULL,
+                SOURCE_FILE       VARCHAR2(512 CHAR),
+                SOURCE_LINE       NUMBER(12),
+                STATUS            VARCHAR2(20 CHAR)   DEFAULT 'PROCESSED' NOT NULL,
+                CHARGE_LOG_ID     NUMBER,
+                PROCESSED_TIME    TIMESTAMP(6)        DEFAULT SYSTIMESTAMP NOT NULL,
+                RETRY_COUNT       NUMBER(10)          DEFAULT 0 NOT NULL,
+                LAST_ERROR        VARCHAR2(2000 CHAR),
+                CREATED_TIME      TIMESTAMP(6)        DEFAULT SYSTIMESTAMP NOT NULL,
+                UPDATED_TIME      TIMESTAMP(6),
+                CONSTRAINT PK_RESULT_REQUEST_INBOX PRIMARY KEY (EVENT_ID),
+                CONSTRAINT CK_RESULT_REQUEST_STATUS CHECK
+                    (STATUS IN ('PROCESSING', 'PROCESSED', 'SKIPPED', 'FAILED')),
+                CONSTRAINT CK_RESULT_REQUEST_RETRY CHECK (RETRY_COUNT >= 0)
+            )
+        ]';
+        DBMS_OUTPUT.PUT_LINE('Created table RESULT_REQUEST_INBOX.');
+    ELSE
+        DBMS_OUTPUT.PUT_LINE('Table RESULT_REQUEST_INBOX already exists; skipped.');
+    END IF;
+END;
+/
+
+PROMPT [2/4] Creating lookup indexes when missing...
+
+DECLARE
+    v_count NUMBER;
+BEGIN
+    SELECT COUNT(*)
+      INTO v_count
+      FROM USER_INDEXES
+     WHERE INDEX_NAME = 'IX_RRI_TRANSACTION_SERVICE';
+
+    IF v_count = 0 THEN
+        EXECUTE IMMEDIATE
+            'CREATE INDEX IX_RRI_TRANSACTION_SERVICE '
+            || 'ON RESULT_REQUEST_INBOX (TRANSACTION_ID, SERVICE_ID)';
+        DBMS_OUTPUT.PUT_LINE('Created index IX_RRI_TRANSACTION_SERVICE.');
+    ELSE
+        DBMS_OUTPUT.PUT_LINE('Index IX_RRI_TRANSACTION_SERVICE already exists; skipped.');
+    END IF;
+
+    SELECT COUNT(*)
+      INTO v_count
+      FROM USER_INDEXES
+     WHERE INDEX_NAME = 'IX_RRI_REQUEST_TIME';
+
+    IF v_count = 0 THEN
+        EXECUTE IMMEDIATE
+            'CREATE INDEX IX_RRI_REQUEST_TIME '
+            || 'ON RESULT_REQUEST_INBOX (REQUEST_TIME)';
+        DBMS_OUTPUT.PUT_LINE('Created index IX_RRI_REQUEST_TIME.');
+    ELSE
+        DBMS_OUTPUT.PUT_LINE('Index IX_RRI_REQUEST_TIME already exists; skipped.');
+    END IF;
+
+    SELECT COUNT(*)
+      INTO v_count
+      FROM USER_INDEXES
+     WHERE INDEX_NAME = 'IX_RRI_STATUS_PROCESSED';
+
+    IF v_count = 0 THEN
+        EXECUTE IMMEDIATE
+            'CREATE INDEX IX_RRI_STATUS_PROCESSED '
+            || 'ON RESULT_REQUEST_INBOX (STATUS, PROCESSED_TIME)';
+        DBMS_OUTPUT.PUT_LINE('Created index IX_RRI_STATUS_PROCESSED.');
+    ELSE
+        DBMS_OUTPUT.PUT_LINE('Index IX_RRI_STATUS_PROCESSED already exists; skipped.');
+    END IF;
+END;
+/
+
+PROMPT [3/4] Adding metadata comments...
+
+COMMENT ON TABLE RESULT_REQUEST_INBOX IS
+    'Idempotency and audit ledger for KittyFallWs resultRequest file processing';
+COMMENT ON COLUMN RESULT_REQUEST_INBOX.EVENT_ID IS
+    'SHA-256 idempotency key derived from transaction and request identity';
+COMMENT ON COLUMN RESULT_REQUEST_INBOX.PAYLOAD_HASH IS
+    'SHA-256 hash of the normalized durable file payload';
+COMMENT ON COLUMN RESULT_REQUEST_INBOX.CHARGE_LOG_ID IS
+    'Optional reference to CHARGE_LOG.ID without altering the hot CHARGE_LOG table';
+COMMENT ON COLUMN RESULT_REQUEST_INBOX.SOURCE_FILE IS
+    'Spool/archive file containing the original durable event';
+COMMENT ON COLUMN RESULT_REQUEST_INBOX.SOURCE_LINE IS
+    'Line number of the event in the spool segment';
+
+PROMPT [4/4] Verifying created objects...
+
+COLUMN TABLE_NAME FORMAT A30
+COLUMN INDEX_NAME FORMAT A32
+COLUMN STATUS FORMAT A10
+
+SELECT TABLE_NAME, STATUS
+  FROM USER_TABLES
+ WHERE TABLE_NAME = 'RESULT_REQUEST_INBOX';
+
+SELECT INDEX_NAME, STATUS
+  FROM USER_INDEXES
+ WHERE TABLE_NAME = 'RESULT_REQUEST_INBOX'
+ ORDER BY INDEX_NAME;
+
+SELECT CONSTRAINT_NAME, CONSTRAINT_TYPE, STATUS
+  FROM USER_CONSTRAINTS
+ WHERE TABLE_NAME = 'RESULT_REQUEST_INBOX'
+ ORDER BY CONSTRAINT_TYPE, CONSTRAINT_NAME;
+
+PROMPT Migration completed successfully.
+
+-- ============================================================================
+-- IMPORTANT - transaction usage expected from the future worker:
+--
+--   1. INSERT RESULT_REQUEST_INBOX(EVENT_ID, ..., STATUS='PROCESSING')
+--   2. UPDATE REG_INFO ...
+--   3. INSERT CHARGE_LOG ...
+--   4. UPDATE RESULT_REQUEST_INBOX
+--         SET STATUS='PROCESSED', CHARGE_LOG_ID=:id, PROCESSED_TIME=SYSTIMESTAMP
+--   5. COMMIT
+--
+-- Tat ca cac buoc tren phai dung CUNG MOT Oracle Connection/transaction.
+-- Neu EVENT_ID trung PK, worker coi event da duoc xu ly va khong renew lan hai.
+-- ============================================================================
+

KittyFallWs/KittyFallWs/src/com/vas/kittyfallwsvas/wsfw/database/WsProcessUtils.java

@@ -29,6 +29,7 @@ import com.vas.kittyfallwsvas.wsfw.obj.HistoryPlay;
 import com.vas.kittyfallwsvas.wsfw.obj.PlayingTimeHistory;
 import com.vas.kittyfallwsvas.wsfw.obj.SpinGift;
 import com.vas.kittyfallwsvas.wsfw.obj.RegisterInfo;
+import com.vas.kittyfallwsvas.wsfw.resultrequest.ResultRequestEvent;
 import com.vas.kittyfallwsvas.wsfw.common.Common;
 import com.vas.kittyfallwsvas.wsfw.common.WebserviceManager;
 
@@ -143,6 +144,191 @@ public class WsProcessUtils extends DbProcessorAbstract {
         super();
     }
 
+    /**
+     * Processes one durable resultRequest event in one Oracle transaction.
+     * The inbox primary key makes replay safe after a crash or manual rerun.
+     *
+     * @return 0 processed, 1 already processed, 2 registration not found
+     */
+    public int processResultRequestRenew(ResultRequestEvent event, Timestamp expireTime,
+            String sourceFile, long sourceLine) throws SQLException {
+        Connection connection = null;
+        PreparedStatement insertInbox = null;
+        PreparedStatement checkDuplicate = null;
+        PreparedStatement selectRegister = null;
+        PreparedStatement updateRegister = null;
+        PreparedStatement selectChargeId = null;
+        PreparedStatement insertCharge = null;
+        PreparedStatement updateInbox = null;
+        ResultSet registerRs = null;
+        ResultSet chargeIdRs = null;
+        ResultSet duplicateRs = null;
+        boolean oldAutoCommit = true;
+        long chargeLogId = 0L;
+
+        try {
+            connection = getConnection(dbName);
+            if (connection == null) {
+                throw new SQLException("Cannot obtain database connection for resultRequest");
+            }
+            oldAutoCommit = connection.getAutoCommit();
+            connection.setAutoCommit(false);
+
+            String inboxSql = "INSERT INTO RESULT_REQUEST_INBOX "
+                    + "(EVENT_ID, TRANSACTION_ID, SERVICE_ID, MSISDN, REQUEST_TIME, "
+                    + "CHARGE_TIME_RAW, PARAMS_VALUE, MODE_VALUE, AMOUNT_VALUE, COMMAND_VALUE, "
+                    + "PAYLOAD_HASH, SOURCE_FILE, SOURCE_LINE, STATUS, RETRY_COUNT) "
+                    + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'PROCESSING', ?)";
+            insertInbox = connection.prepareStatement(inboxSql);
+            setQueryTimeout(insertInbox);
+            insertInbox.setString(1, limit(event.getEventId(), 64));
+            insertInbox.setString(2, limit(event.getTransactionId(), 128));
+            insertInbox.setString(3, limit(event.getServiceId(), 100));
+            insertInbox.setString(4, limit(event.getMsisdn(), 32));
+            insertInbox.setTimestamp(5, new Timestamp(event.getReceivedAt()));
+            insertInbox.setString(6, limit(event.getChargeTimeRaw(), 64));
+            insertInbox.setString(7, limit(event.getParamsValue(), 32));
+            insertInbox.setString(8, limit(event.getModeValue(), 32));
+            insertInbox.setString(9, limit(event.getAmountValue(), 64));
+            insertInbox.setString(10, limit(event.getCommandValue(), 100));
+            insertInbox.setString(11, limit(event.getPayloadHash(), 64));
+            insertInbox.setString(12, limit(sourceFile, 512));
+            insertInbox.setLong(13, sourceLine);
+            insertInbox.setInt(14, event.getRetryCount());
+            try {
+                insertInbox.executeUpdate();
+            } catch (SQLException ex) {
+                if (ex.getErrorCode() == 1) {
+                    connection.rollback();
+                    checkDuplicate = connection.prepareStatement("SELECT PAYLOAD_HASH "
+                            + "FROM RESULT_REQUEST_INBOX WHERE EVENT_ID=?");
+                    checkDuplicate.setString(1, event.getEventId());
+                    duplicateRs = checkDuplicate.executeQuery();
+                    if (duplicateRs.next()
+                            && event.getPayloadHash().equals(duplicateRs.getString(1))) {
+                        logger.info("Skip duplicate resultRequest eventId=" + event.getEventId());
+                        return 1;
+                    }
+                    throw new SQLException("EVENT_ID collision with different payload: "
+                            + event.getEventId(), ex);
+                }
+                throw ex;
+            }
+
+            String selectSql = "SELECT REGISTER_ID FROM REG_INFO "
+                    + "WHERE MSISDN = ? AND PRODUCT_NAME = ? AND RENEW = 1 AND END_TIME IS NULL "
+                    + "ORDER BY EXPIRE_TIME DESC FOR UPDATE";
+            selectRegister = connection.prepareStatement(selectSql);
+            setQueryTimeout(selectRegister);
+            selectRegister.setString(1, event.getMsisdn());
+            selectRegister.setString(2, event.getProductName());
+            registerRs = selectRegister.executeQuery();
+            if (!registerRs.next()) {
+                updateInbox = connection.prepareStatement("UPDATE RESULT_REQUEST_INBOX SET "
+                        + "STATUS='SKIPPED', PROCESSED_TIME=SYSTIMESTAMP, UPDATED_TIME=SYSTIMESTAMP "
+                        + "WHERE EVENT_ID=?");
+                updateInbox.setString(1, event.getEventId());
+                updateInbox.executeUpdate();
+                connection.commit();
+                logger.warn("Renew but not registered: " + event.getMsisdn()
+                        + " package: " + event.getProductName());
+                return 2;
+            }
+            long registerId = registerRs.getLong(1);
+
+            updateRegister = connection.prepareStatement("UPDATE REG_INFO SET EXPIRE_TIME=?, "
+                    + "NUMBER_SPIN=?, EXTEND_STATUS=0, STATUS=1, PLAYED_TIMES=0, LAST_EXTEND=? "
+                    + "WHERE REGISTER_ID=?");
+            setQueryTimeout(updateRegister);
+            updateRegister.setTimestamp(1, expireTime);
+            updateRegister.setInt(2, event.getProductNumberSpin());
+            updateRegister.setTimestamp(3, new Timestamp(event.getReceivedAt()));
+            updateRegister.setLong(4, registerId);
+            if (updateRegister.executeUpdate() != 1) {
+                throw new SQLException("REG_INFO renew affected unexpected row count for registerId=" + registerId);
+            }
+
+            if (event.getProductFee() > 0) {
+                selectChargeId = connection.prepareStatement("SELECT CHARGE_LOG_SEQ.NEXTVAL FROM DUAL");
+                chargeIdRs = selectChargeId.executeQuery();
+                if (!chargeIdRs.next()) {
+                    throw new SQLException("Cannot allocate CHARGE_LOG sequence");
+                }
+                chargeLogId = chargeIdRs.getLong(1);
+                insertCharge = connection.prepareStatement("INSERT INTO CHARGE_LOG "
+                        + "(ID, MSISDN, FEE, CHARGE_TIME, INSERT_TIME, DESCRIPTION, ACCOUNT_ID, STATUS) "
+                        + "VALUES (?, ?, ?, ?, SYSDATE, ?, ?, ?)");
+                setQueryTimeout(insertCharge);
+                insertCharge.setLong(1, chargeLogId);
+                insertCharge.setString(2, event.getMsisdn());
+                insertCharge.setDouble(3, event.getProductFee());
+                insertCharge.setTimestamp(4, new Timestamp(event.getReceivedAt()));
+                insertCharge.setString(5, "Renew " + event.getProductName());
+                insertCharge.setNull(6, Types.NUMERIC);
+                insertCharge.setInt(7, Common.Constant.RENEW_STATUS);
+                if (insertCharge.executeUpdate() != 1) {
+                    throw new SQLException("CHARGE_LOG insert affected unexpected row count");
+                }
+            }
+
+            updateInbox = connection.prepareStatement("UPDATE RESULT_REQUEST_INBOX SET "
+                    + "STATUS='PROCESSED', CHARGE_LOG_ID=?, PROCESSED_TIME=SYSTIMESTAMP, "
+                    + "UPDATED_TIME=SYSTIMESTAMP WHERE EVENT_ID=?");
+            if (chargeLogId > 0) {
+                updateInbox.setLong(1, chargeLogId);
+            } else {
+                updateInbox.setNull(1, Types.NUMERIC);
+            }
+            updateInbox.setString(2, event.getEventId());
+            if (updateInbox.executeUpdate() != 1) {
+                throw new SQLException("RESULT_REQUEST_INBOX final update affected unexpected row count");
+            }
+            connection.commit();
+            return 0;
+        } catch (SQLException ex) {
+            if (connection != null) {
+                try {
+                    connection.rollback();
+                } catch (SQLException rollbackEx) {
+                    logger.error("Cannot rollback resultRequest transaction", rollbackEx);
+                }
+            }
+            throw ex;
+        } finally {
+            closeResultSet(chargeIdRs);
+            closeResultSet(registerRs);
+            closeResultSet(duplicateRs);
+            closeStatement(updateInbox);
+            closeStatement(insertCharge);
+            closeStatement(selectChargeId);
+            closeStatement(updateRegister);
+            closeStatement(selectRegister);
+            closeStatement(insertInbox);
+            closeStatement(checkDuplicate);
+            if (connection != null) {
+                try {
+                    connection.setAutoCommit(oldAutoCommit);
+                } catch (SQLException ex) {
+                    logger.warn("Cannot restore autoCommit after resultRequest: " + ex.getMessage());
+                }
+            }
+            closeConnection(connection);
+        }
+    }
+
+    private void setQueryTimeout(PreparedStatement statement) throws SQLException {
+        if (WebserviceManager.enableQueryDbTimeout && WebserviceManager.queryDbTimeout > 0) {
+            statement.setQueryTimeout(WebserviceManager.queryDbTimeout);
+        }
+    }
+
+    private String limit(String value, int maxLength) {
+        if (value == null) {
+            return null;
+        }
+        return value.length() <= maxLength ? value : value.substring(0, maxLength);
+    }
+
     /**
      * Close Statement
      *

KittyFallWs/KittyFallWs/src/com/vas/kittyfallwsvas/wsfw/resultrequest/ResultRequestEvent.java

@@ -0,0 +1,111 @@
+package com.vas.kittyfallwsvas.wsfw.resultrequest;
+
+import com.vas.kittyfallwsvas.wsfw.obj.ProductInfo;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+
+/**
+ * Durable snapshot of a resultRequest call. Credentials are deliberately not
+ * stored in the spool file.
+ */
+public class ResultRequestEvent {
+
+    private int schemaVersion = 1;
+    private String eventId;
+    private String payloadHash;
+    private long receivedAt;
+    private String transactionId;
+    private String serviceId;
+    private String msisdn;
+    private String chargeTimeRaw;
+    private String paramsValue;
+    private String modeValue;
+    private String amountValue;
+    private String commandValue;
+    private String productName;
+    private double productFee;
+    private int productNumberSpin;
+    private int productExpireDays;
+    private int retryCount;
+
+    public ResultRequestEvent() {
+    }
+
+    public static ResultRequestEvent create(long receivedAt, String transactionId,
+            String serviceId, String msisdn, String chargeTimeRaw,
+            String paramsValue, String modeValue, String amountValue,
+            String commandValue, ProductInfo product) {
+        ResultRequestEvent event = new ResultRequestEvent();
+        event.receivedAt = receivedAt;
+        event.transactionId = trim(transactionId);
+        event.serviceId = trim(serviceId);
+        event.msisdn = trim(msisdn);
+        event.chargeTimeRaw = trim(chargeTimeRaw);
+        event.paramsValue = trim(paramsValue);
+        event.modeValue = trim(modeValue);
+        event.amountValue = trim(amountValue);
+        event.commandValue = trim(commandValue);
+        event.productName = product.getProductName();
+        event.productFee = product.getFee();
+        event.productNumberSpin = product.getNumberSpin();
+        event.productExpireDays = product.getExpireDays();
+
+        String payload = join(event.serviceId, event.msisdn, event.chargeTimeRaw,
+                event.paramsValue, event.modeValue, event.amountValue,
+                event.commandValue, event.productName);
+        event.payloadHash = sha256(payload);
+        String identity = event.transactionId.length() > 0
+                ? join(event.serviceId, event.transactionId)
+                : payload;
+        event.eventId = sha256(identity);
+        return event;
+    }
+
+    private static String trim(String value) {
+        return value == null ? "" : value.trim();
+    }
+
+    private static String join(String... values) {
+        StringBuilder value = new StringBuilder();
+        for (String item : values) {
+            if (value.length() > 0) {
+                value.append('|');
+            }
+            value.append(item == null ? "" : item);
+        }
+        return value.toString();
+    }
+
+    private static String sha256(String value) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            byte[] bytes = digest.digest(value.getBytes(StandardCharsets.UTF_8));
+            StringBuilder result = new StringBuilder(bytes.length * 2);
+            for (byte item : bytes) {
+                result.append(String.format("%02x", item & 0xff));
+            }
+            return result.toString();
+        } catch (Exception ex) {
+            throw new IllegalStateException("SHA-256 is not available", ex);
+        }
+    }
+
+    public int getSchemaVersion() { return schemaVersion; }
+    public String getEventId() { return eventId; }
+    public String getPayloadHash() { return payloadHash; }
+    public long getReceivedAt() { return receivedAt; }
+    public String getTransactionId() { return transactionId; }
+    public String getServiceId() { return serviceId; }
+    public String getMsisdn() { return msisdn; }
+    public String getChargeTimeRaw() { return chargeTimeRaw; }
+    public String getParamsValue() { return paramsValue; }
+    public String getModeValue() { return modeValue; }
+    public String getAmountValue() { return amountValue; }
+    public String getCommandValue() { return commandValue; }
+    public String getProductName() { return productName; }
+    public double getProductFee() { return productFee; }
+    public int getProductNumberSpin() { return productNumberSpin; }
+    public int getProductExpireDays() { return productExpireDays; }
+    public int getRetryCount() { return retryCount; }
+    public void incrementRetryCount() { retryCount++; }
+}

KittyFallWs/KittyFallWs/src/com/vas/kittyfallwsvas/wsfw/resultrequest/ResultRequestSpoolService.java

@@ -0,0 +1,370 @@
+package com.vas.kittyfallwsvas.wsfw.resultrequest;
+
+import com.google.gson.Gson;
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.AtomicMoveNotSupportedException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.log4j.Logger;
+import utils.Config;
+
+/**
+ * Persists accepted requests before replying and drains them to the database
+ * with a single, rate-limited worker.
+ */
+public class ResultRequestSpoolService {
+
+    public interface Processor {
+        void process(ResultRequestEvent event, String sourceFile, long sourceLine) throws Exception;
+    }
+
+    private static class Pending {
+        private final ResultRequestEvent event;
+        private final CompletableFuture<Void> durable = new CompletableFuture<Void>();
+
+        Pending(ResultRequestEvent event) {
+            this.event = event;
+        }
+    }
+
+    private final Logger logger;
+    private final Processor processor;
+    private final Gson gson = new Gson();
+    private final boolean enabled;
+    private final Path root;
+    private final Path activeDir;
+    private final Path readyDir;
+    private final Path processingDir;
+    private final Path doneDir;
+    private final Path deadDir;
+    private final Path activeFile;
+    private final ArrayBlockingQueue<Pending> queue;
+    private final int batchSize;
+    private final long enqueueTimeoutMs;
+    private final long writeTimeoutMs;
+    private final int maxRetry;
+    private final long rateDelayMs;
+    private final long intervalSeconds;
+    private final Object activeLock = new Object();
+    private final AtomicLong sequence = new AtomicLong();
+    private volatile boolean running = true;
+    private Thread writerThread;
+    private ScheduledExecutorService scheduler;
+
+    public static ResultRequestSpoolService create(Logger logger, Processor processor) throws Exception {
+        Properties properties = new Properties();
+        File configFile = new File(Config.configDir, "app.conf");
+        FileReader reader = new FileReader(configFile);
+        try {
+            properties.load(reader);
+        } finally {
+            reader.close();
+        }
+        return new ResultRequestSpoolService(logger, processor, properties);
+    }
+
+    ResultRequestSpoolService(Logger logger, Processor processor, Properties properties) throws Exception {
+        this.logger = logger;
+        this.processor = processor;
+        this.enabled = "SPOOL".equalsIgnoreCase(get(properties, "RESULT_REQUEST_MODE", "DIRECT"));
+        this.root = new File(get(properties, "RESULT_REQUEST_SPOOL_DIR", "../data/result-request-spool")).toPath();
+        this.activeDir = root.resolve("active");
+        this.readyDir = root.resolve("ready");
+        this.processingDir = root.resolve("processing");
+        this.doneDir = root.resolve("done");
+        this.deadDir = root.resolve("dead-letter");
+        this.activeFile = activeDir.resolve("result-request.open");
+        this.batchSize = positiveInt(properties, "RESULT_REQUEST_WRITE_BATCH_SIZE", 100);
+        this.enqueueTimeoutMs = positiveLong(properties, "RESULT_REQUEST_ENQUEUE_TIMEOUT_MS", 2000L);
+        this.writeTimeoutMs = positiveLong(properties, "RESULT_REQUEST_WRITE_TIMEOUT_MS", 10000L);
+        this.maxRetry = positiveInt(properties, "RESULT_REQUEST_MAX_RETRY", 20);
+        this.intervalSeconds = positiveLong(properties, "RESULT_REQUEST_INTERVAL_SECONDS", 300L);
+        int rate = positiveInt(properties, "RESULT_REQUEST_RATE_LIMIT_PER_SECOND", 10);
+        this.rateDelayMs = Math.max(1L, 1000L / rate);
+        this.queue = new ArrayBlockingQueue<Pending>(positiveInt(properties, "RESULT_REQUEST_QUEUE_CAPACITY", 10000));
+
+        if (enabled) {
+            Files.createDirectories(activeDir);
+            Files.createDirectories(readyDir);
+            Files.createDirectories(processingDir);
+            Files.createDirectories(doneDir);
+            Files.createDirectories(deadDir);
+            recoverFiles();
+            startWriter();
+            startProcessor();
+            logger.info("resultRequest SPOOL enabled. directory=" + root.toAbsolutePath()
+                    + ", intervalSeconds=" + intervalSeconds + ", dbRate=" + rate + "/s");
+        } else {
+            logger.info("resultRequest mode is DIRECT (spool worker is not started)");
+        }
+    }
+
+    public boolean isEnabled() {
+        return enabled;
+    }
+
+    public void enqueue(ResultRequestEvent event) throws Exception {
+        if (!enabled) {
+            throw new IllegalStateException("resultRequest spool is disabled");
+        }
+        Pending pending = new Pending(event);
+        if (!queue.offer(pending, enqueueTimeoutMs, TimeUnit.MILLISECONDS)) {
+            throw new IOException("resultRequest spool queue is full");
+        }
+        try {
+            pending.durable.get(writeTimeoutMs, TimeUnit.MILLISECONDS);
+        } catch (TimeoutException ex) {
+            throw new IOException("Timed out waiting for durable spool write", ex);
+        }
+    }
+
+    private void startWriter() {
+        writerThread = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                writerLoop();
+            }
+        }, "result-request-spool-writer");
+        writerThread.setDaemon(true);
+        writerThread.start();
+    }
+
+    private void writerLoop() {
+        while (running) {
+            List<Pending> batch = new ArrayList<Pending>(batchSize);
+            try {
+                Pending first = queue.take();
+                batch.add(first);
+                queue.drainTo(batch, batchSize - 1);
+                writeBatch(batch);
+                for (Pending pending : batch) {
+                    pending.durable.complete(null);
+                }
+            } catch (InterruptedException ex) {
+                Thread.currentThread().interrupt();
+                return;
+            } catch (Exception ex) {
+                logger.error("Cannot write resultRequest spool batch", ex);
+                for (Pending pending : batch) {
+                    pending.durable.completeExceptionally(ex);
+                }
+            }
+        }
+    }
+
+    private void writeBatch(List<Pending> batch) throws IOException {
+        StringBuilder content = new StringBuilder();
+        for (Pending pending : batch) {
+            content.append(gson.toJson(pending.event)).append('\n');
+        }
+        byte[] bytes = content.toString().getBytes(StandardCharsets.UTF_8);
+        synchronized (activeLock) {
+            FileChannel channel = FileChannel.open(activeFile, StandardOpenOption.CREATE,
+                    StandardOpenOption.WRITE, StandardOpenOption.APPEND);
+            try {
+                ByteBuffer buffer = ByteBuffer.wrap(bytes);
+                while (buffer.hasRemaining()) {
+                    channel.write(buffer);
+                }
+                channel.force(true);
+            } finally {
+                channel.close();
+            }
+        }
+    }
+
+    private void startProcessor() {
+        scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+            @Override
+            public Thread newThread(Runnable task) {
+                Thread thread = new Thread(task, "result-request-db-worker");
+                thread.setDaemon(true);
+                return thread;
+            }
+        });
+        scheduler.scheduleWithFixedDelay(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    drainOnce();
+                } catch (Exception ex) {
+                    logger.error("resultRequest spool drain error", ex);
+                }
+            }
+        }, intervalSeconds, intervalSeconds, TimeUnit.SECONDS);
+    }
+
+    private void drainOnce() throws Exception {
+        recoverProcessingFiles();
+        rotateActiveFile();
+        List<Path> files = listFiles(readyDir, "*.ready");
+        for (Path ready : files) {
+            Path claimed = processingDir.resolve(ready.getFileName().toString().replace(".ready", ".processing"));
+            move(ready, claimed);
+            processFile(claimed);
+        }
+    }
+
+    private void rotateActiveFile() throws IOException {
+        synchronized (activeLock) {
+            if (!Files.exists(activeFile) || Files.size(activeFile) == 0) {
+                return;
+            }
+            Path ready = readyDir.resolve(fileName("batch", ".ready"));
+            move(activeFile, ready);
+        }
+    }
+
+    private void processFile(Path file) throws IOException {
+        BufferedReader reader = Files.newBufferedReader(file, StandardCharsets.UTF_8);
+        long lineNumber = 0;
+        try {
+            String line;
+            while ((line = reader.readLine()) != null) {
+                lineNumber++;
+                if (line.trim().length() == 0) {
+                    continue;
+                }
+                ResultRequestEvent event;
+                try {
+                    event = gson.fromJson(line, ResultRequestEvent.class);
+                    if (event == null || event.getEventId() == null) {
+                        throw new IllegalArgumentException("Missing eventId");
+                    }
+                } catch (Exception ex) {
+                    logger.error("Invalid resultRequest spool JSON " + file + ":" + lineNumber, ex);
+                    writeDeadLine(line, file.getFileName().toString(), lineNumber);
+                    continue;
+                }
+                try {
+                    processor.process(event, file.getFileName().toString(), lineNumber);
+                } catch (Exception ex) {
+                    logger.error("resultRequest DB processing failed eventId=" + event.getEventId(), ex);
+                    event.incrementRetryCount();
+                    if (event.getRetryCount() > maxRetry) {
+                        writeEvent(deadDir, "failed", ".dead", event);
+                    } else {
+                        writeEvent(readyDir, "retry", ".ready", event);
+                    }
+                }
+                try {
+                    Thread.sleep(rateDelayMs);
+                } catch (InterruptedException ex) {
+                    Thread.currentThread().interrupt();
+                    throw new IOException("Spool processor interrupted", ex);
+                }
+            }
+        } finally {
+            reader.close();
+        }
+        move(file, doneDir.resolve(file.getFileName().toString().replace(".processing", ".done")));
+    }
+
+    private void recoverFiles() throws IOException {
+        if (Files.exists(activeFile) && Files.size(activeFile) > 0) {
+            move(activeFile, readyDir.resolve(fileName("recovered", ".ready")));
+        }
+        recoverProcessingFiles();
+    }
+
+    private void recoverProcessingFiles() throws IOException {
+        for (Path processing : listFiles(processingDir, "*.processing")) {
+            move(processing, readyDir.resolve(fileName("recovered-processing", ".ready")));
+        }
+    }
+
+    private void writeEvent(Path directory, String prefix, String extension, ResultRequestEvent event) throws IOException {
+        writeAtomic(directory.resolve(fileName(prefix, extension)), gson.toJson(event) + "\n");
+    }
+
+    private void writeDeadLine(String line, String source, long lineNumber) throws IOException {
+        String value = "{\"source\":\"" + source.replace("\"", "") + "\",\"line\":"
+                + lineNumber + ",\"raw\":" + gson.toJson(line) + "}\n";
+        writeAtomic(deadDir.resolve(fileName("invalid", ".dead")), value);
+    }
+
+    private void writeAtomic(Path target, String content) throws IOException {
+        Path temp = target.resolveSibling(target.getFileName().toString() + ".tmp");
+        FileChannel channel = FileChannel.open(temp, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
+        try {
+            ByteBuffer buffer = ByteBuffer.wrap(content.getBytes(StandardCharsets.UTF_8));
+            while (buffer.hasRemaining()) {
+                channel.write(buffer);
+            }
+            channel.force(true);
+        } finally {
+            channel.close();
+        }
+        move(temp, target);
+    }
+
+    private List<Path> listFiles(Path directory, String glob) throws IOException {
+        List<Path> result = new ArrayList<Path>();
+        DirectoryStream<Path> stream = Files.newDirectoryStream(directory, glob);
+        try {
+            for (Path file : stream) {
+                result.add(file);
+            }
+        } finally {
+            stream.close();
+        }
+        java.util.Collections.sort(result);
+        return result;
+    }
+
+    private void move(Path source, Path target) throws IOException {
+        try {
+            Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);
+        } catch (AtomicMoveNotSupportedException ex) {
+            Files.move(source, target);
+        }
+    }
+
+    private String fileName(String prefix, String extension) {
+        return prefix + "-" + System.currentTimeMillis() + "-" + sequence.incrementAndGet() + extension;
+    }
+
+    private static String get(Properties properties, String name, String defaultValue) {
+        String value = properties.getProperty(name);
+        return value == null || value.trim().length() == 0 ? defaultValue : value.trim();
+    }
+
+    private static int positiveInt(Properties properties, String name, int defaultValue) {
+        try {
+            int value = Integer.parseInt(get(properties, name, String.valueOf(defaultValue)));
+            return value > 0 ? value : defaultValue;
+        } catch (Exception ex) {
+            return defaultValue;
+        }
+    }
+
+    private static long positiveLong(Properties properties, String name, long defaultValue) {
+        try {
+            long value = Long.parseLong(get(properties, name, String.valueOf(defaultValue)));
+            return value > 0 ? value : defaultValue;
+        } catch (Exception ex) {
+            return defaultValue;
+        }
+    }
+}

KittyFallWs/KittyFallWs/src/com/vas/webservices/KittyFallWs.java

@@ -22,6 +22,8 @@ import com.vas.kittyfallwsvas.wsfw.common.MessageResponse;
 import com.vas.kittyfallwsvas.wsfw.common.WebserviceAbstract;
 import com.vas.kittyfallwsvas.wsfw.common.WebserviceManager;
 import com.vas.kittyfallwsvas.wsfw.database.WsProcessUtils;
+import com.vas.kittyfallwsvas.wsfw.resultrequest.ResultRequestEvent;
+import com.vas.kittyfallwsvas.wsfw.resultrequest.ResultRequestSpoolService;
 import com.vas.kittyfallwsvas.wsfw.obj.draw.LuckySprin;
 import com.vas.kittyfallwsvas.wsfw.obj.draw.PrizeObj;
 import com.vas.kittyfallwsvas.wsfw.obj.draw.PrizeWinner;
@@ -62,6 +64,7 @@ public class KittyFallWs extends WebserviceAbstract {
     private SimpleDateFormat fullDf = new SimpleDateFormat("yyyyMMddHHmmss");
     private SimpleDateFormat reqDf = new SimpleDateFormat("yyyyMMddHHmmssSSS");
     public WSProcessor ws;
+    private final ResultRequestSpoolService resultRequestSpool;
     
     @Resource
     protected WebServiceContext ctx;
@@ -79,6 +82,23 @@ public class KittyFallWs extends WebserviceAbstract {
             Common.mapMpsConfig = db.loadMpsConfig();
             Common.loadConfig();
         }
+        resultRequestSpool = ResultRequestSpoolService.create(logger,
+                new ResultRequestSpoolService.Processor() {
+            @Override
+            public void process(ResultRequestEvent event, String sourceFile, long sourceLine) throws Exception {
+                processDurableResultRequest(event, sourceFile, sourceLine);
+            }
+        });
+    }
+
+    private void processDurableResultRequest(ResultRequestEvent event,
+            String sourceFile, long sourceLine) throws Exception {
+        Timestamp expireTime = getExpireTime(event.getReceivedAt(), event.getProductExpireDays());
+        int result = db.processResultRequestRenew(event, expireTime, sourceFile, sourceLine);
+        if (result == 0) {
+            logger.info("Durable resultRequest processed eventId=" + event.getEventId()
+                    + ", msisdn=" + event.getMsisdn() + ", package=" + event.getProductName());
+        }
     }
 
     //
@@ -712,8 +732,8 @@ public class KittyFallWs extends WebserviceAbstract {
         try {
             String ip = getIpClient();
 //            UserInfo userInfo = null;
-            br.setLength(0);
-            br.append("receiveresult Params:").
+            StringBuilder requestLog = new StringBuilder();
+            requestLog.append("receiveresult Params:").
                     append("\nmsisdn:").append(msisdn).
                     append("\nparams:").append(params).
                     append("\nchargetime:").append(chargetime).
@@ -722,7 +742,7 @@ public class KittyFallWs extends WebserviceAbstract {
                     append("\ntransactionId:").append(transactionId).
                     append("\ncommand:").append(command).
                     append("\nmode:").append(mode);
-            logger.info(br);
+            logger.info(requestLog);
 
             Request request = new Request();
             request.setMsisdn(msisdn);
@@ -756,9 +776,17 @@ public class KittyFallWs extends WebserviceAbstract {
                     return response.getErrorCode();
                 }
                 if (params.equals("0")) {
-                    renewMps(msisdn, productInfo);
-                    // return
-                    logger.info("Renew account success: " + msisdn);
+                    if (resultRequestSpool.isEnabled()) {
+                        ResultRequestEvent event = ResultRequestEvent.create(reqTime.getTime(),
+                                transactionId, serviceid, msisdn, chargetime, params,
+                                mode, fee, command, productInfo);
+                        resultRequestSpool.enqueue(event);
+                        logger.info("resultRequest durably queued eventId=" + event.getEventId()
+                                + ", msisdn=" + msisdn);
+                    } else {
+                        renewMps(msisdn, productInfo);
+                        logger.info("Renew account success (DIRECT): " + msisdn);
+                    }
                     response.setErrorCode(Common.ErrorCode.SUCCESS);
                     response.setContent(Common.ResultCode.SUCCESS);
                     return response.getErrorCode();
@@ -3149,12 +3177,17 @@ public class KittyFallWs extends WebserviceAbstract {
     }
 
     private Timestamp getExpireTime(ProductInfo productInfo) {
+        return getExpireTime(System.currentTimeMillis(), productInfo.getExpireDays());
+    }
+
+    private Timestamp getExpireTime(long baseTime, int expireDays) {
         Calendar cal = Calendar.getInstance();
+        cal.setTimeInMillis(baseTime);
         cal.set(Calendar.HOUR_OF_DAY, 23);
         cal.set(Calendar.MINUTE, 59);
         cal.set(Calendar.SECOND, 59);
         cal.set(Calendar.MILLISECOND, 0);
-        cal.add(Calendar.DAY_OF_MONTH, productInfo.getExpireDays() - 1);
+        cal.add(Calendar.DAY_OF_MONTH, expireDays - 1);
         return new Timestamp(cal.getTimeInMillis());
     }
 

KittyFallWs/etc/app.conf

@@ -16,4 +16,18 @@ LOG_DATABASE_CONF=../etc/database.xml
 #if true, config database id to insert log
 #LOG_TO_DATABASE_ID=dbProcess
 LOG_TO_DATABASE_ID=dbProcess
-NUM_THREAD_LOG_TRANS = 1
+NUM_THREAD_LOG_TRANS = 1
+
+##################################################
+# resultRequest durable spool
+# Deploy code with DIRECT first. After running the inbox SQL and preparing the
+# spool disk, change to SPOOL and restart KittyFallWs.
+RESULT_REQUEST_MODE=DIRECT
+RESULT_REQUEST_SPOOL_DIR=../data/result-request-spool
+RESULT_REQUEST_INTERVAL_SECONDS=300
+RESULT_REQUEST_QUEUE_CAPACITY=10000
+RESULT_REQUEST_WRITE_BATCH_SIZE=100
+RESULT_REQUEST_ENQUEUE_TIMEOUT_MS=2000
+RESULT_REQUEST_WRITE_TIMEOUT_MS=10000
+RESULT_REQUEST_MAX_RETRY=20
+RESULT_REQUEST_RATE_LIMIT_PER_SECOND=10