package vip.mate.wiki.job; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import vip.mate.llm.failover.AvailableProviderPool; import vip.mate.llm.model.ModelConfigEntity; import vip.mate.llm.service.ModelConfigService; import vip.mate.wiki.repository.WikiProcessingJobMapper; import vip.mate.wiki.job.model.WikiProcessingJobEntity; import java.time.LocalDateTime; import java.util.Map; /** * RFC-030: Wiki processing job lifecycle service — creates, transitions, * and records errors for processing jobs. */ @Slf4j @Service public class WikiProcessingJobService { private final WikiProcessingJobMapper jobMapper; private final WikiModelRoutingService routingService; private final ObjectMapper objectMapper; @Autowired(required = false) private AvailableProviderPool providerPool; @Autowired private ModelConfigService modelConfigService; public WikiProcessingJobService(WikiProcessingJobMapper jobMapper, WikiModelRoutingService routingService, ObjectMapper objectMapper) { this.jobMapper = jobMapper; this.routingService = routingService; this.objectMapper = objectMapper; } public WikiProcessingJobEntity createHeavyIngest(Long kbId, Long rawId) { WikiProcessingJobEntity job = new WikiProcessingJobEntity(); job.setKbId(kbId); job.setRawId(rawId); job.setJobType(WikiJobType.HEAVY_INGEST.name().toLowerCase()); job.setStage(WikiJobStage.QUEUED.name().toLowerCase()); job.setStatus(WikiJobStatus.QUEUED.name().toLowerCase()); job.setMaxRetries(3); job.setRetryCount(0); jobMapper.insert(job); return job; } public WikiProcessingJobEntity createLightEnrich(Long kbId, Long rawId) { WikiProcessingJobEntity job = new WikiProcessingJobEntity(); job.setKbId(kbId); job.setRawId(rawId); job.setJobType(WikiJobType.LIGHT_ENRICH.name().toLowerCase()); job.setStage(WikiJobStage.QUEUED.name().toLowerCase()); job.setStatus(WikiJobStatus.QUEUED.name().toLowerCase()); job.setMaxRetries(2); job.setRetryCount(0); jobMapper.insert(job); return job; } public WikiProcessingJobEntity createLocalRepair(Long kbId, Long rawId, Long targetPageId) { WikiProcessingJobEntity job = new WikiProcessingJobEntity(); job.setKbId(kbId); job.setRawId(rawId); job.setJobType(WikiJobType.LOCAL_REPAIR.name().toLowerCase()); job.setStage(WikiJobStage.QUEUED.name().toLowerCase()); job.setStatus(WikiJobStatus.QUEUED.name().toLowerCase()); job.setMaxRetries(2); job.setRetryCount(0); try { job.setMetaJson(objectMapper.writeValueAsString(Map.of("targetPageId", targetPageId))); } catch (Exception e) { job.setMetaJson("{}"); } jobMapper.insert(job); return job; } public WikiProcessingJobEntity transition(Long jobId, WikiJobStage newStage) { WikiProcessingJobEntity job = jobMapper.selectById(jobId); if (job == null) return null; job.setStage(newStage.name().toLowerCase()); if (newStage == WikiJobStage.ROUTING) { job.setStartedAt(LocalDateTime.now()); job.setStatus(WikiJobStatus.RUNNING.name().toLowerCase()); job.setFallbackChainJson(routingService.buildFallbackChainJson(job.getKbId())); } else if (newStage == WikiJobStage.COMPLETED) { job.setFinishedAt(LocalDateTime.now()); job.setStatus(WikiJobStatus.COMPLETED.name().toLowerCase()); } else if (newStage == WikiJobStage.FAILED) { job.setFinishedAt(LocalDateTime.now()); job.setStatus(WikiJobStatus.FAILED.name().toLowerCase()); } else if (newStage == WikiJobStage.PARTIAL) { job.setFinishedAt(LocalDateTime.now()); job.setStatus(WikiJobStatus.PARTIAL.name().toLowerCase()); } else if (!newStage.isTerminal()) { // Non-terminal intermediate stage: mark as running job.setStatus(WikiJobStatus.RUNNING.name().toLowerCase()); } jobMapper.updateById(job); return job; } public void recordHardError(Long jobId, String errorCode, String errorMessage) { WikiProcessingJobEntity job = jobMapper.selectById(jobId); if (job == null) return; job.setErrorCode(errorCode); job.setErrorMessage(truncate(errorMessage, 2000)); job.setFinishedAt(LocalDateTime.now()); if (job.getRetryCount() < job.getMaxRetries()) { job.setRetryCount(job.getRetryCount() + 1); job.setResumeFromStage(job.getStage()); job.setStage(WikiJobStage.QUEUED.name().toLowerCase()); job.setStatus(WikiJobStatus.QUEUED.name().toLowerCase()); } else { job.setStatus(WikiJobStatus.FAILED.name().toLowerCase()); job.setStage(WikiJobStage.FAILED.name().toLowerCase()); } jobMapper.updateById(job); if (providerPool != null && isHardError(errorCode) && job.getCurrentModelId() != null) { notifyPoolHardError(job.getCurrentModelId(), errorCode); } } public void recordSoftError(Long jobId, String errorCode, String errorMessage) { WikiProcessingJobEntity job = jobMapper.selectById(jobId); if (job == null) return; job.setErrorCode(errorCode); job.setErrorMessage(truncate(errorMessage, 2000)); job.setFinishedAt(LocalDateTime.now()); if (job.getRetryCount() < job.getMaxRetries()) { job.setRetryCount(job.getRetryCount() + 1); job.setResumeFromStage(job.getStage()); job.setStage(WikiJobStage.QUEUED.name().toLowerCase()); job.setStatus(WikiJobStatus.QUEUED.name().toLowerCase()); } else { job.setStatus(WikiJobStatus.FAILED.name().toLowerCase()); job.setStage(WikiJobStage.FAILED.name().toLowerCase()); } jobMapper.updateById(job); } /** * Recover stuck jobs on startup (routing or *_running → queued). */ public void recoverOnStartup() { int recovered = jobMapper.recoverStuckJobs(); if (recovered > 0) { log.info("[WikiJob] Recovered {} stuck jobs on startup", recovered); } } private static boolean isHardError(String errorCode) { return errorCode != null && ( errorCode.equals("AUTH_ERROR") || errorCode.equals("BILLING") || errorCode.equals("MODEL_NOT_FOUND")); } private void notifyPoolHardError(Long modelId, String errorCode) { try { ModelConfigEntity model = modelConfigService.getModel(modelId); if (model != null) { AvailableProviderPool.RemovalSource source; try { source = AvailableProviderPool.RemovalSource.valueOf(errorCode); } catch (IllegalArgumentException e) { source = AvailableProviderPool.RemovalSource.AUTH_ERROR; } providerPool.remove(model.getProvider(), source, "Wiki job hard error: " + errorCode); } } catch (Exception e) { log.warn("[WikiJob] Failed to notify provider pool: {}", e.getMessage()); } } private static String truncate(String s, int maxLen) { if (s == null) return null; return s.length() <= maxLen ? s : s.substring(0, maxLen); } }