Subversion Repositories SmartDukaan

package com.spice.profitmandi.service.cron;

import com.spice.profitmandi.dao.entity.transaction.CronBatch;

import com.spice.profitmandi.dao.entity.transaction.CronBatchItem;

import com.spice.profitmandi.dao.enumuration.transaction.CronBatchItemStatus;

import com.spice.profitmandi.dao.enumuration.transaction.CronBatchStatus;

import com.spice.profitmandi.dao.repository.transaction.CronBatchItemRepository;

import com.spice.profitmandi.dao.repository.transaction.CronBatchRepository;

import com.spice.profitmandi.service.mail.MailOutboxService;

import org.apache.logging.log4j.LogManager;

import org.apache.logging.log4j.Logger;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import org.springframework.transaction.annotation.Propagation;

import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.List;

import java.util.Map;

import java.util.stream.Collectors;

@Service

public class CronBatchService {

    private static final Logger LOGGER = LogManager.getLogger(CronBatchService.class);

    private static final String[] TECHNOLOGY_EMAIL = {"amit.gupta@smartdukaan.com"};

    private static final DateTimeFormatter DTF = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    /** Cap on how many never-attempted items the alert lists by name. */

    private static final int PENDING_LIST_CAP = 50;

    @Autowired

    private CronBatchRepository cronBatchRepository;

    @Autowired

    private CronBatchItemRepository cronBatchItemRepository;

    @Autowired

    private MailOutboxService mailOutboxService;

    /**

     * Read-only guard used by OfferBatchService before creating a new batch.

     * Returns any RUNNING batch for this offer (sellin or activation job name),

     * or null if none. Wrapped in a read-only tx so callers running outside an

     * existing transaction (e.g. background worker thread) can query safely.

     */

    @Transactional(readOnly = true)

    public CronBatch findRunningForOffer(int offerId) {

        List<CronBatch> running = cronBatchRepository.selectRunningForOffer(offerId);

        return running.isEmpty() ? null : running.get(0);

    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)

    public CronBatch createBatch(String jobName, Map<Integer, String> fofoIdPartnerNameMap) {

        CronBatch batch = new CronBatch(jobName);

        batch.setTotalCount(fofoIdPartnerNameMap.size());

        cronBatchRepository.persist(batch);

        for (Map.Entry<Integer, String> entry : fofoIdPartnerNameMap.entrySet()) {

            CronBatchItem item = new CronBatchItem(batch.getId(), entry.getKey(), entry.getValue());

            cronBatchItemRepository.persist(item);

        }

        LOGGER.info("Created batch {} for job {} with {} items", batch.getId(), jobName, fofoIdPartnerNameMap.size());

        return batch;

    }

    /**

     * REQUIRED — deliberately NOT REQUIRES_NEW.

     *

     * Every caller is the last statement of a REQUIRES_NEW per-item work method, so this

     * joins that work transaction and the SUCCESS marker commits (or rolls back) with the

     * work it describes. Under REQUIRES_NEW the marker committed on its own, so work that

     * later failed at flush/commit — a constraint violation, a wallet deadlock — was left

     * recorded as a successful payout.

     */

    @Transactional(propagation = Propagation.REQUIRED)

    public void markItemSuccess(int batchId, int fofoId) {

        CronBatchItem item = cronBatchItemRepository.selectByBatchIdAndFofoId(batchId, fofoId);

        if (item != null && item.getStatus() == CronBatchItemStatus.PENDING) {

            item.markSuccess();

        }

    }

    /**

     * REQUIRES_NEW — the opposite of markItemSuccess, and intentionally so: this is called

     * from the orchestrator's catch block after the work transaction has already rolled

     * back, so the failure record must commit independently of it.

     */

    @Transactional(propagation = Propagation.REQUIRES_NEW)

    public void markItemFailed(int batchId, int fofoId, String errorMessage) {

        CronBatchItem item = cronBatchItemRepository.selectByBatchIdAndFofoId(batchId, fofoId);

        if (item != null && item.getStatus() == CronBatchItemStatus.PENDING) {

            item.markFailed(errorMessage);

        }

    }

    /**

     * RUNNING batches older than {@code staleHours} — a run that died before finalizing.

     * Returns ids only; the caller must finalize each through the proxy (see

     * BatchScheduledTasks.reapStaleBatches) so REQUIRES_NEW actually applies.

     */

    @Transactional(readOnly = true)

    public List<Integer> selectStaleRunningBatchIds(int staleHours) {

        return cronBatchRepository.selectStaleRunning(LocalDateTime.now().minusHours(staleHours))

                .stream().map(CronBatch::getId).collect(Collectors.toList());

    }

    /**

     * Closes a batch using the item rows as the source of truth.

     *

     * successCount is COUNTED, never derived as (total - failures). Deriving it reported

     * every item that never reported at all — the JVM killed mid-loop, a lost marker write —

     * as a successful payout, and suppressed the alert, because failureCount stayed 0.

     *

     * Items still PENDING were never attempted. That leaves the batch INCOMPLETE (not

     * PARTIAL_FAILURE, which means "attempted and failed") and always raises an email.

     * PENDING rows are left as they are: "never tried" and "tried and failed" are

     * different facts and the retry decision depends on which one it was.

     */

