From abae0a61cf2a8f1ee7148d3dee95ced551179ee5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=81ukasz=20Kami=C5=84ski?= Date: Tue, 11 Feb 2025 09:31:34 +0100 Subject: [PATCH] Search job state storage for Data Lake Preview queries (#21525) * Skeleton for search job state storage AutoValues instead of records, first, humble methods in the service Additional fields for SearchJobState Update methods in the service, with unit tests Added TIMEOUT to status Periodical for old jobs removal and expiration. Expiration is for now only a known concept for Data Lake jobs... Helper creators for new and done/finished jobs * Reflecting a naming choice dilemma in the comment for future * SearchType.Result instead of message list in the SearchJobState object --- .../graylog/plugins/views/ViewsBindings.java | 2 + .../views/search/SearchJobIdentifier.java | 16 +- .../views/search/jobs/SearchJobState.java | 142 +++++++++++ .../search/jobs/SearchJobStateService.java | 127 ++++++++++ .../views/search/jobs/SearchJobStatus.java | 28 +++ .../views/search/jobs/SearchJobType.java | 22 ++ .../SearchJobStateCleanupPeriodical.java | 98 ++++++++ .../jobs/SearchJobStateServiceTest.java | 232 ++++++++++++++++++ 8 files changed, 663 insertions(+), 4 deletions(-) create mode 100644 graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobState.java create mode 100644 graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStateService.java create mode 100644 graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStatus.java create mode 100644 graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobType.java create mode 100644 graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/periodical/SearchJobStateCleanupPeriodical.java create mode 100644 graylog2-server/src/test/java/org/graylog/plugins/views/search/jobs/SearchJobStateServiceTest.java diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/ViewsBindings.java b/graylog2-server/src/main/java/org/graylog/plugins/views/ViewsBindings.java index f6e8ea40db78..de7ea7085c08 100644 --- a/graylog2-server/src/main/java/org/graylog/plugins/views/ViewsBindings.java +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/ViewsBindings.java @@ -58,6 +58,7 @@ import org.graylog.plugins.views.search.filter.QueryStringFilter; import org.graylog.plugins.views.search.filter.StreamCategoryFilter; import org.graylog.plugins.views.search.filter.StreamFilter; +import org.graylog.plugins.views.search.jobs.periodical.SearchJobStateCleanupPeriodical; import org.graylog.plugins.views.search.querystrings.LastUsedQueryStringsService; import org.graylog.plugins.views.search.querystrings.MongoLastUsedQueryStringsService; import org.graylog.plugins.views.search.rest.DashboardsResource; @@ -239,6 +240,7 @@ protected void configure() { registerVisualizationConfigSubtypes(); addPeriodical(SearchesCleanUpJob.class); + addPeriodical(SearchJobStateCleanupPeriodical.class); addMigration(V20181220133700_AddViewsAdminRole.class); addMigration(V20190304102700_MigrateMessageListStructure.class); diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/SearchJobIdentifier.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/SearchJobIdentifier.java index 2fc1df5c4fbb..398c5ab13d92 100644 --- a/graylog2-server/src/main/java/org/graylog/plugins/views/search/SearchJobIdentifier.java +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/SearchJobIdentifier.java @@ -17,9 +17,17 @@ package org.graylog.plugins.views.search; import com.fasterxml.jackson.annotation.JsonProperty; +import org.mongojack.Id; +import org.mongojack.ObjectId; -public record SearchJobIdentifier(@JsonProperty("id") String id, - @JsonProperty("search_id") String searchId, - @JsonProperty("owner") String owner, - @JsonProperty("executing_node") String executingNodeId) {} +public record SearchJobIdentifier(@JsonProperty("id") + @ObjectId + @Id + String id, + @JsonProperty(SEARCH_ID_FIELD) String searchId, + @JsonProperty(OWNER_FIELD) String owner, + @JsonProperty("executing_node") String executingNodeId) { + public static final String SEARCH_ID_FIELD = "search_id"; + public static final String OWNER_FIELD = "owner"; +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobState.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobState.java new file mode 100644 index 000000000000..2e9ded305463 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobState.java @@ -0,0 +1,142 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.plugins.views.search.jobs; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonUnwrapped; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.google.auto.value.AutoValue; +import org.graylog.plugins.views.search.SearchJobIdentifier; +import org.graylog.plugins.views.search.SearchType; +import org.graylog2.database.MongoEntity; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; +import org.mongojack.Id; +import org.mongojack.ObjectId; + +import javax.annotation.Nullable; + +@AutoValue +@JsonDeserialize(builder = SearchJobState.Builder.class) +public abstract class SearchJobState implements MongoEntity { + + public static final String CREATED_AT_FIELD = "created_at"; + public static final String UPDATED_AT_FIELD = "updated_at"; + public static final String STATUS_FIELD = "status"; + public static final String TYPE_FIELD = "type"; + public static final String RESULT_FIELD = "result"; + + @JsonUnwrapped + public abstract SearchJobIdentifier identifier(); + + @JsonProperty(STATUS_FIELD) + public abstract SearchJobStatus status(); + + @JsonProperty(TYPE_FIELD) + public abstract SearchJobType type(); + + @JsonProperty("error_message") + public abstract String errorMessage(); + + @JsonProperty("progress") + public abstract int progress(); + + @JsonProperty(RESULT_FIELD) + @Nullable + public abstract SearchType.Result result(); + + @JsonProperty(CREATED_AT_FIELD) + public abstract DateTime createdAt(); + + @JsonProperty(UPDATED_AT_FIELD) + public abstract DateTime updatedAt(); + + public abstract Builder toBuilder(); + + public static Builder builder() { + return Builder.create(); + } + + @AutoValue.Builder + public abstract static class Builder { + + @JsonUnwrapped + public abstract Builder identifier(final SearchJobIdentifier identifier); + + @JsonProperty(STATUS_FIELD) + public abstract Builder status(final SearchJobStatus status); + + @JsonProperty(TYPE_FIELD) + public abstract Builder type(final SearchJobType type); + + @JsonProperty("error_message") + public abstract Builder errorMessage(final String errorMessage); + + @JsonProperty("progress") + public abstract Builder progress(final int progress); + + @JsonProperty(RESULT_FIELD) + public abstract Builder result(final SearchType.Result result); + + @JsonProperty(CREATED_AT_FIELD) + public abstract Builder createdAt(final DateTime createdAt); + + @JsonProperty(UPDATED_AT_FIELD) + public abstract Builder updatedAt(final DateTime updatedAt); + + public abstract SearchJobState build(); + + @JsonCreator + public static Builder create() { + return new AutoValue_SearchJobState.Builder() + .progress(0) + .type(SearchJobType.DATA_LAKE) + .errorMessage(""); + } + } + + public static SearchJobState createNewJob(final SearchJobIdentifier searchJobIdentifier) { + return SearchJobState.builder() + .identifier(searchJobIdentifier) + .result(null) + .status(SearchJobStatus.RUNNING) + .progress(0) + .createdAt(DateTime.now(DateTimeZone.UTC)) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build(); + } + + public static SearchJobState createDoneJobFrom(final SearchJobState existingSearchJob, + final SearchType.Result result) { + return existingSearchJob.toBuilder() + .result(result) + .status(SearchJobStatus.DONE) + .progress(100) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build(); + } + + @Override + @ObjectId + @Id + @JsonIgnore + public String id() { + return identifier().id(); + } +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStateService.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStateService.java new file mode 100644 index 000000000000..9346a9e9c035 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStateService.java @@ -0,0 +1,127 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.plugins.views.search.jobs; + +import com.mongodb.client.MongoCollection; +import com.mongodb.client.model.Filters; +import com.mongodb.client.model.Indexes; +import com.mongodb.client.model.ReplaceOptions; +import com.mongodb.client.model.Sorts; +import com.mongodb.client.model.Updates; +import com.mongodb.client.result.DeleteResult; +import com.mongodb.client.result.InsertOneResult; +import com.mongodb.client.result.UpdateResult; +import jakarta.inject.Inject; +import org.graylog2.database.MongoCollections; +import org.graylog2.database.utils.MongoUtils; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; + +import java.util.Optional; + +import static com.mongodb.client.model.Filters.and; +import static com.mongodb.client.model.Filters.eq; +import static com.mongodb.client.model.Filters.lte; +import static org.graylog.plugins.views.search.SearchJobIdentifier.OWNER_FIELD; +import static org.graylog.plugins.views.search.SearchJobIdentifier.SEARCH_ID_FIELD; +import static org.graylog.plugins.views.search.jobs.SearchJobState.CREATED_AT_FIELD; +import static org.graylog.plugins.views.search.jobs.SearchJobState.STATUS_FIELD; +import static org.graylog.plugins.views.search.jobs.SearchJobState.TYPE_FIELD; +import static org.graylog.plugins.views.search.jobs.SearchJobType.DATA_LAKE; + +public class SearchJobStateService { + + public static final String COLLECTION_NAME = "search_job_states"; + + private final MongoCollection collection; + private final MongoUtils mongoUtils; + + @Inject + public SearchJobStateService(final MongoCollections mongoCollections) { + this.collection = mongoCollections.collection(COLLECTION_NAME, SearchJobState.class); + this.mongoUtils = mongoCollections.utils(collection); + this.collection.createIndex(Indexes.ascending(SEARCH_ID_FIELD)); + this.collection.createIndex(Indexes.ascending(OWNER_FIELD)); + this.collection.createIndex(Indexes.ascending(CREATED_AT_FIELD)); + } + + public Optional get(final String id) { + return mongoUtils.getById(id); + } + + public Optional getLatestForUser(final String user) { + return Optional.ofNullable( + collection.find(Filters.eq(OWNER_FIELD, user)) + .sort(Sorts.descending(CREATED_AT_FIELD)) + .first() + ); + } + + public boolean delete(final String id) { + return mongoUtils.deleteById(id); + } + + public long deleteOlderThan(final DateTime dateTime) { + final DeleteResult deleteResult = collection.deleteMany(lte(CREATED_AT_FIELD, dateTime)); + return deleteResult.getDeletedCount(); + } + + public long expireOlderThan(final DateTime dateTime) { + final UpdateResult updateResult = collection.updateMany( + and( + eq(TYPE_FIELD, DATA_LAKE), + //eq(STATUS_FIELD, DONE) //TODO: should all jobs be expired, or only DONE ones? + lte(CREATED_AT_FIELD, dateTime) + ), + Updates.set(STATUS_FIELD, SearchJobStatus.EXPIRED) + + ); + return updateResult.getModifiedCount(); + } + + public SearchJobState create(final SearchJobState searchJobState) { + if (searchJobState.identifier().id() != null) { + throw new IllegalStateException("ID should be null for a call to create a new SearchJobState"); + } + final InsertOneResult insertOneResult = this.collection.insertOne(searchJobState); + return get(MongoUtils.insertedIdAsString(insertOneResult)).orElseThrow(() -> new IllegalStateException("Unable to retrieve saved search job state!")); + } + + public boolean update(final SearchJobState searchJobState) { + if (searchJobState.identifier().id() == null) { + throw new IllegalStateException("Missing ID of SearchJobState to update"); + } + final UpdateResult updateResult = collection.replaceOne( + MongoUtils.idEq(searchJobState.identifier().id()), + searchJobState.toBuilder().updatedAt(DateTime.now(DateTimeZone.UTC)).build(), + new ReplaceOptions().upsert(true) + ); + return updateResult.getModifiedCount() > 0; + } + + public boolean changeStatus(final String searchJobStateID, + final SearchJobStatus searchJobStatus) { + return get(searchJobStateID) + .map(searchJobState -> searchJobState.toBuilder() + .status(searchJobStatus) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build()) + .map(this::update) + .orElse(false); + + } +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStatus.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStatus.java new file mode 100644 index 000000000000..4327aabca5ca --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobStatus.java @@ -0,0 +1,28 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.plugins.views.search.jobs; + +public enum SearchJobStatus { + RUNNING, + DONE, + CANCELLATION_REQUESTED, + CANCELLED, + TIMEOUT, + EXPIRED, + ERROR, + RESET +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobType.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobType.java new file mode 100644 index 000000000000..1b42dc1aee57 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/SearchJobType.java @@ -0,0 +1,22 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.plugins.views.search.jobs; + +public enum SearchJobType { + DATA_LAKE, + //SEARCH_ENGINE or INDEXER - TBD in future +} diff --git a/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/periodical/SearchJobStateCleanupPeriodical.java b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/periodical/SearchJobStateCleanupPeriodical.java new file mode 100644 index 000000000000..09a2b9179ba0 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog/plugins/views/search/jobs/periodical/SearchJobStateCleanupPeriodical.java @@ -0,0 +1,98 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.plugins.views.search.jobs.periodical; + +import jakarta.inject.Inject; +import org.graylog.plugins.views.search.jobs.SearchJobStateService; +import org.graylog2.plugin.periodical.Periodical; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; +import org.joda.time.Duration; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SearchJobStateCleanupPeriodical extends Periodical { + + private static final Logger LOG = LoggerFactory.getLogger(SearchJobStateCleanupPeriodical.class); + static final Duration MIN_AGE_TO_REMOVE = Duration.standardDays(2); + static final Duration MIN_AGE_TO_EXPIRE = Duration.standardDays(1); + + private final SearchJobStateService searchJobStateService; + + @Inject + public SearchJobStateCleanupPeriodical(final SearchJobStateService searchJobStateService) { + this.searchJobStateService = searchJobStateService; + } + + @Override + public void doRun() { + removeOldJobs(); + expireOldEnoughJobs(); + } + + private void removeOldJobs() { + final DateTime dateTime = DateTime.now(DateTimeZone.UTC).minus(MIN_AGE_TO_REMOVE); + final long numRemoved = searchJobStateService.deleteOlderThan(dateTime); + LOG.debug("Removed search job state documents : " + numRemoved); + } + + private void expireOldEnoughJobs() { + final DateTime dateTime = DateTime.now(DateTimeZone.UTC).minus(MIN_AGE_TO_EXPIRE); + final long numExpired = searchJobStateService.expireOlderThan(dateTime); + LOG.debug("Expired search job state documents : " + numExpired); + } + + @Override + public boolean runsForever() { + return false; + } + + @Override + public boolean stopOnGracefulShutdown() { + return true; + } + + @Override + public boolean masterOnly() { + return true; + } + + @Override + public boolean startOnThisNode() { + return true; + } + + @Override + public boolean isDaemon() { + return true; + } + + @Override + public int getInitialDelaySeconds() { + return 60; + } + + @Override + public int getPeriodSeconds() { + return 5 * 60; + } + + @Override + protected Logger getLogger() { + return LOG; + } +} diff --git a/graylog2-server/src/test/java/org/graylog/plugins/views/search/jobs/SearchJobStateServiceTest.java b/graylog2-server/src/test/java/org/graylog/plugins/views/search/jobs/SearchJobStateServiceTest.java new file mode 100644 index 000000000000..fca3300d4454 --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog/plugins/views/search/jobs/SearchJobStateServiceTest.java @@ -0,0 +1,232 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.plugins.views.search.jobs; + +import org.graylog.plugins.views.search.SearchJobIdentifier; +import org.graylog.plugins.views.search.SearchType; +import org.graylog.testing.mongodb.MongoDBInstance; +import org.graylog2.bindings.providers.MongoJackObjectMapperProvider; +import org.graylog2.database.MongoCollections; +import org.graylog2.database.MongoConnection; +import org.graylog2.shared.bindings.providers.ObjectMapperProvider; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +import java.util.Optional; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + + +public class SearchJobStateServiceTest { + + @Rule + public final MongoDBInstance mongodb = MongoDBInstance.createForClass(); + + private SearchJobStateService toTest; + + @Before + public void setUp() { + final MongoConnection mongoConnection = mongodb.mongoConnection(); + final MongoJackObjectMapperProvider objectMapperProvider = new MongoJackObjectMapperProvider(new ObjectMapperProvider().get()); + this.toTest = new SearchJobStateService(new MongoCollections(objectMapperProvider, mongoConnection)); + } + + @Test + public void testSaveAndGet() { + final SearchJobState toBeSaved = SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, "677fd86ae6db8b71a8e10e3e", "john", "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.RUNNING) + .progress(42) + .createdAt(DateTime.now(DateTimeZone.UTC)) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build(); + final SearchJobState saved = toTest.create(toBeSaved); + final Optional retrieved = toTest.get(saved.id()); + assertTrue(retrieved.isPresent()); + assertEquals(toBeSaved.toBuilder() + .identifier(new SearchJobIdentifier(saved.id(), "677fd86ae6db8b71a8e10e3e", "john", "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .build(), retrieved.get()); + } + + @Test + public void testSaveAndDelete() { + final SearchJobState toBeSaved = SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, "677fd86ae6db8b71a8e10e3e", "john", "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.RUNNING) + .progress(42) + .createdAt(DateTime.now(DateTimeZone.UTC)) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build(); + final SearchJobState saved = toTest.create(toBeSaved); + final Optional retrieved = toTest.get(saved.id()); + assertTrue(retrieved.isPresent()); + assertTrue(toTest.delete(saved.id())); + assertTrue(toTest.get(saved.id()).isEmpty()); + } + + @Test + public void testSaveAndUpdate() { + final SearchJobState toBeSaved = SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, "677fd86ae6db8b71a8e10e3e", "john", "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.RUNNING) + .progress(42) + .createdAt(DateTime.now(DateTimeZone.UTC)) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build(); + final SearchJobState saved = toTest.create(toBeSaved); + Optional retrieved = toTest.get(saved.id()); + assertTrue(retrieved.isPresent()); + + final boolean updated = toTest.update(saved.toBuilder().progress(77).build()); + assertTrue(updated); + retrieved = toTest.get(saved.id()); + assertTrue(retrieved.isPresent()); + assertEquals(77, retrieved.get().progress()); + assertTrue(retrieved.get().updatedAt().isAfter(toBeSaved.updatedAt())); + } + + @Test + public void testChangeStatus() { + final SearchJobState toBeSaved = SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, "677fd86ae6db8b71a8e10e3e", "john", "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.RUNNING) + .progress(42) + .createdAt(DateTime.now(DateTimeZone.UTC)) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build(); + final SearchJobState saved = toTest.create(toBeSaved); + Optional retrieved = toTest.get(saved.id()); + assertTrue(retrieved.isPresent()); + + final boolean updated = toTest.changeStatus(saved.identifier().id(), SearchJobStatus.DONE); + assertTrue(updated); + retrieved = toTest.get(saved.id()); + assertTrue(retrieved.isPresent()); + assertEquals(SearchJobStatus.DONE, retrieved.get().status()); + assertTrue(retrieved.get().updatedAt().isAfter(toBeSaved.updatedAt())); + } + + @Test + public void testGetLatestForUser() { + + toTest.create(SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, + "677fd86ae6db8b71a8e10001", + "john", + "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.RUNNING) + .progress(42) + .createdAt(DateTime.parse("1999-01-01T11:11:11")) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build()); + + toTest.create(SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, + "677fd86ae6db8b71a8e10002", + "john", + "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.RUNNING) + .progress(42) + .createdAt(DateTime.parse("2000-01-01T11:11:11")) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build()); + + toTest.create(SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, + "677fd86ae6db8b71a8e10003", + "bob", + "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.RUNNING) + .progress(42) + .createdAt(DateTime.parse("2020-01-01T11:11:11")) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build()); + + assertTrue(toTest.getLatestForUser("andy").isEmpty()); //Andy has no search jobs + assertEquals("677fd86ae6db8b71a8e10003", toTest.getLatestForUser("bob").get().identifier().searchId()); //Bob has only one, we choose it + assertEquals("677fd86ae6db8b71a8e10002", toTest.getLatestForUser("john").get().identifier().searchId()); //John has 2, we choose the one with the latest "created at" date + + } + + @Test + public void testOldJobsRemovalAndExpiration() { + final SearchJobState oldJob = toTest.create(SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, + "677fd86ae6db8b71a8e10001", + "john", + "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.DONE) + .progress(42) + .createdAt(DateTime.parse("1999-01-01T11:11:11")) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build()); + final SearchJobState jobThatNeedsToBeExpired = toTest.create(SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, + "677fd86ae6db8b71a8e10002", + "john", + "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.DONE) + .progress(42) + .createdAt(DateTime.now(DateTimeZone.UTC).minusDays(1)) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build()); + final SearchJobState freshJob = toTest.create(SearchJobState.builder() + .identifier(new SearchJobIdentifier(null, + "677fd86ae6db8b71a8e10003", + "john", + "dcae52e4-777e-4e3f-8e69-61df7a607016")) + .result(noResult()) + .status(SearchJobStatus.DONE) + .progress(42) + .createdAt(DateTime.now(DateTimeZone.UTC)) + .updatedAt(DateTime.now(DateTimeZone.UTC)) + .build()); + + long numRemoved = toTest.deleteOlderThan(DateTime.now(DateTimeZone.UTC).minusDays(2)); + assertEquals(1, numRemoved); + assertTrue(toTest.get(oldJob.id()).isEmpty()); + assertTrue(toTest.get(jobThatNeedsToBeExpired.id()).isPresent()); + assertTrue(toTest.get(freshJob.id()).isPresent()); + + long numExpired = toTest.expireOlderThan(DateTime.now(DateTimeZone.UTC).minusHours(7)); + assertEquals(1, numExpired); + assertSame(toTest.get(jobThatNeedsToBeExpired.id()).get().status(), SearchJobStatus.EXPIRED); + assertSame(toTest.get(freshJob.id()).get().status(), SearchJobStatus.DONE); + + numRemoved = toTest.deleteOlderThan(DateTime.now(DateTimeZone.UTC).minusDays(2)); + assertEquals(0, numRemoved); + + } + + private SearchType.Result noResult() { + return null; + } +}