Subversion Repositories SmartDukaan

package com.spice.profitmandi.service.lms;

import com.spice.profitmandi.dao.entity.user.LeadCall;

import com.spice.profitmandi.dao.repository.dtr.LeadCallRepository;

import com.spice.profitmandi.service.storage.ObjectStoreService;

import org.apache.logging.log4j.LogManager;

import org.apache.logging.log4j.Logger;

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

import org.springframework.scheduling.annotation.Async;

import org.springframework.stereotype.Service;

import org.springframework.transaction.annotation.Propagation;

import org.springframework.transaction.annotation.Transactional;

import java.io.ByteArrayInputStream;

import java.io.ByteArrayOutputStream;

import java.io.InputStream;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

/**

 * Archives a single call recording into the object store.

 *

 * <p><b>Why this is its own bean.</b> Spring's {@code @Async} and {@code @Transactional} are applied

 * by a proxy around the bean, so a self-invocation from a sweep method in the same class would run

 * inline on the scheduler thread, in the scheduler's transaction — silently losing both annotations.

 * Keeping the per-item work in a separate injected bean is what makes them actually take effect.

 */

@Service

public class RecordingArchiveWorker {

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

    private static final DateTimeFormatter KEY_DATE = DateTimeFormatter.ofPattern("yyyy/MM/dd");

    /**

     * Guard against a runaway download. Vonage mp3 runs roughly 1 MB/minute, so this allows a very

     * long call while still refusing to buffer something pathological into heap.

     */

    private static final long MAX_RECORDING_BYTES = 200L * 1024 * 1024;

    @Autowired

    private LeadCallRepository leadCallRepository;

    @Autowired

    private ObjectStoreService objectStoreService;

    @Autowired

    private LmsDialerProvider dialerProvider;

    /** Separate bean on purpose — see its javadoc; a self-call here would lose REQUIRES_NEW. */

    @Autowired

    private RecordingArchiveFailureRecorder failureRecorder;

    /**

     * Fetch one recording from the provider and store it.

     *

     * <p>Runs in its own transaction so one bad recording cannot roll back a whole sweep. Failures

     * are logged and swallowed: the row keeps its null {@code recording_document_id}, stays in the

     * pending set, and the next sweep retries it. That is the retry mechanism — there is no queue to

     * lose across a restart.

     */

    @Async("recordingArchiveExecutor")

    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Throwable.class)

    public void archiveOne(int leadCallId) {

        LeadCall call = leadCallRepository.selectById(leadCallId);

        if (call == null || call.getRecordingUuid() == null || call.getRecordingObjectKey() != null) {

            return;

        }

        InputStream stream = null;

        try {

            stream = dialerProvider.fetchRecording(call);

            if (stream == null) {

                // Counted as an attempt: at the provider's retention boundary this becomes permanent,

                // and without the counter the row would be retried every ten minutes forever.

                LOGGER.warn("Provider has no recording for call {}", leadCallId);

                failureRecorder.recordFailure(leadCallId);

                return;

            }

            byte[] audio = readCapped(stream);

            if (audio.length == 0) {

                LOGGER.warn("Empty recording body for call {} — leaving it pending", leadCallId);

                failureRecorder.recordFailure(leadCallId);

                return;

            }

            String key = objectKey(call);

            objectStoreService.put(key, new ByteArrayInputStream(audio), audio.length, "audio/mpeg");

            // The key goes on the call row itself. It used to be written to dtr.document.path, which

            // is VARCHAR(30) against a ~58-character key — under STRICT_TRANS_TABLES that threw on

            // every single upload, and because failures here are swallowed to drive the retry, the

            // symptom was an endless loop that stored nothing while the provider window expired.

            call.setRecordingObjectKey(key);

            leadCallRepository.persist(call);

            LOGGER.info("Archived recording for call {} to {} ({} bytes)", leadCallId, key, audio.length);

        } catch (Exception e) {

            LOGGER.error("Could not archive the recording for call {} — will retry", leadCallId, e);

            failureRecorder.recordFailure(leadCallId);

        } finally {

            closeQuietly(stream);

        }

    }

    /**

     * {@code yyyy/MM/dd/lead-<leadId>/call-<callId>-<recordingUuid>.mp3}

     *

     * <p>Date-partitioned so a lifecycle rule can expire a day's prefix without scanning, and the ids

     * are in the key so an object found on its own is still traceable back to a lead.

     *

     * <p>No {@code lms-recordings/} prefix: the bucket is already named that, and prefixing would

     * give every object a path of {@code lms-recordings/lms-recordings/...}.

     */

    private String objectKey(LeadCall call) {

        LocalDateTime when = call.getStartedAt() != null ? call.getStartedAt() : call.getCreatedTimestamp();

        if (when == null) {

            when = LocalDateTime.now();

        }

        return KEY_DATE.format(when)

                + "/lead-" + call.getLeadId()

                + "/call-" + call.getId() + "-" + call.getRecordingUuid() + ".mp3";

    }

    /** Buffered because S3 needs an accurate content length up front, but bounded so it cannot OOM. */

    private byte[] readCapped(InputStream in) throws Exception {

        ByteArrayOutputStream out = new ByteArrayOutputStream();

        byte[] buffer = new byte[8192];

        long total = 0;

        int read;

        while ((read = in.read(buffer)) != -1) {

            total += read;

            if (total > MAX_RECORDING_BYTES) {

                throw new IllegalStateException("Recording exceeds " + MAX_RECORDING_BYTES + " bytes");

            }

            out.write(buffer, 0, read);

        }

        return out.toByteArray();

    }

    private void closeQuietly(InputStream stream) {

        if (stream == null) {

            return;

        }

        try {

            stream.close();

        } catch (Exception e) {

            LOGGER.debug("Ignoring a failure closing the recording stream", e);

        }

    }

}