    @Transactional(propagation = Propagation.REQUIRES_NEW)

    public BatchOutcome finalizeBatch(int batchId) {

        CronBatch batch = cronBatchRepository.selectById(batchId);

        if (batch == null) {

            LOGGER.warn("finalizeBatch called for unknown batch {}", batchId);

            return null;

        }

        List<CronBatchItem> failedItems = cronBatchItemRepository.selectFailedByBatchId(batchId);

        int failureCount = failedItems.size();

        int successCount = (int) cronBatchItemRepository

                .countByBatchIdAndStatus(batchId, CronBatchItemStatus.SUCCESS);

        int pendingCount = batch.getTotalCount() - successCount - failureCount;

        batch.setSuccessCount(successCount);

        batch.setFailureCount(failureCount);

        batch.setCompletedAt(LocalDateTime.now());

        if (pendingCount > 0) {

            batch.setStatus(CronBatchStatus.INCOMPLETE);

        } else if (failureCount > 0) {

            batch.setStatus(CronBatchStatus.PARTIAL_FAILURE);

        } else {

            batch.setStatus(CronBatchStatus.COMPLETED);

        }

        LOGGER.info("Batch {} finalized as {}: {} success, {} failed, {} never attempted",

                batchId, batch.getStatus(), successCount, failureCount, pendingCount);

        if (failureCount > 0 || pendingCount > 0) {

            sendFailureEmail(batch, failedItems, pendingCount);

        }

        return new BatchOutcome(batch.getStatus(), batch.getTotalCount(),

                successCount, failureCount, pendingCount);

    }

    /**

     * Result of a finalize, returned rather than re-read by the caller: finalizeBatch

     * commits in its own session, so a caller that re-reads inside its own transaction

     * gets the stale pre-finalize entity back out of Hibernate's first-level cache.

     */

    public static class BatchOutcome {

        private final CronBatchStatus status;

        private final int totalCount;

        private final int successCount;

        private final int failureCount;

        private final int neverAttemptedCount;

        BatchOutcome(CronBatchStatus status, int totalCount, int successCount,

                     int failureCount, int neverAttemptedCount) {

            this.status = status;

            this.totalCount = totalCount;

            this.successCount = successCount;

            this.failureCount = failureCount;

            this.neverAttemptedCount = neverAttemptedCount;

        }

        public CronBatchStatus getStatus() { return status; }

        public int getTotalCount() { return totalCount; }

        public int getSuccessCount() { return successCount; }

        public int getFailureCount() { return failureCount; }

        public int getNeverAttemptedCount() { return neverAttemptedCount; }

    }

    private void sendFailureEmail(CronBatch batch, List<CronBatchItem> failedItems, int pendingCount) {

        StringBuilder body = new StringBuilder();

        body.append("Cron job: ").append(batch.getJobName()).append("\n");

        body.append("Batch id: ").append(batch.getId()).append("\n");

        body.append("Run time: ").append(batch.getStartedAt().format(DTF)).append("\n");

        body.append("Outcome: ").append(batch.getStatus()).append("\n");

        body.append("Total: ").append(batch.getTotalCount()).append("\n");

        body.append("Success: ").append(batch.getSuccessCount()).append("\n");

        body.append("Failed: ").append(batch.getFailureCount()).append("\n");

        body.append("Never attempted: ").append(pendingCount).append("\n\n");

        if (!failedItems.isEmpty()) {

            body.append("Partner failures:\n");

            for (CronBatchItem item : failedItems) {

                body.append("- ").append(item.getPartnerName())

                        .append(" (fofoId: ").append(item.getFofoId()).append(")")

                        .append(" — ").append(item.getErrorMessage())

                        .append("\n");

            }

            body.append("\n");

        }

        if (pendingCount > 0) {

            body.append("The run did not reach these ").append(pendingCount)

                    .append(" item(s) — they were never attempted and no work was done for them:\n");

            List<CronBatchItem> pendingItems = cronBatchItemRepository

                    .selectByBatchIdAndStatus(batch.getId(), CronBatchItemStatus.PENDING);

            for (CronBatchItem item : pendingItems.subList(0, Math.min(pendingItems.size(), PENDING_LIST_CAP))) {

                body.append("- ").append(item.getPartnerName())

                        .append(" (fofoId: ").append(item.getFofoId()).append(")\n");

            }

            if (pendingItems.size() > PENDING_LIST_CAP) {

                body.append("... and ").append(pendingItems.size() - PENDING_LIST_CAP).append(" more\n");

            }

            body.append("\nRe-run the job to process them.\n");

        }

        String subject = pendingCount > 0

                ? String.format("[CRON ALERT] %s — INCOMPLETE, %d of %d never attempted",

                        batch.getJobName(), pendingCount, batch.getTotalCount())

                : String.format("[CRON ALERT] %s — %d of %d partners failed",

                        batch.getJobName(), batch.getFailureCount(), batch.getTotalCount());

        try {

            mailOutboxService.queueMailViaGoogle(TECHNOLOGY_EMAIL, null, subject, body.toString(),

                    "CronBatchService." + batch.getJobName());

        } catch (Exception e) {

            LOGGER.error("Failed to send batch failure email for batch {}: {}", batch.getId(), e.getMessage());

        }

    }

}