diff --git a/code/common/message-queue/src/main/java/nu/marginalia/mq/persistence/MqPersistence.java b/code/common/message-queue/src/main/java/nu/marginalia/mq/persistence/MqPersistence.java
index dce9d402..3413ffea 100644
--- a/code/common/message-queue/src/main/java/nu/marginalia/mq/persistence/MqPersistence.java
+++ b/code/common/message-queue/src/main/java/nu/marginalia/mq/persistence/MqPersistence.java
@@ -12,6 +12,8 @@ import java.sql.SQLException;
import java.time.Duration;
import java.util.*;
+import static nu.marginalia.mq.MqMessageState.NEW;
+
@Singleton
public class MqPersistence {
private final HikariDataSource dataSource;
@@ -100,12 +102,18 @@ public class MqPersistence {
/** Modifies the state of a message by id */
public void updateMessageState(long id, MqMessageState mqMessageState) throws SQLException {
+ if (NEW == mqMessageState) {
+ reinitializeMessage(id);
+ return;
+ }
+
try (var conn = dataSource.getConnection();
var stmt = conn.prepareStatement("""
UPDATE MESSAGE_QUEUE
SET STATE=?, UPDATED_TIME=CURRENT_TIMESTAMP(6)
WHERE ID=?
""")) {
+
stmt.setString(1, mqMessageState.name());
stmt.setLong(2, id);
@@ -115,6 +123,26 @@ public class MqPersistence {
}
}
+ /** Sets the message to 'NEW' state and removes any owner */
+ public void reinitializeMessage(long id) throws SQLException {
+ try (var conn = dataSource.getConnection();
+ var stmt = conn.prepareStatement("""
+ UPDATE MESSAGE_QUEUE
+ SET STATE='NEW',
+ OWNER_INSTANCE=NULL,
+ OWNER_TICK=NULL,
+ UPDATED_TIME=CURRENT_TIMESTAMP(6)
+ WHERE ID=?
+ """)) {
+
+ stmt.setLong(1, id);
+
+ if (stmt.executeUpdate() != 1) {
+ throw new IllegalArgumentException("No rows updated");
+ }
+ }
+ }
+
/** Creates a new message in the queue referencing as a reply to an existing message
* This message will have it's RELATED_ID set to the original message's ID.
*/
@@ -207,7 +235,8 @@ public class MqPersistence {
AND RECIPIENT_INBOX=?
LIMIT ?
""")
- ) {
+ )
+ {
queryStmt.setString(1, inboxName);
queryStmt.setInt(2, n);
var rs = queryStmt.executeQuery();
@@ -230,9 +259,44 @@ public class MqPersistence {
}
return messages;
+ }
+
}
-}
+ public MqMessage getMessage(long id) throws SQLException {
+ try (var conn = dataSource.getConnection();
+ var queryStmt = conn.prepareStatement("""
+ SELECT
+ ID,
+ RELATED_ID,
+ FUNCTION,
+ PAYLOAD,
+ STATE,
+ SENDER_INBOX IS NOT NULL AS EXPECTS_RESPONSE
+ FROM MESSAGE_QUEUE
+ WHERE ID=?
+ """)
+ )
+ {
+ queryStmt.setLong(1, id);
+ var rs = queryStmt.executeQuery();
+
+ if (rs.next()) {
+ long msgId = rs.getLong("ID");
+ long relatedId = rs.getLong("RELATED_ID");
+
+ String function = rs.getString("FUNCTION");
+ String payload = rs.getString("PAYLOAD");
+
+ MqMessageState state = MqMessageState.valueOf(rs.getString("STATE"));
+ boolean expectsResponse = rs.getBoolean("EXPECTS_RESPONSE");
+
+ return new MqMessage(msgId, relatedId, function, payload, state, expectsResponse);
+ }
+ }
+
+ throw new IllegalArgumentException("No message with id " + id);
+ }
/** Marks unclaimed messages addressed to this inbox with instanceUUID and tick,
* then returns these messages.
*/
@@ -378,4 +442,5 @@ public class MqPersistence {
}
}
+
}
diff --git a/code/services-satellite/control-service/src/main/java/nu/marginalia/control/ControlService.java b/code/services-satellite/control-service/src/main/java/nu/marginalia/control/ControlService.java
index 3411058c..e1efc3e4 100644
--- a/code/services-satellite/control-service/src/main/java/nu/marginalia/control/ControlService.java
+++ b/code/services-satellite/control-service/src/main/java/nu/marginalia/control/ControlService.java
@@ -8,6 +8,8 @@ import nu.marginalia.control.svc.*;
import nu.marginalia.db.storage.model.FileStorageId;
import nu.marginalia.db.storage.model.FileStorageType;
import nu.marginalia.model.gson.GsonFactory;
+import nu.marginalia.mq.MqMessageState;
+import nu.marginalia.mq.persistence.MqPersistence;
import nu.marginalia.renderer.RendererFactory;
import nu.marginalia.service.server.*;
import org.slf4j.Logger;
@@ -43,7 +45,8 @@ public class ControlService extends Service {
ControlActorService controlActorService,
StaticResources staticResources,
MessageQueueViewService messageQueueViewService,
- ControlFileStorageService controlFileStorageService
+ ControlFileStorageService controlFileStorageService,
+ MqPersistence persistence
) throws IOException {
super(params);
@@ -60,7 +63,9 @@ public class ControlService extends Service {
var storageSpecsRenderer = rendererFactory.renderer("control/storage-specs");
var storageCrawlsRenderer = rendererFactory.renderer("control/storage-crawls");
var storageProcessedRenderer = rendererFactory.renderer("control/storage-processed");
+
var storageDetailsRenderer = rendererFactory.renderer("control/storage-details");
+ var updateMessageStateRenderer = rendererFactory.renderer("control/dialog-update-message-state");
this.controlActorService = controlActorService;
@@ -102,6 +107,14 @@ public class ControlService extends Service {
Spark.post("/public/storage/specs", controlActorService::createCrawlSpecification, redirectToStorage);
Spark.post("/public/storage/:fid/delete", controlFileStorageService::flagFileForDeletionRequest, redirectToStorage);
+ Spark.get("/public/message/:id/state", (rq, rsp) -> persistence.getMessage(Long.parseLong(rq.params("id"))), updateMessageStateRenderer::render);
+ Spark.post("/public/message/:id/state", (rq, rsp) -> {
+ MqMessageState state = MqMessageState.valueOf(rq.queryParams("state"));
+ long id = Long.parseLong(rq.params("id"));
+ persistence.updateMessageState(id, state);
+ return "";
+ }, redirectToProcesses);
+
Spark.get("/public/:resource", this::serveStatic);
monitors.subscribe(this::logMonitorStateChange);
diff --git a/code/services-satellite/control-service/src/main/java/nu/marginalia/control/model/FileStorageFileModel.java b/code/services-satellite/control-service/src/main/java/nu/marginalia/control/model/FileStorageFileModel.java
index c8b513ee..41da73e8 100644
--- a/code/services-satellite/control-service/src/main/java/nu/marginalia/control/model/FileStorageFileModel.java
+++ b/code/services-satellite/control-service/src/main/java/nu/marginalia/control/model/FileStorageFileModel.java
@@ -1,15 +1,7 @@
package nu.marginalia.control.model;
-import nu.marginalia.db.storage.model.FileStorage;
-
-import java.util.List;
-
public record FileStorageFileModel(String filename,
- String type,
- String size
- ) {
-
- public boolean isDownloadable() {
- return type.equals("file");
- }
+ String mTime,
+ String size)
+{
}
diff --git a/code/services-satellite/control-service/src/main/java/nu/marginalia/control/svc/ControlFileStorageService.java b/code/services-satellite/control-service/src/main/java/nu/marginalia/control/svc/ControlFileStorageService.java
index 06bf240d..f80287f4 100644
--- a/code/services-satellite/control-service/src/main/java/nu/marginalia/control/svc/ControlFileStorageService.java
+++ b/code/services-satellite/control-service/src/main/java/nu/marginalia/control/svc/ControlFileStorageService.java
@@ -116,11 +116,9 @@ public class ControlFileStorageService {
try (var filesStream = Files.list(storage.asPath())) {
filesStream
+ .filter(Files::isRegularFile)
.map(this::createFileModel)
- .sorted(Comparator
- .comparing(FileStorageFileModel::type)
- .thenComparing(FileStorageFileModel::filename)
- )
+ .sorted(Comparator.comparing(FileStorageFileModel::filename))
.forEach(files::add);
}
catch (IOException ex) {
@@ -132,7 +130,7 @@ public class ControlFileStorageService {
private FileStorageFileModel createFileModel(Path p) {
try {
- String type = Files.isRegularFile(p) ? "file" : "directory";
+ String mTime = Files.getLastModifiedTime(p).toInstant().toString();
String size;
if (Files.isDirectory(p)) {
size = "-";
@@ -146,7 +144,7 @@ public class ControlFileStorageService {
else size = sizeBytes / (1024 * 1024 * 1024) + " GB";
}
- return new FileStorageFileModel(p.toFile().getName(), type, size);
+ return new FileStorageFileModel(p.toFile().getName(), mTime, size);
}
catch (IOException ex) {
throw new RuntimeException(ex);
diff --git a/code/services-satellite/control-service/src/main/resources/templates/control/dialog-update-message-state.hdb b/code/services-satellite/control-service/src/main/resources/templates/control/dialog-update-message-state.hdb
new file mode 100644
index 00000000..e5ee806f
--- /dev/null
+++ b/code/services-satellite/control-service/src/main/resources/templates/control/dialog-update-message-state.hdb
@@ -0,0 +1,41 @@
+
+
+
+Update ID
+
+{{> control/partials/nav}}
+
+
Update Message State
+
Update the of a message in the message queue. This may be useful to prevent an actor
+from resuming an action when this is not desirable. Setting an old message to 'NEW' will
+erase information about its owner, and inboxes will consider the message new again.
+
+
+
\ No newline at end of file
diff --git a/code/services-satellite/control-service/src/main/resources/templates/control/partials/message-queue-table.hdb b/code/services-satellite/control-service/src/main/resources/templates/control/partials/message-queue-table.hdb
index 5c3397f4..cc8d98a2 100644
--- a/code/services-satellite/control-service/src/main/resources/templates/control/partials/message-queue-table.hdb
+++ b/code/services-satellite/control-service/src/main/resources/templates/control/partials/message-queue-table.hdb
@@ -11,8 +11,8 @@
{{#each messages}}