PeakInsight: an AI-native operating system for travel agencies

/**
 * Sarvam STT -> Claude intent classification/extraction -> entity resolution ->
 * precondition checks -> a preview-ready draft. No write API is ever called from here —
 * the response is always a draft the frontend shows the user for confirmation ...
 */
static final int MAX_CLARIFICATION_ROUNDS = 3;
static final int MAX_VOICE_CLIPS_PER_REQUEST = 5;

public ApiResponse<VoiceProcessResponse> process(List<MultipartFile> audioFiles,
        VoiceProcessRequest payload, HttpServletRequest httpRequest, String requestId) {
    long overallStartMs = System.currentTimeMillis();
    try {
        Long userId = JwtAuthUtil.getTokenUserId(httpRequest);
        Long agencyId = JwtAuthUtil.getAgencyId(httpRequest);
        MDC.put(LoggingFields.REQUEST_ID.getField(), requestId);
        MDC.put(LoggingFields.AGENCY_ID.getField(), agencyId.toString());
        // … validate clips, load user scoped to agency …
        claudeUsageService.beginSoft(ClaudeUsageLabels.OPERATION_VOICE_PROCESS,
                agencyId, userId, claudeLlmProperties.resolveModel(), null, null);
        // … CONTEXTUAL / UNIVERSAL / follow-up flows, each returning a draft …
    } finally {
        log.info("[VOICE_PROCESS_TIMING] stage=TOTAL durationMs={}",
                System.currentTimeMillis() - overallStartMs);
        claudeUsageService.endSoft();
        MDC.clear();
    }
}

if (whatsAppInboundSqsQueueUrlProvider.isFifoQueue()) {
    requestBuilder
            .messageGroupId(resolveFifoMessageGroupId(payloadJson))
            .messageDeduplicationId(sha256Hex(payloadJson));
}

// resolveFifoMessageGroupId:
//  - messages webhooks → our line + customer WhatsApp id (ordering per thread)
//  - statuses-only     → provider message id, so one message's lifecycle stays ordered
JsonNode messages = value.path("messages");
if (messages.isArray() && !messages.isEmpty()) {
    String from = messages.get(0).path("from").asText("");
    String ourLine = !displayPhoneNumber.isBlank()
            ? displayPhoneNumber
            : (!phoneNumberId.isBlank() ? "pid_" + phoneNumberId : "");
    if (!ourLine.isBlank() && !from.isBlank()) {
        return sanitizeFifoToken("wa_msg_" + ourLine + "_" + from);
    }
}

@Scheduled(fixedDelayString = "${hj.sqs.whatsapp-inbound.poll-interval-ms:1000}")
public void poll() {
    int batchSize = SqsWorkerSupport.receiveBatchSize(
            whatsAppInboundSqsInFlight, properties.getMaxNumberOfMessages());
    if (batchSize <= 0) {
        return;
    }
    // … long-poll receive (≤20s) …
    for (Message message : messages) {
        boolean accepted = SqsWorkerSupport.dispatch(
                whatsAppInboundSqsInFlight, whatsAppInboundSqsExecutor,
                () -> processMessage(message));
        if (!accepted) {
            log.warn("[BUSINESS_ALERT] SQS whatsapp worker at capacity; leaving message in-flight ...");
        }
    }
}

private void processMessage(Message message) {
    try {
        // … deserialize …
        whatsAppInboundSqsMessageHandler.handleInboundWebhookPayload(payload.getPayloadJson());
        deleteMessage(message);          // only on success
    } catch (Exception e) {
        log.error("[CRITICAL_ALERT] Failed processing whatsapp inbound SQS message ...", e);
    }
}

/**
 * Freezes ai_usage_operations.credits_used from hits × credit_tariff,
 * then debits the attributed user's wallet (expirable → non-expiring → negative expirable).
 */
