diff --git a/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeContext.java b/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeContext.java new file mode 100644 index 000000000..ff4c94a13 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeContext.java @@ -0,0 +1,26 @@ +package stirling.software.saas.payg.charge; + +import stirling.software.saas.payg.model.JobSource; +import stirling.software.saas.payg.model.ProcessType; + +/** + * Per-call context for {@link JobChargeService#openProcess}. Carries the caller's identity and what + * kind of process this is. Does NOT carry policy fields — the charge service resolves the effective + * policy from {@code PricingPolicyService} so a stale snapshot from the caller can't desync from + * the live policy. + */ +public record ChargeContext( + Long ownerUserId, Long ownerTeamId, JobSource source, ProcessType processType) { + + public ChargeContext { + if (ownerUserId == null) { + throw new IllegalArgumentException("ownerUserId is required"); + } + if (source == null) { + throw new IllegalArgumentException("source is required"); + } + if (processType == null) { + throw new IllegalArgumentException("processType is required"); + } + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeOutcome.java b/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeOutcome.java new file mode 100644 index 000000000..b5bba569c --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeOutcome.java @@ -0,0 +1,19 @@ +package stirling.software.saas.payg.charge; + +import java.util.UUID; + +/** + * Result of {@link JobChargeService#openProcess}. {@code processId} is the job the caller's tool + * call belongs to (newly opened or joined). {@code units} is what would have been debited had the + * call been in PAYG (live) mode — for OPENED, the actual classification × policy units; for JOINED, + * always 0 since the parent process already paid. + */ +public record ChargeOutcome(UUID processId, int units, Disposition disposition) { + + public enum Disposition { + /** Started a new process; {@code units} reflects the would-be charge. */ + OPENED, + /** Joined an existing process within window; no incremental charge. */ + JOINED + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java b/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java new file mode 100644 index 000000000..ee342f4e9 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java @@ -0,0 +1,143 @@ +package stirling.software.saas.payg.charge; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import java.util.Objects; + +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.web.multipart.MultipartFile; + +import lombok.extern.slf4j.Slf4j; + +import stirling.software.saas.payg.docs.DocumentClassifier; +import stirling.software.saas.payg.docs.DocumentMetrics; +import stirling.software.saas.payg.job.JobContext; +import stirling.software.saas.payg.job.JobService; +import stirling.software.saas.payg.job.JoinOrOpenResult; +import stirling.software.saas.payg.model.JobSource; +import stirling.software.saas.payg.policy.PricingPolicy; +import stirling.software.saas.payg.policy.PricingPolicyService; +import stirling.software.saas.payg.repository.PaygShadowChargeRepository; +import stirling.software.saas.payg.shadow.PaygShadowCharge; + +/** + * Orchestrates a tool call's open-process decision: look up the team's effective policy, resolve + * the step-limit ceiling for this caller surface, delegate the join-or-open decision to {@link + * JobService}, and — when a new process opens — compute the would-be charge and record a {@link + * PaygShadowCharge} row for comparison against the legacy engine. + * + *

Shadow mode only: this service never debits the wallet ledger or posts a Stripe meter event. + * The real-charging path lives in a separate follow-up and reuses the same orchestration — only the + * side-effect (shadow row vs ledger entry + Stripe call) differs. + * + *

The {@code legacyCreditsCharged} field on the shadow row is set to {@code 0} here. When the + * legacy {@code CreditService} is wired to call this service (separate PR), the legacy debit amount + * becomes available and {@code diffPct} can be computed against it; until then the shadow row + * captures the PAYG units only. + */ +@Service +@Profile("saas") +@Slf4j +public class JobChargeService { + + private final JobService jobService; + private final PricingPolicyService policyService; + private final DocumentClassifier classifier; + private final PaygShadowChargeRepository shadowRepository; + + public JobChargeService( + JobService jobService, + PricingPolicyService policyService, + DocumentClassifier classifier, + PaygShadowChargeRepository shadowRepository) { + this.jobService = Objects.requireNonNull(jobService, "jobService"); + this.policyService = Objects.requireNonNull(policyService, "policyService"); + this.classifier = Objects.requireNonNull(classifier, "classifier"); + this.shadowRepository = Objects.requireNonNull(shadowRepository, "shadowRepository"); + } + + /** + * Open a process (or join an existing one) for this tool call. Side effects: persists a {@code + * ProcessingJob} row plus input signatures, and — on OPENED — writes a {@code + * payg_shadow_charge} row carrying the would-be PAYG units. + */ + @Transactional + public ChargeOutcome openProcess(ChargeContext ctx, List inputs) throws IOException { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(inputs, "inputs"); + if (inputs.isEmpty()) { + throw new IllegalArgumentException("inputs must not be empty"); + } + + PricingPolicy policy = policyService.getEffectivePolicy(ctx.ownerTeamId()); + int stepLimit = resolveStepLimit(policy, ctx.source()); + + JobContext jobCtx = + new JobContext( + ctx.ownerUserId(), + ctx.ownerTeamId(), + ctx.source(), + ctx.processType(), + policy.getId(), + stepLimit); + + List paths = inputs.stream().map(JobInput::path).toList(); + JoinOrOpenResult result = jobService.joinOrOpen(jobCtx, paths); + + if (result.disposition() == JoinOrOpenResult.Disposition.JOINED) { + return new ChargeOutcome(result.job().getId(), 0, ChargeOutcome.Disposition.JOINED); + } + + int units = computeUnits(inputs, policy); + result.job().setDocUnits(units); + + recordShadowRow(ctx, result.job().getId(), policy.getId(), units); + + return new ChargeOutcome(result.job().getId(), units, ChargeOutcome.Disposition.OPENED); + } + + private int resolveStepLimit(PricingPolicy policy, JobSource source) { + Integer fromPolicy = + policy.getStepLimits() == null ? null : policy.getStepLimits().get(source); + if (fromPolicy != null && fromPolicy > 0) { + return fromPolicy; + } + // Defensive default — every JobSource should have an entry per the V12 seed, but a + // hand-edited policy could be missing one. Fall back to the smallest documented limit + // (10 — WEB/API/DESKTOP_APP default) so an admin slip-up never spawns unbounded chains. + log.debug( + "PricingPolicy {} missing stepLimit for source={}; using fallback of 10.", + policy.getId(), + source); + return 10; + } + + private int computeUnits(List inputs, PricingPolicy policy) { + List multiparts = inputs.stream().map(JobInput::multipart).toList(); + DocumentMetrics metrics = + multiparts.size() == 1 + ? classifier.classify(multiparts.get(0), policy) + : classifier.classify(multiparts, policy); + // Apply the policy-level minChargeUnits floor per design § 3.4. The classifier returns + // raw docUnits with a "non-empty input → ≥1" floor; the charge formula's + // max(min_charge_units, docUnits) layers on top. + return Math.max(policy.getMinChargeUnits(), metrics.docUnits()); + } + + private void recordShadowRow( + ChargeContext ctx, java.util.UUID jobId, Long policyId, int units) { + PaygShadowCharge row = new PaygShadowCharge(); + row.setTeamId(ctx.ownerTeamId()); + row.setJobId(jobId); + row.setPolicyId(policyId); + row.setPaygUnits(units); + // No legacy comparison yet — wired when the shadow path is connected to the legacy + // CreditService in the follow-up PR. Until then, diff stays at 0. + row.setLegacyCreditsCharged(0); + row.setDiffPct(0); + shadowRepository.save(row); + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/charge/JobInput.java b/app/saas/src/main/java/stirling/software/saas/payg/charge/JobInput.java new file mode 100644 index 000000000..09daed075 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/charge/JobInput.java @@ -0,0 +1,24 @@ +package stirling.software.saas.payg.charge; + +import java.nio.file.Path; + +import org.springframework.web.multipart.MultipartFile; + +/** + * One input file to a tool call, materialised. {@code path} is the on-disk copy the lineage + * detector hashes; {@code multipart} carries the size + content-type metadata the classifier needs. + * The pair is constructed by the ingress filter (PR-I7a filter) which materialises the request body + * exactly once. Tests construct it from a fixture file plus a {@code MockMultipartFile} wrapping + * the same bytes. + */ +public record JobInput(MultipartFile multipart, Path path) { + + public JobInput { + if (multipart == null) { + throw new IllegalArgumentException("multipart is required"); + } + if (path == null) { + throw new IllegalArgumentException("path is required"); + } + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/job/JobContext.java b/app/saas/src/main/java/stirling/software/saas/payg/job/JobContext.java new file mode 100644 index 000000000..7840effa2 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/job/JobContext.java @@ -0,0 +1,37 @@ +package stirling.software.saas.payg.job; + +import stirling.software.saas.payg.model.JobSource; +import stirling.software.saas.payg.model.ProcessType; + +/** + * Input to {@link JobService#joinOrOpen}: who owns the request, what process shape it is, which + * policy version applies, and the step-limit ceiling derived from that policy for this caller + * surface. {@code stepLimit} is resolved by the caller (typically {@code JobChargeService}) rather + * than re-fetched here so the service stays free of policy-lookup concerns. + */ +public record JobContext( + Long ownerUserId, + Long ownerTeamId, + JobSource source, + ProcessType processType, + Long policyId, + int stepLimit) { + + public JobContext { + if (ownerUserId == null) { + throw new IllegalArgumentException("ownerUserId is required"); + } + if (source == null) { + throw new IllegalArgumentException("source is required"); + } + if (processType == null) { + throw new IllegalArgumentException("processType is required"); + } + if (policyId == null) { + throw new IllegalArgumentException("policyId is required"); + } + if (stepLimit <= 0) { + throw new IllegalArgumentException("stepLimit must be > 0"); + } + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/job/JobService.java b/app/saas/src/main/java/stirling/software/saas/payg/job/JobService.java new file mode 100644 index 000000000..005833c70 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/job/JobService.java @@ -0,0 +1,243 @@ +package stirling.software.saas.payg.job; + +import java.io.IOException; +import java.nio.file.Path; +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import lombok.extern.slf4j.Slf4j; + +import stirling.software.saas.payg.lineage.HashLineageDetector; +import stirling.software.saas.payg.lineage.LineageMatch; +import stirling.software.saas.payg.model.ArtifactKind; +import stirling.software.saas.payg.model.JobStatus; +import stirling.software.saas.payg.model.JobStepStatus; +import stirling.software.saas.payg.repository.ProcessingJobRepository; +import stirling.software.saas.payg.repository.ProcessingJobStepRepository; + +/** + * Persistence + lineage policy layer for the PAYG process model. The hot-path entry point is {@link + * #joinOrOpen}, which applies the "any-match-joins, newest wins" multi-input rule. + * + *

The charge service ({@code JobChargeService}) orchestrates around this — it resolves the + * pricing policy, computes the step-limit ceiling for the current {@code JobSource}, calls {@link + * #joinOrOpen}, and decides whether to record a shadow charge based on the resulting {@link + * JoinOrOpenResult.Disposition}. + */ +@Service +@Profile("saas") +@Slf4j +public class JobService { + + private final HashLineageDetector detector; + private final ProcessingJobRepository jobRepository; + private final ProcessingJobStepRepository stepRepository; + private final Duration workflowWindow; + + public JobService( + HashLineageDetector detector, + ProcessingJobRepository jobRepository, + ProcessingJobStepRepository stepRepository, + @Value("${payg.lineage.workflow-window:PT5M}") Duration workflowWindow) { + this.detector = Objects.requireNonNull(detector, "detector"); + this.jobRepository = Objects.requireNonNull(jobRepository, "jobRepository"); + this.stepRepository = Objects.requireNonNull(stepRepository, "stepRepository"); + Objects.requireNonNull(workflowWindow, "workflowWindow"); + if (workflowWindow.isNegative() || workflowWindow.isZero()) { + throw new IllegalArgumentException( + "payg.lineage.workflow-window must be positive, got " + workflowWindow); + } + this.workflowWindow = workflowWindow; + } + + /** + * Decide whether {@code inputs} should join an existing open process or start a new one, and + * persist that decision. Returns the job to attach to plus the disposition. + * + *

Rule: hash every input. If any input's signatures match an open process owned by the same + * user within the workflow window, attach to that process. When multiple inputs match different + * processes, the one with the freshest {@code lastStepAt} wins — preserves the + * "most-recent-job-wins" invariant the lineage primitives test for. When the matched process is + * already at its step-limit ceiling, fall through to opening a fresh process; the new job's + * input signatures are still recorded so downstream calls lineage-match to it. + * + *

Race note: the read of candidate matches and the subsequent write happen in one + * transaction at default isolation (READ COMMITTED in Postgres). A concurrent admin write or + * another tool call could change the match set between read and commit. For shadow mode the + * resulting diff is a low-grade comparison artefact; tighten isolation when real charging lands + * if it materialises as a real-money issue. + */ + @Transactional + public JoinOrOpenResult joinOrOpen(JobContext ctx, List inputs) throws IOException { + Objects.requireNonNull(ctx, "ctx"); + Objects.requireNonNull(inputs, "inputs"); + if (inputs.isEmpty()) { + throw new IllegalArgumentException("inputs must not be empty"); + } + + Optional bestMatch = findBestMatch(ctx.ownerUserId(), inputs); + + if (bestMatch.isPresent()) { + ProcessingJob existing = + jobRepository + .findById(bestMatch.get().jobId()) + .orElseThrow( + () -> + new IllegalStateException( + "Lineage match returned jobId=" + + bestMatch.get().jobId() + + " but no such ProcessingJob row" + + " exists (stale signature?)")); + if (existing.getStepCount() < ctx.stepLimit()) { + return joinExisting(existing, inputs); + } + // Step-limit hit: spawn a fresh job. The new job will share input signatures with + // the existing chain so future tool calls still lineage-match into the workflow, + // but the freshest jobLastStepAt wins so they target the new one. + log.debug( + "Job {} hit step limit ({}); spawning new process within workflow.", + existing.getId(), + ctx.stepLimit()); + } + + return openFresh(ctx, inputs); + } + + /** + * Records {@code outputFile}'s signatures against {@code jobId} as {@code OUTPUT}. Called after + * a tool runs successfully — failed tools don't record outputs. Multi-output tools call this + * once per output. + */ + @Transactional + public void recordOutput(UUID jobId, Path outputFile) throws IOException { + Objects.requireNonNull(jobId, "jobId"); + Objects.requireNonNull(outputFile, "outputFile"); + detector.record(jobId, outputFile, ArtifactKind.OUTPUT); + } + + /** Appends an audit-trail step row after a tool call completes. */ + @Transactional + public ProcessingJobStep appendStep( + UUID jobId, + String toolId, + JobStepStatus status, + Integer inputPages, + Long inputBytes, + String errorCode) { + Objects.requireNonNull(jobId, "jobId"); + Objects.requireNonNull(toolId, "toolId"); + Objects.requireNonNull(status, "status"); + ProcessingJobStep step = new ProcessingJobStep(); + step.setJobId(jobId); + step.setToolId(toolId); + step.setStatus(status); + LocalDateTime now = LocalDateTime.now(); + step.setStartedAt(now); + step.setCompletedAt(now); + step.setInputPages(inputPages); + step.setInputBytes(inputBytes); + step.setErrorCode(errorCode); + return stepRepository.save(step); + } + + /** + * Closes a job. Idempotent — closing an already-closed job is a silent no-op returning the + * existing row, so multiple close paths (explicit caller, FE on-unload, stale scheduler) all + * compose safely. + */ + @Transactional + public ProcessingJob close(UUID jobId) { + Objects.requireNonNull(jobId, "jobId"); + ProcessingJob job = + jobRepository + .findById(jobId) + .orElseThrow( + () -> + new IllegalArgumentException( + "No ProcessingJob with id " + jobId)); + if (job.getStatus() != JobStatus.OPEN) { + return job; + } + job.setStatus(JobStatus.CLOSED); + job.setClosedAt(LocalDateTime.now()); + return jobRepository.save(job); + } + + /** Returns open jobs whose {@code last_step_at} is older than the workflow window. */ + @Transactional(readOnly = true) + public List findStale() { + return jobRepository.findStale(JobStatus.OPEN, LocalDateTime.now().minus(workflowWindow)); + } + + /** + * Closes all jobs returned by {@link #findStale}. Returns the count closed. Each transition + * goes through {@link #close} semantics (idempotent), so re-running this on an empty stale set + * is safe. + */ + @Transactional + public int closeStale() { + List stale = findStale(); + LocalDateTime now = LocalDateTime.now(); + for (ProcessingJob j : stale) { + j.setStatus(JobStatus.CLOSED); + j.setClosedAt(now); + } + if (!stale.isEmpty()) { + jobRepository.saveAll(stale); + } + return stale.size(); + } + + private Optional findBestMatch(Long userId, List inputs) + throws IOException { + List matches = new ArrayList<>(inputs.size()); + for (Path input : inputs) { + detector.detect(userId, input).ifPresent(matches::add); + } + return matches.stream().max(Comparator.comparing(LineageMatch::jobLastStepAt)); + } + + private JoinOrOpenResult joinExisting(ProcessingJob existing, List inputs) + throws IOException { + existing.setStepCount(existing.getStepCount() + 1); + existing.setLastStepAt(LocalDateTime.now()); + ProcessingJob saved = jobRepository.save(existing); + recordAllInputs(saved.getId(), inputs); + return new JoinOrOpenResult(saved, JoinOrOpenResult.Disposition.JOINED); + } + + private JoinOrOpenResult openFresh(JobContext ctx, List inputs) throws IOException { + ProcessingJob fresh = new ProcessingJob(); + fresh.setId(UUID.randomUUID()); + fresh.setOwnerUserId(ctx.ownerUserId()); + fresh.setOwnerTeamId(ctx.ownerTeamId()); + fresh.setProcessType(ctx.processType()); + fresh.setSource(ctx.source()); + fresh.setPolicyId(ctx.policyId()); + fresh.setStepCount(1); + LocalDateTime now = LocalDateTime.now(); + fresh.setStartedAt(now); + fresh.setLastStepAt(now); + fresh.setStatus(JobStatus.OPEN); + ProcessingJob saved = jobRepository.save(fresh); + recordAllInputs(saved.getId(), inputs); + return new JoinOrOpenResult(saved, JoinOrOpenResult.Disposition.OPENED); + } + + private void recordAllInputs(UUID jobId, List inputs) throws IOException { + for (Path input : inputs) { + detector.record(jobId, input, ArtifactKind.INPUT); + } + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/job/JoinOrOpenResult.java b/app/saas/src/main/java/stirling/software/saas/payg/job/JoinOrOpenResult.java new file mode 100644 index 000000000..e7347f146 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/job/JoinOrOpenResult.java @@ -0,0 +1,17 @@ +package stirling.software.saas.payg.job; + +/** + * Output of {@link JobService#joinOrOpen}. {@code OPENED} means the charge layer should compute + * units and record a shadow charge (or, post-shadow, actually debit). {@code JOINED} means the call + * attached to an existing process — no further charge, just an audit-trail step append. + * + *

Step-limit overflow on a matched job surfaces as {@code OPENED}: the lineage tree is preserved + * via shared input signatures, but the limit means we start a fresh billable process. + */ +public record JoinOrOpenResult(ProcessingJob job, Disposition disposition) { + + public enum Disposition { + OPENED, + JOINED + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/job/StaleJobCloser.java b/app/saas/src/main/java/stirling/software/saas/payg/job/StaleJobCloser.java new file mode 100644 index 000000000..6c4813988 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/job/StaleJobCloser.java @@ -0,0 +1,36 @@ +package stirling.software.saas.payg.job; + +import org.springframework.context.annotation.Profile; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * Auto-closes {@code OPEN} jobs whose {@code last_step_at} is older than the workflow window. Runs + * every minute. API users never have to call {@code close()} explicitly — this scheduler is the + * safety net. + * + *

Single-fire only at V1: not {@code @SchedulerLock}'d, consistent with the other + * {@code @Scheduled} tasks in {@code :saas} (none of them are guarded against multi-pod + * double-fires today either). Multi-pod cluster-correctness for all schedulers is tracked in design + * § 9 as a separate cleanup. The underlying {@code closeStale()} call is idempotent — duplicate + * firings read an empty stale set on the second pod, no data corruption risk. + */ +@Component +@Profile("saas") +@RequiredArgsConstructor +@Slf4j +public class StaleJobCloser { + + private final JobService jobService; + + @Scheduled(fixedRateString = "${payg.job.stale-close-interval-ms:60000}") + public void closeStale() { + int closed = jobService.closeStale(); + if (closed > 0) { + log.info("StaleJobCloser closed {} job(s) idle past the workflow window.", closed); + } + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineagePruneScheduler.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineagePruneScheduler.java new file mode 100644 index 000000000..d69c722ca --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineagePruneScheduler.java @@ -0,0 +1,54 @@ +package stirling.software.saas.payg.lineage; + +import java.time.Duration; +import java.time.Instant; +import java.util.Objects; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Profile; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import lombok.extern.slf4j.Slf4j; + +/** + * Deletes {@code job_artifact_hash} rows older than the configured retention window. The workflow + * window is 5 minutes; we keep an extra hour of headroom as a safety margin (covers clock skew, + * in-flight requests, brief outages of this scheduler). Runs hourly. + * + *

Like {@link stirling.software.saas.payg.job.StaleJobCloser}, this is not + * {@code @SchedulerLock}'d — duplicate firings on multi-pod deploys would both run a {@code DELETE + * WHERE created_at < cutoff}, the second seeing zero rows to delete. Idempotent, wasted IO at + * worst. + */ +@Component +@Profile("saas") +@Slf4j +public class LineagePruneScheduler { + + private final JobLineageStore store; + private final Duration retention; + + public LineagePruneScheduler( + JobLineageStore store, @Value("${payg.lineage.retention:PT1H}") Duration retention) { + this.store = Objects.requireNonNull(store, "store"); + Objects.requireNonNull(retention, "retention"); + if (retention.isNegative() || retention.isZero()) { + throw new IllegalArgumentException( + "payg.lineage.retention must be positive, got " + retention); + } + this.retention = retention; + } + + @Scheduled(cron = "${payg.lineage.prune-cron:0 0 * * * *}", zone = "UTC") + public void prune() { + Instant cutoff = Instant.now().minus(retention); + int deleted = store.pruneOlderThan(cutoff); + if (deleted > 0) { + log.info( + "LineagePruneScheduler deleted {} job_artifact_hash row(s) older than {}.", + deleted, + cutoff); + } + } +} diff --git a/app/saas/src/test/java/stirling/software/saas/payg/charge/JobChargeServiceTest.java b/app/saas/src/test/java/stirling/software/saas/payg/charge/JobChargeServiceTest.java new file mode 100644 index 000000000..00e584316 --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/charge/JobChargeServiceTest.java @@ -0,0 +1,263 @@ +package stirling.software.saas.payg.charge; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; +import org.springframework.mock.web.MockMultipartFile; +import org.springframework.web.multipart.MultipartFile; + +import stirling.software.saas.payg.docs.DocumentClassifier; +import stirling.software.saas.payg.docs.DocumentMetrics; +import stirling.software.saas.payg.job.JobContext; +import stirling.software.saas.payg.job.JobService; +import stirling.software.saas.payg.job.JoinOrOpenResult; +import stirling.software.saas.payg.job.ProcessingJob; +import stirling.software.saas.payg.model.JobSource; +import stirling.software.saas.payg.model.JobStatus; +import stirling.software.saas.payg.model.ProcessType; +import stirling.software.saas.payg.policy.PricingPolicy; +import stirling.software.saas.payg.policy.PricingPolicyService; +import stirling.software.saas.payg.repository.PaygShadowChargeRepository; +import stirling.software.saas.payg.shadow.PaygShadowCharge; + +/** + * Exercises {@link JobChargeService} as an orchestrator: policy lookup, step-limit resolution, + * delegate to {@code JobService}, write shadow row when OPENED, skip everything when JOINED. + */ +class JobChargeServiceTest { + + private JobService jobService; + private PricingPolicyService policyService; + private DocumentClassifier classifier; + private PaygShadowChargeRepository shadowRepo; + private JobChargeService service; + + @BeforeEach + void setUp() { + jobService = Mockito.mock(JobService.class); + policyService = Mockito.mock(PricingPolicyService.class); + classifier = Mockito.mock(DocumentClassifier.class); + shadowRepo = Mockito.mock(PaygShadowChargeRepository.class); + service = new JobChargeService(jobService, policyService, classifier, shadowRepo); + } + + @Test + void openProcess_joinedDisposition_skipsClassifierAndShadowWrite(@TempDir Path tmp) + throws IOException { + // Setup: policy + a JOINED result. + PricingPolicy policy = stubPolicy(/*minCharge*/ 1, Map.of(JobSource.WEB, 10)); + when(policyService.getEffectivePolicy(100L)).thenReturn(policy); + ProcessingJob joinedJob = openJob(UUID.randomUUID()); + when(jobService.joinOrOpen(any(JobContext.class), anyList())) + .thenReturn(new JoinOrOpenResult(joinedJob, JoinOrOpenResult.Disposition.JOINED)); + + JobInput in = jobInput(tmp, "in.pdf", "application/pdf"); + + ChargeOutcome out = + service.openProcess( + new ChargeContext(42L, 100L, JobSource.WEB, ProcessType.SINGLE_TOOL), + List.of(in)); + + assertThat(out.disposition()).isEqualTo(ChargeOutcome.Disposition.JOINED); + assertThat(out.processId()).isEqualTo(joinedJob.getId()); + assertThat(out.units()).isZero(); + verify(classifier, never()).classify(any(MultipartFile.class), any()); + verify(classifier, never()).classify(anyList(), any()); + verify(shadowRepo, never()).save(any()); + } + + @Test + void openProcess_openedSingleFile_classifiesAndWritesShadowRow(@TempDir Path tmp) + throws IOException { + PricingPolicy policy = stubPolicy(/*minCharge*/ 1, Map.of(JobSource.WEB, 10)); + when(policyService.getEffectivePolicy(100L)).thenReturn(policy); + ProcessingJob newJob = openJob(UUID.randomUUID()); + when(jobService.joinOrOpen(any(JobContext.class), anyList())) + .thenReturn(new JoinOrOpenResult(newJob, JoinOrOpenResult.Disposition.OPENED)); + + JobInput in = jobInput(tmp, "in.pdf", "application/pdf"); + when(classifier.classify(any(MultipartFile.class), eq(policy))) + .thenReturn(new DocumentMetrics(50, 1024L, "application/pdf", 4)); + + ChargeOutcome out = + service.openProcess( + new ChargeContext(42L, 100L, JobSource.WEB, ProcessType.SINGLE_TOOL), + List.of(in)); + + assertThat(out.disposition()).isEqualTo(ChargeOutcome.Disposition.OPENED); + assertThat(out.units()).isEqualTo(4); + // Single-file path called single-file classifier overload, not the list one. + verify(classifier, times(1)).classify(any(MultipartFile.class), eq(policy)); + verify(classifier, never()).classify(anyList(), eq(policy)); + + ArgumentCaptor captor = ArgumentCaptor.forClass(PaygShadowCharge.class); + verify(shadowRepo).save(captor.capture()); + PaygShadowCharge row = captor.getValue(); + assertThat(row.getTeamId()).isEqualTo(100L); + assertThat(row.getJobId()).isEqualTo(newJob.getId()); + assertThat(row.getPolicyId()).isEqualTo(policy.getId()); + assertThat(row.getPaygUnits()).isEqualTo(4); + // Legacy comparison not wired yet — zeroed until CreditService is wired in the follow-up. + assertThat(row.getLegacyCreditsCharged()).isZero(); + assertThat(row.getDiffPct()).isZero(); + + // Job entity carries the classified docUnits so close-time receipts can render correctly. + assertThat(newJob.getDocUnits()).isEqualTo(4); + } + + @Test + void openProcess_openedMultiFile_usesListClassifier(@TempDir Path tmp) throws IOException { + PricingPolicy policy = stubPolicy(/*minCharge*/ 1, Map.of(JobSource.WEB, 10)); + when(policyService.getEffectivePolicy(100L)).thenReturn(policy); + ProcessingJob newJob = openJob(UUID.randomUUID()); + when(jobService.joinOrOpen(any(JobContext.class), anyList())) + .thenReturn(new JoinOrOpenResult(newJob, JoinOrOpenResult.Disposition.OPENED)); + + JobInput a = jobInput(tmp, "a.pdf", "application/pdf"); + JobInput b = jobInput(tmp, "b.pdf", "application/pdf"); + when(classifier.classify(anyList(), eq(policy))) + .thenReturn(new DocumentMetrics(100, 2048L, "application/pdf", 7)); + + ChargeOutcome out = + service.openProcess( + new ChargeContext(42L, 100L, JobSource.WEB, ProcessType.AUTOMATION), + List.of(a, b)); + + assertThat(out.units()).isEqualTo(7); + verify(classifier, never()).classify(any(MultipartFile.class), any()); + verify(classifier, times(1)).classify(anyList(), eq(policy)); + } + + @Test + void openProcess_minChargeUnitsFloorApplied(@TempDir Path tmp) throws IOException { + // Policy says min 5 units. Classifier returns 2. Expect 5 to be charged. + PricingPolicy policy = stubPolicy(/*minCharge*/ 5, Map.of(JobSource.WEB, 10)); + when(policyService.getEffectivePolicy(100L)).thenReturn(policy); + ProcessingJob newJob = openJob(UUID.randomUUID()); + when(jobService.joinOrOpen(any(JobContext.class), anyList())) + .thenReturn(new JoinOrOpenResult(newJob, JoinOrOpenResult.Disposition.OPENED)); + when(classifier.classify(any(MultipartFile.class), eq(policy))) + .thenReturn(new DocumentMetrics(10, 1024L, "application/pdf", 2)); + + ChargeOutcome out = + service.openProcess( + new ChargeContext(42L, 100L, JobSource.WEB, ProcessType.SINGLE_TOOL), + List.of(jobInput(tmp, "in.pdf", "application/pdf"))); + + assertThat(out.units()).isEqualTo(5); + } + + @Test + void openProcess_resolvesStepLimitFromPolicy_perJobSource(@TempDir Path tmp) + throws IOException { + // Different limits per source: WEB=10, PIPELINE=20. Verify the right one is passed. + PricingPolicy policy = + stubPolicy(/*minCharge*/ 1, Map.of(JobSource.WEB, 10, JobSource.PIPELINE, 20)); + when(policyService.getEffectivePolicy(100L)).thenReturn(policy); + ProcessingJob newJob = openJob(UUID.randomUUID()); + when(jobService.joinOrOpen(any(JobContext.class), anyList())) + .thenReturn(new JoinOrOpenResult(newJob, JoinOrOpenResult.Disposition.OPENED)); + when(classifier.classify(any(MultipartFile.class), eq(policy))) + .thenReturn(new DocumentMetrics(1, 100L, "application/pdf", 1)); + + service.openProcess( + new ChargeContext(42L, 100L, JobSource.PIPELINE, ProcessType.AUTOMATION), + List.of(jobInput(tmp, "in.pdf", "application/pdf"))); + + ArgumentCaptor ctxCaptor = ArgumentCaptor.forClass(JobContext.class); + verify(jobService).joinOrOpen(ctxCaptor.capture(), anyList()); + assertThat(ctxCaptor.getValue().stepLimit()).isEqualTo(20); + assertThat(ctxCaptor.getValue().source()).isEqualTo(JobSource.PIPELINE); + assertThat(ctxCaptor.getValue().policyId()).isEqualTo(policy.getId()); + } + + @Test + void openProcess_missingStepLimitForSource_fallsBackToConservativeDefault(@TempDir Path tmp) + throws IOException { + // Policy has no entry for DESKTOP_APP — should fall through to the conservative default + // (10) rather than throwing or treating as unlimited. + PricingPolicy policy = stubPolicy(/*minCharge*/ 1, Map.of(JobSource.WEB, 10)); + when(policyService.getEffectivePolicy(100L)).thenReturn(policy); + ProcessingJob newJob = openJob(UUID.randomUUID()); + when(jobService.joinOrOpen(any(JobContext.class), anyList())) + .thenReturn(new JoinOrOpenResult(newJob, JoinOrOpenResult.Disposition.OPENED)); + when(classifier.classify(any(MultipartFile.class), eq(policy))) + .thenReturn(new DocumentMetrics(1, 100L, "application/pdf", 1)); + + service.openProcess( + new ChargeContext(42L, 100L, JobSource.DESKTOP_APP, ProcessType.SINGLE_TOOL), + List.of(jobInput(tmp, "in.pdf", "application/pdf"))); + + ArgumentCaptor ctxCaptor = ArgumentCaptor.forClass(JobContext.class); + verify(jobService).joinOrOpen(ctxCaptor.capture(), anyList()); + assertThat(ctxCaptor.getValue().stepLimit()).isEqualTo(10); + } + + @Test + void openProcess_emptyInputs_throws() { + assertThatThrownBy( + () -> + service.openProcess( + new ChargeContext( + 42L, 100L, JobSource.WEB, ProcessType.SINGLE_TOOL), + List.of())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("inputs must not be empty"); + } + + // --- helpers -------------------------------------------------------------------------------- + + private static PricingPolicy stubPolicy(int minCharge, Map stepLimits) { + PricingPolicy p = new PricingPolicy(); + p.setId(42L); + p.setVersion("v1-test"); + p.setEffectiveFrom(LocalDateTime.now()); + p.setDocPagesPerUnit(25); + p.setDocBytesPerUnit(5L * 1024 * 1024); + p.setMinChargeUnits(minCharge); + p.setFileUnitCap(1000); + p.setStepLimits(new HashMap<>(stepLimits)); + p.setIsDefault(true); + return p; + } + + private static ProcessingJob openJob(UUID id) { + ProcessingJob j = new ProcessingJob(); + j.setId(id); + j.setStatus(JobStatus.OPEN); + j.setStepCount(1); + j.setStartedAt(LocalDateTime.now()); + j.setLastStepAt(LocalDateTime.now()); + j.setSource(JobSource.WEB); + j.setProcessType(ProcessType.SINGLE_TOOL); + return j; + } + + private static JobInput jobInput(Path tmp, String name, String contentType) throws IOException { + Path p = tmp.resolve(name); + Files.writeString(p, "fixture-" + name); + MultipartFile mp = new MockMultipartFile("file", name, contentType, Files.readAllBytes(p)); + return new JobInput(mp, p); + } +} diff --git a/app/saas/src/test/java/stirling/software/saas/payg/job/JobServiceTest.java b/app/saas/src/test/java/stirling/software/saas/payg/job/JobServiceTest.java new file mode 100644 index 000000000..864860e8b --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/job/JobServiceTest.java @@ -0,0 +1,348 @@ +package stirling.software.saas.payg.job; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +import stirling.software.saas.payg.lineage.HashLineageDetector; +import stirling.software.saas.payg.lineage.LineageMatch; +import stirling.software.saas.payg.model.ArtifactKind; +import stirling.software.saas.payg.model.JobSource; +import stirling.software.saas.payg.model.JobStatus; +import stirling.software.saas.payg.model.ProcessType; +import stirling.software.saas.payg.repository.ProcessingJobRepository; +import stirling.software.saas.payg.repository.ProcessingJobStepRepository; + +/** + * Drives {@link JobService} through the multi-input "any-match-joins" lineage policy. Uses a fake + * {@link HashLineageDetector} that pretends each file's signatures are pre-registered with + * particular jobs; persistence goes through mocked repositories. + */ +class JobServiceTest { + + private static final Duration WINDOW = Duration.ofMinutes(5); + + private FakeDetector detector; + private ProcessingJobRepository jobRepo; + private ProcessingJobStepRepository stepRepo; + private JobService service; + + @BeforeEach + void setUp() { + detector = new FakeDetector(); + jobRepo = Mockito.mock(ProcessingJobRepository.class); + stepRepo = Mockito.mock(ProcessingJobStepRepository.class); + // save() returns the same instance so the service can introspect post-write state. + when(jobRepo.save(any(ProcessingJob.class))).thenAnswer(inv -> inv.getArgument(0)); + when(stepRepo.save(any(ProcessingJobStep.class))).thenAnswer(inv -> inv.getArgument(0)); + service = new JobService(detector, jobRepo, stepRepo, WINDOW); + } + + @Test + void constructor_rejectsNonPositiveWindow() { + assertThatThrownBy(() -> new JobService(detector, jobRepo, stepRepo, Duration.ZERO)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy( + () -> + new JobService( + detector, + jobRepo, + stepRepo, + Duration.ofMinutes(5).negated())) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void joinOrOpen_noMatch_opensNewJob(@TempDir Path tmp) throws IOException { + Path input = givenFile(tmp, "in.bin"); + // No detector matches. + + JoinOrOpenResult result = service.joinOrOpen(ctx(42L, 100L, 10), List.of(input)); + + assertThat(result.disposition()).isEqualTo(JoinOrOpenResult.Disposition.OPENED); + assertThat(result.job().getOwnerUserId()).isEqualTo(42L); + assertThat(result.job().getOwnerTeamId()).isEqualTo(100L); + assertThat(result.job().getSource()).isEqualTo(JobSource.WEB); + assertThat(result.job().getStatus()).isEqualTo(JobStatus.OPEN); + assertThat(result.job().getStepCount()).isEqualTo(1); + assertThat(result.job().getId()).isNotNull(); + // Input was recorded under the new job. + assertThat(detector.recorded(result.job().getId(), input, ArtifactKind.INPUT)).isTrue(); + } + + @Test + void joinOrOpen_singleMatch_joinsExistingJob(@TempDir Path tmp) throws IOException { + UUID existingId = UUID.randomUUID(); + ProcessingJob existing = openJob(existingId, 42L, 3, LocalDateTime.now()); + when(jobRepo.findById(existingId)).thenReturn(Optional.of(existing)); + + Path input = givenFile(tmp, "in.bin"); + detector.willMatch( + input, new LineageMatch(existingId, ArtifactKind.INPUT, existing.getLastStepAt())); + + JoinOrOpenResult result = service.joinOrOpen(ctx(42L, 100L, 10), List.of(input)); + + assertThat(result.disposition()).isEqualTo(JoinOrOpenResult.Disposition.JOINED); + assertThat(result.job().getId()).isEqualTo(existingId); + assertThat(result.job().getStepCount()).isEqualTo(4); // 3 -> 4 + // Input is recorded under the joined job too — so future calls on the same input still + // lineage-match to it. + assertThat(detector.recorded(existingId, input, ArtifactKind.INPUT)).isTrue(); + } + + @Test + void joinOrOpen_multiInputAnyMatchJoins(@TempDir Path tmp) throws IOException { + // Inputs A and B; A matches existing job; B is unrelated. Should join A's job. + UUID jobAId = UUID.randomUUID(); + ProcessingJob jobA = openJob(jobAId, 42L, 2, LocalDateTime.now()); + when(jobRepo.findById(jobAId)).thenReturn(Optional.of(jobA)); + + Path inA = givenFile(tmp, "a.bin"); + Path inB = givenFile(tmp, "b.bin"); + detector.willMatch( + inA, new LineageMatch(jobAId, ArtifactKind.OUTPUT, jobA.getLastStepAt())); + // inB has no detector match. + + JoinOrOpenResult result = service.joinOrOpen(ctx(42L, 100L, 10), List.of(inA, inB)); + + assertThat(result.disposition()).isEqualTo(JoinOrOpenResult.Disposition.JOINED); + assertThat(result.job().getId()).isEqualTo(jobAId); + // BOTH inputs recorded under the joined job — inB is now part of jobA's lineage tree. + assertThat(detector.recorded(jobAId, inA, ArtifactKind.INPUT)).isTrue(); + assertThat(detector.recorded(jobAId, inB, ArtifactKind.INPUT)).isTrue(); + } + + @Test + void joinOrOpen_multiMatchPicksNewestByLastStepAt(@TempDir Path tmp) throws IOException { + UUID olderId = UUID.randomUUID(); + UUID newerId = UUID.randomUUID(); + LocalDateTime older = LocalDateTime.now().minus(Duration.ofMinutes(2)); + LocalDateTime newer = LocalDateTime.now(); + ProcessingJob olderJob = openJob(olderId, 42L, 1, older); + ProcessingJob newerJob = openJob(newerId, 42L, 1, newer); + when(jobRepo.findById(newerId)).thenReturn(Optional.of(newerJob)); + + Path inA = givenFile(tmp, "a.bin"); + Path inB = givenFile(tmp, "b.bin"); + detector.willMatch(inA, new LineageMatch(olderId, ArtifactKind.OUTPUT, older)); + detector.willMatch(inB, new LineageMatch(newerId, ArtifactKind.OUTPUT, newer)); + + JoinOrOpenResult result = service.joinOrOpen(ctx(42L, 100L, 10), List.of(inA, inB)); + + assertThat(result.disposition()).isEqualTo(JoinOrOpenResult.Disposition.JOINED); + assertThat(result.job().getId()).isEqualTo(newerId); + // Older job was NOT looked up — newest wins purely by lastStepAt comparison from the + // LineageMatch payload, before any DB lookup. + verify(jobRepo, never()).findById(olderId); + } + + @Test + void joinOrOpen_stepLimitHit_spawnsNewJob(@TempDir Path tmp) throws IOException { + UUID existingId = UUID.randomUUID(); + // stepCount equal to limit (10) → can't append, must spawn new. + ProcessingJob existing = openJob(existingId, 42L, 10, LocalDateTime.now()); + when(jobRepo.findById(existingId)).thenReturn(Optional.of(existing)); + + Path input = givenFile(tmp, "in.bin"); + detector.willMatch( + input, new LineageMatch(existingId, ArtifactKind.OUTPUT, existing.getLastStepAt())); + + JoinOrOpenResult result = service.joinOrOpen(ctx(42L, 100L, 10), List.of(input)); + + assertThat(result.disposition()).isEqualTo(JoinOrOpenResult.Disposition.OPENED); + assertThat(result.job().getId()).isNotEqualTo(existingId); + assertThat(result.job().getStepCount()).isEqualTo(1); + // The original job's stepCount must NOT have been incremented (we abandoned it). + assertThat(existing.getStepCount()).isEqualTo(10); + // Input is recorded under the NEW job so subsequent calls lineage-match to it + // (mostRecentMatchWins shifts the chain forward). + assertThat(detector.recorded(result.job().getId(), input, ArtifactKind.INPUT)).isTrue(); + } + + @Test + void joinOrOpen_emptyInputs_throws() { + assertThatThrownBy(() -> service.joinOrOpen(ctx(42L, 100L, 10), List.of())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("inputs must not be empty"); + } + + @Test + void joinOrOpen_matchPointsAtMissingJob_throwsStateException(@TempDir Path tmp) + throws IOException { + UUID staleId = UUID.randomUUID(); + when(jobRepo.findById(staleId)).thenReturn(Optional.empty()); + + Path input = givenFile(tmp, "in.bin"); + detector.willMatch( + input, new LineageMatch(staleId, ArtifactKind.OUTPUT, LocalDateTime.now())); + + assertThatThrownBy(() -> service.joinOrOpen(ctx(42L, 100L, 10), List.of(input))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("stale signature"); + } + + @Test + void recordOutput_delegatesToDetector(@TempDir Path tmp) throws IOException { + UUID jobId = UUID.randomUUID(); + Path output = givenFile(tmp, "out.bin"); + + service.recordOutput(jobId, output); + + assertThat(detector.recorded(jobId, output, ArtifactKind.OUTPUT)).isTrue(); + } + + @Test + void close_idempotent() { + UUID jobId = UUID.randomUUID(); + ProcessingJob open = openJob(jobId, 42L, 5, LocalDateTime.now()); + when(jobRepo.findById(jobId)).thenReturn(Optional.of(open)); + + ProcessingJob first = service.close(jobId); + assertThat(first.getStatus()).isEqualTo(JobStatus.CLOSED); + assertThat(first.getClosedAt()).isNotNull(); + + // Re-call: status already CLOSED, should return the same row without saving again. + ProcessingJob second = service.close(jobId); + assertThat(second.getStatus()).isEqualTo(JobStatus.CLOSED); + // save was called exactly once across both close() calls. + verify(jobRepo, times(1)).save(any(ProcessingJob.class)); + } + + @Test + void close_unknownJob_throws() { + UUID unknown = UUID.randomUUID(); + when(jobRepo.findById(unknown)).thenReturn(Optional.empty()); + assertThatThrownBy(() -> service.close(unknown)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("No ProcessingJob"); + } + + @Test + void closeStale_closesAllStaleJobs() { + ProcessingJob a = + openJob( + UUID.randomUUID(), + 42L, + 1, + LocalDateTime.now().minus(Duration.ofMinutes(10))); + ProcessingJob b = + openJob( + UUID.randomUUID(), + 42L, + 1, + LocalDateTime.now().minus(Duration.ofMinutes(20))); + when(jobRepo.findStale(eq(JobStatus.OPEN), any(LocalDateTime.class))) + .thenReturn(List.of(a, b)); + + int closed = service.closeStale(); + + assertThat(closed).isEqualTo(2); + assertThat(a.getStatus()).isEqualTo(JobStatus.CLOSED); + assertThat(b.getStatus()).isEqualTo(JobStatus.CLOSED); + verify(jobRepo).saveAll(List.of(a, b)); + } + + @Test + void closeStale_emptyResult_noSave() { + when(jobRepo.findStale(eq(JobStatus.OPEN), any(LocalDateTime.class))).thenReturn(List.of()); + assertThat(service.closeStale()).isZero(); + verify(jobRepo, never()).saveAll(any()); + } + + @Test + void appendStep_persistsStepRow() { + UUID jobId = UUID.randomUUID(); + ArgumentCaptor captor = ArgumentCaptor.forClass(ProcessingJobStep.class); + + service.appendStep( + jobId, + "/api/v1/general/split", + stirling.software.saas.payg.model.JobStepStatus.OK, + 42, + 12_345L, + null); + + verify(stepRepo, atLeastOnce()).save(captor.capture()); + ProcessingJobStep saved = captor.getValue(); + assertThat(saved.getJobId()).isEqualTo(jobId); + assertThat(saved.getToolId()).isEqualTo("/api/v1/general/split"); + assertThat(saved.getInputPages()).isEqualTo(42); + assertThat(saved.getInputBytes()).isEqualTo(12_345L); + } + + // --- helpers -------------------------------------------------------------------------------- + + private static JobContext ctx(long userId, long teamId, int stepLimit) { + return new JobContext( + userId, teamId, JobSource.WEB, ProcessType.SINGLE_TOOL, 1L, stepLimit); + } + + private static ProcessingJob openJob( + UUID id, long userId, int stepCount, LocalDateTime lastStepAt) { + ProcessingJob j = new ProcessingJob(); + j.setId(id); + j.setOwnerUserId(userId); + j.setStatus(JobStatus.OPEN); + j.setStepCount(stepCount); + j.setStartedAt(lastStepAt.minus(Duration.ofMinutes(1))); + j.setLastStepAt(lastStepAt); + j.setProcessType(ProcessType.SINGLE_TOOL); + j.setSource(JobSource.WEB); + j.setPolicyId(1L); + return j; + } + + private static Path givenFile(Path tmp, String name) throws IOException { + Path p = tmp.resolve(name); + Files.writeString(p, "fixture-" + name); + return p; + } + + /** Test double: programmable detector + records observations. */ + private static class FakeDetector implements HashLineageDetector { + private final Map matches = new HashMap<>(); + private final Set observations = new java.util.HashSet<>(); + + void willMatch(Path input, LineageMatch match) { + matches.put(input, match); + } + + boolean recorded(UUID jobId, Path file, ArtifactKind kind) { + return observations.contains(jobId + "|" + file + "|" + kind); + } + + @Override + public Optional detect(Long userId, Path inputFile) { + return Optional.ofNullable(matches.get(inputFile)); + } + + @Override + public void record(UUID jobId, Path file, ArtifactKind kind) { + observations.add(jobId + "|" + file + "|" + kind); + } + } +} diff --git a/app/saas/src/test/java/stirling/software/saas/payg/job/StaleJobCloserTest.java b/app/saas/src/test/java/stirling/software/saas/payg/job/StaleJobCloserTest.java new file mode 100644 index 000000000..e526bf7ca --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/job/StaleJobCloserTest.java @@ -0,0 +1,35 @@ +package stirling.software.saas.payg.job; + +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +/** + * Smoke test for the scheduler wiring. The interesting close logic is exercised in {@link + * JobServiceTest#closeStale_closesAllStaleJobs}; here we just confirm the scheduler bean delegates + * to {@code JobService.closeStale} and tolerates an empty result without erroring. + */ +class StaleJobCloserTest { + + @Test + void closeStale_invokesJobService() { + JobService jobService = Mockito.mock(JobService.class); + when(jobService.closeStale()).thenReturn(3); + + new StaleJobCloser(jobService).closeStale(); + + verify(jobService).closeStale(); + } + + @Test + void closeStale_zeroClosedDoesNotThrow() { + JobService jobService = Mockito.mock(JobService.class); + when(jobService.closeStale()).thenReturn(0); + + new StaleJobCloser(jobService).closeStale(); + + verify(jobService).closeStale(); + } +} diff --git a/app/saas/src/test/java/stirling/software/saas/payg/lineage/LineagePruneSchedulerTest.java b/app/saas/src/test/java/stirling/software/saas/payg/lineage/LineagePruneSchedulerTest.java new file mode 100644 index 000000000..f7920afad --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/lineage/LineagePruneSchedulerTest.java @@ -0,0 +1,52 @@ +package stirling.software.saas.payg.lineage; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Duration; +import java.time.Instant; + +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +class LineagePruneSchedulerTest { + + @Test + void prune_passesNowMinusRetentionToStore() { + JobLineageStore store = Mockito.mock(JobLineageStore.class); + when(store.pruneOlderThan(any(Instant.class))).thenReturn(7); + Duration retention = Duration.ofHours(1); + + Instant before = Instant.now(); + new LineagePruneScheduler(store, retention).prune(); + Instant after = Instant.now(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(Instant.class); + verify(store).pruneOlderThan(captor.capture()); + Instant cutoff = captor.getValue(); + // cutoff should sit inside [before-retention, after-retention] — bounds the wall-clock + // drift the test itself introduces between the captor and our reference reads. + assertThat(cutoff).isBetween(before.minus(retention), after.minus(retention)); + } + + @Test + void constructor_rejectsNonPositiveRetention() { + JobLineageStore store = Mockito.mock(JobLineageStore.class); + assertThatThrownBy(() -> new LineagePruneScheduler(store, Duration.ZERO)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> new LineagePruneScheduler(store, Duration.ofMinutes(5).negated())) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void prune_zeroDeletedDoesNotThrow() { + JobLineageStore store = Mockito.mock(JobLineageStore.class); + when(store.pruneOlderThan(any(Instant.class))).thenReturn(0); + new LineagePruneScheduler(store, Duration.ofHours(1)).prune(); + verify(store).pruneOlderThan(any(Instant.class)); + } +}