diff --git a/app/proprietary/build.gradle b/app/proprietary/build.gradle index 7fb12de5ce..fb21ed0d0c 100644 --- a/app/proprietary/build.gradle +++ b/app/proprietary/build.gradle @@ -80,6 +80,10 @@ dependencies { implementation "software.amazon.awssdk:s3:${awsSdkVersion}" implementation "software.amazon.awssdk:url-connection-client:${awsSdkVersion}" + // @DataJpaTest slice (Boot 4 ships test slices as separate starters, like webmvc-test at the + // root) so policy.source repositories can be exercised against embedded H2. + testImplementation 'org.springframework.boot:spring-boot-starter-data-jpa-test' + // Testcontainers: real MinIO/LocalStack (S3) and Valkey for integration tests in CI without // manually-started instances. Tests skip cleanly when Docker is unavailable. testImplementation "org.testcontainers:testcontainers:${testcontainersMinioVersion}" diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java index 819b6cae02..fa2a190e8f 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java @@ -21,6 +21,7 @@ import stirling.software.proprietary.policy.model.PolicyRun; import stirling.software.proprietary.policy.model.PolicyRunStatus; import stirling.software.proprietary.policy.progress.PolicyProgressListener; import stirling.software.proprietary.policy.source.Source; +import stirling.software.proprietary.policy.source.SourceDocCounter; import stirling.software.proprietary.policy.source.SourceStore; /** @@ -37,6 +38,7 @@ public class PolicyRunner { private final PolicyEngine policyEngine; private final List inputSources; private final SourceStore sourceStore; + private final SourceDocCounter docCounter; /** * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so @@ -65,7 +67,7 @@ public class PolicyRunner { policy.id()); continue; } - runIds.addAll(pullAndRun(policy, source.toInputSpec())); + runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec())); } return runIds; } @@ -82,7 +84,11 @@ public class PolicyRunner { return policyEngine.submit(definition, inputs, listener); } - private List pullAndRun(Policy policy, InputSpec spec) { + /** + * Resolves the source and starts a run per unit; records how many documents the source fed and + * returns the ids of the runs started. + */ + private List pullAndRun(Policy policy, String sourceId, InputSpec spec) { InputSource source = sourceFor(spec); if (source == null) { log.warn( @@ -103,9 +109,12 @@ public class PolicyRunner { return List.of(); } List runIds = new ArrayList<>(); + long docsFed = 0; for (ResolvedInput unit : work) { runIds.add(startRun(policy, unit.inputs(), unit.onComplete())); + docsFed += unit.inputs().primary().size(); } + docCounter.record(sourceId, docsFed); return runIds; } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/DocStats.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/DocStats.java new file mode 100644 index 0000000000..3f01275bd0 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/DocStats.java @@ -0,0 +1,15 @@ +package stirling.software.proprietary.policy.source; + +/** + * Per-source document throughput for the overview row: how many documents a source has fed into + * runs in total and over the trailing 24-hour and 30-day windows. Counts documents fed + * (picked up by a run), so a snapshot-mode source that re-reads the same files each run counts them + * per run. + */ +public record DocStats(long total, long last24h, long last30d) { + + /** Number of trailing daily buckets in a source's daily series. */ + public static final int DAYS = 30; + + public static final DocStats ZERO = new DocStats(0, 0, 0); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounter.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounter.java new file mode 100644 index 0000000000..4f05dcca6e --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounter.java @@ -0,0 +1,64 @@ +package stirling.software.proprietary.policy.source; + +import java.time.Instant; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; + +/** + * In-memory {@link SourceDocCounter} for tests and any future no-database mode. Holds hourly + * buckets per source; the clock is injectable so window boundaries can be exercised + * deterministically. {@link JpaSourceDocCounter} is the runtime bean. + */ +public class InProcessSourceDocCounter implements SourceDocCounter { + + private final Supplier clock; + private final Map> bucketsBySource = new ConcurrentHashMap<>(); + + public InProcessSourceDocCounter() { + this(Instant::now); + } + + public InProcessSourceDocCounter(Supplier clock) { + this.clock = clock; + } + + @Override + public void record(String sourceId, long docs) { + if (docs <= 0) { + return; + } + bucketsBySource + .computeIfAbsent(sourceId, key -> new ConcurrentHashMap<>()) + .merge(currentHour(), docs, Long::sum); + } + + @Override + public Map statsFor(Collection sourceIds) { + long now = currentHour(); + Map stats = new HashMap<>(); + for (String id : sourceIds) { + Map buckets = bucketsBySource.getOrDefault(id, Map.of()); + long total = buckets.values().stream().mapToLong(Long::longValue).sum(); + long last24h = + SourceDocWindows.sumSince(buckets, now - (SourceDocWindows.HOURS_IN_24H - 1)); + long last30d = SourceDocWindows.sumSince(buckets, SourceDocWindows.firstDayHour(now)); + stats.put(id, new DocStats(total, last24h, last30d)); + } + return stats; + } + + @Override + public List dailySeriesFor(String sourceId) { + long now = currentHour(); + Map buckets = bucketsBySource.getOrDefault(sourceId, Map.of()); + return SourceDocWindows.series(SourceDocWindows.byDay(buckets), now / 24); + } + + private long currentHour() { + return clock.get().getEpochSecond() / 3600; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/JpaSourceDocCounter.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/JpaSourceDocCounter.java new file mode 100644 index 0000000000..af67953d77 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/JpaSourceDocCounter.java @@ -0,0 +1,154 @@ +package stirling.software.proprietary.policy.source; + +import java.time.Instant; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.function.IntSupplier; +import java.util.function.Supplier; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +/** + * Durable {@link SourceDocCounter}; the runtime bean. {@code record} keeps two things in step: an + * hourly bucket ({@link SourceDocCountEntity}) that feeds the rolling 24h / 30d / daily-series + * windows, and a denormalized lifetime total ({@link SourceDocTotalEntity}) read directly for the + * all-time figure. The lifetime counter means the overview never scans a source's whole bucket + * history, and lets {@link #pruneOldBuckets()} retire buckets past the 30-day window so the hourly + * table stays bounded (~one row per source per active hour, for at most 30 days). + */ +@Service +@ConditionalOnBooleanProperty(name = "policies.enabled") +public class JpaSourceDocCounter implements SourceDocCounter { + + private final SourceDocCountRepository countRepository; + private final SourceDocTotalRepository totalRepository; + private final Supplier clock; + + @Autowired + public JpaSourceDocCounter( + SourceDocCountRepository countRepository, SourceDocTotalRepository totalRepository) { + this(countRepository, totalRepository, Instant::now); + } + + // Clock seam so tests can pin "now"; the runtime bean uses the wall clock above. + JpaSourceDocCounter( + SourceDocCountRepository countRepository, + SourceDocTotalRepository totalRepository, + Supplier clock) { + this.countRepository = countRepository; + this.totalRepository = totalRepository; + this.clock = clock; + } + + @Override + public void record(String sourceId, long docs) { + if (docs <= 0) { + return; + } + long bucketHour = currentHour(); + upsert( + () -> totalRepository.increment(sourceId, docs), + () -> totalRepository.saveAndFlush(new SourceDocTotalEntity(sourceId, docs))); + upsert( + () -> countRepository.increment(sourceId, bucketHour, docs), + () -> + countRepository.saveAndFlush( + new SourceDocCountEntity(sourceId, bucketHour, docs))); + } + + /** + * Add {@code docs} to a per-source running total: increment the existing row, else insert a new + * one. The insert is flushed now (in its own transaction, since {@code record} is not + * {@code @Transactional}) so a concurrent run's winning insert surfaces as a constraint + * violation we retry as an increment, rather than as a silent {@code merge} overwrite or a + * later doomed commit. + */ + private static void upsert(IntSupplier increment, Runnable insert) { + if (increment.getAsInt() > 0) { + return; + } + try { + insert.run(); + } catch (DataIntegrityViolationException concurrentInsert) { + increment.getAsInt(); + } + } + + @Override + public Map statsFor(Collection sourceIds) { + if (sourceIds.isEmpty()) { + return Map.of(); + } + long now = currentHour(); + Map totals = sums(totalRepository.totalsFor(sourceIds)); + Map last24h = + sums( + countRepository.sumBySourceSince( + sourceIds, now - (SourceDocWindows.HOURS_IN_24H - 1))); + Map last30d = + sums( + countRepository.sumBySourceSince( + sourceIds, SourceDocWindows.firstDayHour(now))); + + Map stats = new HashMap<>(); + for (String id : sourceIds) { + stats.put( + id, + new DocStats( + totals.getOrDefault(id, 0L), + last24h.getOrDefault(id, 0L), + last30d.getOrDefault(id, 0L))); + } + return stats; + } + + @Override + public List dailySeriesFor(String sourceId) { + long now = currentHour(); + Collection ids = List.of(sourceId); + Map dailyCounts = + dailyBySource( + countRepository.dailyCountsSince( + ids, SourceDocWindows.firstDayHour(now))) + .getOrDefault(sourceId, Map.of()); + return SourceDocWindows.series(dailyCounts, now / 24); + } + + /** + * Retire hourly buckets older than the 30-day window; the lifetime total is held separately, so + * nothing reported is lost. Daily is ample - the read queries already ignore older buckets, so + * this is purely storage hygiene. + */ + @Scheduled(fixedDelay = 1, timeUnit = TimeUnit.DAYS) + public void pruneOldBuckets() { + countRepository.deleteOlderThan(SourceDocWindows.firstDayHour(currentHour())); + } + + private long currentHour() { + return clock.get().getEpochSecond() / 3600; + } + + private static Map sums(List rows) { + Map map = new HashMap<>(); + for (SourceDocSum row : rows) { + map.put(row.sourceId(), row.count() == null ? 0L : row.count()); + } + return map; + } + + private static Map> dailyBySource(List rows) { + Map> bySource = new HashMap<>(); + for (SourceDayDocSum row : rows) { + bySource.computeIfAbsent(row.sourceId(), key -> new HashMap<>()) + .merge(row.day(), row.docs() == null ? 0L : row.docs(), Long::sum); + } + return bySource; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java index 5755eb332f..27d60703de 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java @@ -74,6 +74,20 @@ public class SourceController { .orElseGet(() -> ResponseEntity.notFound().build()); } + @GetMapping("/{sourceId}/document-counts") + @Operation( + summary = "Daily document counts for a source", + description = + "The trailing 30-day per-day document series (oldest first) for the source's" + + " sparkline.") + public ResponseEntity> documentCounts(@PathVariable String sourceId) { + return sourceStore + .get(sourceId) + .filter(sourceAccessGuard::canAccess) + .map(source -> ResponseEntity.ok(overviewService.dailySeries(source.id()))) + .orElseGet(() -> ResponseEntity.notFound().build()); + } + @PostMapping(consumes = MediaType.APPLICATION_JSON_VALUE) @Operation( summary = "Create or update a source", diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDayDocSum.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDayDocSum.java new file mode 100644 index 0000000000..923040ac62 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDayDocSum.java @@ -0,0 +1,7 @@ +package stirling.software.proprietary.policy.source; + +/** + * A {@code (sourceId, epoch-day, summed document count)} row from the daily-aggregate query. The + * day is {@code floor(bucketHour / 24)}, i.e. hours-since-epoch collapsed to days-since-epoch. + */ +public record SourceDayDocSum(String sourceId, Long day, Long docs) {} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountEntity.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountEntity.java new file mode 100644 index 0000000000..59fb5d954b --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountEntity.java @@ -0,0 +1,62 @@ +package stirling.software.proprietary.policy.source; + +import java.io.Serializable; + +import org.springframework.data.domain.Persistable; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.IdClass; +import jakarta.persistence.Table; +import jakarta.persistence.Transient; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +/** + * One hour's document tally for a source: {@code bucketHour} is the hours-since-epoch the documents + * were fed, {@code docCount} the running total for that hour. Rolling-window totals are summed from + * these buckets. {@code sourceId} is a plain value, not a foreign key, matching the rest of the + * subsystem so it stays decoupled from the security entities. + */ +@Entity +@Table(name = "policy_source_doc_counts") +@IdClass(SourceDocCountId.class) +@NoArgsConstructor +@Getter +@Setter +public class SourceDocCountEntity implements Serializable, Persistable { + + private static final long serialVersionUID = 1L; + + @Id + @Column(name = "source_id") + private String sourceId; + + @Id + @Column(name = "bucket_hour") + private long bucketHour; + + @Column(name = "doc_count") + private long docCount; + + public SourceDocCountEntity(String sourceId, long bucketHour, long docCount) { + this.sourceId = sourceId; + this.bucketHour = bucketHour; + this.docCount = docCount; + } + + @Override + @Transient + public SourceDocCountId getId() { + return new SourceDocCountId(sourceId, bucketHour); + } + + @Override + @Transient + public boolean isNew() { + return true; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountId.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountId.java new file mode 100644 index 0000000000..549902901d --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountId.java @@ -0,0 +1,36 @@ +package stirling.software.proprietary.policy.source; + +import java.io.Serializable; +import java.util.Objects; + +/** Composite key for {@link SourceDocCountEntity}: one row per source per hour bucket. */ +public class SourceDocCountId implements Serializable { + + private static final long serialVersionUID = 1L; + + private String sourceId; + private long bucketHour; + + public SourceDocCountId() {} + + public SourceDocCountId(String sourceId, long bucketHour) { + this.sourceId = sourceId; + this.bucketHour = bucketHour; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof SourceDocCountId other)) { + return false; + } + return bucketHour == other.bucketHour && Objects.equals(sourceId, other.sourceId); + } + + @Override + public int hashCode() { + return Objects.hash(sourceId, bucketHour); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountRepository.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountRepository.java new file mode 100644 index 0000000000..53a6965e28 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountRepository.java @@ -0,0 +1,69 @@ +package stirling.software.proprietary.policy.source; + +import java.util.Collection; +import java.util.List; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.annotation.Transactional; + +@Repository +public interface SourceDocCountRepository + extends JpaRepository { + + /** + * Add to an existing bucket; returns the number of rows updated (0 when the bucket is new). + * Transactional per call so {@code JpaSourceDocCounter.record} can run it (and the retry after + * a concurrent insert) without an enclosing transaction. + */ + @Modifying + @Transactional + @Query( + "update SourceDocCountEntity e set e.docCount = e.docCount + :docs" + + " where e.sourceId = :sourceId and e.bucketHour = :bucketHour") + int increment( + @Param("sourceId") String sourceId, + @Param("bucketHour") long bucketHour, + @Param("docs") long docs); + + /** + * Delete hourly buckets older than {@code floor} (hours-since-epoch). Nothing reads buckets + * before the 30-day window ({@code SourceDocWindows.firstDayHour}); the lifetime total lives in + * {@code policy_source_doc_totals}, so retiring old buckets keeps the table bounded without + * losing any reported figure. + */ + @Modifying + @Transactional + @Query("delete from SourceDocCountEntity e where e.bucketHour < :floor") + int deleteOlderThan(@Param("floor") long floor); + + /** + * Document total per source restricted to buckets at or after {@code since} (the 24h window). + */ + @Query( + "select new stirling.software.proprietary.policy.source.SourceDocSum(" + + "e.sourceId, sum(e.docCount))" + + " from SourceDocCountEntity e" + + " where e.sourceId in :ids and e.bucketHour >= :since" + + " group by e.sourceId") + List sumBySourceSince( + @Param("ids") Collection ids, @Param("since") long since); + + /** + * Per-source, per-day document totals for buckets at or after {@code since}, summed in the + * database so the overview reads ~one row per source per active day instead of per active hour. + * The day is {@code cast(floor(bucketHour / 24.0) as long)}: {@code 24.0} forces decimal + * division and the cast pins the result to a whole day on every dialect. + */ + @Query( + "select new stirling.software.proprietary.policy.source.SourceDayDocSum(" + + "e.sourceId, cast(floor(e.bucketHour / 24.0) as long), sum(e.docCount))" + + " from SourceDocCountEntity e" + + " where e.sourceId in :ids and e.bucketHour >= :since" + + " group by cast(floor(e.bucketHour / 24.0) as long), e.sourceId") + List dailyCountsSince( + @Param("ids") Collection ids, @Param("since") long since); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCounter.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCounter.java new file mode 100644 index 0000000000..08eb1c277f --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCounter.java @@ -0,0 +1,28 @@ +package stirling.software.proprietary.policy.source; + +import java.util.Collection; +import java.util.List; +import java.util.Map; + +/** + * Records and reports how many documents each source feeds into runs. Counting is bucketed by hour + * so the overview can report rolling totals ({@link DocStats}) cheaply; {@link JpaSourceDocCounter} + * is the runtime bean and {@link InProcessSourceDocCounter} backs tests. + */ +public interface SourceDocCounter { + + /** Record that {@code docs} documents were fed from {@code sourceId} at the current time. */ + void record(String sourceId, long docs); + + /** + * Document totals for each given source; a source with no recorded docs maps to {@link + * DocStats#ZERO}. + */ + Map statsFor(Collection sourceIds); + + /** + * The trailing {@link DocStats#DAYS}-day daily document series for one source, oldest first, + * for the detail-panel sparkline. + */ + List dailySeriesFor(String sourceId); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocSum.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocSum.java new file mode 100644 index 0000000000..8324397401 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocSum.java @@ -0,0 +1,6 @@ +package stirling.software.proprietary.policy.source; + +/** + * A {@code (sourceId, summed document count)} row, populated by the doc-count aggregate queries. + */ +public record SourceDocSum(String sourceId, Long count) {} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalEntity.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalEntity.java new file mode 100644 index 0000000000..afff11d43a --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalEntity.java @@ -0,0 +1,59 @@ +package stirling.software.proprietary.policy.source; + +import java.io.Serializable; + +import org.springframework.data.domain.Persistable; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.Table; +import jakarta.persistence.Transient; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +/** + * A source's lifetime document total, denormalized so the overview reads the all-time count in one + * row instead of scanning the source's whole hourly-bucket history, and so {@link + * SourceDocCountEntity} buckets can be pruned to the rolling 30-day window without losing it. + * + *

Like {@link SourceDocCountEntity}, implements {@link Persistable} reporting {@code isNew() == + * true} so a new source's first {@code save} {@code persist}s (a raw INSERT) and a concurrent + * insert surfaces as a constraint violation the counter retries as an increment, rather than {@code + * merge} silently overwriting it. + */ +@Entity +@Table(name = "policy_source_doc_totals") +@NoArgsConstructor +@Getter +@Setter +public class SourceDocTotalEntity implements Serializable, Persistable { + + private static final long serialVersionUID = 1L; + + @Id + @Column(name = "source_id") + private String sourceId; + + @Column(name = "doc_total") + private long docTotal; + + public SourceDocTotalEntity(String sourceId, long docTotal) { + this.sourceId = sourceId; + this.docTotal = docTotal; + } + + @Override + @Transient + public String getId() { + return sourceId; + } + + @Override + @Transient + public boolean isNew() { + return true; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalRepository.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalRepository.java new file mode 100644 index 0000000000..6542e98787 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalRepository.java @@ -0,0 +1,34 @@ +package stirling.software.proprietary.policy.source; + +import java.util.Collection; +import java.util.List; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.annotation.Transactional; + +@Repository +public interface SourceDocTotalRepository extends JpaRepository { + + /** + * Add to a source's lifetime total; returns the number of rows updated (0 when the source has + * no total row yet). Transactional per call so {@code JpaSourceDocCounter.record} can run it + * (and the retry after a concurrent insert) without an enclosing transaction. + */ + @Modifying + @Transactional + @Query( + "update SourceDocTotalEntity e set e.docTotal = e.docTotal + :docs" + + " where e.sourceId = :sourceId") + int increment(@Param("sourceId") String sourceId, @Param("docs") long docs); + + /** Lifetime totals for the given sources, as {@code (sourceId, total)} rows. */ + @Query( + "select new stirling.software.proprietary.policy.source.SourceDocSum(" + + "e.sourceId, e.docTotal)" + + " from SourceDocTotalEntity e where e.sourceId in :ids") + List totalsFor(@Param("ids") Collection ids); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocWindows.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocWindows.java new file mode 100644 index 0000000000..d52de7024b --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocWindows.java @@ -0,0 +1,61 @@ +package stirling.software.proprietary.policy.source; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Shared rolling-window math over a source's document buckets, so the JPA and in-memory counters + * agree on the window boundaries and the daily series. Both define "last 30 days" as the buckets at + * or after {@link #firstDayHour}, and build the series from per-day counts with {@link #series} + * (the JPA counter aggregates by day in SQL; the in-memory counter groups its hourly buckets with + * {@link #byDay}). + */ +final class SourceDocWindows { + + static final long HOURS_IN_24H = 24; + + private SourceDocWindows() {} + + /** + * The hours-since-epoch at the start of the oldest day in the 30-day window, given the current + * hour bucket. Both the "last 30 days" total and the daily series are measured from here, so + * the KPI and the sparkline always cover the same buckets. + */ + static long firstDayHour(long nowHour) { + return ((nowHour / 24) - (DocStats.DAYS - 1)) * 24; + } + + /** + * Build the {@link DocStats#DAYS}-day daily series (oldest first) from per-day document counts + * (keyed by epoch-day, i.e. hours-since-epoch / 24). {@code currentDay} is today's epoch-day; + * the series runs back {@code DAYS} days from it. + */ + static List series(Map dailyCounts, long currentDay) { + long firstDay = currentDay - (DocStats.DAYS - 1); + long[] daily = new long[DocStats.DAYS]; + for (Map.Entry day : dailyCounts.entrySet()) { + int dayIndex = (int) (day.getKey() - firstDay); + if (dayIndex >= 0 && dayIndex < DocStats.DAYS) { + daily[dayIndex] += day.getValue(); + } + } + return Arrays.stream(daily).boxed().toList(); + } + + /** Collapse hourly buckets (keyed by hours-since-epoch) into per-day counts (keyed by day). */ + static Map byDay(Map hourlyCounts) { + Map dailyCounts = new HashMap<>(); + hourlyCounts.forEach((hour, count) -> dailyCounts.merge(hour / 24, count, Long::sum)); + return dailyCounts; + } + + /** Sum hourly buckets at or after {@code since} (hours-since-epoch). */ + static long sumSince(Map hourlyCounts, long since) { + return hourlyCounts.entrySet().stream() + .filter(entry -> entry.getKey() >= since) + .mapToLong(Map.Entry::getValue) + .sum(); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java index b9bc7c1cb2..6e5f92ce44 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java @@ -30,12 +30,15 @@ public class SourceOverviewService { private final PolicyStore policyStore; private final SourceAccessGuard sourceAccessGuard; private final PolicyAccessGuard policyAccessGuard; + private final SourceDocCounter docCounter; public SourcesResponse overview() { List sources = sourceAccessGuard.visibleFrom(sourceStore); List policies = policyAccessGuard.visibleFrom(policyStore); Map> referencesBySource = referencesBySource(policies); + Map docStats = + docCounter.statsFor(sources.stream().map(Source::id).toList()); List views = sources.stream() @@ -44,7 +47,8 @@ public class SourceOverviewService { toView( source, referencesBySource.getOrDefault( - source.id(), List.of()))) + source.id(), List.of()), + docStats.getOrDefault(source.id(), DocStats.ZERO))) .sorted( Comparator.comparingInt(SourceView::referenceCount) .reversed() @@ -54,6 +58,14 @@ public class SourceOverviewService { return new SourcesResponse(buildKpis(views), views); } + /** + * The 30-day daily document series for one source (oldest first), for the expanded row's + * sparkline. + */ + public List dailySeries(String sourceId) { + return docCounter.dailySeriesFor(sourceId); + } + /** Policies referencing each source id, across the caller's visible policies. */ private static Map> referencesBySource(List policies) { Map> bySource = new HashMap<>(); @@ -65,7 +77,8 @@ public class SourceOverviewService { return bySource; } - private static SourceView toView(Source source, List referencingPolicies) { + private static SourceView toView( + Source source, List referencingPolicies, DocStats docs) { List refs = referencingPolicies.stream() .map(policy -> new SourceView.PolicyRef(policy.id(), policy.name())) @@ -78,7 +91,9 @@ public class SourceOverviewService { refs.size(), refs, configRows(source), - null); + docs.total(), + docs.last24h(), + docs.last30d()); } /** A disabled (paused) source reads as "disabled"; an unreferenced one reads as "unused". */ diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceView.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceView.java index bb1fdbf9d0..6c6a9c4c6d 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceView.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceView.java @@ -4,8 +4,8 @@ import java.util.List; /** * One row in the Sources overview: a persisted input connection shown exactly once, with how many - * policies reference it (and which). {@code docsTotal} is {@code null} - per-source document volume - * is not tracked yet; the field is reserved so a later doc-accounting pass is additive. + * policies reference it (and which) and how many documents it has fed into runs ({@code docsTotal} + * lifetime plus the trailing 24-hour and 30-day windows). */ public record SourceView( String id, @@ -15,7 +15,9 @@ public record SourceView( int referenceCount, List referencingPolicies, List config, - Long docsTotal) { + long docsTotal, + long docs24h, + long docs30d) { /** A policy that references this source. */ public record PolicyRef(String id, String name) {} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java index cff169581b..b1ab05f8ba 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java @@ -33,6 +33,7 @@ import stirling.software.proprietary.policy.model.PolicyInputs; import stirling.software.proprietary.policy.model.PolicyRun; import stirling.software.proprietary.policy.model.PolicyRunStatus; import stirling.software.proprietary.policy.progress.PolicyProgressListener; +import stirling.software.proprietary.policy.source.InProcessSourceDocCounter; import stirling.software.proprietary.policy.source.InProcessSourceStore; import stirling.software.proprietary.policy.source.Source; import stirling.software.proprietary.policy.source.SourceStore; @@ -53,7 +54,12 @@ class PolicyRunnerTest { @BeforeEach void setUp() { - runner = new PolicyRunner(policyEngine, List.of(folderSource), sourceStore); + runner = + new PolicyRunner( + policyEngine, + List.of(folderSource), + sourceStore, + new InProcessSourceDocCounter()); } @Test diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounterTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounterTest.java new file mode 100644 index 0000000000..b598b44c80 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounterTest.java @@ -0,0 +1,62 @@ +package stirling.software.proprietary.policy.source; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.time.Instant; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; + +/** + * Tests the rolling-window aggregation: documents recorded at different times roll into the total, + * the 24-hour / 30-day windows, and the daily series correctly as the clock advances. + */ +class InProcessSourceDocCounterTest { + + private static InProcessSourceDocCounter seededCounter(AtomicReference clock) { + InProcessSourceDocCounter counter = new InProcessSourceDocCounter(clock::get); + // 40 days ago: outside the 30-day window. + clock.set(Instant.parse("2026-05-21T12:00:00Z")); + counter.record("s", 100); + // 10 days ago: inside 30 days, outside 24 hours. + clock.set(Instant.parse("2026-06-20T12:00:00Z")); + counter.record("s", 50); + // 2 hours ago: inside both windows. + clock.set(Instant.parse("2026-06-30T10:00:00Z")); + counter.record("s", 7); + // Query as of "now". + clock.set(Instant.parse("2026-06-30T12:00:00Z")); + return counter; + } + + @Test + void rollsCountsIntoTotalAnd24hAnd30dWindows() { + AtomicReference clock = new AtomicReference<>(); + DocStats stats = seededCounter(clock).statsFor(List.of("s")).get("s"); + + assertEquals(157, stats.total()); + assertEquals(7, stats.last24h()); + assertEquals(57, stats.last30d()); + } + + @Test + void buildsTheDailySeriesOldestFirst() { + AtomicReference clock = new AtomicReference<>(); + // Today (index 29) and 10 days ago (index 19) only; 40 days ago is outside the window. + List series = seededCounter(clock).dailySeriesFor("s"); + + assertEquals(30, series.size()); + assertEquals(7L, series.get(29)); + assertEquals(50L, series.get(19)); + assertEquals(57L, series.stream().mapToLong(Long::longValue).sum()); + } + + @Test + void aSourceWithNoRecordedDocsIsZero() { + InProcessSourceDocCounter counter = new InProcessSourceDocCounter(); + assertEquals(DocStats.ZERO, counter.statsFor(List.of("unknown")).get("unknown")); + assertEquals(Collections.nCopies(DocStats.DAYS, 0L), counter.dailySeriesFor("unknown")); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/JpaSourceDocCounterDbTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/JpaSourceDocCounterDbTest.java new file mode 100644 index 0000000000..4df0ab0ced --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/JpaSourceDocCounterDbTest.java @@ -0,0 +1,93 @@ +package stirling.software.proprietary.policy.source; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.time.Instant; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.AutoConfigurationPackage; +import org.springframework.boot.data.jpa.test.autoconfigure.DataJpaTest; + +/** + * Exercises {@link JpaSourceDocCounter} against a real (H2) database so the daily-aggregate query + * ({@code cast(floor(bucketHour / 24.0) as long)} grouping) and the record upsert are actually run, + * not just asserted against a mock. The unit-level window math lives in {@link + * InProcessSourceDocCounterTest}. + */ +@DataJpaTest +class JpaSourceDocCounterDbTest { + + // Pinned "now" so seeding and statsFor agree regardless of when the test runs (no hour-tick + // flake at the boundary of a wall-clock hour). + private static final Instant NOW = Instant.parse("2026-06-30T12:00:00Z"); + private static final long NOW_HOUR = NOW.getEpochSecond() / 3600; + + @Autowired private SourceDocCountRepository repository; + @Autowired private SourceDocTotalRepository totalRepository; + + private JpaSourceDocCounter counter() { + return new JpaSourceDocCounter(repository, totalRepository, () -> NOW); + } + + @Test + void recordIncrementsBothTheHourlyBucketAndTheLifetimeTotal() { + JpaSourceDocCounter counter = counter(); + counter.record("s", 5); + counter.record("s", 3); + + DocStats stats = counter.statsFor(List.of("s")).get("s"); + assertEquals(8, stats.total()); // from the denormalized lifetime row + assertEquals(8, stats.last24h()); + assertEquals(8, stats.last30d()); + assertEquals(8L, counter.dailySeriesFor("s").get(DocStats.DAYS - 1)); + } + + @Test + void statsBucketDocsByDayAndWindowFromTheDatabase() { + // Seed buckets directly at controlled hours: today, 10 days ago, 40 days ago, plus the + // lifetime row (record() would write it, but here we seed history directly). + repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR, 7)); + repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR - 24L * 10, 50)); + repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR - 24L * 40, 100)); + totalRepository.saveAndFlush(new SourceDocTotalEntity("s", 157)); + + DocStats stats = counter().statsFor(List.of("s")).get("s"); + assertEquals(157, stats.total()); // lifetime, including the out-of-window 40-days-ago docs + assertEquals(7, stats.last24h()); + assertEquals(57, stats.last30d()); + + // The daily series (fetched separately, per source) covers the same 30-day window. + List series = counter().dailySeriesFor("s"); + assertEquals(DocStats.DAYS, series.size()); + assertEquals(7L, series.get(DocStats.DAYS - 1)); // today + assertEquals(50L, series.get(DocStats.DAYS - 11)); // 10 days ago + assertEquals(57L, series.stream().mapToLong(Long::longValue).sum()); + } + + @Test + void pruneRetiresOutOfWindowBucketsButKeepsTheLifetimeTotal() { + repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR, 7)); // today + repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR - 24L * 40, 100)); // 40d ago + totalRepository.saveAndFlush(new SourceDocTotalEntity("s", 107)); + + counter().pruneOldBuckets(); + + assertEquals(1, repository.count()); // only the in-window bucket remains + DocStats stats = counter().statsFor(List.of("s")).get("s"); + assertEquals(107, stats.total()); // lifetime survives pruning + assertEquals(7, stats.last24h()); + assertEquals(7, stats.last30d()); + } + + @Test + void aSourceWithNoRecordedDocsIsZero() { + assertEquals(DocStats.ZERO, counter().statsFor(List.of("unknown")).get("unknown")); + } + + @SpringBootConfiguration + @AutoConfigurationPackage + static class TestApp {} +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java index c01df84a55..d1d62b7dbe 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java @@ -50,7 +50,12 @@ class SourceControllerTest { SourceAccessGuard sourceGuard = new SourceAccessGuard(userService, properties, authority); PolicyAccessGuard policyGuard = new PolicyAccessGuard(userService, properties, authority); SourceOverviewService overviewService = - new SourceOverviewService(sourceStore, policyStore, sourceGuard, policyGuard); + new SourceOverviewService( + sourceStore, + policyStore, + sourceGuard, + policyGuard, + new InProcessSourceDocCounter()); triggerManager = mock(PolicyTriggerManager.class); // A permissive input source so config validation passes and save can be exercised. InputSource folderInput = mock(InputSource.class); diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java index 986fd8c636..275f076db3 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java @@ -1,7 +1,6 @@ package stirling.software.proprietary.policy.source; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -31,6 +30,7 @@ class SourceOverviewServiceTest { private final SourceStore sourceStore = new InProcessSourceStore(); private final PolicyStore policyStore = new InProcessPolicyStore(); + private final SourceDocCounter docCounter = new InProcessSourceDocCounter(); private SourceOverviewService service; @BeforeEach @@ -41,7 +41,9 @@ class SourceOverviewServiceTest { PolicyManagementAuthority authority = mock(PolicyManagementAuthority.class); SourceAccessGuard sourceGuard = new SourceAccessGuard(userService, properties, authority); PolicyAccessGuard policyGuard = new PolicyAccessGuard(userService, properties, authority); - service = new SourceOverviewService(sourceStore, policyStore, sourceGuard, policyGuard); + service = + new SourceOverviewService( + sourceStore, policyStore, sourceGuard, policyGuard, docCounter); } @Test @@ -111,7 +113,8 @@ class SourceOverviewServiceTest { SourceAccessGuard sourceGuard = new SourceAccessGuard(userService, properties, authority); PolicyAccessGuard policyGuard = new PolicyAccessGuard(userService, properties, authority); SourceOverviewService scoped = - new SourceOverviewService(sourceStore, policyStore, sourceGuard, policyGuard); + new SourceOverviewService( + sourceStore, policyStore, sourceGuard, policyGuard, docCounter); Source ours = teamSource("Ours", "/ours", 1L); teamSource("Theirs", "/theirs", 2L); @@ -128,9 +131,19 @@ class SourceOverviewServiceTest { } @Test - void documentVolumeIsNotTrackedYet() { + void documentCountsReflectRecordedDocs() { Source a = source("A", "/a"); - assertNull(find(service.overview(), a.id()).docsTotal()); + Source b = source("B", "/b"); + docCounter.record(a.id(), 5); + docCounter.record(a.id(), 3); + + SourceView av = find(service.overview(), a.id()); + assertEquals(8, av.docsTotal()); + assertEquals(8, av.docs24h()); + assertEquals(8, av.docs30d()); + + // A source with no recorded documents reads as zero, not null. + assertEquals(0, find(service.overview(), b.id()).docsTotal()); } private Source source(String name, String directory) { diff --git a/app/saas/src/main/resources/db/migration/saas/V23__policy_source_doc_counts.sql b/app/saas/src/main/resources/db/migration/saas/V23__policy_source_doc_counts.sql new file mode 100644 index 0000000000..7dcbaf1c0e --- /dev/null +++ b/app/saas/src/main/resources/db/migration/saas/V23__policy_source_doc_counts.sql @@ -0,0 +1,26 @@ +-- Per-source document throughput, in two tables: +-- +-- policy_source_doc_counts one row per source per hour bucket (hours-since-epoch), holding how +-- many documents that source fed into runs in that hour. Feeds the +-- rolling last-24h / last-30d windows and the 30-day daily series, and +-- is pruned to that window so it stays bounded. +-- policy_source_doc_totals a denormalized lifetime total per source, incremented alongside the +-- hourly bucket, so the overview reads the all-time figure in one row +-- instead of scanning a source's whole bucket history - and so the +-- hourly buckets can be pruned without losing it. +-- +-- Gated by policies.enabled like the rest of the subsystem; Hibernate ddl-auto would also create +-- these, but the migration keeps the schema explicit for the Flyway-managed deployments. + +CREATE TABLE IF NOT EXISTS policy_source_doc_counts ( + source_id VARCHAR(255) NOT NULL, + bucket_hour BIGINT NOT NULL, + doc_count BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (source_id, bucket_hour) +); + +CREATE TABLE IF NOT EXISTS policy_source_doc_totals ( + source_id VARCHAR(255) NOT NULL, + doc_total BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (source_id) +); diff --git a/frontend/portal/public/locales/en-US/translation.toml b/frontend/portal/public/locales/en-US/translation.toml index fb85a21a27..b32f19c303 100644 --- a/frontend/portal/public/locales/en-US/translation.toml +++ b/frontend/portal/public/locales/en-US/translation.toml @@ -330,7 +330,11 @@ subtitle = "{{type}} ยท {{status}}" closeAriaLabel = "Close detail" usedBy = "Used by" notReferenced = "Not referenced by any policy, so it's safe to delete." -docsUntracked = "Per-source document volume isn't tracked yet." +documents = "Documents" +docsTotal = "Total seen" +docs24h = "Last 24h" +docs30d = "Last 30 days" +docsTrend = "Documents over the last 30 days" edit = "Edit" pause = "Pause" resume = "Resume" diff --git a/frontend/portal/src/api/sources.ts b/frontend/portal/src/api/sources.ts index 9f9839d7a1..9a66aeef2d 100644 --- a/frontend/portal/src/api/sources.ts +++ b/frontend/portal/src/api/sources.ts @@ -28,8 +28,10 @@ export interface SourceView { referenceCount: number; referencingPolicies: SourcePolicyRef[]; config: SourceDetailRow[]; - /** Per-source document volume: not tracked yet (always null for now). */ - docsTotal: number | null; + /** Documents this source has fed into runs: lifetime, plus trailing 24h / 30d windows. */ + docsTotal: number; + docs24h: number; + docs30d: number; } export interface SourceKpi { @@ -69,6 +71,17 @@ export async function fetchSource(id: string): Promise { ); } +/** + * GET /api/v1/sources/{id}/document-counts: the trailing 30-day daily document + * series (oldest first) for the source's sparkline. Fetched only for the expanded + * row, so the overview list stays lightweight. + */ +export async function fetchSourceDocCounts(id: string): Promise { + return apiClient.local.json( + `/api/v1/sources/${encodeURIComponent(id)}/document-counts`, + ); +} + /** POST /api/v1/sources: create (blank id) or update (matched id) a source. */ export async function createSource(source: Source): Promise { return apiClient.local.json("/api/v1/sources", { diff --git a/frontend/portal/src/components/sources/SourceDetailCard.stories.tsx b/frontend/portal/src/components/sources/SourceDetailCard.stories.tsx index 0f9ddee521..118c084d14 100644 --- a/frontend/portal/src/components/sources/SourceDetailCard.stories.tsx +++ b/frontend/portal/src/components/sources/SourceDetailCard.stories.tsx @@ -1,6 +1,9 @@ import type { Meta, StoryObj } from "@storybook/react-vite"; import type { SourceView } from "@portal/api/sources"; import { SourceDetailCard } from "@portal/components/sources/SourceDetailCard"; +import { sampleDailySeries } from "@portal/mocks/sampleDailySeries"; + +const SAMPLE_SERIES = sampleDailySeries(330); const IN_USE: SourceView = { id: "src-claims", @@ -16,7 +19,9 @@ const IN_USE: SourceView = { { label: "Directory", value: "/data/claims-intake" }, { label: "Mode", value: "consume" }, ], - docsTotal: null, + docsTotal: 45230, + docs24h: 312, + docs30d: 9870, }; const ORPHANED: SourceView = { @@ -27,7 +32,9 @@ const ORPHANED: SourceView = { referenceCount: 0, referencingPolicies: [], config: [{ label: "Directory", value: "/data/archive" }], - docsTotal: null, + docsTotal: 45230, + docs24h: 312, + docs30d: 9870, }; const meta: Meta = { @@ -35,6 +42,7 @@ const meta: Meta = { component: SourceDetailCard, parameters: { layout: "padded" }, args: { + docSeries: SAMPLE_SERIES, onClose: () => {}, onEdit: () => {}, onTogglePause: () => {}, diff --git a/frontend/portal/src/components/sources/SourceDetailCard.tsx b/frontend/portal/src/components/sources/SourceDetailCard.tsx index 6169116c0f..b32f1ec36b 100644 --- a/frontend/portal/src/components/sources/SourceDetailCard.tsx +++ b/frontend/portal/src/components/sources/SourceDetailCard.tsx @@ -7,6 +7,7 @@ import "@portal/views/Sources.css"; interface SourceDetailCardProps { source: SourceView; + docSeries: number[]; onClose: () => void; onEdit: (source: SourceView) => void; onTogglePause: (source: SourceView) => void; @@ -18,6 +19,7 @@ interface SourceDetailCardProps { /** Expanded detail for the selected source row, with edit/pause/delete actions. */ export function SourceDetailCard({ source, + docSeries, onClose, onEdit, onTogglePause, @@ -55,7 +57,7 @@ export function SourceDetailCard({ - +

); } diff --git a/frontend/portal/src/components/sources/SourcesTable.stories.tsx b/frontend/portal/src/components/sources/SourcesTable.stories.tsx index 30f1e5bc5f..0a556ca63b 100644 --- a/frontend/portal/src/components/sources/SourcesTable.stories.tsx +++ b/frontend/portal/src/components/sources/SourcesTable.stories.tsx @@ -17,7 +17,9 @@ const SOURCES: SourceView[] = [ { label: "Directory", value: "/data/claims-intake" }, { label: "Mode", value: "consume" }, ], - docsTotal: null, + docsTotal: 45230, + docs24h: 312, + docs30d: 9870, }, { id: "src-archive", @@ -27,7 +29,9 @@ const SOURCES: SourceView[] = [ referenceCount: 0, referencingPolicies: [], config: [{ label: "Directory", value: "/data/archive" }], - docsTotal: null, + docsTotal: 1180, + docs24h: 0, + docs30d: 0, }, { id: "src-legacy", @@ -37,7 +41,9 @@ const SOURCES: SourceView[] = [ referenceCount: 0, referencingPolicies: [], config: [{ label: "Directory", value: "/mnt/legacy" }], - docsTotal: null, + docsTotal: 48600, + docs24h: 0, + docs30d: 0, }, ]; diff --git a/frontend/portal/src/components/sources/Sparkline.test.tsx b/frontend/portal/src/components/sources/Sparkline.test.tsx new file mode 100644 index 0000000000..45db5eca44 --- /dev/null +++ b/frontend/portal/src/components/sources/Sparkline.test.tsx @@ -0,0 +1,48 @@ +import { describe, expect, it } from "vitest"; +import { render } from "@testing-library/react"; +import { Sparkline } from "@portal/components/sources/Sparkline"; + +function pointsOf(container: HTMLElement): string[] { + const poly = container.querySelector("polyline"); + return (poly?.getAttribute("points") ?? "") + .trim() + .split(/\s+/) + .filter(Boolean); +} + +describe("Sparkline", () => { + it("draws one point per value", () => { + const { container } = render(); + expect(pointsOf(container)).toHaveLength(5); + }); + + it("renders nothing for an empty series", () => { + const { container } = render(); + expect(container.querySelector("svg")).toBeNull(); + }); + + it("renders a flat finite line for an all-zero series (no divide-by-zero)", () => { + const { container } = render(); + const ys = pointsOf(container).map((p) => Number(p.split(",")[1])); + expect(ys).toHaveLength(4); + expect(ys.every(Number.isFinite)).toBe(true); + // All equal: a zero series is a single horizontal line, not NaN-laden. + expect(new Set(ys).size).toBe(1); + }); + + it("puts the peak value at the top of the band", () => { + const { container } = render(); + const ys = pointsOf(container).map((p) => Number(p.split(",")[1])); + // y grows downward in SVG, so the larger value (10) sits at the smaller y. + expect(ys[1]).toBeLessThan(ys[0]); + }); + + it("exposes its aria-label", () => { + const { container } = render( + , + ); + expect(container.querySelector("svg")?.getAttribute("aria-label")).toBe( + "Docs trend", + ); + }); +}); diff --git a/frontend/portal/src/components/sources/Sparkline.tsx b/frontend/portal/src/components/sources/Sparkline.tsx new file mode 100644 index 0000000000..7877d21fd4 --- /dev/null +++ b/frontend/portal/src/components/sources/Sparkline.tsx @@ -0,0 +1,53 @@ +import "@portal/views/Sources.css"; + +interface SparklineProps { + /** Series values, oldest first. */ + data: number[]; + /** Drawing height in px; the width fills the container. */ + height?: number; + ariaLabel?: string; +} + +/** + * A tiny dependency-free trend line: normalises {@code data} to its own peak and draws a single + * polyline. The viewBox is fixed while the rendered width fills the container (the stroke stays + * crisp via non-scaling-stroke), so it adapts to any column without distorting the line weight. + */ +export function Sparkline({ data, height = 36, ariaLabel }: SparklineProps) { + if (data.length === 0) { + return null; + } + const width = 240; + const pad = 3; + const max = Math.max(...data, 1); + const stepX = data.length > 1 ? width / (data.length - 1) : 0; + const points = data + .map((value, i) => { + const x = i * stepX; + const y = pad + (1 - value / max) * (height - 2 * pad); + return `${x.toFixed(1)},${y.toFixed(1)}`; + }) + .join(" "); + + return ( + + + + ); +} diff --git a/frontend/portal/src/mocks/handlers/sources.ts b/frontend/portal/src/mocks/handlers/sources.ts index 90418caf56..4839c17720 100644 --- a/frontend/portal/src/mocks/handlers/sources.ts +++ b/frontend/portal/src/mocks/handlers/sources.ts @@ -7,6 +7,7 @@ import type { SourceView, SourcesResponse, } from "@portal/api/sources"; +import { sampleDailySeries } from "@portal/mocks/sampleDailySeries"; /** * Stateful mock for the Sources surface so the portal works fully offline with @@ -66,6 +67,27 @@ const references: Record = { "src-contracts": [{ id: "pol_contract", name: "Contract Review" }], }; +/** Per-source document throughput, mirroring the backend's docsTotal / 24h / 30d. */ +const docCounts: Record< + string, + { total: number; last24h: number; last30d: number } +> = { + "src-claims": { total: 45230, last24h: 312, last30d: 9870 }, + "src-contracts": { total: 12840, last24h: 96, last30d: 2310 }, + "src-archive": { total: 1180, last24h: 0, last30d: 0 }, + "src-legacy": { total: 48600, last24h: 0, last30d: 0 }, +}; + +function docsFor(id: string): { + total: number; + last24h: number; + last30d: number; + daily: number[]; +} { + const counts = docCounts[id] ?? { total: 0, last24h: 0, last30d: 0 }; + return { ...counts, daily: sampleDailySeries(counts.last30d / 30) }; +} + let store: StoredSource[] = seedSources(); let idCounter = 0; @@ -97,6 +119,7 @@ function toSourceView( source: StoredSource, refs: SourcePolicyRef[], ): SourceView { + const docs = docsFor(source.id); return { id: source.id, name: source.name, @@ -105,7 +128,9 @@ function toSourceView( referenceCount: refs.length, referencingPolicies: refs, config: configRows(source.options), - docsTotal: null, + docsTotal: docs.total, + docs24h: docs.last24h, + docs30d: docs.last30d, }; } @@ -142,6 +167,13 @@ export const sourcesHandlers = [ return HttpResponse.json(source); }), + http.get("/api/v1/sources/:id/document-counts", async ({ params }) => { + await delay(120); + const source = store.find((s) => s.id === params.id); + if (!source) return new HttpResponse(null, { status: 404 }); + return HttpResponse.json(docsFor(source.id).daily); + }), + http.post("/api/v1/sources", async ({ request }) => { await delay(120); const incoming = (await request.json()) as Source; diff --git a/frontend/portal/src/mocks/sampleDailySeries.ts b/frontend/portal/src/mocks/sampleDailySeries.ts new file mode 100644 index 0000000000..20b5c3e716 --- /dev/null +++ b/frontend/portal/src/mocks/sampleDailySeries.ts @@ -0,0 +1,10 @@ +/** + * A gentle, deterministic 30-point daily series averaging ~`avg`/day, shaped by a sine wave so mock + * and Storybook sparklines have something to draw. Shared so the mock handler and the source stories + * stay in sync instead of each carrying their own copy of the formula. + */ +export function sampleDailySeries(avg: number): number[] { + return Array.from({ length: 30 }, (_, i) => + Math.round(avg * (0.5 + Math.abs(Math.sin((i + 1) / 3)))), + ); +} diff --git a/frontend/portal/src/views/Pipelines.test.tsx b/frontend/portal/src/views/Pipelines.test.tsx index afcad9be07..579181b1ce 100644 --- a/frontend/portal/src/views/Pipelines.test.tsx +++ b/frontend/portal/src/views/Pipelines.test.tsx @@ -93,7 +93,9 @@ const SOURCES: SourcesResponse = { referenceCount: 1, referencingPolicies: [], config: [], - docsTotal: null, + docsTotal: 0, + docs24h: 0, + docs30d: 0, }, ], }; diff --git a/frontend/portal/src/views/Sources.css b/frontend/portal/src/views/Sources.css index 7c5a594c0b..d265a17135 100644 --- a/frontend/portal/src/views/Sources.css +++ b/frontend/portal/src/views/Sources.css @@ -238,6 +238,13 @@ font-weight: 600; } +.portal-sparkline { + display: block; + width: 100%; + margin-top: 0.5rem; + color: var(--color-blue); +} + .portal-sources__chips { display: flex; flex-wrap: wrap; diff --git a/frontend/portal/src/views/Sources.test.tsx b/frontend/portal/src/views/Sources.test.tsx index 98ce7175e4..191e30bbd8 100644 --- a/frontend/portal/src/views/Sources.test.tsx +++ b/frontend/portal/src/views/Sources.test.tsx @@ -16,11 +16,13 @@ vi.mock("react-i18next", () => ({ const fetchSources = vi.fn(); const fetchSource = vi.fn(); +const fetchSourceDocCounts = vi.fn(); const createSource = vi.fn(); const deleteSource = vi.fn(); vi.mock("@portal/api/sources", () => ({ fetchSources: () => fetchSources(), fetchSource: (id: string) => fetchSource(id), + fetchSourceDocCounts: (id: string) => fetchSourceDocCounts(id), createSource: (source: unknown) => createSource(source), deleteSource: (id: string) => deleteSource(id), })); @@ -43,7 +45,9 @@ const RESPONSE: SourcesResponse = { { id: "pol-2", name: "Classification" }, ], config: [{ label: "Directory", value: "/data/incoming" }], - docsTotal: null, + docsTotal: 1240, + docs24h: 18, + docs30d: 540, }, { id: "src-orphan", @@ -53,7 +57,9 @@ const RESPONSE: SourcesResponse = { referenceCount: 0, referencingPolicies: [], config: [{ label: "Directory", value: "/tmp/scratch" }], - docsTotal: null, + docsTotal: 1240, + docs24h: 18, + docs30d: 540, }, ], }; @@ -70,6 +76,8 @@ describe("Sources view", () => { beforeEach(() => { fetchSources.mockReset(); fetchSource.mockReset(); + fetchSourceDocCounts.mockReset(); + fetchSourceDocCounts.mockResolvedValue([]); createSource.mockReset(); deleteSource.mockReset(); }); diff --git a/frontend/portal/src/views/Sources.tsx b/frontend/portal/src/views/Sources.tsx index 230dc3abb7..7df5f1680e 100644 --- a/frontend/portal/src/views/Sources.tsx +++ b/frontend/portal/src/views/Sources.tsx @@ -14,6 +14,7 @@ import { createSource, deleteSource, fetchSource, + fetchSourceDocCounts, fetchSources, type Source, type SourcesResponse, @@ -49,6 +50,21 @@ export function Sources() { const sources = data?.sources ?? []; const expanded = sources.find((s) => s.id === expandedId) ?? null; + // The 30-day sparkline series lives off the list endpoint; fetch it for the one + // expanded row only (empty while collapsed, so no request fires). + const docSeriesState = useAsync<{ id: string; series: number[] }>( + () => + expandedId + ? fetchSourceDocCounts(expandedId).then((series) => ({ + id: expandedId, + series, + })) + : Promise.resolve({ id: "", series: [] }), + [expandedId], + ); + const docSeries = + docSeriesState.data?.id === expandedId ? docSeriesState.data.series : []; + function openCreate() { setEditingSource(null); setWizardOpen(true); @@ -166,6 +182,7 @@ export function Sources() { {expanded && ( setExpandedId(null)} onEdit={openEdit} onTogglePause={togglePause}