@Transactional
public void finalizeOperationCredits(Long operationId) {
    // …
    ClaudeUsageOperation op = opOpt.get();
    if (op.getCreditsUsed() != null) {
        return;                                   // idempotent
    }
    List<ClaudeUsageHit> hits = hitRepository.findByOperationIdOrderByIdAsc(operationId);
    int credits = computeCredits(op.getModel(), hits, tariff);
    op.setCreditsUsed(credits);
    DebitSnapshot snapshot = debitWallet(op.getCrmUserId(), op.getAgencyId(), credits);
    op.setExpirableDelta(snapshot.expirableDelta());
    op.setNonExpiringDelta(snapshot.nonExpiringDelta());
    // …
}

private DebitSnapshot debitWallet(Long crmUserId, Long agencyId, int credits) {
    // …
    if (remaining > 0 && exp > 0) { takenExp = Math.min(exp, remaining); exp -= takenExp; remaining -= takenExp; }
    if (remaining > 0 && non > 0) { takenNon = Math.min(non, remaining); non -= takenNon; remaining -= takenNon; }
    if (remaining > 0) {
        // Overdraw lands on expirable (may go negative).
        takenExp += remaining;
        exp -= remaining;
    }
    // …
}

Long agencyId = jwtTokenService.getAgencyIdFromToken(claims);
CrmUser user = crmUserRepository.findByIdAndAgencyId(userId, agencyId).orElse(null);
// … reject inactive users …

AgencyStatus agencyStatus = user.getAgency() != null ? user.getAgency().getStatus() : null;
if (agencyStatus == null || agencyStatus != AgencyStatus.ACTIVE) {
    log.warn("[CRITICAL_ALERT] CRM access blocked for agency {} (status={}) user {}", ...);
    response.setStatus(HttpServletResponse.SC_UNAUTHORIZED);
    deviceTokenService.deleteAllTokensForUser(UserType.CRM, userId);
    refreshTokenService.deleteAllTokensForUser(UserType.CRM, userId);
    refreshTokenService.clearRefreshTokenCookie(response);
    return;
}
request.setAttribute(JwtClaim.AGENCY_ID.getClaimName(), agencyId);

@Transactional
public void processEnrollment(Long enrollmentId) {
    AutomationEnrollment enrollment = /* … */;
    if (enrollment.getStatus().isTerminal()) {
        return;
    }
    if (enrollment.getStatus() == AutomationEnrollmentStatus.WAITING) {
        if (conversationSignals.customerReplied(enrollment.getLeadId(), enrollment.getStartedAt())) {
            cancel(enrollment, AutomationCancelReason.CUSTOMER_REPLIED);
            return;
        }
        if (enrollment.getNextRunAt() != null && enrollment.getNextRunAt().isAfter(LocalDateTime.now())) {
            return;
        }
        advancePastCurrent(enrollment);          // wait completed
        enrollment.setStatus(AutomationEnrollmentStatus.ACTIVE);
        enrollment.setNextRunAt(null);
        enrollmentRepository.save(enrollment);
    }
    int guard = 0;
    while (enrollment.getStatus() == AutomationEnrollmentStatus.ACTIVE && guard++ < MAX_STEPS_PER_PASS) {
        // … load step; WAIT → enterWait(...) and return …
        AutomationActionExecutor.ActionResult result = actionExecutor.execute(enrollment, step, lead);
        recordRun(enrollment, step, result);
        advancePastCurrent(enrollment);
        enrollmentRepository.save(enrollment);
    }
}

AutomationRoundRobinCursor cursor = cursorRepository
        .findForUpdate(agencyId, workflowId, stepId)       // @Lock(PESSIMISTIC_WRITE)
        .orElse(null);
if (cursor == null) {
    try {
        cursorRepository.saveAndFlush(newCursor(agencyId, workflowId, stepId));
    } catch (DataIntegrityViolationException ignored) {
        // Concurrent create — fall through to locked read.
    }
    cursor = cursorRepository.findForUpdate(agencyId, workflowId, stepId).orElseThrow(...);
}
int index = Math.floorMod(cursor.getNextIndex(), userIds.size());
cursor.setNextIndex((index + 1) % userIds.size());