diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractor.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractor.java new file mode 100644 index 000000000..86e6e4ab4 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractor.java @@ -0,0 +1,61 @@ +package stirling.software.saas.payg.lineage; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.DigestInputStream; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; +import java.util.Set; + +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Component; + +/** + * SHA-256 of the file's bytes. The simplest universally-applicable signature — works for every + * content type, doesn't parse, doesn't allocate proportional to file size (fixed 64 KiB read + * buffer), hardware-accelerated by the JVM on modern hardware (Intel SHA-NI, ARM SHA extensions). + * + *

Always returns exactly one {@link LineageSignature} of type {@code "sha256"}. A future {@code + * PdfMetadataSignatureExtractor} would be a separate bean and add its own signature type — composed + * at the detector layer, no interaction needed here. + */ +@Component +@Profile("saas") +public class ByteHashSignatureExtractor implements LineageSignatureExtractor { + + private static final String ALGORITHM = "SHA-256"; + private static final String SIGNATURE_TYPE = "sha256"; + private static final int BUFFER_SIZE = 64 * 1024; + + @Override + public Set extract(Path file) throws IOException { + MessageDigest digest = newDigest(); + try (InputStream raw = Files.newInputStream(file); + DigestInputStream in = new DigestInputStream(raw, digest)) { + byte[] buf = new byte[BUFFER_SIZE]; + // Drain through the digest stream; we only care about side effects on the digest. + while (in.read(buf) != -1) { + // no-op + } + } + String hex = HexFormat.of().formatHex(digest.digest()); + return Set.of(new LineageSignature(SIGNATURE_TYPE, hex)); + } + + @Override + public String name() { + return SIGNATURE_TYPE; + } + + private static MessageDigest newDigest() { + try { + return MessageDigest.getInstance(ALGORITHM); + } catch (NoSuchAlgorithmException e) { + // SHA-256 is mandated by every JDK; unreachable in practice. + throw new IllegalStateException(ALGORITHM + " unavailable — JDK is misconfigured", e); + } + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java new file mode 100644 index 000000000..2d261ee02 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java @@ -0,0 +1,111 @@ +package stirling.software.saas.payg.lineage; + +import java.io.IOException; +import java.nio.file.Path; +import java.time.Duration; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Component; + +import lombok.extern.slf4j.Slf4j; + +import stirling.software.saas.payg.model.ArtifactKind; + +/** + * Default detector. Composes signatures from every registered {@link LineageSignatureExtractor} + * (currently just SHA-256 byte hash; future PDF-aware extractors are drop-in additions) and + * delegates lookup + persistence to a {@link JobLineageStore}. + */ +@Slf4j +@Component +@Profile("saas") +public class DefaultHashLineageDetector implements HashLineageDetector { + + private final List extractors; + private final JobLineageStore store; + private final Duration workflowWindow; + + public DefaultHashLineageDetector( + List extractors, + JobLineageStore store, + @Value("${payg.lineage.workflow-window:PT5M}") Duration workflowWindow) { + Objects.requireNonNull(extractors, "extractors"); + Objects.requireNonNull(store, "store"); + Objects.requireNonNull(workflowWindow, "workflowWindow"); + if (extractors.isEmpty()) { + throw new IllegalStateException( + "DefaultHashLineageDetector requires at least one LineageSignatureExtractor" + + " bean — none registered."); + } + if (workflowWindow.isNegative() || workflowWindow.isZero()) { + // A non-positive window would mean "since = now + |window|", so the > comparison + // only matches jobs in the future — i.e. nothing matches, silently. Fail loud + // instead. + throw new IllegalArgumentException( + "payg.lineage.workflow-window must be positive, got " + workflowWindow); + } + this.extractors = List.copyOf(extractors); + this.store = store; + this.workflowWindow = workflowWindow; + } + + @Override + public Optional detect(Long userId, Path inputFile) throws IOException { + Objects.requireNonNull(userId, "userId"); + Objects.requireNonNull(inputFile, "inputFile"); + + Set signatures = extractAll(inputFile); + if (signatures.isEmpty()) { + // No extractor produced anything for this content. Treat as no-match. + log.debug("No signatures extracted from {}; lineage check returns empty.", inputFile); + return Optional.empty(); + } + + return store.findOpenJobForSignatures(userId, signatures, workflowWindow); + } + + @Override + public void record(UUID jobId, Path file, ArtifactKind kind) throws IOException { + Objects.requireNonNull(jobId, "jobId"); + Objects.requireNonNull(file, "file"); + Objects.requireNonNull(kind, "kind"); + + Set signatures = extractAll(file); + if (signatures.isEmpty()) { + log.debug( + "No signatures extracted from {} for job {} ({}); nothing recorded.", + file, + jobId, + kind); + return; + } + store.record(jobId, signatures, kind); + } + + private Set extractAll(Path file) { + Set union = new HashSet<>(); + for (LineageSignatureExtractor extractor : extractors) { + try { + union.addAll(extractor.extract(file)); + } catch (IOException e) { + // A single extractor failing on file IO / format parse (e.g. PDF-aware extractor + // on a malformed PDF) must not block other extractors from contributing. Log and + // continue. RuntimeExceptions deliberately propagate — they signal bugs we want + // to surface, not "expected" extractor-doesn't-fit-this-content failures. + log.debug( + "Extractor '{}' failed on {} ({}); continuing with other extractors.", + extractor.name(), + file, + e.getMessage()); + } + } + return union; + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java new file mode 100644 index 000000000..60aa589bf --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java @@ -0,0 +1,36 @@ +package stirling.software.saas.payg.lineage; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.Optional; +import java.util.UUID; + +import stirling.software.saas.payg.model.ArtifactKind; + +/** + * High-level lineage API. Decides whether a tool call should join an existing open process by + * comparing the incoming file's signatures against signatures previously recorded for that user's + * open processes. + * + *

Two operations: + * + *

+ * + *

Both methods take a {@link Path}; the caller is expected to have already materialised the + * upload / response body to a managed temp file (via {@code TempFileManager}). The detector + * delegates signature extraction to one or more {@link LineageSignatureExtractor}s and storage to a + * {@link JobLineageStore}, both of which are swappable. + */ +public interface HashLineageDetector { + + /** Looks up an open process for this user whose recorded signatures match the input file. */ + Optional detect(Long userId, Path inputFile) throws IOException; + + /** Records the file's signatures against the given job as either INPUT or OUTPUT. */ + void record(UUID jobId, Path file, ArtifactKind kind) throws IOException; +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java new file mode 100644 index 000000000..7ab46d9bd --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java @@ -0,0 +1,39 @@ +package stirling.software.saas.payg.lineage; + +import java.time.Duration; +import java.time.Instant; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; + +import stirling.software.saas.payg.model.ArtifactKind; + +/** + * Persistence boundary for lineage data. Two operations: record signatures against a job, and look + * up the open job (if any) that previously recorded a matching signature for the same user. + * + *

This interface deliberately knows nothing about the storage technology — the production impl + * is JPA-backed against {@code job_artifact_hash}, and a future Redis-backed impl will be a + * straight swap. The detector and any callers only see this interface. + * + *

Signature values are stored using {@link LineageSignature#asStorageKey()}, so multiple + * signature types (SHA-256, PDF {@code /ID}, etc.) coexist on the same table without a separate + * column. + */ +public interface JobLineageStore { + + /** Records every signature in {@code signatures} against the given job + artifact kind. */ + void record(UUID jobId, Set signatures, ArtifactKind kind); + + /** + * Finds the most-recently-active open job (owned by {@code userId}, whose {@code last_step_at} + * is within {@code workflowWindow}) whose recorded artifacts include any of the supplied + * candidate signatures. Returns the first match; ordering across multiple matches is by job + * activity recency. + */ + Optional findOpenJobForSignatures( + Long userId, Set candidates, Duration workflowWindow); + + /** Deletes records created before {@code cutoff}. Returns the number of rows removed. */ + int pruneOlderThan(Instant cutoff); +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java new file mode 100644 index 000000000..4da4cd0b5 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java @@ -0,0 +1,88 @@ +package stirling.software.saas.payg.lineage; + +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; + +import org.springframework.context.annotation.Profile; +import org.springframework.data.domain.Limit; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; + +import lombok.RequiredArgsConstructor; + +import stirling.software.saas.payg.job.JobArtifactHash; +import stirling.software.saas.payg.job.JobArtifactHash.JobArtifactHashId; +import stirling.software.saas.payg.model.ArtifactKind; +import stirling.software.saas.payg.model.JobStatus; +import stirling.software.saas.payg.repository.JobArtifactHashRepository; + +/** + * JPA-backed {@link JobLineageStore} against the {@code job_artifact_hash} table. The lookup runs + * as a single joined query against {@code processing_job} so status + window filtering happens at + * the database, not in-process. + * + *

Signatures are persisted using {@link LineageSignature#asStorageKey()} ({@code "type:value"}) + * so multiple signature types coexist on the same column without a schema change. + */ +@Component +@Profile("saas") +@RequiredArgsConstructor +public class JpaJobLineageStore implements JobLineageStore { + + private final JobArtifactHashRepository hashRepository; + + @Override + @Transactional + public void record(UUID jobId, Set signatures, ArtifactKind kind) { + Objects.requireNonNull(jobId, "jobId"); + Objects.requireNonNull(signatures, "signatures"); + Objects.requireNonNull(kind, "kind"); + if (signatures.isEmpty()) { + return; + } + // saveAll + @Transactional → one transaction, all-or-nothing. Without this, a multi- + // signature record() could leave partial state if a save mid-way fails. + List rows = new ArrayList<>(signatures.size()); + for (LineageSignature signature : signatures) { + JobArtifactHash row = new JobArtifactHash(); + row.setId(new JobArtifactHashId(jobId, signature.asStorageKey(), kind)); + rows.add(row); + } + hashRepository.saveAll(rows); + } + + @Override + public Optional findOpenJobForSignatures( + Long userId, Set candidates, Duration workflowWindow) { + Objects.requireNonNull(userId, "userId"); + Objects.requireNonNull(candidates, "candidates"); + Objects.requireNonNull(workflowWindow, "workflowWindow"); + if (candidates.isEmpty()) { + return Optional.empty(); + } + + List storageKeys = candidates.stream().map(LineageSignature::asStorageKey).toList(); + LocalDateTime since = LocalDateTime.now().minus(workflowWindow); + + List matches = + hashRepository.findOpenJobsForSignatures( + userId, JobStatus.OPEN, since, storageKeys, Limit.of(1)); + return matches.isEmpty() ? Optional.empty() : Optional.of(matches.get(0)); + } + + @Override + @Transactional + public int pruneOlderThan(Instant cutoff) { + Objects.requireNonNull(cutoff, "cutoff"); + return hashRepository.deleteOlderThan( + LocalDateTime.ofInstant(cutoff, ZoneId.systemDefault())); + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageMatch.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageMatch.java new file mode 100644 index 000000000..c22759614 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageMatch.java @@ -0,0 +1,19 @@ +package stirling.software.saas.payg.lineage; + +import java.time.LocalDateTime; +import java.util.UUID; + +import stirling.software.saas.payg.model.ArtifactKind; + +/** + * Result of a successful lineage lookup: the open job a tool call should join, plus the {@link + * ArtifactKind} of the recorded artifact that produced the match (was the matched hash the prior + * call's input or its output?) and the job's {@code last_step_at} so the caller can decide whether + * to extend the workflow window. + * + *

If a single job has multiple recorded signatures that all match the candidate set (e.g. both + * its INPUT and OUTPUT hashes align with what the caller offered), {@code matchedKind} reflects one + * of those matches — ordering across same-job-different-kind rows is unspecified. Callers needing + * to enumerate every matched kind should query {@link JobLineageStore} directly. + */ +public record LineageMatch(UUID jobId, ArtifactKind matchedKind, LocalDateTime jobLastStepAt) {} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignature.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignature.java new file mode 100644 index 000000000..3402b4485 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignature.java @@ -0,0 +1,50 @@ +package stirling.software.saas.payg.lineage; + +import java.util.Objects; + +/** + * A single identity signal for a piece of content. The detector treats {@code (type, value)} as an + * opaque equality key — two signatures match if both components are equal. Format: + * + *

+ * + *

Persisted as the concatenation {@code "{type}:{value}"} so multiple signature types can + * coexist on the same {@code job_artifact_hash} table without a separate column. + */ +public record LineageSignature(String type, String value) { + + public LineageSignature { + Objects.requireNonNull(type, "type"); + Objects.requireNonNull(value, "value"); + if (type.isBlank()) { + throw new IllegalArgumentException("signature type must not be blank"); + } + if (type.contains(":")) { + throw new IllegalArgumentException("signature type must not contain ':'"); + } + if (value.isBlank()) { + throw new IllegalArgumentException("signature value must not be blank"); + } + } + + /** Storage form: {@code "type:value"}. */ + public String asStorageKey() { + return type + ":" + value; + } + + /** Parses a storage-form key back into a {@code LineageSignature}. */ + public static LineageSignature fromStorageKey(String key) { + Objects.requireNonNull(key, "key"); + int colon = key.indexOf(':'); + if (colon <= 0 || colon == key.length() - 1) { + throw new IllegalArgumentException( + "Storage key must be of the form 'type:value': " + key); + } + return new LineageSignature(key.substring(0, colon), key.substring(colon + 1)); + } +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignatureExtractor.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignatureExtractor.java new file mode 100644 index 000000000..7b5f94b65 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/LineageSignatureExtractor.java @@ -0,0 +1,34 @@ +package stirling.software.saas.payg.lineage; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.Set; + +/** + * Extracts one or more {@link LineageSignature}s from a file. Lineage is decided at the file level: + * the extractor sees the whole content and returns whatever identity signals it can pull out. + * + *

The default {@link ByteHashSignatureExtractor} returns a single SHA-256 of the bytes. A future + * format-aware extractor (e.g. for PDFs, opening with jpdfium to read {@code /ID[0]} or a + * content-stream hash) is just another {@link org.springframework.stereotype.Component} bean + * implementing this interface — no changes to the detector or the store. Multiple extractors can be + * composed; the detector unions their signatures and matches on any of them. + * + *

Callers materialise upload bodies / response bodies to a {@link Path} (via {@code + * TempFileManager}) before extraction so the extractor can stream-hash and, optionally, + * parse the file with a format-specific library — both from the same on-disk byte sequence. + */ +public interface LineageSignatureExtractor { + + /** + * Returns every signature this extractor can derive from {@code file}. Empty set is legal — + * meaning "this extractor doesn't know how to fingerprint this content" — and is normal for + * format-specific extractors that decline to handle non-matching inputs. + */ + Set extract(Path file) throws IOException; + + /** + * Short, stable name for logging and diagnostics ({@code "sha256"}, {@code "pdf-id"}, etc.). + */ + String name(); +} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java b/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java index c9cb2af9c..0c3af74db 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java +++ b/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java @@ -1,8 +1,10 @@ package stirling.software.saas.payg.repository; import java.time.LocalDateTime; +import java.util.Collection; import java.util.List; +import org.springframework.data.domain.Limit; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; @@ -11,6 +13,7 @@ import org.springframework.stereotype.Repository; import stirling.software.saas.payg.job.JobArtifactHash; import stirling.software.saas.payg.job.JobArtifactHash.JobArtifactHashId; +import stirling.software.saas.payg.lineage.LineageMatch; import stirling.software.saas.payg.model.JobStatus; @Repository @@ -18,21 +21,27 @@ public interface JobArtifactHashRepository extends JpaRepository { /** - * Lineage lookup: find the open job (if any) whose recorded input/output hashes include the - * supplied content hash, scoped to one user and the workflow window. + * Lineage lookup: find open jobs (owned by {@code userId}, {@code last_step_at > since}) whose + * recorded artifacts include any of the supplied storage-form signature keys. Ordered by job + * activity recency. Caller passes {@link Limit#of(int)} to bound the result set — for the + * single-match hot path use {@code Limit.of(1)} so the DB doesn't materialise unwanted rows. */ @Query( - "SELECT j.ownerUserId, h.id.jobId FROM JobArtifactHash h" + "SELECT new stirling.software.saas.payg.lineage.LineageMatch(" + + " h.id.jobId, h.id.kind, j.lastStepAt)" + + " FROM JobArtifactHash h" + " JOIN ProcessingJob j ON j.id = h.id.jobId" + " WHERE j.ownerUserId = :userId" + " AND j.status = :openStatus" + " AND j.lastStepAt > :since" - + " AND h.id.contentHash = :contentHash") - List findLineageMatches( + + " AND h.id.contentHash IN :signatures" + + " ORDER BY j.lastStepAt DESC") + List findOpenJobsForSignatures( @Param("userId") Long userId, @Param("openStatus") JobStatus openStatus, @Param("since") LocalDateTime since, - @Param("contentHash") String contentHash); + @Param("signatures") Collection signatures, + Limit limit); /** Prunes rows older than {@code cutoff}; run from a scheduled task. */ @Modifying diff --git a/app/saas/src/test/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractorTest.java b/app/saas/src/test/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractorTest.java new file mode 100644 index 000000000..d25617469 --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/lineage/ByteHashSignatureExtractorTest.java @@ -0,0 +1,80 @@ +package stirling.software.saas.payg.lineage; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Set; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class ByteHashSignatureExtractorTest { + + private final ByteHashSignatureExtractor extractor = new ByteHashSignatureExtractor(); + + @Test + void identicalBytes_produceIdenticalSignatures(@TempDir Path tmp) throws IOException { + Path a = tmp.resolve("a.bin"); + Path b = tmp.resolve("b.bin"); + byte[] contents = "hello, lineage".getBytes(); + Files.write(a, contents); + Files.write(b, contents); + + Set sigsA = extractor.extract(a); + Set sigsB = extractor.extract(b); + + assertThat(sigsA).isEqualTo(sigsB); + assertThat(sigsA).hasSize(1); + assertThat(sigsA.iterator().next().type()).isEqualTo("sha256"); + } + + @Test + void differentBytes_produceDifferentSignatures(@TempDir Path tmp) throws IOException { + Path a = tmp.resolve("a.bin"); + Path b = tmp.resolve("b.bin"); + Files.write(a, "one".getBytes()); + Files.write(b, "two".getBytes()); + + Set sigsA = extractor.extract(a); + Set sigsB = extractor.extract(b); + + assertThat(sigsA).isNotEqualTo(sigsB); + } + + @Test + void emptyFile_isHashable(@TempDir Path tmp) throws IOException { + Path empty = tmp.resolve("empty.bin"); + Files.write(empty, new byte[0]); + + Set sigs = extractor.extract(empty); + + // SHA-256 of the empty string is a well-known constant. + assertThat(sigs).hasSize(1); + assertThat(sigs.iterator().next().value()) + .isEqualTo("e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"); + } + + @Test + void largeFile_streamsWithoutLoadingFullyInMemory(@TempDir Path tmp) throws IOException { + // 10 MiB file. The extractor uses a 64 KiB buffer so this should hash without trouble. + Path big = tmp.resolve("big.bin"); + byte[] block = new byte[64 * 1024]; + try (var out = Files.newOutputStream(big)) { + for (int i = 0; i < 10 * 16; i++) { + out.write(block); + } + } + + Set sigs = extractor.extract(big); + + assertThat(sigs).hasSize(1); + assertThat(sigs.iterator().next().type()).isEqualTo("sha256"); + } + + @Test + void name_isStable() { + assertThat(extractor.name()).isEqualTo("sha256"); + } +} diff --git a/app/saas/src/test/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetectorTest.java b/app/saas/src/test/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetectorTest.java new file mode 100644 index 000000000..8516416f6 --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetectorTest.java @@ -0,0 +1,277 @@ +package stirling.software.saas.payg.lineage; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +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 stirling.software.saas.payg.model.ArtifactKind; +import stirling.software.saas.payg.model.JobStatus; + +/** + * Exercises the detector end-to-end against the in-memory store + a fake extractor with + * predetermined signatures. Covers the matching rules: same-user, within window, status=OPEN, + * multi-signature, and post-record retrieval. The contract is shared with the JPA store; if Redis + * (or any future) impl satisfies the same {@link JobLineageStore} interface, these tests are the + * canonical "did I get the abstraction right?" check. + */ +class DefaultHashLineageDetectorTest { + + private static final Duration WINDOW = Duration.ofMinutes(5); + + private FakeFileSignatureExtractor extractor; + private InMemoryJobLineageStore store; + private DefaultHashLineageDetector detector; + + @BeforeEach + void setUp() { + extractor = new FakeFileSignatureExtractor(); + store = new InMemoryJobLineageStore(); + detector = new DefaultHashLineageDetector(List.of(extractor), store, WINDOW); + } + + @Test + void noPriorJobs_returnsEmpty(@TempDir Path tmp) throws IOException { + Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc"); + + Optional match = detector.detect(42L, file); + + assertThat(match).isEmpty(); + } + + @Test + void matchingInputSignature_returnsLineageMatch(@TempDir Path tmp) throws IOException { + UUID job = openJobForUser(42L, LocalDateTime.now()); + recordSignatureForJob(job, "sha256", "abc", ArtifactKind.INPUT); + Path file = givenFileWithSignatures(tmp, "next.bin", "sha256", "abc"); + + Optional match = detector.detect(42L, file); + + assertThat(match).isPresent(); + assertThat(match.get().jobId()).isEqualTo(job); + assertThat(match.get().matchedKind()).isEqualTo(ArtifactKind.INPUT); + } + + @Test + void matchingOutputSignature_alsoCounts(@TempDir Path tmp) throws IOException { + // OCR's output hash matches Compress's input hash — that's the lineage chain. + UUID ocr = openJobForUser(42L, LocalDateTime.now()); + recordSignatureForJob(ocr, "sha256", "ocr-output-hash", ArtifactKind.OUTPUT); + Path compressInput = + givenFileWithSignatures(tmp, "to-compress.bin", "sha256", "ocr-output-hash"); + + Optional match = detector.detect(42L, compressInput); + + assertThat(match).isPresent(); + assertThat(match.get().jobId()).isEqualTo(ocr); + assertThat(match.get().matchedKind()).isEqualTo(ArtifactKind.OUTPUT); + } + + @Test + void differentUser_doesNotMatch(@TempDir Path tmp) throws IOException { + UUID job = openJobForUser(/* userA= */ 1L, LocalDateTime.now()); + recordSignatureForJob(job, "sha256", "abc", ArtifactKind.INPUT); + Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc"); + + Optional match = detector.detect(/* userB= */ 2L, file); + + assertThat(match).isEmpty(); + } + + @Test + void outsideWorkflowWindow_doesNotMatch(@TempDir Path tmp) throws IOException { + // Job last touched 10 minutes ago; window is 5 minutes. + UUID job = openJobForUser(42L, LocalDateTime.now().minus(Duration.ofMinutes(10))); + recordSignatureForJob(job, "sha256", "abc", ArtifactKind.INPUT); + Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc"); + + Optional match = detector.detect(42L, file); + + assertThat(match).isEmpty(); + } + + @Test + void closedJob_doesNotMatch(@TempDir Path tmp) throws IOException { + UUID job = UUID.randomUUID(); + store.registerJob(job, 42L, JobStatus.CLOSED, LocalDateTime.now()); + recordSignatureForJob(job, "sha256", "abc", ArtifactKind.OUTPUT); + Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc"); + + Optional match = detector.detect(42L, file); + + assertThat(match).isEmpty(); + } + + @Test + void multipleSignatures_matchIfAnyMatches(@TempDir Path tmp) throws IOException { + // The job was recorded with one signature (sha256). The incoming file produces TWO + // signatures (sha256 + pdf-id). Different sha256 (bytes changed) but same pdf-id → match. + UUID job = openJobForUser(42L, LocalDateTime.now()); + recordSignatureForJob(job, "pdf-id", "stable-uuid", ArtifactKind.OUTPUT); + + Path file = + givenFileWithSignatures( + tmp, + "input.pdf", + // Two signatures from the same file: + "sha256", + "differs-because-bytes-changed", + "pdf-id", + "stable-uuid"); + + Optional match = detector.detect(42L, file); + + assertThat(match).isPresent(); + assertThat(match.get().jobId()).isEqualTo(job); + } + + @Test + void mostRecentMatchWins_whenMultipleJobsHaveTheSignature(@TempDir Path tmp) + throws IOException { + UUID older = openJobForUser(42L, LocalDateTime.now().minus(Duration.ofMinutes(2))); + UUID newer = openJobForUser(42L, LocalDateTime.now()); + recordSignatureForJob(older, "sha256", "shared", ArtifactKind.OUTPUT); + recordSignatureForJob(newer, "sha256", "shared", ArtifactKind.OUTPUT); + Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "shared"); + + Optional match = detector.detect(42L, file); + + assertThat(match).isPresent(); + assertThat(match.get().jobId()).isEqualTo(newer); + } + + @Test + void record_persistsSignatures(@TempDir Path tmp) throws IOException { + UUID job = openJobForUser(42L, LocalDateTime.now()); + Path file = givenFileWithSignatures(tmp, "output.bin", "sha256", "fresh-hash"); + + detector.record(job, file, ArtifactKind.OUTPUT); + + // Now a fresh detect() call against a different file with the same hash should match. + Path subsequent = givenFileWithSignatures(tmp, "next.bin", "sha256", "fresh-hash"); + Optional match = detector.detect(42L, subsequent); + + assertThat(match).isPresent(); + assertThat(match.get().jobId()).isEqualTo(job); + assertThat(match.get().matchedKind()).isEqualTo(ArtifactKind.OUTPUT); + } + + @Test + void extractorThrowing_doesNotBreakRecording(@TempDir Path tmp) throws IOException { + // Composite scenario: one of two extractors throws; the other still contributes. + FakeFileSignatureExtractor throwing = + new FakeFileSignatureExtractor() { + @Override + public Set extract(Path file) throws IOException { + throw new IOException( + "synthetic — pretend the file is malformed for this type"); + } + + @Override + public String name() { + return "throwing-extractor"; + } + }; + DefaultHashLineageDetector composite = + new DefaultHashLineageDetector(List.of(throwing, extractor), store, WINDOW); + UUID job = openJobForUser(42L, LocalDateTime.now()); + + Path file = givenFileWithSignatures(tmp, "input.bin", "sha256", "abc"); + composite.record(job, file, ArtifactKind.INPUT); + + // The throwing extractor's signatures didn't land, but the byte-hash one's did, so a + // subsequent detect on a matching sha256 hash still finds the job. + Path next = givenFileWithSignatures(tmp, "next.bin", "sha256", "abc"); + Optional match = composite.detect(42L, next); + assertThat(match).isPresent(); + } + + @Test + void constructor_rejectsEmptyExtractorList() { + assertThatThrownBy(() -> new DefaultHashLineageDetector(List.of(), store, WINDOW)) + .isInstanceOf(IllegalStateException.class); + } + + @Test + void constructor_rejectsNonPositiveWorkflowWindow() { + // Zero window: since == now, `> :since` never matches. + assertThatThrownBy( + () -> + new DefaultHashLineageDetector( + List.of(extractor), store, Duration.ZERO)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must be positive"); + + // Negative window: since == now + |window|, `> :since` only matches the future. Silent + // no-match — must fail loud at startup instead. + assertThatThrownBy( + () -> + new DefaultHashLineageDetector( + List.of(extractor), store, Duration.ofMinutes(5).negated())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must be positive"); + } + + // --- helpers -------------------------------------------------------------------------------- + + private UUID openJobForUser(long userId, LocalDateTime lastStepAt) { + UUID id = UUID.randomUUID(); + store.registerJob(id, userId, JobStatus.OPEN, lastStepAt); + return id; + } + + private void recordSignatureForJob(UUID jobId, String type, String value, ArtifactKind kind) { + store.record(jobId, Set.of(new LineageSignature(type, value)), kind); + } + + private Path givenFileWithSignatures(Path tmp, String filename, String... typesAndValues) + throws IOException { + // typesAndValues are pairs: type1, value1, type2, value2, ... + if (typesAndValues.length % 2 != 0) { + throw new IllegalArgumentException("typesAndValues must be in pairs"); + } + Set signatures = + java.util.stream.IntStream.iterate(0, i -> i < typesAndValues.length, i -> i + 2) + .mapToObj( + i -> new LineageSignature(typesAndValues[i], typesAndValues[i + 1])) + .collect(java.util.stream.Collectors.toSet()); + + Path file = tmp.resolve(filename); + Files.write(file, ("placeholder bytes for " + filename).getBytes()); + extractor.setSignaturesFor(file, signatures); + return file; + } + + /** Test double: returns whatever signatures the test registered for a given file. */ + private static class FakeFileSignatureExtractor implements LineageSignatureExtractor { + private final Map> programmed = new HashMap<>(); + + void setSignaturesFor(Path file, Set sigs) { + programmed.put(file, sigs); + } + + @Override + public Set extract(Path file) throws IOException { + return programmed.getOrDefault(file, Set.of()); + } + + @Override + public String name() { + return "fake"; + } + } +} diff --git a/app/saas/src/test/java/stirling/software/saas/payg/lineage/InMemoryJobLineageStore.java b/app/saas/src/test/java/stirling/software/saas/payg/lineage/InMemoryJobLineageStore.java new file mode 100644 index 000000000..e0a7ba4be --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/lineage/InMemoryJobLineageStore.java @@ -0,0 +1,90 @@ +package stirling.software.saas.payg.lineage; + +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +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 stirling.software.saas.payg.model.ArtifactKind; +import stirling.software.saas.payg.model.JobStatus; + +/** + * In-memory test double for {@link JobLineageStore}. Same interface as the JPA store, plus a {@link + * #registerJob} hook so tests can model "this job exists with this user/status/lastStepAt" without + * standing up a database. + * + *

The store keeps two maps: one of recorded hash entries, one of registered job state. The + * findOpenJobForSignatures impl joins them in-process the same way the JPA query does at the DB. + * + *

Lives in {@code src/test} because it's only useful for unit testing the detector. A future + * Redis-backed production impl would live in {@code src/main} alongside {@link JpaJobLineageStore}. + */ +public class InMemoryJobLineageStore implements JobLineageStore { + + // All access is synchronized on `this` — a plain HashMap is fine here; ConcurrentHashMap + // would duplicate the locking the synchronized methods already provide. + private final List entries = new ArrayList<>(); + private final Map jobs = new HashMap<>(); + + public synchronized void registerJob( + UUID jobId, Long ownerUserId, JobStatus status, LocalDateTime lastStepAt) { + jobs.put(jobId, new JobState(ownerUserId, status, lastStepAt)); + } + + @Override + public synchronized void record( + UUID jobId, Set signatures, ArtifactKind kind) { + Instant now = Instant.now(); + for (LineageSignature sig : signatures) { + entries.add(new Entry(jobId, sig.asStorageKey(), kind, now)); + } + } + + @Override + public synchronized Optional findOpenJobForSignatures( + Long userId, Set candidates, Duration workflowWindow) { + LocalDateTime since = LocalDateTime.now().minus(workflowWindow); + Set candidateKeys = + candidates.stream() + .map(LineageSignature::asStorageKey) + .collect(java.util.stream.Collectors.toSet()); + + return entries.stream() + .filter(e -> candidateKeys.contains(e.storageKey())) + .map( + e -> { + JobState job = jobs.get(e.jobId()); + if (job == null) return null; + if (!userId.equals(job.ownerUserId())) return null; + if (job.status() != JobStatus.OPEN) return null; + if (!job.lastStepAt().isAfter(since)) return null; + return new LineageMatch(e.jobId(), e.kind(), job.lastStepAt()); + }) + .filter(java.util.Objects::nonNull) + .max(Comparator.comparing(LineageMatch::jobLastStepAt)); + } + + @Override + public synchronized int pruneOlderThan(Instant cutoff) { + int before = entries.size(); + entries.removeIf(e -> e.recordedAt().isBefore(cutoff)); + return before - entries.size(); + } + + /** Test helper — clears all state. */ + public synchronized void clear() { + entries.clear(); + jobs.clear(); + } + + private record Entry(UUID jobId, String storageKey, ArtifactKind kind, Instant recordedAt) {} + + private record JobState(Long ownerUserId, JobStatus status, LocalDateTime lastStepAt) {} +} diff --git a/app/saas/src/test/java/stirling/software/saas/payg/lineage/LineageSignatureTest.java b/app/saas/src/test/java/stirling/software/saas/payg/lineage/LineageSignatureTest.java new file mode 100644 index 000000000..5bc592c9e --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/lineage/LineageSignatureTest.java @@ -0,0 +1,65 @@ +package stirling.software.saas.payg.lineage; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import org.junit.jupiter.api.Test; + +class LineageSignatureTest { + + @Test + void storageKey_isTypeAndValueColonJoined() { + LineageSignature sig = new LineageSignature("sha256", "abc123"); + + assertThat(sig.asStorageKey()).isEqualTo("sha256:abc123"); + } + + @Test + void fromStorageKey_roundTrips() { + LineageSignature original = new LineageSignature("pdf-id", "deadbeef-cafe-1234"); + + LineageSignature parsed = LineageSignature.fromStorageKey(original.asStorageKey()); + + assertThat(parsed).isEqualTo(original); + } + + @Test + void typeContainingColon_isRejected() { + // Otherwise the storage-key parse would be ambiguous. + assertThatThrownBy(() -> new LineageSignature("foo:bar", "value")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void valueCanContainColon() { + // Values are free-form because they're whatever the extractor produces. The parser only + // splits on the FIRST colon, so colons in the value are preserved. + LineageSignature sig = new LineageSignature("custom", "a:b:c"); + + LineageSignature parsed = LineageSignature.fromStorageKey(sig.asStorageKey()); + + assertThat(parsed.value()).isEqualTo("a:b:c"); + } + + @Test + void blankType_isRejected() { + assertThatThrownBy(() -> new LineageSignature("", "value")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void blankValue_isRejected() { + assertThatThrownBy(() -> new LineageSignature("sha256", "")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void malformedStorageKey_isRejected() { + assertThatThrownBy(() -> LineageSignature.fromStorageKey("nocolon")) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> LineageSignature.fromStorageKey(":nothingbefore")) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> LineageSignature.fromStorageKey("nothingafter:")) + .isInstanceOf(IllegalArgumentException.class); + } +}