diff --git a/commafeed-server/src/main/java/com/commafeed/backend/dao/FeedEntryDAO.java b/commafeed-server/src/main/java/com/commafeed/backend/dao/FeedEntryDAO.java index 61b33121..0cf2962b 100644 --- a/commafeed-server/src/main/java/com/commafeed/backend/dao/FeedEntryDAO.java +++ b/commafeed-server/src/main/java/com/commafeed/backend/dao/FeedEntryDAO.java @@ -1,7 +1,10 @@ package com.commafeed.backend.dao; import java.time.Instant; +import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; import jakarta.inject.Singleton; import jakarta.persistence.EntityManager; @@ -9,6 +12,7 @@ import jakarta.persistence.EntityManager; import com.commafeed.backend.model.Feed; import com.commafeed.backend.model.FeedEntry; import com.commafeed.backend.model.QFeedEntry; +import com.google.common.collect.Lists; import com.querydsl.core.Tuple; import com.querydsl.core.types.dsl.NumberExpression; import com.querydsl.jpa.impl.JPAQuery; @@ -17,6 +21,7 @@ import com.querydsl.jpa.impl.JPAQuery; public class FeedEntryDAO extends GenericDAO { private static final QFeedEntry ENTRY = QFeedEntry.feedEntry; + private static final int IN_CLAUSE_BATCH_SIZE = 1000; public FeedEntryDAO(EntityManager entityManager) { super(entityManager, FeedEntry.class); @@ -26,6 +31,18 @@ public class FeedEntryDAO extends GenericDAO { return query().select(ENTRY).from(ENTRY).where(ENTRY.guidHash.eq(guidHash), ENTRY.feed.eq(feed)).limit(1).fetchOne(); } + public Set findExistingGuids(Feed feed, Set guidHashes) { + if (guidHashes.isEmpty()) { + return Set.of(); + } + + Set result = new HashSet<>(); + for (List batch : Lists.partition(new ArrayList<>(guidHashes), IN_CLAUSE_BATCH_SIZE)) { + result.addAll(query().select(ENTRY.guidHash).from(ENTRY).where(ENTRY.feed.eq(feed), ENTRY.guidHash.in(batch)).fetch()); + } + return result; + } + public List findFeedsExceedingCapacity(long maxCapacity, long max, boolean keepStarredEntries) { NumberExpression count = ENTRY.id.count(); JPAQuery query = query().select(ENTRY.feed.id, count).from(ENTRY); diff --git a/commafeed-server/src/main/java/com/commafeed/backend/feed/FeedRefreshUpdater.java b/commafeed-server/src/main/java/com/commafeed/backend/feed/FeedRefreshUpdater.java index e2c0d5ba..7440c3fa 100644 --- a/commafeed-server/src/main/java/com/commafeed/backend/feed/FeedRefreshUpdater.java +++ b/commafeed-server/src/main/java/com/commafeed/backend/feed/FeedRefreshUpdater.java @@ -18,6 +18,7 @@ import org.apache.commons.lang3.StringUtils; import com.codahale.metrics.Meter; import com.codahale.metrics.MetricRegistry; import com.commafeed.backend.Digests; +import com.commafeed.backend.dao.FeedEntryDAO; import com.commafeed.backend.dao.FeedSubscriptionDAO; import com.commafeed.backend.dao.UnitOfWork; import com.commafeed.backend.feed.parser.FeedParserResult.Content; @@ -42,6 +43,7 @@ public class FeedRefreshUpdater { private final UnitOfWork unitOfWork; private final FeedService feedService; private final FeedEntryService feedEntryService; + private final FeedEntryDAO feedEntryDAO; private final FeedSubscriptionDAO feedSubscriptionDAO; private final Striped locks; @@ -49,11 +51,12 @@ public class FeedRefreshUpdater { private final Meter feedUpdated; private final Meter entryInserted; - public FeedRefreshUpdater(UnitOfWork unitOfWork, FeedService feedService, FeedEntryService feedEntryService, MetricRegistry metrics, - FeedSubscriptionDAO feedSubscriptionDAO) { + public FeedRefreshUpdater(UnitOfWork unitOfWork, FeedService feedService, FeedEntryService feedEntryService, FeedEntryDAO feedEntryDAO, + MetricRegistry metrics, FeedSubscriptionDAO feedSubscriptionDAO) { this.unitOfWork = unitOfWork; this.feedService = feedService; this.feedEntryService = feedEntryService; + this.feedEntryDAO = feedEntryDAO; this.feedSubscriptionDAO = feedSubscriptionDAO; locks = Striped.lazyWeakLock(100000); @@ -126,8 +129,19 @@ public class FeedRefreshUpdater { Map> insertedUnreadEntriesBySubscription = new HashMap<>(); if (!entries.isEmpty()) { - List subscriptions = null; + Map entriesByGuidHash = new HashMap<>(); for (Entry entry : entries) { + entriesByGuidHash.put(Digests.sha1Hex(entry.guid()), entry); + } + Set existingGuids = unitOfWork.call(() -> feedEntryDAO.findExistingGuids(feed, entriesByGuidHash.keySet())); + List newEntries = entriesByGuidHash.entrySet() + .stream() + .filter(e -> !existingGuids.contains(e.getKey())) + .map(Map.Entry::getValue) + .toList(); + + List subscriptions = null; + for (Entry entry : newEntries) { if (subscriptions == null) { subscriptions = unitOfWork.call(() -> feedSubscriptionDAO.findByFeed(feed)); } diff --git a/commafeed-server/src/test/java/com/commafeed/integration/rest/LargeDatasetIT.java b/commafeed-server/src/test/java/com/commafeed/integration/rest/LargeDatasetIT.java new file mode 100644 index 00000000..40105f53 --- /dev/null +++ b/commafeed-server/src/test/java/com/commafeed/integration/rest/LargeDatasetIT.java @@ -0,0 +1,134 @@ +package com.commafeed.integration.rest; + +import java.time.Duration; +import java.time.Instant; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockserver.model.HttpRequest; +import org.mockserver.model.HttpResponse; +import org.mockserver.model.MediaType; + +import com.commafeed.TestConstants; +import com.commafeed.frontend.model.Entries; +import com.commafeed.frontend.model.Entry; +import com.commafeed.frontend.model.request.MarkRequest; +import com.commafeed.frontend.resource.CategoryREST; +import com.commafeed.integration.BaseIT; + +import io.quarkus.test.junit.QuarkusTest; +import io.restassured.RestAssured; +import io.restassured.http.ContentType; + +@QuarkusTest +class LargeDatasetIT extends BaseIT { + + private static final int FEED_COUNT = 10; + private static final int ENTRIES_PER_FEED = 20; + private static final int TOTAL_ENTRIES = FEED_COUNT * ENTRIES_PER_FEED; + + private Long firstSubscriptionId; + + @BeforeEach + void setup() { + initialSetup(TestConstants.ADMIN_USERNAME, TestConstants.ADMIN_PASSWORD); + RestAssured.authentication = RestAssured.preemptive().basic(TestConstants.ADMIN_USERNAME, TestConstants.ADMIN_PASSWORD); + + for (int i = 0; i < FEED_COUNT; i++) { + String path = "/feed/" + i; + getMockServerClient().when(HttpRequest.request().withMethod("GET").withPath(path)) + .respond(HttpResponse.response().withBody(generateFeed(i)).withContentType(MediaType.APPLICATION_XML)); + Long subscriptionId = subscribe("http://localhost:" + getMockServerClient().getPort() + path); + if (i == 0) { + firstSubscriptionId = subscriptionId; + } + } + + Awaitility.await().atMost(Duration.ofSeconds(60)).until(() -> getAllEntries().getEntries().size(), count -> count >= TOTAL_ENTRIES); + } + + @AfterEach + void cleanup() { + RestAssured.reset(); + } + + @Test + void markAllAsRead() { + Entries before = getAllEntries(); + Assertions.assertEquals(TOTAL_ENTRIES, before.getEntries().size()); + Assertions.assertTrue(before.getEntries().stream().noneMatch(Entry::isRead)); + + MarkRequest markRequest = new MarkRequest(); + markRequest.setId(CategoryREST.ALL); + markRequest.setRead(true); + RestAssured.given().body(markRequest).contentType(ContentType.JSON).post("rest/category/mark").then().statusCode(200); + + Entries after = getAllEntries(); + Assertions.assertEquals(TOTAL_ENTRIES, after.getEntries().size()); + Assertions.assertTrue(after.getEntries().stream().allMatch(Entry::isRead)); + } + + @Test + void refreshDoesNotCreateDuplicateEntries() { + Assertions.assertEquals(TOTAL_ENTRIES, getAllEntries().getEntries().size()); + Instant threshold = Instant.now().minus(Duration.ofSeconds(1)); + forceRefreshAllFeeds(); + + Awaitility.await() + .atMost(Duration.ofSeconds(15)) + .until(() -> getSubscription(firstSubscriptionId), f -> f.getLastRefresh().isAfter(threshold)); + Assertions.assertEquals(TOTAL_ENTRIES, getAllEntries().getEntries().size()); + } + + @Test + void paginationHasMore() { + Entries firstPage = RestAssured.given() + .get("rest/category/entries?id=all&readType=all&limit=20&offset=0") + .then() + .statusCode(200) + .extract() + .as(Entries.class); + Assertions.assertEquals(20, firstPage.getEntries().size()); + Assertions.assertTrue(firstPage.isHasMore()); + + Entries lastPage = RestAssured.given() + .get("rest/category/entries?id=all&readType=all&limit=20&offset={offset}", TOTAL_ENTRIES - 20) + .then() + .statusCode(200) + .extract() + .as(Entries.class); + Assertions.assertEquals(20, lastPage.getEntries().size()); + Assertions.assertFalse(lastPage.isHasMore()); + } + + private Entries getAllEntries() { + return RestAssured.given() + .get("rest/category/entries?id=all&readType=all&limit=1000") + .then() + .statusCode(200) + .extract() + .as(Entries.class); + } + + private String generateFeed(int feedIndex) { + StringBuilder sb = new StringBuilder(); + sb.append("\n\n\n"); + sb.append("Feed ").append(feedIndex).append("\n"); + sb.append("https://hostname.local/feed/").append(feedIndex).append("\n"); + sb.append("Test feed ").append(feedIndex).append("\n"); + Instant base = Instant.parse("2024-01-01T00:00:00Z"); + for (int i = 0; i < ENTRIES_PER_FEED; i++) { + sb.append("\n"); + sb.append("Feed ").append(feedIndex).append(" Item ").append(i).append("\n"); + sb.append("https://hostname.local/feed/").append(feedIndex).append("/item/").append(i).append("\n"); + sb.append("Description for feed ").append(feedIndex).append(" item ").append(i).append("\n"); + sb.append("").append(base.minus(Duration.ofHours(i))).append("\n"); + sb.append("\n"); + } + sb.append("\n"); + return sb.toString(); + } +}